"""Security utilities for VoIdea. JWT handling, password hashing, and authentication helpers. """ from datetime import datetime, timedelta, timezone from typing import Any from jose import JWTError, jwt from passlib.context import CryptContext from app.core.config import get_settings settings = get_settings() pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") def get_password_hash(password: str) -> str: """Hash password using bcrypt.""" return pwd_context.hash(password) def verify_password(plain_password: str, hashed_password: str) -> bool: """Verify password against hash.""" return pwd_context.verify(plain_password, hashed_password) def create_access_token( data: dict[str, Any], expires_delta: timedelta | None = None, ) -> str: """Create JWT access token. Args: data: Payload data to encode in token expires_delta: Optional custom expiration time Returns: Encoded JWT token string """ to_encode = data.copy() if expires_delta: expire = datetime.now(timezone.utc) + expires_delta else: expire = datetime.now(timezone.utc) + timedelta( minutes=settings.jwt_access_token_expire_minutes ) to_encode.update({ "exp": expire, "iat": datetime.now(timezone.utc), "type": "access", }) encoded_jwt = jwt.encode( to_encode, settings.jwt_secret_key, algorithm=settings.jwt_algorithm, ) return encoded_jwt def create_refresh_token( data: dict[str, Any], expires_delta: timedelta | None = None, ) -> str: """Create JWT refresh token. Args: data: Payload data to encode in token expires_delta: Optional custom expiration time Returns: Encoded JWT refresh token string """ to_encode = data.copy() if expires_delta: expire = datetime.now(timezone.utc) + expires_delta else: expire = datetime.now(timezone.utc) + timedelta( days=settings.jwt_refresh_token_expire_days ) to_encode.update({ "exp": expire, "iat": datetime.now(timezone.utc), "type": "refresh", }) encoded_jwt = jwt.encode( to_encode, settings.jwt_secret_key, algorithm=settings.jwt_algorithm, ) return encoded_jwt def create_reset_token(data: dict[str, Any]) -> str: """Create JWT password reset token (1 hour expiry, separate key). Args: data: Payload data to encode in token Returns: Encoded JWT reset token string """ to_encode = data.copy() expire = datetime.now(timezone.utc) + timedelta(hours=1) to_encode.update({ "exp": expire, "iat": datetime.now(timezone.utc), "type": "reset", }) secret = settings.jwt_reset_secret_key or settings.jwt_secret_key return jwt.encode(to_encode, secret, algorithm=settings.jwt_algorithm) def decode_token( token: str, secret: str | None = None, ) -> dict[str, Any] | None: """Decode and verify JWT token. Args: token: JWT token string secret: Optional secret key (defaults to jwt_secret_key) Returns: Decoded payload or None if invalid """ try: payload = jwt.decode( token, secret or settings.jwt_secret_key, algorithms=[settings.jwt_algorithm], ) return payload except JWTError: return None def decode_reset_token(token: str) -> dict[str, Any] | None: """Decode and verify password reset JWT token (uses reset secret key). Args: token: JWT reset token string Returns: Decoded payload or None if invalid """ secret = settings.jwt_reset_secret_key or settings.jwt_secret_key return decode_token(token, secret) def verify_token_type(token_data: dict[str, Any], expected_type: str) -> bool: """Verify token type. Args: token_data: Decoded token payload expected_type: Expected token type (access/refresh/reset) Returns: True if token type matches """ return token_data.get("type") == expected_type