Spaces:
Runtime error
Runtime error
File size: 2,613 Bytes
cdc55f4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 | """
Authentication endpoints.
POST /auth/register — create a user account (email + password + role).
Duplicate email returns HTTP 409.
POST /auth/login — validate credentials and return a signed JWT.
Rate-limited to 10 requests/minute per IP to slow
brute-force attacks.
"""
from datetime import datetime
from fastapi import APIRouter, Depends, HTTPException, Request, status
from pydantic import BaseModel
from sqlmodel import Session, select
from app.config import settings
from app.deps import get_db
from app.limiter import limiter
from app.models.db import User, UserRole
from app.security import hash_password, verify_password, create_access_token
router = APIRouter()
class RegisterRequest(BaseModel):
email: str
password: str
role: UserRole = UserRole.user
class RegisterResponse(BaseModel):
id: str
email: str
role: str
created_at: datetime
class LoginRequest(BaseModel):
email: str
password: str
class LoginResponse(BaseModel):
access_token: str
token_type: str
role: str
@router.post("/register", response_model=RegisterResponse, status_code=201)
def register(req: RegisterRequest, db: Session = Depends(get_db)):
existing = db.exec(select(User).where(User.email == req.email)).first()
if existing:
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Email already registered")
user = User(
email=req.email,
hashed_password=hash_password(req.password),
role=req.role,
created_at=datetime.utcnow(),
)
db.add(user)
db.commit()
db.refresh(user)
return RegisterResponse(
id=str(user.id),
email=user.email,
role=user.role.value,
created_at=user.created_at,
)
@router.post("/login", response_model=LoginResponse)
@limiter.limit("10/minute")
def login(request: Request, req: LoginRequest, db: Session = Depends(get_db)):
user = db.exec(select(User).where(User.email == req.email)).first()
if not user or not verify_password(req.password, user.hashed_password):
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid credentials")
if not user.is_active:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Account is inactive")
token = create_access_token(
data={"sub": str(user.id), "role": user.role.value},
expires_minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES,
)
return LoginResponse(access_token=token, token_type="bearer", role=user.role.value)
|