Gitea-Tools/webui/policy_inventory.py

"""Read-only workflow policy and guardrail inventory for the web UI (#646).

Policy and guardrails live in code, profiles, docs, and skills. An operator
cannot *see* the active workflow policy configuration from the console without
reading the repository tree. This module projects the major guardrails into a
redacted, machine-readable inventory with source attribution (file / module /
doc), so the console can render them as HTML tables with source pointers.

Design constraints (Phase 3, #646):

- **Read-only projection.** Nothing here edits policy or exposes a toggle that
  could weaken a gate. It reports what is already enforced elsewhere.
- **Source attribution without secrets.** Every guardrail carries pointers to
  the file/module/doc that owns it. Live values are compact summaries derived
  from the safe policy accessors that already exist (``rbac_matrix``,
  ``redaction_policy``, ``audit_policy``); raw regex, tokens, and endpoints are
  never embedded.
- **Redact before emit.** ``snapshot_to_dict`` runs the whole payload through
  ``console_redaction.redact_payload`` so a planted or accidental secret in any
  projected value degrades to the placeholder rather than reaching a client.
- **Fail soft.** A projection that raises is recorded as a per-entry error and
  never takes the page down; a guardrail is still listed with its sources.
- **Diff vs documented defaults where feasible.** When a guardrail declares a
  documented invariant, the active projection is compared against it and the
  result is reported; otherwise the diff is explicitly ``None`` with a reason.
"""

from __future__ import annotations

from dataclasses import dataclass
from typing import Any, Callable

from webui import console_audit
from webui import console_authz
from webui import console_redaction

SCHEMA_VERSION = 1

READ_ONLY_NOTE = (
    "Read-only projection of guardrails enforced in code, profiles, docs, and "
    "skills. This view never edits policy and exposes no gate-weakening toggle."
)


@dataclass(frozen=True)
class SourcePointer:
    """Where a guardrail is defined. Attribution only — never a secret."""

    label: str
    path: str
    kind: str  # "module" | "doc" | "script" | "config"
    anchor: str | None = None

    def to_dict(self) -> dict[str, Any]:
        return {
            "label": self.label,
            "path": self.path,
            "kind": self.kind,
            "anchor": self.anchor,
        }


@dataclass(frozen=True)
class PolicyEntry:
    key: str
    title: str
    category: str
    summary: str
    sources: tuple[SourcePointer, ...]
    active: dict[str, Any] | None
    documented_default: dict[str, Any] | None
    diff: dict[str, Any] | None
    error: str | None = None

    def to_dict(self) -> dict[str, Any]:
        return {
            "key": self.key,
            "title": self.title,
            "category": self.category,
            "summary": self.summary,
            "sources": [s.to_dict() for s in self.sources],
            "active": self.active,
            "documented_default": self.documented_default,
            "diff": self.diff,
            "error": self.error,
        }


@dataclass(frozen=True)
class PolicyInventorySnapshot:
    schema_version: int
    read_only: bool
    note: str
    entries: tuple[PolicyEntry, ...]
    categories: tuple[str, ...]
    build_errors: tuple[str, ...]


def _diff_active_vs_default(
    active: dict[str, Any] | None,
    documented_default: dict[str, Any] | None,
) -> dict[str, Any] | None:
    """Compare only the keys the documented default declares.

    Returns ``None`` when no documented default is declared (diff not feasible)
    or when the active projection is unavailable. Otherwise reports, per
    declared key, whether the active value matches the documented invariant.
    """
    if not documented_default:
        return None
    if not active:
        return {"status": "active_unavailable", "checked": {}}
    checked: dict[str, Any] = {}
    matches = True
    for key, expected in documented_default.items():
        observed = active.get(key)
        ok = observed == expected
        matches = matches and ok
        checked[key] = {"expected": expected, "observed": observed, "matches": ok}
    return {
        "status": "matches_documented_default" if matches else "drift_detected",
        "checked": checked,
    }


# ── Live projections (compact, safe, fail-soft) ──────────────────────────────
# Each returns a small dict of already-safe machine values. They are module
# level so tests can substitute one to prove the redaction pass runs.


def _project_role_separation() -> dict[str, Any]:
    matrix = console_authz.rbac_matrix()
    return {
        "model_version": matrix.get("model_version"),
        "active_phase": matrix.get("active_phase"),
        "roles": [r.get("role") for r in matrix.get("roles", [])],
        "privileged_action_count": len(matrix.get("privileged_actions", [])),
        "default_decision": matrix.get("default_decision"),
        "execution_enabled": matrix.get("execution_enabled"),
    }


