from datetime import datetime, timedelta, timezone from typing import Optional import bcrypt from jose import JWTError, jwt from fastapi import Depends, HTTPException, status from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from app.core.config import get_settings settings = get_settings() bearer_scheme = HTTPBearer() def hash_password(plain: str) -> str: return bcrypt.hashpw(plain.encode("utf-8"), bcrypt.gensalt()).decode("utf-8") def verify_password(plain: str, hashed: str) -> bool: return bcrypt.checkpw(plain.encode("utf-8"), hashed.encode("utf-8")) def _create_token(subject: str, kind: str, expires_delta: timedelta) -> str: expire = datetime.now(timezone.utc) + expires_delta payload = {"sub": subject, "kind": kind, "exp": expire, "iat": datetime.now(timezone.utc)} return jwt.encode(payload, settings.secret_key, algorithm=settings.algorithm) def create_access_token(user_id: str) -> str: return _create_token(user_id, "access", timedelta(minutes=settings.access_token_expire_minutes)) def create_refresh_token(user_id: str) -> str: return _create_token(user_id, "refresh", timedelta(days=settings.refresh_token_expire_days)) def decode_token(token: str) -> dict: try: return jwt.decode(token, settings.secret_key, algorithms=[settings.algorithm]) except JWTError as exc: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate credentials", headers={"WWW-Authenticate": "Bearer"}, ) from exc def get_current_user_id( credentials: HTTPAuthorizationCredentials = Depends(bearer_scheme), ) -> str: payload = decode_token(credentials.credentials) if payload.get("kind") != "access": raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token type") user_id: Optional[str] = payload.get("sub") if not user_id: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token payload") return user_id