180 lines
6.4 KiB
Python
180 lines
6.4 KiB
Python
import secrets
|
||
from datetime import datetime, timedelta, timezone
|
||
|
||
from fastapi import APIRouter, Depends, HTTPException, status
|
||
from pydantic import BaseModel, EmailStr, Field
|
||
from sqlalchemy import select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from config import settings
|
||
from deps import get_current_user, get_user_perms
|
||
from email_utils import password_reset_template, send_email, verify_email_template
|
||
from models import PasswordResetToken, User, get_db
|
||
from security import create_access_token, hash_password, verify_password
|
||
|
||
router = APIRouter(prefix="/auth", tags=["auth"])
|
||
|
||
|
||
class RegisterIn(BaseModel):
|
||
email: EmailStr
|
||
password: str = Field(min_length=8, max_length=128)
|
||
display_name: str | None = Field(None, max_length=64)
|
||
|
||
|
||
class LoginIn(BaseModel):
|
||
email: EmailStr
|
||
password: str
|
||
|
||
|
||
class TokenOut(BaseModel):
|
||
access_token: str
|
||
token_type: str = "bearer"
|
||
expires_in_minutes: int = settings.ACCESS_TOKEN_EXPIRE_MINUTES
|
||
|
||
|
||
class UserOut(BaseModel):
|
||
id: int
|
||
email: str
|
||
display_name: str | None
|
||
email_verified: bool
|
||
is_admin: bool
|
||
role: str | None = None
|
||
permissions: list[str] = []
|
||
|
||
class Config:
|
||
from_attributes = True
|
||
|
||
|
||
class RequestResetIn(BaseModel):
|
||
email: EmailStr
|
||
|
||
|
||
class ConfirmResetIn(BaseModel):
|
||
token: str
|
||
new_password: str = Field(min_length=8, max_length=128)
|
||
|
||
|
||
@router.post("/register", response_model=UserOut, status_code=status.HTTP_201_CREATED)
|
||
async def register(data: RegisterIn, db: AsyncSession = Depends(get_db)):
|
||
exists = (await db.execute(select(User).where(User.email == str(data.email).lower()))).scalar_one_or_none()
|
||
if exists:
|
||
raise HTTPException(409, "Bu e-posta adresi zaten kayıtlı")
|
||
|
||
token = secrets.token_urlsafe(32)
|
||
user = User(
|
||
email=str(data.email).lower(),
|
||
password_hash=hash_password(data.password),
|
||
display_name=data.display_name,
|
||
email_verified=False,
|
||
email_verify_token=token,
|
||
email_verify_sent_at=datetime.now(timezone.utc),
|
||
)
|
||
db.add(user)
|
||
await db.commit()
|
||
await db.refresh(user)
|
||
|
||
verify_url = f"{settings.APP_URL}/verify-email?token={token}"
|
||
body, html = verify_email_template(verify_url)
|
||
try:
|
||
await send_email(user.email, "Stepstead — E-posta Doğrulama", body, html)
|
||
except Exception as e:
|
||
print(f"[email] verify mail failed: {e}")
|
||
|
||
return user
|
||
|
||
|
||
@router.post("/login", response_model=TokenOut)
|
||
async def login(data: LoginIn, db: AsyncSession = Depends(get_db)):
|
||
user = (await db.execute(select(User).where(User.email == str(data.email).lower()))).scalar_one_or_none()
|
||
if not user or not verify_password(data.password, user.password_hash):
|
||
raise HTTPException(401, "Geçersiz e-posta veya şifre")
|
||
if not user.is_active:
|
||
raise HTTPException(403, "Hesap askıya alınmış")
|
||
user.last_login_at = datetime.now(timezone.utc)
|
||
await db.commit()
|
||
return TokenOut(access_token=create_access_token(user.id, {"email": user.email}))
|
||
|
||
|
||
@router.get("/verify-email")
|
||
async def verify_email(token: str, db: AsyncSession = Depends(get_db)):
|
||
user = (await db.execute(select(User).where(User.email_verify_token == token))).scalar_one_or_none()
|
||
if not user:
|
||
raise HTTPException(400, "Geçersiz veya kullanılmış token")
|
||
|
||
sent_at = user.email_verify_sent_at
|
||
if sent_at and (datetime.now(timezone.utc) - sent_at) > timedelta(hours=24):
|
||
raise HTTPException(400, "Token süresi dolmuş")
|
||
|
||
user.email_verified = True
|
||
user.email_verify_token = None
|
||
await db.commit()
|
||
return {"status": "ok", "message": "E-posta doğrulandı"}
|
||
|
||
|
||
@router.post("/resend-verification")
|
||
async def resend_verification(data: RequestResetIn, db: AsyncSession = Depends(get_db)):
|
||
user = (await db.execute(select(User).where(User.email == str(data.email).lower()))).scalar_one_or_none()
|
||
if not user:
|
||
return {"status": "ok"}
|
||
if user.email_verified:
|
||
return {"status": "already_verified"}
|
||
|
||
token = secrets.token_urlsafe(32)
|
||
user.email_verify_token = token
|
||
user.email_verify_sent_at = datetime.now(timezone.utc)
|
||
await db.commit()
|
||
body, html = verify_email_template(f"{settings.APP_URL}/verify-email?token={token}")
|
||
await send_email(user.email, "Stepstead — E-posta Doğrulama", body, html)
|
||
return {"status": "ok"}
|
||
|
||
|
||
@router.post("/request-password-reset")
|
||
async def request_password_reset(data: RequestResetIn, db: AsyncSession = Depends(get_db)):
|
||
user = (await db.execute(select(User).where(User.email == str(data.email).lower()))).scalar_one_or_none()
|
||
if not user:
|
||
return {"status": "ok"}
|
||
|
||
token = secrets.token_urlsafe(32)
|
||
db.add(PasswordResetToken(
|
||
user_id=user.id,
|
||
token=token,
|
||
expires_at=datetime.now(timezone.utc) + timedelta(hours=1),
|
||
))
|
||
await db.commit()
|
||
body, html = password_reset_template(f"{settings.APP_URL}/reset-password?token={token}")
|
||
try:
|
||
await send_email(user.email, "Stepstead — Şifre Sıfırlama", body, html)
|
||
except Exception as e:
|
||
print(f"[email] reset mail failed: {e}")
|
||
return {"status": "ok"}
|
||
|
||
|
||
@router.post("/reset-password")
|
||
async def reset_password(data: ConfirmResetIn, db: AsyncSession = Depends(get_db)):
|
||
rt = (await db.execute(select(PasswordResetToken).where(PasswordResetToken.token == data.token))).scalar_one_or_none()
|
||
if not rt or rt.used_at is not None:
|
||
raise HTTPException(400, "Geçersiz veya kullanılmış token")
|
||
if rt.expires_at < datetime.now(timezone.utc):
|
||
raise HTTPException(400, "Token süresi dolmuş")
|
||
|
||
user = (await db.execute(select(User).where(User.id == rt.user_id))).scalar_one_or_none()
|
||
if not user:
|
||
raise HTTPException(400, "Kullanıcı bulunamadı")
|
||
|
||
user.password_hash = hash_password(data.new_password)
|
||
rt.used_at = datetime.now(timezone.utc)
|
||
await db.commit()
|
||
return {"status": "ok", "message": "Şifre değiştirildi"}
|
||
|
||
|
||
@router.get("/me", response_model=UserOut)
|
||
async def me(user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db)):
|
||
out = UserOut.model_validate(user, from_attributes=True)
|
||
perms = await get_user_perms(user, db)
|
||
out.permissions = sorted(perms)
|
||
if user.role_id:
|
||
from models import Role
|
||
out.role = (await db.execute(select(Role.name).where(Role.id == user.role_id))).scalar_one_or_none()
|
||
elif "*" in perms:
|
||
out.role = "admin"
|
||
return out
|