feat: remove unused authentication dependencies and guards, add comprehensive...

"""

FastAPI authentication dependencies.

Bearer-token path for machine-to-machine / API clients:

  * bearer  — `Authorization: Bearer <jwt>`

When AUTH__SECURITY_MODE=none every check is bypassed and the dependency returns

None, so the app runs without any Keycloak instance.

"""

from typing import Annotated

import structlog

from fastapi import Depends, HTTPException, status

from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer

from portal_service.api.auth.verification import _decode_token

from portal_service.config import AuthSettings, SecurityMode, get_settings

from portal_service.domain.errors import (

    AuthenticationError,

    IdentityProviderUnavailableError,

    TokenExpiredError,

)

from portal_service.domain.models import TokenClaims

logger: structlog.BoundLogger = structlog.get_logger(__name__)

_bearer = HTTPBearer(auto_error=False)

def _auth_settings() -> AuthSettings:

    return get_settings().auth

BearerDep = Annotated[HTTPAuthorizationCredentials | None, Depends(_bearer)]

def get_current_user(

    credentials: BearerDep,

    auth_settings: Annotated[AuthSettings, Depends(_auth_settings)],

) -> TokenClaims | None:

    if auth_settings.security_mode == SecurityMode.NONE:

        logger.warning("security disabled — skipping auth")

        return None

    if credentials is None:

        raise HTTPException(

            status_code=status.HTTP_401_UNAUTHORIZED,

            detail="Not authenticated",

            headers={"WWW-Authenticate": "Bearer"},

        )

    try:

        return _decode_token(credentials.credentials)

    except TokenExpiredError:

        raise HTTPException(

            status_code=status.HTTP_401_UNAUTHORIZED,

            detail="Token has expired",

            headers={"WWW-Authenticate": "Bearer"},

        )

    except AuthenticationError as exc:

        logger.warning("bearer token invalid", error=str(exc))

        raise HTTPException(

            status_code=status.HTTP_401_UNAUTHORIZED,

            detail="Could not validate credentials",

            headers={"WWW-Authenticate": "Bearer"},

        )

    except IdentityProviderUnavailableError as exc:

        logger.error("identity provider unreachable", error=str(exc))

        raise HTTPException(

            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,

            detail="Authentication service unavailable",

        )

CurrentUserDep = Annotated[TokenClaims | None, Depends(get_current_user)]

"""

Role/scope authorization guards.

These live in the HTTP layer and raise HTTPException directly (no domain-error

indirection needed here). Each guard bypasses checks when security is disabled.

"""

from typing import Annotated, Any

import structlog

from fastapi import Depends, HTTPException, status

from portal_service.api.auth.dependencies import (

    CurrentUserDep,

    _auth_settings,

)

from portal_service.config import AuthSettings, SecurityMode

from portal_service.domain.models import Role, TokenClaims

logger: structlog.BoundLogger = structlog.get_logger(__name__)

def _require(kind: str, required_set: frozenset[str], actual: set[str], user: TokenClaims) -> None:

    """All-of check: raise 403 if any item from required_set is absent from actual."""

    missing = required_set - actual

    if missing:

        logger.warning(

            f"authorization failed: missing {kind}",

            required=list(required_set),

            present=list(actual),

            sub=user.sub,

        )

        raise HTTPException(

            status_code=status.HTTP_403_FORBIDDEN,

            detail=f"Missing required {kind}: {', '.join(sorted(missing))}",

        )

def _require_any(kind: str, accepted: frozenset[str], actual: set[str], user: TokenClaims) -> None:

    """Any-of check: raise 403 unless at least one item from accepted is in actual."""

    if not accepted & actual:

        logger.warning(

            f"authorization failed: no accepted {kind}",

            accepted=sorted(accepted),

            present=sorted(actual),

            sub=user.sub,

        )

        raise HTTPException(

            status_code=status.HTTP_403_FORBIDDEN,

            detail=f"Requires one of {kind}: {', '.join(sorted(accepted))}",

        )

def require_any_role(*accepted: Role) -> Any:

    """Allow the request if the user holds at least one of the given roles."""

    if not accepted:

        raise ValueError("require_any_role() needs at least one role")

    accepted_set: frozenset[str] = frozenset(r.value for r in accepted)

    def _check(

        user: CurrentUserDep,

        auth_settings: Annotated[AuthSettings, Depends(_auth_settings)],

    ) -> TokenClaims | None:

        if auth_settings.security_mode == SecurityMode.NONE:

            return None

        if user is None:

            raise HTTPException(

                status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated"

            )

        _require_any("roles", accepted_set, user.roles, user)

        return user

    return Depends(_check)

def require_scopes(*required: str) -> Any:

    """Allow the request only if the token carries every one of the given scopes."""

    required_set: frozenset[str] = frozenset(required)

    def _check(

        user: CurrentUserDep,

        auth_settings: Annotated[AuthSettings, Depends(_auth_settings)],

    ) -> TokenClaims | None:

        if auth_settings.security_mode == SecurityMode.NONE:

            return None

        if user is None:

            raise HTTPException(

                status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated"

            )

        _require("scopes", required_set, user.scopes, user)

        return user

    return Depends(_check)