ops-mason/src/ops_mason/reuse_surface_hub_writer.py

"""Decide whether a reuse-surface hub-writer token is the dedicated caller.

The check is metadata only. It never accepts a secret value, a login token, or
a command that reads ``platform/data/*``.
"""

from __future__ import annotations

POLICY = "workload-kv-read-reuse-surface-runtime"
ROLE = "reuse-surface-hub-writer-workload-kv-read"
MOUNT = "netkingdom"
# The role binds /assurance/level=aal2 (CCR-2026-0044), so a login that
# succeeds on it carried aal2 for that request.
ROLE_REQUIRES_AAL2 = True
FORBIDDEN_ENTITY_ID = "4a1f663c-0621-0483-edd0-f852274f4b6b"
FORBIDDEN_POLICIES = frozenset(
    {
        "default",
        "platform-admin",
        "operator-custody",
        "workload-kv-read-reuse-surface-runtime-eso",
    }
)

DATA = "platform/data/workloads/reuse/reuse-surface/runtime-secrets"
METADATA = "platform/metadata/workloads/reuse/reuse-surface/runtime-secrets"
PARENT_DATA = "platform/data/workloads/reuse/reuse-surface"
PARENT_METADATA = "platform/metadata/workloads/reuse/reuse-surface"
SIBLING = "platform/data/workloads/issue-core/issue-core/issue-core-runtime"

EXACT_READ = (DATA, METADATA)
MUST_DENY = (PARENT_DATA, PARENT_METADATA, SIBLING)


def _caps(report: dict, path: str) -> list[str]:
    raw = (report.get("capabilities") or {}).get(path)
    if not isinstance(raw, list):
        return []
    return sorted(str(item) for item in raw)


def assess(report: dict) -> list[str]:
    """Return the reasons a login report fails. An empty list is a pass.

    ``non-member`` passes only when login was refused. ``member`` passes only
    when the token's policy set is exactly the workload policy, the token is
    not the platform-admin entity, the two exact paths are read-only, the
    parent and a sibling workload are denied, and the token was revoked.
    """
    mode = report.get("mode")
    if mode not in {"member", "non-member"}:
        return [f"unknown mode {mode!r}"]
    login = report.get("login")
    if mode == "non-member":
        if login != "refused":
            return ["non-member login succeeded; the role accepted the wrong principal"]
        if not str(report.get("openbao_error") or "").strip():
            return ["non-member refusal recorded no openbao_error text"]
        return []

    failures: list[str] = []
    if login != "ok":
        failures.append(f"member login was {login!r}")
        return failures

    policies = set(report.get("policies") or [])
    identity_policies = set(report.get("identity_policies") or [])
    effective = policies | identity_policies
    if effective != {POLICY}:
        failures.append(
            "token policies are not exactly "
            f"{POLICY} (effective={sorted(effective)})"
        )
    overlap = effective & FORBIDDEN_POLICIES
    if overlap:
        failures.append(f"forbidden policies attached: {sorted(overlap)}")
    entity_id = str(report.get("entity_id") or "")
    if entity_id == FORBIDDEN_ENTITY_ID:
        failures.append(
            "token belongs to the platform-admin entity; do not attach this policy to it"
        )
    if not entity_id:
        failures.append("token lookup did not report an entity id")

    for path in EXACT_READ:
        caps = _caps(report, path)
        if caps != ["read"]:
            failures.append(f"{path} capabilities {caps or ['<missing>']}, want ['read']")
    for path in MUST_DENY:
        caps = _caps(report, path)
        if caps != ["deny"]:
            failures.append(f"{path} capabilities {caps or ['<missing>']}, want ['deny']")
    if report.get("revoked") is not True:
        failures.append("token was not revoked")
    return failures