diff --git a/docs/webui-local-dev.md b/docs/webui-local-dev.md index 15c00e4..cf4bfad 100644 --- a/docs/webui-local-dev.md +++ b/docs/webui-local-dev.md @@ -66,6 +66,8 @@ status, onboarding checklist state, and the fail-closed error payloads (#635). | `/api/prompts` | JSON prompt export with workflow hashes | | `/runtime` | MCP runtime health and stale detection (#430) | | `/api/runtime` | JSON runtime health export | +| `/policy` | Workflow policy and guardrail configuration visibility (#646) | +| `/api/v1/policy` | Versioned JSON guardrail inventory (redacted, read-only) | | `/audit` | Report audit paste + validator preview (#431) | | `/api/audit` | JSON validator preview (POST `report_text`, optional `task_kind`) | | `/worktrees` | Worktree hygiene dashboard (#432) | @@ -78,7 +80,6 @@ status, onboarding checklist state, and the fail-closed error payloads (#635). | `/sessions` | Phase 1 shell stub — session inventory (backed by #636) | | `/inventory` | Phase 1 shell stub — unified inventory (backed by #636) | | `/timeline` | Phase 1 shell stub — workflow event timeline | -| `/policy` | Phase 1 shell stub — capability/role policy placeholder | | `/insights` | Phase 1 shell stub — operational insights placeholder | Most routes are GET-only. POST/PUT/PATCH/DELETE return `405` with @@ -239,12 +240,25 @@ health, workflow/schema SHA-256 hashes, and stale-runtime warnings when the checkout is behind merged safety-gate changes. Restart guidance links to #420; no tokens or MCP restart actions are exposed. +## Policy & guardrail visibility (#646) + +`/policy` (HTML) and `/api/v1/policy` (JSON) surface a **read-only** projection +of the major workflow guardrails — role separation/RBAC, lease lifecycle, +author worktree binding, merge confirmation, secret redaction, contamination +containment, allocator policy, audit logging, and mutation gating. Each entry +carries source pointers to the file/module/doc that owns it, a compact active +value derived from the existing safe policy accessors, and — where a documented +default is declared — a diff of active vs documented. The whole payload is run +through the console redaction pass before it is emitted, so a planted or +accidental secret degrades to the placeholder rather than reaching a client. +The view never edits policy and exposes no gate-weakening toggle. + ## Application shell — Phase 1 (#638) The console shell (`webui/layout.py`) renders a grouped navigation driven by a single nav-config module, `webui/nav.py`. Nav groups follow the epic #631 Phase 1 information architecture: **Health, Traffic, Runtime/Sessions, -Projects, Inventory, Timeline, Policy** (placeholder), and **Insights** +Projects, Inventory, Timeline, Policy** (live via #646), and **Insights** (placeholder). Live views and Phase 1 placeholders (`stub`) are declared in one place so the layout and the route table cannot drift. @@ -254,10 +268,10 @@ a **mode: read-only** badge — plus a **Docs** link to this document. No privileged action controls are present in the Phase 1 shell. Not-yet-implemented surfaces (`/sessions`, `/inventory`, `/timeline`, -`/policy`, `/insights`) resolve to graceful read-only stub pages instead of -404s; their backing views land in later child issues of #631 (the inventory -surfaces are backed by #636). Mutating methods on stub routes still fail closed -with `read-only-mvp`. +`/insights`) resolve to graceful read-only stub pages instead of 404s; their +backing views land in later child issues of #631 (the inventory surfaces are +backed by #636). `/policy` is a live read-only surface (#646), not a stub. +Mutating methods on stub routes still fail closed with `read-only-mvp`. ## System-health dashboard (#639) diff --git a/tests/test_webui_policy_visibility.py b/tests/test_webui_policy_visibility.py new file mode 100644 index 0000000..d07c43d --- /dev/null +++ b/tests/test_webui_policy_visibility.py @@ -0,0 +1,243 @@ +"""Tests for the read-only workflow policy/guardrail visibility view (#646). + +Covers issue #646 acceptance criteria: + +1. Console lists major guardrails with source pointers. +2. Secrets redacted. +3. Tests ensure sample secrets never appear. +4. Docs explain read-only nature (asserted here for the page copy; the doc + itself is covered by inspection). +""" + +import json +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from starlette.testclient import TestClient + +from webui import console_redaction +from webui import policy_inventory +from webui.app import create_app +from webui.policy_inventory import ( + PolicyEntry, + PolicyInventorySnapshot, + SourcePointer, + load_policy_inventory, + snapshot_to_dict, +) +from webui.policy_views import render_policy_page + + +def _entry(key, category, *, active=None, error=None): + return PolicyEntry( + key=key, + title=key.replace("_", " ").title(), + category=category, + summary=f"summary for {key}", + sources=(SourcePointer("src", f"{key}.py", "module"),), + active=active, + documented_default=None, + diff=None, + error=error, + ) + + +def _snapshot(entries): + return PolicyInventorySnapshot( + schema_version=1, + read_only=True, + note="read-only projection", + entries=tuple(entries), + categories=tuple(dict.fromkeys(e.category for e in entries)), + build_errors=(), + ) + +# The guardrail categories issue #646 names as in-scope. +_EXPECTED_CATEGORIES = { + "role_separation", + "lease_rules", + "worktree_rules", + "merge_confirmation", + "redaction", + "contamination", + "allocator_policy", + "audit_logging", + "mutation_gating", +} + + +class TestPolicyInventoryModel(unittest.TestCase): + def test_major_guardrails_present(self): + snapshot = load_policy_inventory() + categories = {e.category for e in snapshot.entries} + self.assertEqual(_EXPECTED_CATEGORIES, categories) + self.assertGreaterEqual(len(snapshot.entries), len(_EXPECTED_CATEGORIES)) + + def test_every_guardrail_has_source_pointers(self): + # AC1: source attribution (file/module/doc) for every guardrail. + snapshot = load_policy_inventory() + for entry in snapshot.entries: + with self.subTest(entry=entry.key): + self.assertTrue(entry.sources, "guardrail must carry source pointers") + for source in entry.sources: + self.assertTrue(source.path) + self.assertIn(source.kind, {"module", "doc", "script", "config"}) + + def test_diff_reported_where_documented_default_declared(self): + snapshot = load_policy_inventory() + checked_any = False + for entry in snapshot.entries: + if entry.documented_default is None: + self.assertIsNone(entry.diff) + continue + checked_any = True + self.assertIsNotNone(entry.diff) + self.assertEqual( + entry.diff["status"], + "matches_documented_default", + f"{entry.key} drifted from its documented default: {entry.diff}", + ) + self.assertTrue(checked_any, "at least one guardrail should declare a default") + + def test_live_projections_populate_active(self): + snapshot = load_policy_inventory() + by_key = {e.key: e for e in snapshot.entries} + for key in ("role_separation", "redaction", "audit_logging"): + self.assertIsNone(by_key[key].error, f"{key} projection failed") + self.assertIsInstance(by_key[key].active, dict) + + def test_build_entry_is_fail_soft_on_projection_error(self): + def _boom(): + raise RuntimeError("projection exploded") + + row = ( + "redaction", + "Secret redaction", + "redaction", + "summary", + (SourcePointer("x", "webui/console_redaction.py", "module"),), + _boom, + {"redact_before_persist": True}, + ) + entry = policy_inventory._build_entry(row) + self.assertIsNone(entry.active) + self.assertIsNotNone(entry.error) + self.assertEqual(entry.diff["status"], "active_unavailable") + + +class TestPolicyRedaction(unittest.TestCase): + def test_real_snapshot_has_no_secret_shapes(self): + # AC3: the real emitted payload never carries a known secret shape. + payload = snapshot_to_dict(load_policy_inventory()) + self.assertEqual(console_redaction.scan_for_secrets(payload), []) + + def test_planted_keychain_secret_is_redacted(self): + # AC2/AC3: a secret planted in an active projection is masked before emit. + snapshot = _snapshot([ + _entry( + "redaction", + "redaction", + active={"leaked": "keychain:prgs-author-super-secret", "roles": ["author"]}, + ) + ]) + payload = snapshot_to_dict(snapshot) + blob = json.dumps(payload) + self.assertNotIn("keychain:prgs-author-super-secret", blob) + self.assertEqual(console_redaction.scan_for_secrets(payload), []) + + def test_planted_credential_assignment_is_redacted(self): + snapshot = _snapshot([ + _entry( + "audit_logging", + "audit_logging", + active={"leaked": "token=abcd1234efgh5678", "append_only": True}, + ) + ]) + payload = snapshot_to_dict(snapshot) + blob = json.dumps(payload) + self.assertNotIn("abcd1234efgh5678", blob) + self.assertEqual(console_redaction.scan_for_secrets(payload), []) + + def test_planted_secret_is_redacted_in_html_emit(self): + # AC2/AC3: HTML emit path runs redaction before rendering HTML cards. + snapshot = _snapshot([ + _entry( + "redaction", + "redaction", + active={"leaked": "keychain:prgs-author-super-secret", "roles": ["author"]}, + ) + ]) + html_output = render_policy_page(snapshot) + self.assertNotIn("keychain:prgs-author-super-secret", html_output) + self.assertEqual(console_redaction.scan_for_secrets(html_output), []) + + +class TestPolicyRoutes(unittest.TestCase): + def setUp(self): + self.client = TestClient(create_app()) + + def test_policy_html_lists_guardrails_with_sources(self): + response = self.client.get("/policy") + self.assertEqual(response.status_code, 200) + text = response.text + self.assertIn("Workflow policy", text) + self.assertIn("Role separation and RBAC", text) + self.assertIn("Source pointers", text) + self.assertIn("task_capability_map.py", text) + self.assertIn("docs/safety-model.md", text) + + def test_policy_html_states_read_only(self): + # AC4: the page explains its read-only nature. + text = self.client.get("/policy").text + self.assertIn("read-only", text.lower()) + self.assertNotIn("Policy', text) + self.assertNotIn('href="/policy" class="nav-stub"', text) + + +class TestPolicyViewFailSoft(unittest.TestCase): + def test_page_renders_when_a_projection_errors(self): + snapshot = _snapshot([ + _entry("role_separation", "role_separation", error="active projection unavailable: boom"), + _entry("redaction", "redaction", active={"redact_before_persist": True}), + ]) + page = render_policy_page(snapshot) + # The errored guardrail surfaces its error; other guardrails still render. + self.assertIn("Active value unavailable", page) + self.assertIn("Redaction", page) + self.assertIn("Workflow policy", page) + + +if __name__ == "__main__": + unittest.main() diff --git a/webui/app.py b/webui/app.py index 2c7b146..4167364 100644 --- a/webui/app.py +++ b/webui/app.py @@ -46,6 +46,8 @@ from webui.worktree_scanner import load_hygiene_snapshot, snapshot_to_dict as wo from webui.worktree_views import render_worktrees_page from webui.runtime_health import load_runtime_snapshot, snapshot_to_dict as runtime_snapshot_to_dict from webui.runtime_views import render_runtime_page +from webui.policy_inventory import load_policy_inventory, snapshot_to_dict as policy_snapshot_to_dict +from webui.policy_views import render_policy_page from webui.inventory import ( SECTION_NAMES as _INVENTORY_SECTIONS, load_inventory_snapshot, @@ -313,6 +315,17 @@ async def api_runtime(_request: Request) -> JSONResponse: return JSONResponse(runtime_snapshot_to_dict(load_runtime_snapshot())) +async def policy(_request: Request) -> HTMLResponse: + snapshot = load_policy_inventory() + return HTMLResponse( + render_page(title="Policy", body_html=render_policy_page(snapshot)) + ) + + +async def api_v1_policy(_request: Request) -> JSONResponse: + return JSONResponse(policy_snapshot_to_dict(load_policy_inventory())) + + async def _parse_audit_form(request: Request) -> tuple[str, str | None]: if request.method == "GET": return "", None @@ -750,6 +763,8 @@ def create_app(*, bind_host: str | None = None) -> Starlette: Route("/api/prompts", api_prompts, methods=["GET"]), Route("/runtime", runtime, methods=["GET"]), Route("/api/runtime", api_runtime, methods=["GET"]), + Route("/policy", policy, methods=["GET"]), + Route("/api/v1/policy", api_v1_policy, methods=["GET"]), Route("/api/v1/timeline", api_v1_timeline, methods=["GET"]), Route("/analytics", analytics, methods=["GET"]), Route("/api/analytics", api_v1_analytics, methods=["GET"]), diff --git a/webui/nav.py b/webui/nav.py index 45d1212..a3f21ed 100644 --- a/webui/nav.py +++ b/webui/nav.py @@ -5,7 +5,7 @@ the ``webui/app.py`` route table stay aligned with epic #631. Read-only: every destination is a GET view or a Phase 1 placeholder. No mutation links. Nav groups follow the #631 Phase 1 information architecture: Health, Traffic, -Runtime/Sessions, Projects, Inventory, Timeline, Policy (placeholder), and +Runtime/Sessions, Projects, Inventory, Timeline, Policy (live via #646), and Insights (placeholder). Later-phase surfaces are declared as ``stub`` items and backed by ``STUB_PAGES`` so their nav links resolve to a graceful placeholder instead of a 404. @@ -60,7 +60,7 @@ NAV_GROUPS: tuple[NavGroup, ...] = ( NavItem("/timeline", "Timeline", "stub"), )), NavGroup("Policy", ( - NavItem("/policy", "Policy", "stub"), + NavItem("/policy", "Policy"), NavItem("/prompts", "Prompts"), )), NavGroup("Insights", ( @@ -89,10 +89,6 @@ STUB_PAGES: dict[str, tuple[str, str]] = { "Timeline", "Workflow event timeline across issues and PRs. A later Phase 1 surface.", ), - "/policy": ( - "Policy", - "Capability and role policy surface. Placeholder until a later phase.", - ), "/insights": ( "Insights", "Aggregate operational insights and trends. Placeholder until a later " diff --git a/webui/policy_inventory.py b/webui/policy_inventory.py new file mode 100644 index 0000000..96e853a --- /dev/null +++ b/webui/policy_inventory.py @@ -0,0 +1,387 @@ +"""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 ", "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) diff --git a/webui/policy_views.py b/webui/policy_views.py new file mode 100644 index 0000000..fcfffc3 --- /dev/null +++ b/webui/policy_views.py @@ -0,0 +1,145 @@ +"""HTML views for the workflow policy and guardrail inventory (#646).""" + +from __future__ import annotations + +import html +import json +from typing import Any + +from webui import console_redaction +from webui.policy_inventory import ( + PolicyEntry, + PolicyInventorySnapshot, + SourcePointer, + snapshot_to_dict, +) + + +def _source_pointer(source: dict[str, Any] | SourcePointer) -> str: + if isinstance(source, dict): + label = str(source.get("label") or "") + path = str(source.get("path") or "") + anchor = source.get("anchor") + kind = str(source.get("kind") or "") + else: + label = source.label + path = source.path + anchor = source.anchor + kind = source.kind + + if anchor: + path = f"{path}#{anchor}" + return ( + f"
  • {html.escape(label)} — " + f"{html.escape(path)} " + f"({html.escape(kind)})
  • " + ) + + +def _active_block(entry: dict[str, Any] | PolicyEntry) -> str: + error = entry.get("error") if isinstance(entry, dict) else entry.error + active = entry.get("active") if isinstance(entry, dict) else entry.active + if error: + return ( + "

    Active value unavailable: " + f"{html.escape(error)}

    " + ) + if not active: + return "

    No live projection for this guardrail.

    " + pretty = json.dumps(active, indent=2, sort_keys=True, default=str) + return f"
    {html.escape(pretty)}
    " + + +def _diff_block(entry: dict[str, Any] | PolicyEntry) -> str: + diff = entry.get("diff") if isinstance(entry, dict) else entry.diff + documented_default = entry.get("documented_default") if isinstance(entry, dict) else entry.documented_default + if diff is None: + if documented_default is None: + return "

    Diff vs documented default: not feasible (no declared default).

    " + return "

    Diff vs documented default: unavailable.

    " + status = str(diff.get("status") if isinstance(diff, dict) else "unknown") + badge = "badge-claimed" if status == "matches_documented_default" else "badge-blocked" + rows = [] + checked = diff.get("checked") if isinstance(diff, dict) else {} + if isinstance(checked, dict): + for key, cell in checked.items(): + cell_dict = cell if isinstance(cell, dict) else {} + marker = "✓" if cell_dict.get("matches") else "✗" + rows.append( + "" + f"{html.escape(str(key))}" + f"{html.escape(str(cell_dict.get('expected')))}" + f"{html.escape(str(cell_dict.get('observed')))}" + f"{marker}" + "" + ) + table = "" + if rows: + table = ( + "" + "" + "" + f"{''.join(rows)}
    KeyDocumentedActiveMatch
    " + ) + return ( + f"

    Diff vs documented default: " + f"{html.escape(status)}

    " + f"{table}" + ) + + +def _entry_card(entry: dict[str, Any] | PolicyEntry) -> str: + title = str(entry.get("title") if isinstance(entry, dict) else entry.title) + category = str(entry.get("category") if isinstance(entry, dict) else entry.category) + summary = str(entry.get("summary") if isinstance(entry, dict) else entry.summary) + sources_data = entry.get("sources", []) if isinstance(entry, dict) else entry.sources + + sources = "".join(_source_pointer(s) for s in sources_data) + return ( + "
    " + f"

    {html.escape(title)} " + f"{html.escape(category)}

    " + f"

    {html.escape(summary)}

    " + "

    Source pointers

    " + f"
      {sources}
    " + "

    Active configuration

    " + f"{_active_block(entry)}" + f"{_diff_block(entry)}" + "
    " + ) + + +def render_policy_page(snapshot: PolicyInventorySnapshot | dict[str, Any]) -> str: + if isinstance(snapshot, PolicyInventorySnapshot): + payload = snapshot_to_dict(snapshot) + elif isinstance(snapshot, dict): + payload = console_redaction.redact_payload(snapshot) + else: + payload = {} + + categories_list = payload.get("categories") or [] + categories = ", ".join(html.escape(str(c)) for c in categories_list) or "none" + entries_list = payload.get("entries") or [] + cards = "".join(_entry_card(e) for e in entries_list) + build_errors = "" + errors_list = payload.get("build_errors") or [] + if errors_list: + items = "".join( + f"
  • {html.escape(str(err))}
  • " for err in errors_list + ) + build_errors = ( + "

    Some guardrails could not be built:" + f"

      {items}
    " + ) + note = str(payload.get("note") or "") + schema_version = payload.get("schema_version") or 1 + return ( + "

    Workflow policy & guardrails

    " + f"

    {html.escape(note)}

    " + f"

    Schema v{schema_version} · " + f"{len(entries_list)} guardrails · categories: {categories}

    " + f"{build_errors}" + f"{cards}" + "

    This page is read-only. It reports enforced policy " + "and never edits or weakens a gate. Secret values are redacted.

    " + )