core/security.py

"""JWT Bearer authentication with lightweight ABAC checks.

The API uses symmetric HS256 JWTs. Tokens carry standard claims plus an
``attrs`` object used for attribute-based access control, e.g.:

    {"sub": "alice", "exp": 1700000000,
     "attrs": {"role": "admin", "scopes": ["harvest:read", "harvest:write"]}}

Configuration via environment:
    JWT_SECRET      required HMAC secret for signing / verifying tokens
    JWT_ALGORITHM   optional, defaults to HS256
    JWT_ISSUER      optional issuer claim to validate
    JWT_AUDIENCE    optional audience claim to validate
"""

from __future__ import annotations

import os
from dataclasses import dataclass

import jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer


_bearer_scheme = HTTPBearer(auto_error=False)


@dataclass(frozen=True)
class Principal:
    """Authenticated caller and their attributes."""

    subject: str
    attrs: dict

    def has_scope(self, scope: str) -> bool:
        scopes = self.attrs.get("scopes") or []
        return scope in scopes

    def has_role(self, role: str) -> bool:
        return self.attrs.get("role") == role


def _get_secret() -> str:
    secret = os.environ.get("JWT_SECRET")
    if not secret:
        # Fail closed — never accept tokens without a configured secret.
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="JWT_SECRET is not configured",
        )
    return secret


def _decode(token: str) -> dict:
    algorithm = os.environ.get("JWT_ALGORITHM", "HS256")
    options = {"require": ["exp"]}
    kwargs: dict = {"algorithms": [algorithm], "options": options}
    audience = os.environ.get("JWT_AUDIENCE")
    if audience:
        kwargs["audience"] = audience
    issuer = os.environ.get("JWT_ISSUER")
    if issuer:
        kwargs["issuer"] = issuer
    try:
        return jwt.decode(token, _get_secret(), **kwargs)
    except jwt.ExpiredSignatureError as exc:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Token has expired",
            headers={"WWW-Authenticate": "Bearer"},
        ) from exc
    except jwt.InvalidTokenError as exc:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication token",
            headers={"WWW-Authenticate": "Bearer"},
        ) from exc


def get_current_principal(
    credentials: HTTPAuthorizationCredentials | None = Depends(_bearer_scheme),
) -> Principal:
    """FastAPI dependency that resolves the caller from a Bearer token."""
    if credentials is None or credentials.scheme.lower() != "bearer":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Missing Bearer credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )
    payload = _decode(credentials.credentials)
    subject = payload.get("sub") or payload.get("subject")
    if not subject:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Token is missing 'sub' claim",
            headers={"WWW-Authenticate": "Bearer"},
        )
    attrs = payload.get("attrs") or {}
    if not isinstance(attrs, dict):
        attrs = {}
    # Fall back to top-level scope/role if not nested under attrs.
    if "scopes" not in attrs and "scope" in payload:
        scope = payload["scope"]
        attrs["scopes"] = scope.split() if isinstance(scope, str) else list(scope)
    if "role" not in attrs and "role" in payload:
        attrs["role"] = payload["role"]
    return Principal(subject=str(subject), attrs=attrs)


def require_scopes(*required: str):
    """Return a dependency enforcing that the caller has all given scopes."""
    required_set = tuple(required)

    def _checker(principal: Principal = Depends(get_current_principal)) -> Principal:
        missing: list[str] = [s for s in required_set if not principal.has_scope(s)]
        if missing:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"Missing required scopes: {', '.join(required_set)}",
            )
        return principal

    return _checker