leocrm/app/core/security.py

"""Password hashing and JWT encoding/decoding.

Uses python-jose[cryptography] (per 02-architecture.md Section 13.1) and
passlib[bcrypt] with bcrypt 4.0.1 (per Section 13.6, 03a-patterns R-6).
"""

from __future__ import annotations

from datetime import UTC, datetime, timedelta
from typing import Any

from jose import JWTError, jwt
from passlib.context import CryptContext

from app.core.config import get_settings

_settings = get_settings()

# === Password Hashing ===

# CryptContext auto-upgrades old hashes when verified. bcrypt 4.0.1 is pinned
# because 4.1+ breaks passlib (per 03a-patterns-summary R-6).
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")


def hash_password(plain: str) -> str:
    """Hash a plaintext password using bcrypt with the configured rounds."""
    return pwd_context.hash(plain)


def verify_password(plain: str, hashed: str) -> bool:
    """Verify a plaintext password against a bcrypt hash."""
    try:
        return pwd_context.verify(plain, hashed)
    except (ValueError, TypeError):
        return False


# === JWT ===


def create_access_token(
    user_id: int,
    org_id: int,
    role: str,
    expires_delta: timedelta | None = None,
) -> str:
    """Create a JWT access token with the given user, org, and role claims.

    The token contains:
      - sub: user id (string)
      - org_id: organization id
      - role: user role
      - exp: expiry timestamp
      - iat: issued-at timestamp
    """
    now = datetime.now(UTC)
    if expires_delta is None:
        expires_delta = timedelta(hours=_settings.JWT_EXPIRY_HOURS)
    expire = now + expires_delta

    payload: dict[str, Any] = {
        "sub": str(user_id),
        "org_id": org_id,
        "role": role,
        "iat": int(now.timestamp()),
        "exp": int(expire.timestamp()),
    }
    return jwt.encode(payload, _settings.AUTH_SECRET, algorithm=_settings.JWT_ALGORITHM)


def decode_access_token(token: str) -> dict[str, Any] | None:
    """Decode and validate a JWT access token.

    Returns the payload dict on success, or None on any error
    (invalid signature, malformed token, expired).
    """
    try:
        payload = jwt.decode(
            token,
            _settings.AUTH_SECRET,
            algorithms=[_settings.JWT_ALGORITHM],
        )
        return payload
    except JWTError:
        return None


def is_token_expired(token: str) -> bool:
    """Check whether a token is expired without raising on other errors."""
    try:
        jwt.get_unverified_claims(token)
        # If decode succeeds, it's not expired.
        jwt.decode(token, _settings.AUTH_SECRET, algorithms=[_settings.JWT_ALGORITHM])
        return False
    except JWTError as e:
        return "expired" in str(e).lower() or "exp" in str(e).lower()
    except Exception:
        return True