import hashlib import secrets from dataclasses import dataclass from datetime import UTC, datetime, timedelta from fastapi import Depends, HTTPException from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer from sqlalchemy import select from sqlalchemy.orm import Session from .config import get_settings from .db import get_db from .models import AuthSession, User bearer = HTTPBearer(auto_error=False) def _token_hash(token: str) -> str: return hashlib.sha256(token.encode()).hexdigest() def issue_session(db: Session, user_id: str, app_key: str) -> str: token = secrets.token_urlsafe(48) db.add( AuthSession( token_hash=_token_hash(token), user_id=user_id, app_key=app_key, expires_at=datetime.now(UTC) + timedelta(seconds=get_settings().session_ttl_seconds), ) ) return token @dataclass(frozen=True) class SessionClaims: session_id: str user_id: str app_key: str def current_claims( credentials: HTTPAuthorizationCredentials | None = Depends(bearer), db: Session = Depends(get_db), ) -> SessionClaims: if credentials is None or credentials.scheme.lower() != "bearer": raise HTTPException(status_code=401, detail="authentication required") session = db.scalar( select(AuthSession).where( AuthSession.token_hash == _token_hash(credentials.credentials), AuthSession.revoked_at.is_(None), AuthSession.expires_at > datetime.now(UTC), ) ) if session is None: raise HTTPException(status_code=401, detail="invalid or expired session") return SessionClaims( session_id=session.id, user_id=session.user_id, app_key=session.app_key, ) def current_user( claims: SessionClaims = Depends(current_claims), db: Session = Depends(get_db) ) -> User: user = db.get(User, claims.user_id) if user is None: raise HTTPException(status_code=401, detail="user no longer exists") return user