Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
ab33337a94 |
@@ -64,6 +64,8 @@ status, onboarding checklist state, and the fail-closed error payloads (#635).
|
|||||||
| `/api/prompts` | JSON prompt export with workflow hashes |
|
| `/api/prompts` | JSON prompt export with workflow hashes |
|
||||||
| `/runtime` | MCP runtime health and stale detection (#430) |
|
| `/runtime` | MCP runtime health and stale detection (#430) |
|
||||||
| `/api/runtime` | JSON runtime health export |
|
| `/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) |
|
| `/audit` | Report audit paste + validator preview (#431) |
|
||||||
| `/api/audit` | JSON validator preview (POST `report_text`, optional `task_kind`) |
|
| `/api/audit` | JSON validator preview (POST `report_text`, optional `task_kind`) |
|
||||||
| `/worktrees` | Worktree hygiene dashboard (#432) |
|
| `/worktrees` | Worktree hygiene dashboard (#432) |
|
||||||
@@ -153,6 +155,19 @@ health, workflow/schema SHA-256 hashes, and stale-runtime warnings when the
|
|||||||
checkout is behind merged safety-gate changes. Restart guidance links to #420;
|
checkout is behind merged safety-gate changes. Restart guidance links to #420;
|
||||||
no tokens or MCP restart actions are exposed.
|
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.
|
||||||
|
|
||||||
## Deployment boundary (#435)
|
## Deployment boundary (#435)
|
||||||
|
|
||||||
MVP serves on loopback by default. Binding `0.0.0.0` or `::` is **refused**
|
MVP serves on loopback by default. Binding `0.0.0.0` or `::` is **refused**
|
||||||
|
|||||||
@@ -0,0 +1,222 @@
|
|||||||
|
"""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), [])
|
||||||
|
|
||||||
|
|
||||||
|
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("<form", text.lower())
|
||||||
|
|
||||||
|
def test_policy_html_has_no_secret_shapes(self):
|
||||||
|
text = self.client.get("/policy").text
|
||||||
|
self.assertEqual(console_redaction.scan_for_secrets(text), [])
|
||||||
|
|
||||||
|
def test_api_v1_policy_returns_inventory(self):
|
||||||
|
response = self.client.get("/api/v1/policy")
|
||||||
|
self.assertEqual(response.status_code, 200)
|
||||||
|
data = response.json()
|
||||||
|
self.assertEqual(data["schema_version"], policy_inventory.SCHEMA_VERSION)
|
||||||
|
self.assertTrue(data["read_only"])
|
||||||
|
self.assertEqual(data["entry_count"], len(data["entries"]))
|
||||||
|
self.assertEqual(set(data["categories"]), _EXPECTED_CATEGORIES)
|
||||||
|
|
||||||
|
def test_policy_is_read_only_no_post(self):
|
||||||
|
# AC4 / non-goal: no mutation endpoint.
|
||||||
|
response = self.client.post("/policy")
|
||||||
|
self.assertIn(response.status_code, (404, 405))
|
||||||
|
|
||||||
|
def test_nav_links_policy(self):
|
||||||
|
text = self.client.get("/").text
|
||||||
|
self.assertIn('href="/policy"', 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()
|
||||||
@@ -45,6 +45,8 @@ from webui.worktree_scanner import load_hygiene_snapshot, snapshot_to_dict as wo
|
|||||||
from webui.worktree_views import render_worktrees_page
|
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_health import load_runtime_snapshot, snapshot_to_dict as runtime_snapshot_to_dict
|
||||||
from webui.runtime_views import render_runtime_page
|
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
|
||||||
|
|
||||||
_READ_ONLY_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})
|
_READ_ONLY_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})
|
||||||
_AUDIT_MUTATION_PATHS = frozenset({"/audit", "/api/audit"})
|
_AUDIT_MUTATION_PATHS = frozenset({"/audit", "/api/audit"})
|
||||||
@@ -68,6 +70,7 @@ async def home(_request: Request) -> HTMLResponse:
|
|||||||
"<li><strong>Projects</strong> — registry and onboarding (#427)</li>"
|
"<li><strong>Projects</strong> — registry and onboarding (#427)</li>"
|
||||||
"<li><strong>Prompts</strong> — canonical workflow prompt library (#428)</li>"
|
"<li><strong>Prompts</strong> — canonical workflow prompt library (#428)</li>"
|
||||||
"<li><strong>Runtime</strong> — MCP health and stale-runtime detection (#430)</li>"
|
"<li><strong>Runtime</strong> — MCP health and stale-runtime detection (#430)</li>"
|
||||||
|
"<li><strong>Policy</strong> — workflow guardrail configuration visibility (#646)</li>"
|
||||||
"<li><strong>Audit</strong> — final-report paste and validator preview (#431)</li>"
|
"<li><strong>Audit</strong> — final-report paste and validator preview (#431)</li>"
|
||||||
"<li><strong>Worktrees</strong> — branch hygiene dashboard (#432)</li>"
|
"<li><strong>Worktrees</strong> — branch hygiene dashboard (#432)</li>"
|
||||||
"<li><strong>Leases</strong> — collision and lease visibility (#433)</li>"
|
"<li><strong>Leases</strong> — collision and lease visibility (#433)</li>"
|
||||||
@@ -210,6 +213,17 @@ async def api_runtime(_request: Request) -> JSONResponse:
|
|||||||
return JSONResponse(runtime_snapshot_to_dict(load_runtime_snapshot()))
|
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]:
|
async def _parse_audit_form(request: Request) -> tuple[str, str | None]:
|
||||||
if request.method == "GET":
|
if request.method == "GET":
|
||||||
return "", None
|
return "", None
|
||||||
@@ -415,6 +429,8 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
|
|||||||
Route("/api/prompts", api_prompts, methods=["GET"]),
|
Route("/api/prompts", api_prompts, methods=["GET"]),
|
||||||
Route("/runtime", runtime, methods=["GET"]),
|
Route("/runtime", runtime, methods=["GET"]),
|
||||||
Route("/api/runtime", api_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("/audit", audit, methods=["GET", "POST"]),
|
Route("/audit", audit, methods=["GET", "POST"]),
|
||||||
Route("/api/audit", api_audit, methods=["GET", "POST"]),
|
Route("/api/audit", api_audit, methods=["GET", "POST"]),
|
||||||
Route("/worktrees", worktrees, methods=["GET"]),
|
Route("/worktrees", worktrees, methods=["GET"]),
|
||||||
|
|||||||
@@ -8,6 +8,7 @@ NAV_ITEMS = (
|
|||||||
("/projects", "Projects"),
|
("/projects", "Projects"),
|
||||||
("/prompts", "Prompts"),
|
("/prompts", "Prompts"),
|
||||||
("/runtime", "Runtime"),
|
("/runtime", "Runtime"),
|
||||||
|
("/policy", "Policy"),
|
||||||
("/audit", "Audit"),
|
("/audit", "Audit"),
|
||||||
("/worktrees", "Worktrees"),
|
("/worktrees", "Worktrees"),
|
||||||
("/leases", "Leases"),
|
("/leases", "Leases"),
|
||||||
|
|||||||
@@ -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 <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)
|
||||||
@@ -0,0 +1,104 @@
|
|||||||
|
"""HTML views for the workflow policy and guardrail inventory (#646)."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import html
|
||||||
|
import json
|
||||||
|
|
||||||
|
from webui.policy_inventory import PolicyEntry, PolicyInventorySnapshot
|
||||||
|
|
||||||
|
|
||||||
|
def _source_pointer(source) -> str:
|
||||||
|
path = source.path
|
||||||
|
if source.anchor:
|
||||||
|
path = f"{path}#{source.anchor}"
|
||||||
|
return (
|
||||||
|
f"<li>{html.escape(source.label)} — "
|
||||||
|
f"<code>{html.escape(path)}</code> "
|
||||||
|
f"<span class='muted'>({html.escape(source.kind)})</span></li>"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _active_block(entry: PolicyEntry) -> str:
|
||||||
|
if entry.error:
|
||||||
|
return (
|
||||||
|
"<p class='muted'><strong>Active value unavailable:</strong> "
|
||||||
|
f"{html.escape(entry.error)}</p>"
|
||||||
|
)
|
||||||
|
if not entry.active:
|
||||||
|
return "<p class='muted'>No live projection for this guardrail.</p>"
|
||||||
|
pretty = json.dumps(entry.active, indent=2, sort_keys=True, default=str)
|
||||||
|
return f"<pre class='prompt-text'>{html.escape(pretty)}</pre>"
|
||||||
|
|
||||||
|
|
||||||
|
def _diff_block(entry: PolicyEntry) -> str:
|
||||||
|
if entry.diff is None:
|
||||||
|
if entry.documented_default is None:
|
||||||
|
return "<p class='muted'>Diff vs documented default: not feasible (no declared default).</p>"
|
||||||
|
return "<p class='muted'>Diff vs documented default: unavailable.</p>"
|
||||||
|
status = entry.diff.get("status", "unknown")
|
||||||
|
badge = "badge-claimed" if status == "matches_documented_default" else "badge-blocked"
|
||||||
|
rows = []
|
||||||
|
for key, cell in (entry.diff.get("checked") or {}).items():
|
||||||
|
marker = "✓" if cell.get("matches") else "✗"
|
||||||
|
rows.append(
|
||||||
|
"<tr>"
|
||||||
|
f"<td><code>{html.escape(str(key))}</code></td>"
|
||||||
|
f"<td><code>{html.escape(str(cell.get('expected')))}</code></td>"
|
||||||
|
f"<td><code>{html.escape(str(cell.get('observed')))}</code></td>"
|
||||||
|
f"<td>{marker}</td>"
|
||||||
|
"</tr>"
|
||||||
|
)
|
||||||
|
table = ""
|
||||||
|
if rows:
|
||||||
|
table = (
|
||||||
|
"<table class='detail'><thead><tr>"
|
||||||
|
"<th>Key</th><th>Documented</th><th>Active</th><th>Match</th>"
|
||||||
|
"</tr></thead><tbody>"
|
||||||
|
f"{''.join(rows)}</tbody></table>"
|
||||||
|
)
|
||||||
|
return (
|
||||||
|
f"<p class='meta'>Diff vs documented default: "
|
||||||
|
f"<span class='badge {badge}'>{html.escape(status)}</span></p>"
|
||||||
|
f"{table}"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _entry_card(entry: PolicyEntry) -> str:
|
||||||
|
sources = "".join(_source_pointer(s) for s in entry.sources)
|
||||||
|
return (
|
||||||
|
"<div class='prompt-card'>"
|
||||||
|
f"<h3>{html.escape(entry.title)} "
|
||||||
|
f"<span class='badge'>{html.escape(entry.category)}</span></h3>"
|
||||||
|
f"<p>{html.escape(entry.summary)}</p>"
|
||||||
|
"<p class='meta'><strong>Source pointers</strong></p>"
|
||||||
|
f"<ul>{sources}</ul>"
|
||||||
|
"<p class='meta'><strong>Active configuration</strong></p>"
|
||||||
|
f"{_active_block(entry)}"
|
||||||
|
f"{_diff_block(entry)}"
|
||||||
|
"</div>"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def render_policy_page(snapshot: PolicyInventorySnapshot) -> str:
|
||||||
|
categories = ", ".join(html.escape(c) for c in snapshot.categories) or "none"
|
||||||
|
cards = "".join(_entry_card(e) for e in snapshot.entries)
|
||||||
|
build_errors = ""
|
||||||
|
if snapshot.build_errors:
|
||||||
|
items = "".join(
|
||||||
|
f"<li>{html.escape(err)}</li>" for err in snapshot.build_errors
|
||||||
|
)
|
||||||
|
build_errors = (
|
||||||
|
"<div class='stub'><p><strong>Some guardrails could not be built:"
|
||||||
|
f"</strong></p><ul>{items}</ul></div>"
|
||||||
|
)
|
||||||
|
return (
|
||||||
|
"<h2>Workflow policy & guardrails</h2>"
|
||||||
|
f"<p class='muted'>{html.escape(snapshot.note)}</p>"
|
||||||
|
f"<p class='meta'>Schema v{snapshot.schema_version} · "
|
||||||
|
f"{len(snapshot.entries)} guardrails · categories: {categories}</p>"
|
||||||
|
f"{build_errors}"
|
||||||
|
f"{cards}"
|
||||||
|
"<p class='muted'>This page is read-only. It reports enforced policy "
|
||||||
|
"and never edits or weakens a gate. Secret values are redacted.</p>"
|
||||||
|
)
|
||||||
Reference in New Issue
Block a user