mtgonline/backend/app/core/security.py

"""
Security utilities for authentication and password hashing.

Implements JWT token management and bcrypt password hashing with salt.
Provides functions for creating, validating, and decoding JWT tokens,
as well as password verification and hashing.

## Features

- Bcrypt password hashing with configurable rounds
- JWT access and refresh token creation
- Token decoding and validation
- FastAPI dependency for extracting current user from Authorization header

## Usage

    from app.core.security import (
        verify_password,
        hash_password,
        create_access_token,
        create_refresh_token,
        decode_token,
        get_current_user,
    )

    # Hash and verify passwords
    hashed = hash_password("secure_password")
    is_valid = verify_password("secure_password", hashed)

    # Create and decode tokens
    token = create_access_token(subject="user_id", privlevel="Admin")
    payload = decode_token(token)

    # FastAPI dependency
    @app.get("/protected")
    async def protected_route(user: dict = Depends(get_current_user)):
        ...
"""

from datetime import datetime, timedelta, timezone
from typing import Optional

from fastapi import Header, HTTPException, status
from jose import JWTError, jwt
from passlib.context import CryptContext

from app.core.config import get_config

config = get_config()

# Password hashing context with bcrypt
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")


def verify_password(plain_password: str, hashed_password: str) -> bool:
    """Verify a plain password against a bcrypt hash.

    Args:
        plain_password: The plain text password to verify.
        hashed_password: The bcrypt hash to verify against.

    Returns:
        True if the password matches the hash, False otherwise.

    Example:
        >>> hashed = hash_password("my_password")
        >>> verify_password("my_password", hashed)
        True
        >>> verify_password("wrong_password", hashed)
        False
    """
    return pwd_context.verify(plain_password, hashed_password)


def hash_password(password: str) -> str:
    """Hash a password using bcrypt with configurable rounds.

    Args:
        password: The plain text password to hash.

    Returns:
        The bcrypt hash string.

    Example:
        >>> hashed = hash_password("secure_password")
        >>> len(hashed) > 0
        True
    """
    return pwd_context.hash(password, rounds=config.BCRYPT_ROUNDS)


def create_access_token(
    subject: str,
    privlevel: str = "User",
    expires_delta: Optional[timedelta] = None,
) -> str:
    """Create a JWT access token.

    Args:
        subject: The user identifier (typically user ID or username).
        privlevel: User privilege level (default "User").
        expires_delta: Optional custom expiration time delta.
            If not provided, uses JWT_ACCESS_TOKEN_EXPIRE_MINUTES from config.

    Returns:
        Encoded JWT access token string.

    Example:
        >>> token = create_access_token(subject="user123", privlevel="Admin")
        >>> isinstance(token, str)
        True
    """
    if expires_delta:
        expire = datetime.now(timezone.utc) + expires_delta
    else:
        expire = datetime.now(timezone.utc) + timedelta(
            minutes=config.JWT_ACCESS_TOKEN_EXPIRE_MINUTES
        )

    secret = config.jwt_secret

    payload = {
        "sub": subject,
        "exp": expire,
        "iat": datetime.now(timezone.utc),
        "type": "access",
        "privlevel": privlevel,
    }
    return jwt.encode(payload, secret, algorithm=config.JWT_ALGORITHM)


def create_refresh_token(subject: str, privlevel: str = "User") -> str:
    """Create a JWT refresh token with longer expiry.

    Args:
        subject: The user identifier (typically user ID or username).
        privlevel: User privilege level (default "User").

    Returns:
        Encoded JWT refresh token string.

    Example:
        >>> token = create_refresh_token(subject="user123")
        >>> isinstance(token, str)
        True
    """
    expire = datetime.now(timezone.utc) + timedelta(
        days=config.JWT_REFRESH_TOKEN_EXPIRE_DAYS
    )

    secret = config.jwt_secret

    payload = {
        "sub": subject,
        "exp": expire,
        "iat": datetime.now(timezone.utc),
        "type": "refresh",
        "privlevel": privlevel,
    }
    return jwt.encode(payload, secret, algorithm=config.JWT_ALGORITHM)


def decode_token(token: str) -> Optional[dict]:
    """Decode and validate a JWT token.

    Args:
        token: The JWT token string to decode.

    Returns:
        The token payload dictionary if valid, None if invalid or expired.

    Example:
        >>> token = create_access_token(subject="user123")
        >>> payload = decode_token(token)
        >>> payload["sub"]
        'user123'
    """
    try:
        secret = config.jwt_secret

        payload = jwt.decode(
            token,
            secret,
            algorithms=[config.JWT_ALGORITHM],
        )
        return payload
    except JWTError:
        return None


def get_current_user(authorization: str = Header(...)) -> dict:
    """Extract user info from JWT token in Authorization header.

    FastAPI dependency that validates the Bearer token and extracts
    user information from the decoded payload.

    Args:
        authorization: The Authorization header value (Bearer <token>).

    Returns:
        Dictionary containing user_id, privlevel, and token_type.

    Raises:
        HTTPException: 401 if token is invalid, missing, or wrong type.

    Example:
        >>> @app.get("/protected")
        ... async def protected(user: dict = Depends(get_current_user)):
        ...     return {"user_id": user["user_id"]}
    """
    # Parse Bearer token
    parts = authorization.split()
    if len(parts) != 2 or parts[0].lower() != "bearer":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication scheme",
        )

    token = parts[1]
    payload = decode_token(token)

    if not payload:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid token",
        )

    if payload.get("type") != "access":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid token type",
        )

    return {
        "user_id": payload.get("sub"),
        "privlevel": payload.get("privlevel", "User"),
        "token_type": payload.get("type"),
    }