def _project_redaction() -> dict[str, Any]:
    policy = console_redaction.redaction_policy()
    return {
        "policy_version": policy.get("policy_version"),
        "placeholder": policy.get("placeholder"),
        "applies_to": policy.get("applies_to"),
        "console_detector_count": len(policy.get("console_rules", [])),
        "redact_before_persist": policy.get("redact_before_persist"),
        "failure_mode": policy.get("failure_mode"),
    }


def _project_audit() -> dict[str, Any]:
    policy = console_audit.audit_policy()
    return {
        "schema_version": policy.get("schema_version"),
        "required_field_count": len(policy.get("required_fields", [])),
        "results": policy.get("results"),
        "retention_defaults_days": policy.get("retention_defaults_days"),
        "append_only": policy.get("append_only"),
        "redact_before_persist": policy.get("redact_before_persist"),
        "enabled": policy.get("enabled"),
    }


def _static(value: dict[str, Any]) -> Callable[[], dict[str, Any]]:
    return lambda: dict(value)


# ── Guardrail catalog ────────────────────────────────────────────────────────
# One row per major guardrail. ``project`` yields the active value (may raise;
# caught per entry). ``documented_default`` drives the feasible diff.

_CatalogRow = tuple[
    str,
    str,
    str,
    str,
    tuple[SourcePointer, ...],
    Callable[[], dict[str, Any]] | None,
    dict[str, Any] | None,
]

_CATALOG: tuple[_CatalogRow, ...] = (
    (
        "role_separation",
        "Role separation and RBAC",
        "role_separation",
        "Author, reviewer, merger, and reconciler capabilities are disjoint and "
        "role-exclusive; self-review and self-merge are always blocked. The "
        "console RBAC model defaults to deny.",
        (
            SourcePointer("task capability map", "task_capability_map.py", "module"),
            SourcePointer("role/namespace gate", "role_namespace_gate.py", "module"),
            SourcePointer("console RBAC", "webui/console_authz.py", "module"),
        ),
        _project_role_separation,
        {"default_decision": "deny", "execution_enabled": False},
    ),
    (
        "lease_rules",
        "Issue and PR lease lifecycle",
        "lease_rules",
        "Durable work is claimed through issue locks and control-plane leases "
        "with freshness, expiry, and dead-session recovery; abandoned or stale "
        "claims are reclaimed only through the sanctioned recovery path.",
        (
            SourcePointer("issue lock store", "issue_lock_store.py", "module"),
            SourcePointer("branch cleanup guard", "branch_cleanup_guard.py", "module"),
            SourcePointer("safety model §5", "docs/safety-model.md", "doc", "5-mutation-gating"),
        ),
        None,
        None,
    ),
    (
        "worktree_rules",
        "Author worktree binding",
        "worktree_rules",
        "Author mutations require a validated worktree under branches/ derived "
        "from the active issue lock; silent fallback to the stable control "
        "checkout or master is forbidden (#618).",
        (
            SourcePointer("author worktree gate", "author_mutation_worktree.py", "module"),
            SourcePointer("worktree bootstrap", "scripts/worktree-start", "script"),
            SourcePointer("workflow scope guard", "workflow_scope_guard.py", "module"),
        ),
        None,
        None,
    ),
    (
        "merge_confirmation",
        "Explicit merge confirmation",
        "merge_confirmation",
        "A merge fails closed unless the caller passes the exact confirmation "
        "phrase for that PR; reviewing never implies merging.",
        (
            SourcePointer("merge path", "merge_pr.py", "module"),
            SourcePointer("merge tool gate", "gitea_mcp_server.py", "module"),
        ),
        _static({"required_confirmation_format": "MERGE PR <n>", "auto_merge": False}),
        {"auto_merge": False},
    ),
    (
        "redaction",
        "Secret redaction",
        "redaction",
        "Every console surface runs the shared gitea_audit pass then console "
        "patterns before any payload, HTML, log line, or audit record leaves "
        "the server; unredactable values fail closed to the placeholder.",
        (
            SourcePointer("console redaction", "webui/console_redaction.py", "module"),
            SourcePointer("shared redaction", "gitea_audit.py", "module"),
            SourcePointer("safety model §3", "docs/safety-model.md", "doc", "3-secret-redaction"),
        ),
        _project_redaction,
        {"redact_before_persist": True},
    ),
    (
        "contamination",
        "Contamination containment",
        "contamination",
        "A session contaminated by a direct stable-branch push or a manual MCP "
        "daemon kill is blocked from review, merge, close, and completion "
        "mutations until cleared (reconciler-exempt).",
        (
            SourcePointer("contamination gates", "gitea_mcp_server.py", "module"),
            SourcePointer("stable-branch audit", "workflow_scope_guard.py", "module"),
        ),
        None,
        None,
    ),
    (
        "allocator_policy",
        "Work allocation policy",
        "allocator_policy",
        "Workers do not self-select exclusive work; the controller-owned "
        "allocator ranks the complete queue by priority then PRs-before-issues "
        "then ascending number, honoring dependency edges and foreign claims.",
        (
            SourcePointer("allocator", "gitea_mcp_server.py", "module"),
            SourcePointer("safety model §5", "docs/safety-model.md", "doc", "5-mutation-gating"),
        ),
        _static(
            {
                "self_select_exclusive_work": False,
                "ranking": "priority desc, PRs before issues, number asc",
                "respects_dependency_edges": True,
                "respects_foreign_claims": True,
            }
        ),
        {"self_select_exclusive_work": False},
    ),
    (
        "audit_logging",
        "Audit logging",
        "audit_logging",
        "Console intent and authorization outcomes are recorded to an "
        "append-only, redact-before-persist audit log; MCP mutations are "
        "recorded by gitea_audit and correlated by request id.",
        (
            SourcePointer("console audit", "webui/console_audit.py", "module"),
            SourcePointer("MCP audit", "gitea_audit.py", "module"),
            SourcePointer("safety model §1", "docs/safety-model.md", "doc", "1-audit-logging-and-confirmation"),
        ),
        _project_audit,
        {"append_only": True, "redact_before_persist": True},
    ),
    (
        "mutation_gating",
        "Mutation gating and master parity",
        "mutation_gating",
        "Mutations fail closed while the running server is stale relative to "
        "master, and every mutation is preceded by identity and capability "
        "resolution in a fixed pre-flight order.",
        (
            SourcePointer("mutation gate", "gitea_mcp_server.py", "module"),
            SourcePointer("safety model §5", "docs/safety-model.md", "doc", "5-mutation-gating"),
        ),
        _static(
            {
                "stale_runtime_blocks_mutations": True,
                "preflight_order": "whoami -> resolve_task_capability -> mutation",
            }
        ),
        {"stale_runtime_blocks_mutations": True},
    ),
)


