LegacyHUB/app/integrations/identity.py

"""TeamHUB gateway trusted-identity adapter (D2 / 09_ROLLOUT_AND_SSO, 12_AUTH).

The module never implements its own login/IdP and never talks to AD/LDAP,
HRHUB or the ``staff`` database directly. User identity arrives as trusted
``X-TeamHub-*`` headers injected by the platform gateway/SSO. This adapter is
the single place those headers are parsed and mapped to local permissions.

Trust in these headers depends on network isolation: the module must be
reachable only behind the gateway (see ``docker-compose.teamhub.yml`` and the
RUNBOOK firewall note). The machine-to-machine ``X-API-Key`` layer
(``app/api/security.py``) is intentionally separate and must not be conflated
with this user-context layer.
"""

from __future__ import annotations

from collections.abc import Mapping
from dataclasses import dataclass

from fastapi import HTTPException, Request, status

# Exact trusted-header names injected by the gateway (09 §2.1, 12 Prompt 4).
HEADER_ACTOR = "X-TeamHub-Actor"
HEADER_ACTOR_ID = "X-TeamHub-Actor-Id"
HEADER_APP = "X-TeamHub-App"
HEADER_ROLE = "X-TeamHub-Role"  # primary/default role, kept for ToolsHUB compat
HEADER_ROLES = "X-TeamHub-Roles"
HEADER_SCOPES = "X-TeamHub-Scopes"
HEADER_ENTITLEMENTS_VERSION = "X-TeamHub-Entitlements-Version"

# Domain scopes LegacyHUB authorizes against.
SCOPE_DOCUMENTS_READ = "documents:read"
SCOPE_DOCUMENTS_INGEST = "documents:ingest"
SCOPE_DOCUMENTS_ADMIN = "documents:admin"

# Local role -> scope mapping (app-specific Roles/Scopes per D2). Gateway is
# expected to inject only grants relevant to this app_code; this map is the
# module's interpretation of role names it receives.
ROLE_SCOPE_MAP: dict[str, frozenset[str]] = {
    "admin": frozenset({SCOPE_DOCUMENTS_READ, SCOPE_DOCUMENTS_INGEST, SCOPE_DOCUMENTS_ADMIN}),
    "ingestor": frozenset({SCOPE_DOCUMENTS_READ, SCOPE_DOCUMENTS_INGEST}),
    "qa": frozenset({SCOPE_DOCUMENTS_READ}),
    "viewer": frozenset({SCOPE_DOCUMENTS_READ}),
}


@dataclass(frozen=True)
class TeamHubIdentity:
    actor: str
    actor_id: str | None
    app: str | None
    primary_role: str | None
    roles: tuple[str, ...]
    scopes: tuple[str, ...]
    entitlements_version: str | None

    def has_role(self, role: str) -> bool:
        return role in self.roles

    def resolved_scopes(self) -> set[str]:
        """Union of explicit header scopes and scopes implied by roles."""
        out: set[str] = set(self.scopes)
        for role in self.roles:
            out |= ROLE_SCOPE_MAP.get(role.lower(), frozenset())
        return out

    def has_scope(self, scope: str) -> bool:
        return scope in self.resolved_scopes()


def _split_csv(value: str | None) -> tuple[str, ...]:
    if not value:
        return ()
    return tuple(part.strip() for part in value.split(",") if part.strip())


def parse_identity(headers: Mapping[str, str]) -> TeamHubIdentity | None:
    """Parse trusted headers into a :class:`TeamHubIdentity`.

    Returns ``None`` when no actor header is present (anonymous / direct dev
    access). ``headers`` may be a Starlette ``Headers`` object (case-insensitive)
    or any mapping keyed by the exact header names.
    """
    actor = headers.get(HEADER_ACTOR)
    if not actor or not actor.strip():
        return None

    primary_role = headers.get(HEADER_ROLE)
    primary_role = primary_role.strip() if primary_role and primary_role.strip() else None

    roles = _split_csv(headers.get(HEADER_ROLES))
    if primary_role and primary_role not in roles:
        roles = (primary_role, *roles)

    return TeamHubIdentity(
        actor=actor.strip(),
        actor_id=(headers.get(HEADER_ACTOR_ID) or None),
        app=(headers.get(HEADER_APP) or None),
        primary_role=primary_role,
        roles=roles,
        scopes=_split_csv(headers.get(HEADER_SCOPES)),
        entitlements_version=(headers.get(HEADER_ENTITLEMENTS_VERSION) or None),
    )


# ---- FastAPI dependencies ----

def get_identity(request: Request) -> TeamHubIdentity | None:
    """Optional identity: parsed from the request, ``None`` if absent."""
    return parse_identity(request.headers)


def require_identity(request: Request) -> TeamHubIdentity:
    identity = get_identity(request)
    if identity is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="missing trusted identity headers",
        )
    return identity


def require_scope(scope: str):
    """Dependency factory: require a resolved domain scope.

    No-op when ``AUTH_REQUIRE_IDENTITY`` is off (direct local dev, no gateway):
    identity is not present so scope cannot be enforced. Behind the gateway the
    enforcement middleware guarantees an actor, and this checks the scope.
    """

    def _dependency(request: Request) -> TeamHubIdentity | None:
        from app.config import settings  # noqa: PLC0415 - read current value

        if not settings.auth_require_identity:
            return None
        identity = require_identity(request)
        if not identity.has_scope(scope):
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"missing required scope: {scope}",
            )
        return identity

    return _dependency


__all__ = [
    "HEADER_ACTOR",
    "HEADER_ACTOR_ID",
    "HEADER_APP",
    "HEADER_ENTITLEMENTS_VERSION",
    "HEADER_ROLE",
    "HEADER_ROLES",
    "HEADER_SCOPES",
    "SCOPE_DOCUMENTS_ADMIN",
    "SCOPE_DOCUMENTS_INGEST",
    "SCOPE_DOCUMENTS_READ",
    "TeamHubIdentity",
    "get_identity",
    "parse_identity",
    "require_identity",
    "require_scope",
]