"""Security utilities for authentication and password hashing.""" from datetime import datetime, timedelta from typing import Any from jose import jwt from passlib.context import CryptContext from app.core.config import settings # Password hashing context pwd_context = CryptContext( schemes=[settings.PASSWORD_HASH_ALGORITHM], deprecated="auto", ) def verify_password(plain_password: str, hashed_password: str) -> bool: """ Verify a plain password against a hashed password. Args: plain_password: The plain text password to verify. hashed_password: The hashed password to compare against. Returns: True if the password matches, False otherwise. """ return pwd_context.verify(plain_password, hashed_password) def get_password_hash(password: str) -> str: """ Hash a password using the configured algorithm. Args: password: The plain text password to hash. Returns: The hashed password. """ return pwd_context.hash(password) def create_access_token( subject: str | Any, expires_delta: timedelta | None = None, ) -> str: """ Create a JWT access token. Args: subject: The subject of the token (usually user ID). expires_delta: Optional expiration time delta. Defaults to settings.ACCESS_TOKEN_EXPIRE_MINUTES. Returns: The encoded JWT access token. """ if expires_delta: expire = datetime.utcnow() + expires_delta else: expire = datetime.utcnow() + timedelta( minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES ) to_encode = {"exp": expire, "sub": str(subject), "type": "access"} encoded_jwt = jwt.encode( to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM, ) return encoded_jwt def create_refresh_token( subject: str | Any, expires_delta: timedelta | None = None, ) -> str: """ Create a JWT refresh token. Args: subject: The subject of the token (usually user ID). expires_delta: Optional expiration time delta. Defaults to settings.REFRESH_TOKEN_EXPIRE_DAYS. Returns: The encoded JWT refresh token. """ if expires_delta: expire = datetime.utcnow() + expires_delta else: expire = datetime.utcnow() + timedelta( days=settings.REFRESH_TOKEN_EXPIRE_DAYS ) to_encode = {"exp": expire, "sub": str(subject), "type": "refresh"} encoded_jwt = jwt.encode( to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM, ) return encoded_jwt def decode_token(token: str) -> dict[str, Any]: """ Decode and validate a JWT token. Args: token: The JWT token to decode. Returns: The decoded token payload. Raises: jwt.JWTError: If the token is invalid or expired. """ payload = jwt.decode( token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM], ) return payload