def _build_entry(row: _CatalogRow) -> PolicyEntry:
    key, title, category, summary, sources, project, documented_default = row
    active: dict[str, Any] | None = None
    error: str | None = None
    if project is not None:
        try:
            active = project()
        except Exception as exc:  # noqa: BLE001 — fail soft; never take the page down
            active = None
            error = f"active projection unavailable: {exc}"
    diff = _diff_active_vs_default(active, documented_default)
    return PolicyEntry(
        key=key,
        title=title,
        category=category,
        summary=summary,
        sources=sources,
        active=active,
        documented_default=documented_default,
        diff=diff,
        error=error,
    )


def load_policy_inventory() -> PolicyInventorySnapshot:
    """Build the read-only guardrail inventory. Never raises for one bad entry."""
    entries: list[PolicyEntry] = []
    build_errors: list[str] = []
    for row in _CATALOG:
        try:
            entries.append(_build_entry(row))
        except Exception as exc:  # noqa: BLE001 — one row must not break the rest
            build_errors.append(f"{row[0]}: {exc}")
    categories = tuple(dict.fromkeys(e.category for e in entries))
    return PolicyInventorySnapshot(
        schema_version=SCHEMA_VERSION,
        read_only=True,
        note=READ_ONLY_NOTE,
        entries=tuple(entries),
        categories=categories,
        build_errors=tuple(build_errors),
    )


def snapshot_to_dict(snapshot: PolicyInventorySnapshot) -> dict[str, Any]:
    """Serialize the snapshot, redacting the entire payload before it is emitted."""
    payload = {
        "schema_version": snapshot.schema_version,
        "read_only": snapshot.read_only,
        "note": snapshot.note,
        "categories": list(snapshot.categories),
        "entry_count": len(snapshot.entries),
        "entries": [entry.to_dict() for entry in snapshot.entries],
        "build_errors": list(snapshot.build_errors),
    }
    return console_redaction.redact_payload(payload)