flipventory_v2/app/auth/security.py

"""
Authentication helpers: password hashing and JWT token management.

Password hashing
-----------------
We use passlib's bcrypt scheme.  bcrypt is intentionally slow (tunable cost
factor) which makes brute-force attacks against the hash database expensive.
Never store or log plaintext passwords.

JWT tokens
-----------
Tokens are signed with HS256 (HMAC-SHA256).  The payload contains:
  - ``sub``   – the user's e-mail address (the "subject")
  - ``exp``   – expiry timestamp (unix epoch seconds)

The token is passed by the client in the ``Authorization: Bearer <token>``
header.  FastAPI's OAuth2PasswordBearer extracts the raw token string for us.
"""
from datetime import datetime, timedelta, timezone

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from passlib.context import CryptContext
from sqlalchemy.orm import Session

from app.config import settings
from app.database import get_db
from app.models.user import User

# ---------------------------------------------------------------------------
# Password helpers
# ---------------------------------------------------------------------------

# CryptContext manages the algorithm list.  If we ever want to upgrade (e.g.,
# argon2) we can add it to the list and passlib will migrate hashes on login.
_pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")


def hash_password(plaintext: str) -> str:
    """Return the bcrypt hash of ``plaintext``."""
    return _pwd_context.hash(plaintext)


def verify_password(plaintext: str, hashed: str) -> bool:
    """Return True if ``plaintext`` matches ``hashed``."""
    return _pwd_context.verify(plaintext, hashed)


# ---------------------------------------------------------------------------
# JWT helpers
# ---------------------------------------------------------------------------

def create_access_token(subject: str) -> str:
    """
    Create a signed JWT token for ``subject`` (the user's e-mail).

    The token expires after ``settings.access_token_expire_minutes`` minutes
    from the time of creation.
    """
    expire = datetime.now(timezone.utc) + timedelta(
        minutes=settings.access_token_expire_minutes
    )
    payload = {"sub": subject, "exp": expire}
    return jwt.encode(payload, settings.secret_key, algorithm=settings.algorithm)


def _decode_token(token: str) -> str:
    """
    Decode and validate a JWT token.

    Returns the ``sub`` claim (e-mail) on success.
    Raises HTTPException(401) if the token is invalid or expired.
    """
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(
            token, settings.secret_key, algorithms=[settings.algorithm]
        )
        email: str | None = payload.get("sub")
        if email is None:
            raise credentials_exception
        return email
    except JWTError:
        raise credentials_exception


# ---------------------------------------------------------------------------
# FastAPI dependencies
# ---------------------------------------------------------------------------

# tokenUrl must match the path of our login endpoint so Swagger UI's
# "Authorize" button works.
_oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")


def get_current_user(
    token: str = Depends(_oauth2_scheme),
    db: Session = Depends(get_db),
) -> User:
    """
    Dependency: resolve a Bearer token to a User ORM object.

    Raises 401 if the token is invalid/expired.
    Raises 403 if the account is disabled.
    """
    email = _decode_token(token)
    user = db.query(User).filter(User.email == email).first()
    if user is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="User not found",
            headers={"WWW-Authenticate": "Bearer"},
        )
    if not user.is_active:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Account is disabled",
        )
    return user


def require_admin(current_user: User = Depends(get_current_user)) -> User:
    """
    Dependency: ensure the logged-in user is an admin.

    Raises 403 if they are not.  Use this on any endpoint that should be
    admin-only.
    """
    if not current_user.is_admin:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Admin access required",
        )
    return current_user