import bcrypt import secrets from datetime import UTC, datetime, timedelta from typing import Any from jose import JWTError, jwt from app.config import settings ALGORITHM = "HS256" def hash_password(password: str) -> str: return bcrypt.hashpw(password.encode(), bcrypt.gensalt()).decode() def verify_password(plain_password: str, hashed_password: str) -> bool: return bcrypt.checkpw(plain_password.encode(), hashed_password.encode()) def create_access_token(subject: str, expires_delta: timedelta | None = None) -> str: expire = datetime.now(UTC) + ( expires_delta or timedelta(minutes=settings.access_token_expire_minutes) ) payload = {"sub": subject, "exp": expire, "type": "access"} return jwt.encode(payload, settings.secret_key, algorithm=ALGORITHM) def create_refresh_token(subject: str) -> str: expire = datetime.now(UTC) + timedelta(days=settings.refresh_token_expire_days) payload = {"sub": subject, "exp": expire, "type": "refresh"} return jwt.encode(payload, settings.secret_key, algorithm=ALGORITHM) def create_oauth_state() -> str: expire = datetime.now(UTC) + timedelta(minutes=10) payload = {"nonce": secrets.token_urlsafe(16), "exp": expire, "type": "oauth_state"} return jwt.encode(payload, settings.secret_key, algorithm=ALGORITHM) def verify_oauth_state(state: str) -> bool: try: payload = decode_token(state) return payload.get("type") == "oauth_state" except JWTError: return False def decode_token(token: str) -> dict[str, Any]: return jwt.decode(token, settings.secret_key, algorithms=[ALGORITHM]) def verify_token(token: str, expected_type: str = "access") -> str | None: try: payload = decode_token(token) if payload.get("type") != expected_type: return None subject = payload.get("sub") return subject if isinstance(subject, str) else None except JWTError: return None