src/core/access_control.py

"""Identity & authorization seam (ADR-0012 — AWS-independent).

The OHSU security review describes an intended access model — ~5 users, 1–2
admins, service accounts, and **no shared accounts** — that today has no
technical enforcement: both Spaces call Gradio ``launch()`` with no ``auth=``.
ADR-0012 adds an authentication gate whose *provider* (HuggingFace OAuth now,
OHSU SSO later) sits behind one small, stable seam. This module IS that seam:
the pure, provider-independent authorization logic + role lookup that the gate
and the audit trace both call. Swapping HF-OAuth for OHSU SSO changes only
*where an identity string comes from*, never the functions here.

Config, not code (mirrors ``recipients.yaml`` / ``UPLOAD_ADMIN_IDS``):

- ``ADMIN_IDS``   — comma-separated identities with the **admin** role (register
  uploads per ADR-0011, deploy). Canonical going forward.
- ``ALLOWED_IDS`` — comma-separated identities with the **user** role (may run
  analyses). Admins are implicitly allowed to run analyses too.
- ``UPLOAD_ADMIN_IDS`` — honored as admins for **back-compat** with the ADR-0011
  upload gate, so a deploy that only set that var keeps one coherent admin set.

Two roles only (ADR-0012 decision 4): ``admin`` and ``user``. Role is an
allow-list attribute in config, not a separate auth system.

**Fail-closed.** An unknown or empty identity is neither authorized nor an
admin, and has no role. Empty config therefore denies everyone — the correct
default for a fresh deploy: access opens only once an operator names the
identities, never by accident.
"""

from __future__ import annotations

import os
from dataclasses import dataclass

ADMIN_ROLE = "admin"
USER_ROLE = "user"

# The identity recorded in the audit trace when no authenticated principal was
# forwarded (e.g. the specialist called without a threaded identity). Recorded
# honestly as anonymous with no role — never silently attributed to someone.
ANONYMOUS = "anonymous"

_ADMIN_ENV_VARS = ("ADMIN_IDS", "UPLOAD_ADMIN_IDS")
_ALLOWED_ENV_VAR = "ALLOWED_IDS"


@dataclass(frozen=True)
class Principal:
    """An authenticated caller: an identity string and its resolved role.

    ``role`` is ``None`` when the identity is not on any allow-list (fail-closed).
    ``authorized`` is True when the identity may use the system at all (admin or
    listed user).
    """

    identity: str
    role: str | None

    @property
    def authorized(self) -> bool:
        return self.role is not None

    @property
    def is_admin(self) -> bool:
        return self.role == ADMIN_ROLE


def _ids_from_env(*env_vars: str) -> set[str]:
    """Union of comma-separated identities across ``env_vars`` (blanks dropped)."""
    ids: set[str] = set()
    for var in env_vars:
        raw = os.environ.get(var, "")
        ids.update(part.strip() for part in raw.split(",") if part.strip())
    return ids


def get_admin_ids() -> set[str]:
    """The admin identities (``ADMIN_IDS`` ∪ ``UPLOAD_ADMIN_IDS``, back-compat)."""
    return _ids_from_env(*_ADMIN_ENV_VARS)


def get_allowed_ids() -> set[str]:
    """The non-admin user identities (``ALLOWED_IDS``)."""
    return _ids_from_env(_ALLOWED_ENV_VAR)


def role_for(identity: str | None) -> str | None:
    """Resolve ``identity`` to ``"admin"`` / ``"user"`` / ``None`` (fail-closed).

    Admin membership wins over user membership, so an identity that appears on
    both lists is an admin. An empty/unknown identity has no role.
    """
    if not identity:
        return None
    identity = identity.strip()
    if not identity:
        return None
    if identity in get_admin_ids():
        return ADMIN_ROLE
    if identity in get_allowed_ids():
        return USER_ROLE
    return None


def is_authorized(identity: str | None) -> bool:
    """Whether ``identity`` may use the system at all (admin or listed user)."""
    return role_for(identity) is not None


def is_admin(identity: str | None) -> bool:
    """Whether ``identity`` holds the admin role."""
    return role_for(identity) == ADMIN_ROLE


def resolve_principal(identity: str | None) -> Principal:
    """Build a :class:`Principal` for ``identity`` (fail-closed on unknown)."""
    stripped = identity.strip() if identity else ""
    return Principal(identity=stripped or ANONYMOUS, role=role_for(identity))


_ENFORCE_VALUES = {"1", "on", "true", "enforce", "yes"}


def access_enforced() -> bool:
    """Whether the allow-list gate is enforced (env ``ACCESS_CONTROL``).

    OFF by default so the gate code can ship to a currently-public Space without
    locking anyone out; turn it on only once OAuth is enabled and the allow-list
    env vars are set (mirrors the orchestrator's ``access.access_enforced``).
    """
    return os.environ.get("ACCESS_CONTROL", "").strip().lower() in _ENFORCE_VALUES


def check_access(identity: str | None) -> tuple[bool, str | None]:
    """Gate decision for the specialist's own Gradio UI (ADR-0012).

    Returns ``(allowed, denial_message)``. When enforcement is off, everyone is
    allowed (current behavior). When on, the allow-list is fail-closed: a
    missing identity or a non-listed one is denied with a user-facing message.

    ``identity`` is the effective caller — the direct human's OAuth username, or
    the principal the orchestrator forwards (already gated at its front door).
    """
    if not access_enforced():
        return True, None
    if not identity:
        return False, (
            "🔒 Please sign in with your HuggingFace account (button above) to use this tool."
        )
    if not is_authorized(identity):
        return False, (
            f"🔒 Access denied — '{identity}' is not on the authorized-users list "
            "for this deployment. Contact the study admin to request access."
        )
    return True, None


def principal_trace_fields(identity: str | None) -> dict:
    """The identity fields ADR-0008's audit trace records (ADR-0012 decision 3).

    Returns ``{"principal": <identity or "anonymous">, "role": <role or None>}``.
    A missing/unauthenticated identity is recorded as ``anonymous`` with
    ``role=None`` — the honest "we don't know who ran this" state, not a blank.
    """
    principal = resolve_principal(identity)
    return {"principal": principal.identity, "role": principal.role}