""" web/routes/auth.py - Login, logout, token refresh. POST /login - form submit; sets access_token + refresh_token cookies POST /logout - revokes refresh token, clears cookies POST /refresh - exchanges refresh_token cookie for a new access_token """ from fastapi import APIRouter, Form, HTTPException, Request, Response, status from fastapi.responses import RedirectResponse from fastapi.templating import Jinja2Templates from web import auth as auth_lib from web import db router = APIRouter() def _set_auth_cookies(response: Response, access_token: str, refresh_token: str) -> None: response.set_cookie( "access_token", access_token, httponly=True, samesite="strict", max_age=auth_lib.ACCESS_TOKEN_EXPIRE_MINUTES * 60, ) response.set_cookie( "refresh_token", refresh_token, httponly=True, samesite="strict", max_age=auth_lib.REFRESH_TOKEN_EXPIRE_DAYS * 86400, ) def _clear_auth_cookies(response: Response) -> None: response.delete_cookie("access_token") response.delete_cookie("refresh_token") def _templates(request: Request) -> Jinja2Templates: return request.app.state.templates @router.get("/login") async def login_page(request: Request): return _templates(request).TemplateResponse(request, "login.html") @router.post("/login") async def login( request: Request, username: str = Form(...), password: str = Form(...), ): user = db.get_user_by_username(username) if user is None or not auth_lib.verify_password(password, user["password_hash"]): return _templates(request).TemplateResponse( request, "login.html", {"error": "Invalid username or password"}, status_code=status.HTTP_401_UNAUTHORIZED, ) access_token = auth_lib.create_access_token(user["id"], user["role"]) refresh_token, jti, expires_at = auth_lib.create_refresh_token(user["id"]) db.store_refresh_token(jti, user["id"], expires_at) response = RedirectResponse(url="/dashboard", status_code=status.HTTP_303_SEE_OTHER) _set_auth_cookies(response, access_token, refresh_token) return response @router.post("/logout") async def logout(request: Request): refresh_token = request.cookies.get("refresh_token") if refresh_token: payload = auth_lib.decode_refresh_token(refresh_token) if payload and payload.get("jti"): db.revoke_refresh_token(payload["jti"]) response = RedirectResponse(url="/login", status_code=status.HTTP_303_SEE_OTHER) _clear_auth_cookies(response) return response @router.post("/refresh") async def refresh(request: Request): refresh_token = request.cookies.get("refresh_token") if not refresh_token: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="No refresh token") payload = auth_lib.decode_refresh_token(refresh_token) if payload is None: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid refresh token") jti = payload.get("jti") if not jti or not db.is_refresh_token_valid(jti): raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Refresh token revoked or expired") user = db.get_user_by_id(payload["sub"]) if user is None or not user["is_active"]: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found") # Rotate: revoke old, issue new db.revoke_refresh_token(jti) new_access = auth_lib.create_access_token(user["id"], user["role"]) new_refresh, new_jti, expires_at = auth_lib.create_refresh_token(user["id"]) db.store_refresh_token(new_jti, user["id"], expires_at) response = Response(status_code=status.HTTP_204_NO_CONTENT) _set_auth_cookies(response, new_access, new_refresh) return response