""" Authentication routes — Caller Type 1 (human browser / mobile users). POST /auth/login — bcrypt verify → issue access + refresh tokens POST /auth/refresh — rotate refresh token, detect theft by jti reuse POST /auth/logout — revoke current jti, clear cookie Security decisions inline at each step. """ from __future__ import annotations import uuid from datetime import datetime, timedelta, timezone import structlog from fastapi import APIRouter, Cookie, Depends, HTTPException, Request, Response from jose import JWTError from sqlalchemy import select, update from sqlalchemy.ext.asyncio import AsyncSession from app.config import get_settings from app.core.rate_limit import limiter from app.core.security import ( create_access_token, create_refresh_token, decode_token, hash_password, verify_password, ) from app.db.models import RefreshToken, User from app.db.session import get_db from app.dependencies.auth import _write_audit_event, get_client_ip from app.schemas.auth import LoginRequest, LogoutResponse, RefreshResponse, TokenResponse settings = get_settings() log = structlog.get_logger(__name__) router = APIRouter(prefix="/auth", tags=["auth"]) _COOKIE_NAME = "refresh_token" # Computed ONCE at import time so the hash is always a valid bcrypt string. # Used when the user is not found — verify_password() still runs bcrypt against # this hash to keep response time constant (prevents user-enumeration via timing). _DUMMY_HASH: str = hash_password("__timing_protection_sentinel__") # --------------------------------------------------------------------------- # Cookie helpers # --------------------------------------------------------------------------- def _set_refresh_cookie(response: Response, token: str) -> None: """ Set the HttpOnly refresh-token cookie with all security flags. httponly — JavaScript cannot read this cookie (XSS protection) secure — Sent only over HTTPS (disabled in local dev) samesite — 'strict': never sent on cross-site navigations (CSRF protection) path — Scoped to /auth so it is not sent to /data/* endpoints """ response.set_cookie( key=_COOKIE_NAME, value=token, httponly=True, secure=settings.cookie_secure, samesite=settings.cookie_samesite, domain=settings.cookie_domain, max_age=settings.refresh_token_ttl, path="/auth", # scoped — cookie NOT sent to /data/* or /admin/* ) def _clear_refresh_cookie(response: Response) -> None: """Expire the refresh-token cookie immediately.""" response.delete_cookie( key=_COOKIE_NAME, path="/auth", domain=settings.cookie_domain, httponly=True, secure=settings.cookie_secure, samesite=settings.cookie_samesite, ) # =========================================================================== # POST /auth/login # =========================================================================== @router.post("/login", response_model=TokenResponse, status_code=200) @limiter.limit("5/15minutes") # 5 attempts per 15 min per IP — brute-force guard async def login( request: Request, response: Response, payload: LoginRequest, db: AsyncSession = Depends(get_db), ) -> TokenResponse: """ Authenticate a human user and issue JWT access + refresh tokens. Security steps (in order): 1. Look up user by email (partial index on active accounts) 2. Account lockout check (before bcrypt — prevent timing amplification) 3. bcrypt verify — always runs even when user not found (timing safety) 4. Increment failure counter / set lockout on bad password 5. Reset counter + update last_login_at on success 6. Issue tokens, store refresh_tokens row, set HttpOnly cookie """ ip = get_client_ip(request) ua = request.headers.get("User-Agent", "") # Step 1: fetch user — partial index makes this fast result = await db.execute( select(User).where( User.email == payload.email, User.is_active == True, # noqa: E712 ) ) user = result.scalar_one_or_none() # Step 2: lockout check — must happen BEFORE bcrypt to prevent an attacker # from using repeated requests to measure bcrypt cost differences. if user and user.locked_until and user.locked_until > datetime.now(timezone.utc): await _write_audit_event( db, "account_locked", "user", user_id=str(user.id), ip_address=ip, user_agent=ua, metadata={"locked_until": user.locked_until.isoformat()}, ) raise HTTPException( status_code=429, detail="Account temporarily locked due to too many failed attempts", ) # Step 3: bcrypt verify — always run to keep response time constant # (avoids user-enumeration via timing side-channel) password_ok = verify_password( payload.password, user.password_hash if user else _DUMMY_HASH, ) if not user or not password_ok: if user: # Step 4a: increment failure counter, apply lockout if threshold reached new_attempts = user.failed_login_attempts + 1 lock_until: datetime | None = None if new_attempts >= settings.max_login_attempts: lock_until = datetime.now(timezone.utc) + timedelta( seconds=settings.account_lockout_seconds ) log.warning( "account_locked", user_id=str(user.id), ip=ip, attempts=new_attempts, ) await db.execute( update(User) .where(User.id == user.id) .values( failed_login_attempts=new_attempts, locked_until=lock_until, ) ) await _write_audit_event( db, "login_failure", "user", user_id=str(user.id) if user else None, ip_address=ip, user_agent=ua, # Log the attempted email for ops visibility — NOT the password metadata={"reason": "invalid_credentials", "email": payload.email}, ) log.warning("login_failure", ip=ip, email=payload.email) # Single generic message — do not reveal whether email exists raise HTTPException(status_code=401, detail="Invalid credentials") # Step 5: successful auth — reset failure state await db.execute( update(User) .where(User.id == user.id) .values( failed_login_attempts=0, locked_until=None, last_login_at=datetime.now(timezone.utc), ) ) # Step 6: issue tokens access_token, exp = create_access_token(str(user.id), user.role) jti = str(uuid.uuid4()) family_id = str(uuid.uuid4()) # new family per login session refresh_jwt, refresh_exp = create_refresh_token(jti, str(user.id)) db.add( RefreshToken( jti=uuid.UUID(jti), user_id=user.id, family_id=uuid.UUID(family_id), revoked=False, user_agent=ua[:512], ip_address=ip, expires_at=refresh_exp, ) ) await _write_audit_event( db, "login_success", "user", user_id=str(user.id), ip_address=ip, user_agent=ua, metadata={"role": user.role}, ) log.info("login_success", user_id=str(user.id), role=user.role, ip=ip) _set_refresh_cookie(response, refresh_jwt) return TokenResponse(access_token=access_token, expires_in=settings.access_token_ttl) # =========================================================================== # POST /auth/refresh # =========================================================================== @router.post("/refresh", response_model=RefreshResponse, status_code=200) async def refresh( request: Request, response: Response, db: AsyncSession = Depends(get_db), refresh_token_cookie: str | None = Cookie(None, alias=_COOKIE_NAME), ) -> RefreshResponse: """ Silent token refresh — rotate the refresh token and issue a new access token. Rotation protocol: 1. Verify JWT signature (no DB hit — fast path) 2. Look up jti in DB 3. If jti is revoked → TOKEN THEFT — revoke entire family, force re-login 4. If valid → mark old jti revoked, insert new jti (same family_id), issue tokens Theft detection rationale: Each refresh token is single-use. If a rotated (revoked) token is presented, an attacker stole and replayed it. Revoking the entire family forces all active sessions from this login to re-authenticate, neutralising the stolen token. """ ip = get_client_ip(request) ua = request.headers.get("User-Agent", "") if not refresh_token_cookie: raise HTTPException(status_code=401, detail="No refresh token provided") # Step 1: crypto verify — no DB hit, fast rejection of tampered tokens try: token_payload = decode_token(refresh_token_cookie) except JWTError: _clear_refresh_cookie(response) raise HTTPException(status_code=401, detail="Invalid or expired refresh token") if token_payload.get("type") != "refresh": _clear_refresh_cookie(response) raise HTTPException(status_code=401, detail="Invalid token type") jti_str: str | None = token_payload.get("jti") sub: str | None = token_payload.get("sub") if not jti_str or not sub: _clear_refresh_cookie(response) raise HTTPException(status_code=401, detail="Malformed token") # Step 2: DB lookup — existence + revocation try: jti = uuid.UUID(jti_str) except ValueError: raise HTTPException(status_code=401, detail="Invalid token format") result = await db.execute( select(RefreshToken).where(RefreshToken.jti == jti) ) token_row = result.scalar_one_or_none() if not token_row: _clear_refresh_cookie(response) raise HTTPException(status_code=401, detail="Token not recognised") # Step 3: THEFT DETECTION — token is in DB but already revoked if token_row.revoked: # Revoke the ENTIRE family to invalidate all sessions from this login await db.execute( update(RefreshToken) .where(RefreshToken.family_id == token_row.family_id) .values(revoked=True) ) await _write_audit_event( db, "token_theft", "user", user_id=sub, ip_address=ip, user_agent=ua, metadata={ "reused_jti": jti_str, "family_id": str(token_row.family_id), }, ) log.error( "token_theft_detected", user_id=sub, jti=jti_str, family_id=str(token_row.family_id), ip=ip, ) _clear_refresh_cookie(response) raise HTTPException( status_code=401, detail="Session invalidated — please log in again", ) # Belt-and-suspenders hard expiry (JWT exp already checked above) if token_row.expires_at < datetime.now(timezone.utc): _clear_refresh_cookie(response) raise HTTPException(status_code=401, detail="Refresh token expired") # Step 4: ROTATION — mark old token used, issue fresh tokens await db.execute( update(RefreshToken) .where(RefreshToken.jti == jti) .values(revoked=True) ) new_jti = str(uuid.uuid4()) new_refresh_jwt, new_refresh_exp = create_refresh_token(new_jti, sub) db.add( RefreshToken( jti=uuid.UUID(new_jti), user_id=token_row.user_id, family_id=token_row.family_id, # same family — preserves session chain revoked=False, user_agent=ua[:512], ip_address=ip, expires_at=new_refresh_exp, ) ) # Fetch current role (may have changed since last login, e.g. admin promotion) role_row = await db.execute( select(User.role).where(User.id == token_row.user_id) ) role = role_row.scalar_one_or_none() or "viewer" access_token, _ = create_access_token(sub, role) _set_refresh_cookie(response, new_refresh_jwt) return RefreshResponse( access_token=access_token, expires_in=settings.access_token_ttl, ) # =========================================================================== # POST /auth/logout # =========================================================================== @router.post("/logout", response_model=LogoutResponse, status_code=200) async def logout( request: Request, response: Response, db: AsyncSession = Depends(get_db), refresh_token_cookie: str | None = Cookie(None, alias=_COOKIE_NAME), ) -> LogoutResponse: """ Revoke the current refresh token and clear the HttpOnly cookie. Residual access token window: up to ACCESS_TOKEN_TTL (15 min) before the short-lived JWT expires naturally. Acceptable trade-off — production hardening option: add a Redis blocklist checked in verify_user_jwt(). """ ip = get_client_ip(request) if refresh_token_cookie: try: token_payload = decode_token(refresh_token_cookie) jti_str = token_payload.get("jti") if jti_str: jti = uuid.UUID(jti_str) await db.execute( update(RefreshToken) .where(RefreshToken.jti == jti) .values(revoked=True) ) await _write_audit_event( db, "logout", "user", user_id=token_payload.get("sub"), ip_address=ip, metadata={"jti": jti_str}, ) except (JWTError, ValueError): # Best-effort revocation — if the cookie is already invalid, # just clear it. Do not return an error to the client. pass _clear_refresh_cookie(response) return LogoutResponse()