from datetime import datetime, timezone from uuid import uuid4 import bcrypt from fastapi import Cookie, Depends, HTTPException, Request from fastapi.responses import RedirectResponse from sqlalchemy.orm import Session from app.config import SECRET_KEY, SESSION_COOKIE_NAME from app.database import get_db from app.models import ApiKey, User # ---------- Password ---------- def hash_password(password: str) -> str: return bcrypt.hashpw(password.encode(), bcrypt.gensalt()).decode() def verify_password(password: str, password_hash: str) -> bool: return bcrypt.checkpw(password.encode(), password_hash.encode()) # ---------- Session ---------- # In-memory session store: token -> user_id _sessions: dict[str, int] = {} def create_session(user_id: int) -> str: token = uuid4().hex _sessions[token] = user_id return token def destroy_session(token: str) -> None: _sessions.pop(token, None) def get_session_user_id(token: str | None) -> int | None: if token is None: return None return _sessions.get(token) # ---------- Web Dependencies ---------- def get_current_user_web( request: Request, db: Session = Depends(get_db), ) -> User: token = request.cookies.get(SESSION_COOKIE_NAME) uid = get_session_user_id(token) if uid is None: raise HTTPException(status_code=303, headers={"Location": "/login"}) user = db.query(User).filter(User.id == uid).first() if user is None: raise HTTPException(status_code=303, headers={"Location": "/login"}) return user def redirect_if_not_logged_in(request: Request, db: Session) -> User | None: """Return user if logged in, else None — caller decides redirect.""" token = request.cookies.get(SESSION_COOKIE_NAME) uid = get_session_user_id(token) if uid is None: return None return db.query(User).filter(User.id == uid).first() # ---------- API Dependencies ---------- def get_current_user_api( request: Request, db: Session = Depends(get_db), ) -> User: auth_header = request.headers.get("Authorization", "") if not auth_header.startswith("Bearer "): raise HTTPException(status_code=401, detail="Missing or invalid Authorization header") token = auth_header[7:] api_key = db.query(ApiKey).filter(ApiKey.key == token).first() if api_key is None: raise HTTPException(status_code=401, detail="Invalid API key") user = db.query(User).filter(User.id == api_key.user_id).first() if user is None: raise HTTPException(status_code=401, detail="User not found") return user