import hashlib import hmac import secrets from fastapi import Depends, HTTPException, Security from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer from sqlalchemy import select from sqlalchemy.orm import Session from app.config import get_settings from app.database import get_db from app.models import ApiKey security_scheme = HTTPBearer(auto_error=False) def _legacy_hash_api_key(raw_key: str) -> str: return hashlib.sha256(raw_key.encode("utf-8")).hexdigest() def hash_api_key(raw_key: str) -> str: secret = get_settings().jwt_secret.encode("utf-8") return hmac.new(secret, raw_key.encode("utf-8"), hashlib.sha256).hexdigest() def verify_api_key_hash(raw_key: str, stored_hash: str) -> bool: if hmac.compare_digest(hash_api_key(raw_key), stored_hash): return True return hmac.compare_digest(_legacy_hash_api_key(raw_key), stored_hash) def generate_api_key() -> tuple[str, str, str]: """Returns (full_key, prefix, hash).""" raw = f"sac_{secrets.token_urlsafe(32)}" prefix = raw[:12] return raw, prefix, hash_api_key(raw) def verify_api_key(db: Session, raw_key: str) -> bool: rows = db.scalars(select(ApiKey).where(ApiKey.is_active.is_(True))).all() for row in rows: if verify_api_key_hash(raw_key, row.key_hash): return True return False def get_api_key_auth( credentials: HTTPAuthorizationCredentials | None = Security(security_scheme), db: Session = Depends(get_db), ) -> str: if credentials is None or credentials.scheme.lower() != "bearer": raise HTTPException(status_code=401, detail="Missing or invalid Authorization header") if not verify_api_key(db, credentials.credentials): raise HTTPException(status_code=401, detail="Invalid API key") return credentials.credentials