from datetime import datetime, timedelta, timezone import jwt from argon2 import PasswordHasher from argon2.exceptions import VerifyMismatchError from fastapi import Depends, HTTPException from fastapi.security import OAuth2PasswordBearer from sqlalchemy.orm import Session from .config import get_settings from .db import get_db from .models import User hasher = PasswordHasher() oauth2 = OAuth2PasswordBearer(tokenUrl="/auth/token") def hash_password(password: str) -> str: return hasher.hash(password) def verify_password(password: str, digest: str) -> bool: try: return hasher.verify(digest, password) except VerifyMismatchError: return False def make_token(user_id: str) -> str: cfg = get_settings() now = datetime.now(timezone.utc) return jwt.encode({"sub": user_id, "iat": now, "exp": now + timedelta(minutes=cfg.access_token_minutes)}, cfg.jwt_secret, algorithm=cfg.jwt_algorithm) def current_user(token: str = Depends(oauth2), db: Session = Depends(get_db)) -> User: cfg = get_settings() try: user_id = jwt.decode(token, cfg.jwt_secret, algorithms=[cfg.jwt_algorithm])["sub"] except (jwt.PyJWTError, KeyError): raise HTTPException(401, "Invalid or expired token", headers={"WWW-Authenticate": "Bearer"}) user = db.get(User, user_id) if not user: raise HTTPException(401, "Account no longer exists") return user