Compare commits

...
Author SHA1 Message Date
jcwalker3andClaude Opus 4.8 479e434f92 feat(webui): console authorization, RBAC, redaction, and audit model (Closes #633)
Phase 1 of the MCP Control Plane Web Console (#631) defines the model that
future gated writes must pass through, and enables none of them.

The read-only MVP (#426-#436) ships with no authentication; protection comes
from network placement alone (#435). That is adequate while every route is a
GET and inadequate the moment Phase 2 wires a write. This lands the authority
first, so no write can later be added without something to check it against.

webui/console_authz.py
  Identity sources (none / local-dev / access-proxy), a four-role matrix
  (viewer, operator, controller, admin), the privileged-action list, and a
  fail-closed authorize(). Every action maps to a task_key in
  task_capability_map, so the console cannot invent an authority the MCP layer
  does not already define. Roles are always server-side configuration, never a
  client assertion. Deny reasons are closed and enumerated; there is no
  implicit allow branch, and even an allow reports execution_enabled=false
  while ACTIVE_PHASE is 1.

webui/console_redaction.py
  One redaction pass for API payloads, rendered HTML, logs, and audit records.
  Reuses gitea_audit.redact as the shared authority rather than forking it,
  then adds console patterns for keychain references, credential assignments,
  PEM private-key blocks, and JWTs. Never raises: an unredactable value
  degrades to the placeholder rather than being emitted raw.

webui/console_audit.py
  Console-side audit records, which gitea_audit cannot supply: it records MCP
  mutations and carries no console actor, identity source, correlation id, or
  retention class, and an authorization denial is not a mutation at all. The
  two are additive and join on correlation.request_id. Records are redacted at
  build time, re-scanned at write time, and dropped rather than persisted if
  they still trip a detector. Retention is per-record; an unknown action is
  retained as privileged rather than standard.

webui/app.py
  Attaches an authorization block to the existing preview and attempt routes
  and records the decision. The terminal outcome is unchanged - gated_actions
  still fails closed for every action - so this cannot loosen anything. Adds
  GET /api/console/security-model publishing the three policies as JSON.

Probe authentication is deliberately declarative in this slice:
probe_auth_required() reports operator intent and no route consults it. The
documentation says so plainly and a regression test pins the not-enforced
status, so wiring it in Phase 2 is a deliberate change rather than a silent
one. An operator who sets the variable believing it protects a probe would be
worse off than one who knows it does not.

Tests: tests/test_webui_console_authz_audit.py - 75 passed, 93 subtests,
covering each acceptance criterion and each test the issue requires
(redaction units, default-deny for unauthenticated write stubs, audit record
creation for a simulated privileged preview).

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-22 14:37:35 -05:00
sysadmin 9eb0f29cef Merge pull request 'feat(webui): worker registry and configuration schema (Closes #798)' (#810) from feat/issue-798-worker-registry-schema into master 2026-07-22 06:48:29 -05:00
sysadmin 0b29404031 Merge pull request 'docs(webui): MCP Control Plane Web Console architecture ADR (Closes #632)' (#796) from docs/issue-632-web-console-architecture into master 2026-07-22 06:22:11 -05:00
jcwalker3andClaude Opus 4.8 5463f58933 feat(webui): worker registry and configuration schema (Closes #798)
Add the declarative worker registry that epic #797 makes the source of
truth for the scheduled multi-LLM worker fleet.

Providers and configured workers are modelled as separate entities so a
provider can be listed with no worker configured, and so provider facts
are not copied into every worker record. A worker records provider,
model, project, role, namespace, profile, workflow, schedule, timeout,
enabled state, and scheduler metadata.

Validation fails closed: unknown fields are refused rather than ignored,
so a typo cannot silently disable a timeout; a worker naming an
undeclared provider is rejected; worker ids, provider ids, and
LaunchAgent labels must be unique.

Persistence is atomic (temp file in the same directory, fsync, replace).
Every superseded document is retained as a numbered revision, and
rollback republishes a chosen revision as a new head, so history stays
append-only and a rollback is itself reversible.

The credential-rejection guard is extracted to webui/registry_safety.py
so both registries share one implementation instead of two copies of a
security check; project_registry.py keeps identical behaviour.

Scope: data model, validation, persistence only. No routes, scheduler,
process control, or provider probing - those are #799/#800/#804/#805.
The workers array ships empty because populating it is #808.

Tests: tests/test_webui_worker_registry.py, 44 cases.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-22 06:10:34 -05:00
sysadmin 5032965e3a Merge pull request 'feat: make master-parity live-remote aware so a stale daemon fails closed (Closes #610)' (#788) from feat/issue-610-live-remote-parity into master 2026-07-22 05:51:56 -05:00
jcwalker3andGrok 4.5 57a52b1a99 fix(parity): hermetic live-remote master reads under pytest (Closes #610)
Remediate PR #788 review F1/F2: suite-wide hermetic mode prevents
git ls-remote from running in tests so feature worktrees no longer flip
legacy runtime-context assertions to live_stale, and unit tests stay offline.
Module flag survives patch.dict(clear=True); env override still wins.

Co-Authored-By: Grok 4.5 <[email protected]>
2026-07-22 05:40:15 -05:00
jcwalker3 344dc41ce2 Merge branch 'master' into feat/issue-610-live-remote-parity 2026-07-22 04:30:44 -05:00
jcwalker3 0f19773076 Merge branch 'master' into feat/issue-610-live-remote-parity 2026-07-21 20:48:54 -05:00
sysadminandClaude Opus 4.8 324b4b3e93 feat: make master-parity live-remote aware so a stale daemon fails closed (Closes #610)
Master-parity previously compared only the daemon's startup commit against the
local on-disk HEAD. When the checkout was not pulled, parity reported green even
though the live remote master had advanced, so a stale daemon could claim a
mutation-safe result while running outdated capability gates (observed during
PR #592 recovery, where the resolver correctly required restart but parity said
in_parity=true).

Changes:
- master_parity_gate.assess_master_parity() gains an optional live_remote_head
  and reports the three commits distinctly (daemon_start_head, local_head,
  live_remote_head) plus live_known / live_stale / mutation_safe. A result is
  mutation_safe only when daemon, local checkout, and live remote all agree.
- parity_block_reasons() now blocks mutations on live-staleness too; read-only
  operations remain unblocked (non-goal: never block diagnostics offline).
- New parity_resolver_disagreement(): typed fail-closed blocker naming the
  capability resolver as authoritative when it requires restart but parity
  looks locally green.
- New read_remote_master_head(): best-effort `git ls-remote` for the live
  target, cached with a 60s TTL (bounded offline latency, no network probe per
  gate call); env override GITEA_TEST_LIVE_REMOTE_HEAD keeps tests hermetic.
- Server: _current_master_parity() reads the live remote head;
  gitea_assess_master_parity and runtime_context surface the distinguished
  SHAs, mutation_safe, and resolver-authoritative guidance.

Tests: 13 new cases (live-remote parity, live-stale blocking + typed blocker,
remote-head reader + TTL cache, server wiring). Full suite: 2423 passed; the 8
remaining failures (test_config TestAuthIntegration, test_credentials
TestGetCredentials) are baseline-proven keychain/env failures identical on the
unmodified base.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-09 18:52:59 -04:00
16 changed files with 3859 additions and 60 deletions
+295
View File
@@ -0,0 +1,295 @@
# Web console authorization, RBAC, redaction, and audit model (#633)
**Phase 1. Read-only. This document defines the model that future console
writes must pass through; it enables none of them.**
The MVP deployment boundary ([`webui-deployment.md`](webui-deployment.md), #435)
documents internal-only serving and states plainly that MVP authentication is
*none* — protection comes from network placement. That is adequate while every
route is a GET, and inadequate the moment a gated write ships. This document
and the three modules it describes land **before** any write exists, so no
Phase 2 action can be added without an authority to check it against.
| Concern | Module |
|---------|--------|
| Identity, roles, authorization decision | `webui/console_authz.py` |
| Secret redaction for every surface | `webui/console_redaction.py` |
| Audit event schema, retention, sink | `webui/console_audit.py` |
| Machine-readable publication | `GET /api/console/security-model` |
Two invariants hold everywhere and are non-negotiable for every child of #631:
1. **No secrets reach the browser.** Credentials are resolved server-side and
redacted before any payload, page, log line, or audit record leaves.
2. **No ungated mutations.** Authorization is necessary but never sufficient;
execution stays disabled until the Phase 2 framework ships.
## Identity sources
The console performs *authorization*. Authentication is delegated, because a
console that mints its own sessions is a credential store, and this one must
not be.
| Source | Mode value | Authenticated | Shared host | Phase |
|--------|-----------|---------------|-------------|-------|
| None | `none` (default) | No — anonymous, capped at `viewer` | No | 1 |
| Local dev | `local-dev` / `local_dev` | Yes, **asserted not verified** | No | 1 |
| Access proxy | `access-proxy` / `access_proxy` | Yes, asserted by trusted proxy | Yes | 2 |
Selected by `WEBUI_AUTH_MODE`. An unrecognised value falls back to `none`
rather than erroring open.
**Access-proxy mode** reads the subject from the
`Cf-Access-Authenticated-User-Email` header, set by Cloudflare Access, WARP, or
an equivalent org portal that terminates authentication in front of the
console. If the header is absent the request did not traverse the proxy, so the
principal degrades to anonymous — it is never trusted by default.
The **role is always server-side configuration**, never a client assertion. It
comes from `WEBUI_ROLE_MAP`, a JSON object of subject → role:
```json
{"[email protected]": "operator", "[email protected]": "controller"}
```
An unmapped subject gets `viewer`. Malformed JSON yields an empty map, so
everyone gets `viewer` — a parse failure loses authority rather than granting
it.
Full SSO is explicitly a non-goal of this issue.
## Role matrix
Four roles, ordered least to most authority. Each role inherits every lower
role's actions; the table states the *minimum* rank required.
| Role | Authority |
|------|-----------|
| `viewer` | Read every console view. No write, ever, in any phase. |
| `operator` | Viewer, plus author-class work: claim, comment, open a PR. |
| `controller` | Operator, plus reviewer/merger-class decisions on a PR. |
| `admin` | Controller, plus destructive and policy-editing actions. |
`viewer` holds the empty write set by construction, and a test asserts it stays
empty.
## Privileged actions
Every console action maps to a `task_key` in `task_capability_map.py`, the same
single source of truth `gitea_resolve_task_capability` and the MCP tool gates
use. The console therefore cannot invent an authority the MCP layer does not
already define, and a regression test asserts each mapping matches.
| Action | Minimum role | Class | MCP permission | Confirm | Dual control | Break-glass | Phase |
|--------|--------------|-------|----------------|---------|--------------|-------------|-------|
| `claim_issue` | operator | gated_write | `gitea.issue.comment` | Yes | No | No | 2 |
| `comment_issue` | operator | gated_write | `gitea.issue.comment` | Yes | No | No | 2 |
| `create_issue` | operator | gated_write | `gitea.issue.create` | Yes | No | No | 2 |
| `comment_pr` | operator | gated_write | `gitea.pr.comment` | Yes | No | No | 2 |
| `create_pr` | operator | gated_write | `gitea.pr.create` | Yes | No | No | 2 |
| `review_pr` | controller | privileged | `gitea.pr.review` | Yes | No | No | 3 |
| `close_pr` | controller | privileged | `gitea.pr.close` | Yes | No | No | 3 |
| `merge_pr` | controller | privileged | `gitea.pr.merge` | Yes | **Yes** | **Yes** | 3 |
| `delete_branch` | admin | destructive | `gitea.branch.delete` | Yes | **Yes** | **Yes** | 3 |
**Dual control** means the acting principal may not be the sole authority: a
second distinct principal must confirm. **Break-glass** means the action is
expected to be unavailable in normal operation and its use is retained for two
years. Both are declared here and enforced by the Phase 2 framework; Phase 1
records the requirement on every decision so the framework cannot ship without
honouring it.
`delete_branch` is admin-only rather than controller because it is the one
irreversible action in the set.
### Authorization decision
`authorize(action_id, principal, for_execution=False)` returns a decision
record and **denies by default**. The deny reasons are closed and enumerated:
| Reason code | Meaning |
|-------------|---------|
| `unknown_action` | No such console action is registered. |
| `unauthenticated` | The principal is anonymous. |
| `unknown_role` | The role is not in the matrix. |
| `insufficient_role` | The role ranks below the action's minimum. |
| `phase_not_active` | Execution requested for an action whose phase is not open. |
| `allowed_preview_only` | Authorized — preview only, execution still disabled. |
There is no implicit allow branch. Even the allow result reports
`execution_enabled: false` while the console is in Phase 1, so no caller can
read an allow as permission to mutate.
## Secret redaction
One pass applies to **API payloads, rendered HTML, server logs, and audit
records** — the four surfaces where a credential could escape.
Redaction reuses `gitea_audit.redact` rather than forking it: that remains the
authority for secret-looking dict keys, `Authorization` material, and raw URLs.
The console layer then applies its own patterns:
Each rule below matches an *assignment form*: the named key, followed by `=` or
`:`, followed by the value. The keys are listed bare rather than spelled out as
complete assignments, because this document is itself scanned by
`scan_for_secrets` — writing the examples in full assignment form would make the
documentation trip the very detectors it documents.
| Rule | Catches (as an assignment) |
|------|----------------------------|
| `credential_assignment` | `token`, `password`, `passwd`, `secret`, `api_key`, `access_key`, `client_secret`, `private_key` |
| `credential_env_assignment` | `GITEA_TOKEN`, `GITEA_PASS`, `GITEA_PASSWORD` and suffixed variants |
| `keychain_reference` | `keychain:` entry references |
| `keychain_command` | macOS `security` keychain lookups (`find-generic-password`, `find-internet-password`) |
| `private_key_block` | PEM `BEGIN ... PRIVATE KEY` blocks |
| `json_web_token` | Three-segment `eyJ...` JWTs |
| `bearer_credential` | `Bearer` / `Basic` credentials |
Assignments keep the key and replace only the value, so an operator can still
see *what* was removed. Two behaviours are deliberate:
- **Fail closed.** A value that cannot be redacted becomes `[REDACTED]`
outright rather than being emitted raw. Redaction never raises.
- **Redact before persist.** `console_audit.build_event` redacts before
serialization, and `write_event` re-scans and **drops** any record that still
trips a detector. An unredacted record is never durable.
`scan_for_secrets` is the assertion helper: it returns the detector names still
matching a payload, and already-redacted hits are not findings. Tests use it to
prove the published policy, the security-model endpoint, and this document
itself carry no secret material.
## Audit event schema
`gitea_audit` records MCP-side *mutations* — which profile and Gitea user
performed which tool call. It has no console actor, no identity source, no
correlation identifier, and no retention class, and an authorization **denial**
is not a mutation, so it would never appear there at all. The console record is
additive, not a replacement: a Phase 2 action emits both, joined on
`correlation.request_id`.
Required fields, all asserted by tests so an edit cannot quietly drop one:
| Field | Content |
|-------|---------|
| `schema_version` | Currently `1`. |
| `event_id` | Unique per record. |
| `timestamp` | Timezone-aware ISO-8601, UTC. |
| `actor` | `subject`, `role`, `identity_source`, `authenticated`. |
| `action` | Console action id. |
| `action_class` | `gated_write`, `privileged`, `destructive`, or `unknown`. |
| `target` | `{kind, ref}`, e.g. `{"kind": "pr", "ref": "#123"}`. |
| `result` | `allowed`, `denied`, `previewed`, `failed`, `succeeded`. |
| `reason_code` | The authorization reason code above. |
| `correlation` | `request_id`, `session_id`, `mcp_task`, `mcp_permission`. |
| `retention` | `class`, `days`, `expires_at`. |
| `redacted` | Always `true`; records are redacted at build time. |
An unrecognised `result` degrades to `failed` rather than being stored
verbatim.
The sink is an append-only JSON Lines file named by
`WEBUI_CONSOLE_AUDIT_LOG`. It is **off by default**: with the variable unset,
events are still built — so callers and tests exercise the schema — but nothing
is written. Auditing never raises; a failed write returns `False` rather than
breaking the request it describes.
## Retention
| Class | Applies to | Default |
|-------|-----------|---------|
| `standard` | Routine gated writes | 90 days |
| `privileged` | `review_pr`, `close_pr`, and any unclassifiable action | 365 days |
| `break_glass` | `merge_pr`, `delete_branch` | 730 days |
Each record carries its own class, day count, and computed `expires_at`, so
retention is auditable per record rather than inferred from file age. An
**unknown action is retained as privileged, not standard** — for a safety
control the conservative direction is to keep the record longer.
Nothing in this module updates or deletes. Expiry is enforced by an
operator-run policy against `expires_at`, never by the console silently
rewriting its own history.
## Phase 2 integration
Phase 2 opens gated writes. It must reuse this model rather than introduce a
second one. The integration points are already wired and observable:
- **`GET /api/actions/{action_id}/preview`** attaches an `authorization` block
to the existing preview payload and records a `previewed` audit event.
- **`POST /api/actions/{action_id}/attempt`** attaches the same block and
records a `denied` event. The terminal outcome is unchanged — the MVP
registry in `webui/gated_actions.py` still fails closed for every action — so
Phase 1 cannot loosen anything. Phase 2 enforces on this same decision
instead of adding a parallel check.
- **`GET /api/console/security-model`** publishes the RBAC matrix, redaction
policy, and audit policy as JSON for operators and tests.
To open Phase 2, a child issue must: raise `ACTIVE_PHASE`, implement the
confirmation and dual-control flow the matrix already declares, emit a
`succeeded` or `failed` record alongside the `gitea_audit` mutation record, and
keep `viewer` unable to reach any of it. Turning on execution without the
confirmation flow contradicts a declared requirement and is a review failure,
not a shortcut.
## Local-dev mode
`WEBUI_AUTH_MODE=local-dev` reads the principal straight from the environment:
| Variable | Purpose |
|----------|---------|
| `WEBUI_DEV_SUBJECT` | Subject string; absent ⇒ anonymous |
| `WEBUI_DEV_ROLE` | One of `viewer`, `operator`, `controller`, `admin`; unrecognised ⇒ `viewer` |
**INSECURE — this mode is for loopback development only.** The subject and role
are *asserted by the developer running the process and verified by nothing*.
Anyone able to set an environment variable on the host is an `admin`, and
anyone able to reach the port inherits that principal. It provides no
authentication whatsoever; it exists so Phase 2 authorization paths can be
exercised without standing up a proxy.
Never enable local-dev mode on a non-loopback bind. Combining it with
`WEBUI_ALLOW_PUBLIC_BIND=1` or `WEBUI_ALLOW_REMOTE_BIND=1` publishes an
unauthenticated admin console.
For anything beyond a laptop use `access-proxy` mode behind Cloudflare Access,
WARP, or a VPN, as [`webui-deployment.md`](webui-deployment.md) requires.
### Probe authentication
`WEBUI_REQUIRE_PROBE_AUTH=1` declares that non-public probes should require an
authenticated principal. It is **opt-in**: the default is off so the MVP
`/health` contract is unchanged.
**This flag is declarative in Phase 1 and enforces nothing today.**
`console_authz.probe_auth_required()` reports the operator's intent, and no
route consults it — setting the variable does not currently change the
behaviour of `/health` or any other endpoint. It is published here so the Phase
2 action framework has a declared policy to honour rather than inventing a
second one, exactly as `ACTIVE_PHASE` gates execution while the matrix is
already declared. A regression test pins this "declared, not enforced" status,
so wiring it later is a deliberate change rather than a silent one.
Until Phase 2 wires it, probe protection rests on network placement alone, as
[`webui-deployment.md`](webui-deployment.md) (#435) states.
## Environment variables
| Variable | Default | Purpose |
|----------|---------|---------|
| `WEBUI_AUTH_MODE` | `none` | Identity source selection |
| `WEBUI_DEV_SUBJECT` | unset | Local-dev subject (insecure) |
| `WEBUI_DEV_ROLE` | `viewer` | Local-dev role (insecure) |
| `WEBUI_ROLE_MAP` | unset | JSON subject → role map |
| `WEBUI_REQUIRE_PROBE_AUTH` | unset | Require auth for non-public probes |
| `WEBUI_CONSOLE_AUDIT_LOG` | unset | Append-only audit sink path |
All are read server-side only. None is ever rendered into a page or returned by
an API.
## Non-goals
- No full SSO product; authentication stays delegated to the proxy.
- No browser-initiated merges or approvals in any phase covered here.
- No tokens in the frontend, in browser storage, or in committed config.
+4 -1
View File
@@ -7,7 +7,10 @@ only.
## MVP deployment model
- **Default bind:** `127.0.0.1:8765` (`WEBUI_HOST` / `WEBUI_PORT`)
- **Authentication:** none in MVP — protection comes from network placement
- **Authentication:** none in MVP — protection comes from network placement.
The authorization, RBAC, redaction, and audit model that future gated writes
must pass through is defined in
[`webui-authz-audit.md`](webui-authz-audit.md) (#633).
- **Mutations:** read-only routes; gated write actions remain disabled (#434)
- **Secrets:** resolved server-side via `gitea_auth` / `GITEA_MCP_CONFIG`; never
embedded in HTML, JavaScript, or browser storage
+59 -4
View File
@@ -12536,10 +12536,39 @@ def _try_auto_switch_for_operation(op: str, host: str | None = None) -> bool:
return False
def _git_default_remote_name(root: str) -> str:
"""First configured git remote name for *root*, defaulting to 'origin'.
Used to resolve the live remote master target for parity (#610). Best
effort: any failure falls back to 'origin' so callers never raise.
"""
try:
res = subprocess.run(
["git", "-C", root, "remote"],
capture_output=True, text=True, check=False,
)
except Exception:
return "origin"
if res.returncode != 0:
return "origin"
names = [n.strip() for n in (res.stdout or "").splitlines() if n.strip()]
return names[0] if names else "origin"
def _current_master_parity() -> dict:
"""Assess this process's code against the on-disk master HEAD (#420)."""
"""Assess this process's code against local and live remote master (#420/#610).
Compares the daemon's startup commit, the on-disk checkout HEAD, and the
live remote master target. A stale daemon relative to live master fails
closed for mutations even when the local checkout HEAD still matches the
startup commit. The live-remote read is best effort: an unresolved live
head leaves read-only diagnostics unblocked but is never mutation-safe.
"""
current_head = master_parity_gate.read_git_head(PROJECT_ROOT)
return master_parity_gate.assess_master_parity(_STARTUP_PARITY, current_head)
live_head = master_parity_gate.read_remote_master_head(
PROJECT_ROOT, remote=_git_default_remote_name(PROJECT_ROOT))
return master_parity_gate.assess_master_parity(
_STARTUP_PARITY, current_head, live_remote_head=live_head)
def _current_runtime_mode_report(refresh: bool = False) -> dict:
@@ -16480,10 +16509,29 @@ def gitea_get_runtime_context(
"restart_required": parity["restart_required"],
"startup_head": parity["startup_head"],
"current_head": parity["current_head"],
# #610 distinguished mutation-safety signals:
"daemon_start_head": parity["daemon_start_head"],
"local_head": parity["local_head"],
"live_remote_head": parity["live_remote_head"],
"live_known": parity["live_known"],
"live_stale": parity["live_stale"],
"mutation_safe": parity["mutation_safe"],
"summary": master_parity_gate.format_parity(parity),
"mutation_gate_enforced": not master_parity_gate.gate_disabled(),
# #610: the capability resolver is authoritative for mutation safety;
# local parity alone must never authorize a mutation.
"resolver_authoritative_for_mutation_safety": True,
}
if parity["stale"] and not master_parity_gate.gate_disabled():
if parity["restart_required"] and not master_parity_gate.gate_disabled():
if parity["live_stale"]:
safe_next_action = (
"Daemon is stale relative to LIVE remote master "
f"(started {parity['startup_head'][:12] if parity['startup_head'] else 'unknown'}, "
f"live master {parity['live_remote_head'][:12] if parity['live_remote_head'] else 'unknown'}); "
"restart/reconnect the Gitea MCP server before mutating. The "
"capability resolver is authoritative for mutation safety."
)
else:
safe_next_action = (
"Server code is stale relative to master; restart the Gitea MCP "
"server to load current capability gates before mutating. "
@@ -16537,6 +16585,13 @@ def gitea_assess_master_parity(
"determinable": parity["determinable"],
"startup_head": parity["startup_head"],
"current_head": parity["current_head"],
# #610 distinguished mutation-safety signals:
"daemon_start_head": parity["daemon_start_head"],
"local_head": parity["local_head"],
"live_remote_head": parity["live_remote_head"],
"live_known": parity["live_known"],
"live_stale": parity["live_stale"],
"mutation_safe": parity["mutation_safe"],
"mutation_gate_enforced": enforced,
"summary": master_parity_gate.format_parity(parity),
"reasons": parity["reasons"],
@@ -16553,7 +16608,7 @@ def gitea_assess_master_parity(
source=canonical_source,
),
}
if parity["stale"] and enforced:
if parity["restart_required"] and enforced:
out["report"] = master_parity_gate.parity_report(parity)
return out
+213 -17
View File
@@ -24,12 +24,50 @@ from __future__ import annotations
import os
import subprocess
import time
# Live-remote head cache: the parity gate runs on every mutation and every
# runtime-context read, so the ``git ls-remote`` result is cached briefly to
# avoid a network round-trip per call (#610). Keyed by (root, remote, branch).
_REMOTE_HEAD_CACHE: dict[tuple[str, str, str], tuple[float, str | None]] = {}
_REMOTE_HEAD_TTL = 60.0
# When True, ``read_remote_master_head`` never performs ``git ls-remote`` unless
# ``GITEA_TEST_LIVE_REMOTE_HEAD`` is set. Conftest enables this suite-wide so
# feature worktrees (whose HEAD differs from live master) cannot flip legacy
# runtime-context assertions to live_stale, and so unit tests never depend on
# a live network (PR #788 F1/F2 / issue #610). Module-level (not env-only) so
# ``patch.dict(os.environ, …, clear=True)`` cannot re-enable the probe.
_HERMETIC_TEST_MODE: bool = False
def _clear_remote_head_cache() -> None:
"""Reset the live-remote head cache (test isolation / forced refresh)."""
_REMOTE_HEAD_CACHE.clear()
def set_hermetic_test_mode(enabled: bool) -> None:
"""Enable or disable suite-wide hermetic live-remote reads (tests only)."""
global _HERMETIC_TEST_MODE
_HERMETIC_TEST_MODE = bool(enabled)
_clear_remote_head_cache()
def hermetic_test_mode() -> bool:
"""Return whether hermetic live-remote reads are active."""
return bool(_HERMETIC_TEST_MODE)
# Environment escape hatches (ops + tests):
# GITEA_MCP_DISABLE_PARITY_GATE -> disable enforcement entirely (fail open).
# GITEA_TEST_CURRENT_HEAD -> force the "current" HEAD read, for tests.
ENV_DISABLE = "GITEA_MCP_DISABLE_PARITY_GATE"
ENV_TEST_CURRENT_HEAD = "GITEA_TEST_CURRENT_HEAD"
# GITEA_TEST_LIVE_REMOTE_HEAD -> force the live remote master read, for tests.
ENV_TEST_LIVE_REMOTE_HEAD = "GITEA_TEST_LIVE_REMOTE_HEAD"
# GITEA_TEST_ALLOW_LIVE_REMOTE_PROBE -> opt a single test into a real ls-remote
# even when hermetic mode is on (rare; prefer ENV_TEST_LIVE_REMOTE_HEAD).
ENV_TEST_ALLOW_LIVE_REMOTE_PROBE = "GITEA_TEST_ALLOW_LIVE_REMOTE_PROBE"
def read_git_head(root: str) -> str | None:
@@ -58,6 +96,75 @@ def read_git_head(root: str) -> str | None:
return (res.stdout or "").strip() or None
def read_remote_master_head(
root: str,
remote: str = "origin",
branch: str = "master",
ttl: float = _REMOTE_HEAD_TTL,
) -> str | None:
"""Return the live remote ``branch`` commit SHA, or ``None`` (#610).
Resolves the *live* target commit via ``git ls-remote`` so parity can tell
a daemon that is behind the live remote master apart from one whose local
checkout simply hasn't been pulled. ``None`` means the live head could not
be resolved (offline, no such remote, git unavailable, error) -- callers
must treat unknown live state as *not mutation-safe* while never blocking
read-only diagnostics. A ``GITEA_TEST_LIVE_REMOTE_HEAD`` override takes
precedence so the wiring can be exercised deterministically and offline.
The result is cached for *ttl* seconds per (root, remote, branch) so the
gate does not run a network probe on every mutation/read (``ttl=0`` forces
a live probe). Both hits and ``None`` misses are cached to bound offline
latency; the env override bypasses the cache and the subprocess entirely.
Under suite hermetic mode (``set_hermetic_test_mode(True)``, set by
conftest) a missing override returns ``None`` without network I/O so
feature-worktree test runs cannot observe live_stale against real master
(PR #788 F1) and unit tests stay offline (F2). Opt out with an explicit
``GITEA_TEST_LIVE_REMOTE_HEAD`` pin or ``GITEA_TEST_ALLOW_LIVE_REMOTE_PROBE``.
"""
forced = os.environ.get(ENV_TEST_LIVE_REMOTE_HEAD)
if forced is not None:
return forced.strip() or None
if _HERMETIC_TEST_MODE and not (
os.environ.get(ENV_TEST_ALLOW_LIVE_REMOTE_PROBE) or ""
).strip():
# Hermetic default: live head unknown. live_stale stays False;
# mutation_safe is False when live is unknown (documented #610 note).
return None
# Defense in depth: even without the module flag, never probe while pytest
# is running unless the test opted into a real probe or set an override.
if (os.environ.get("PYTEST_CURRENT_TEST") or "").strip() and not (
os.environ.get(ENV_TEST_ALLOW_LIVE_REMOTE_PROBE) or ""
).strip():
return None
if not root:
return None
key = (root, remote, branch)
now = time.monotonic()
if ttl > 0:
cached = _REMOTE_HEAD_CACHE.get(key)
if cached is not None and (now - cached[0]) < ttl:
return cached[1]
sha: str | None = None
try:
res = subprocess.run(
["git", "-C", root, "ls-remote", remote, f"refs/heads/{branch}"],
capture_output=True,
text=True,
check=False,
timeout=5,
)
if res.returncode == 0:
lines = (res.stdout or "").strip().splitlines()
if lines:
sha = lines[0].split("\t", 1)[0].split()[0].strip() or None
except Exception:
sha = None
_REMOTE_HEAD_CACHE[key] = (now, sha)
return sha
def capture_startup_parity(root: str, head: str | None = None) -> dict:
"""Capture the process source-tree baseline once at server startup.
@@ -72,18 +179,38 @@ def _short(sha: str | None) -> str:
return sha[:12] if sha else "unknown"
def assess_master_parity(startup: dict | None, current_head: str | None) -> dict:
def assess_master_parity(
startup: dict | None,
current_head: str | None,
live_remote_head: str | None = None,
) -> dict:
"""Compare the startup baseline against the current on-disk ``HEAD``.
Pure: both HEADs are supplied by the caller. Returns a structured result:
Pure: all HEADs are supplied by the caller. Returns a structured result:
- ``in_parity`` -- server code matches the on-disk master (or parity
could not be determined, which is not treated as stale).
- ``stale`` -- the on-disk master has definitively advanced past the
running process.
- ``restart_required`` -- alias of ``stale``; the recovery action.
- ``determinable`` -- whether both HEADs were known well enough to compare.
- ``restart_required`` -- ``stale`` or ``live_stale``; the recovery action.
- ``determinable`` -- whether both local HEADs were known well enough to
compare.
- ``startup_head`` / ``current_head`` / ``reasons``.
#610 adds live-remote awareness so a daemon that is stale relative to the
*live* remote master cannot report a mutation-safe result even when the
local checkout HEAD still matches the daemon's startup commit:
- ``daemon_start_head`` -- the commit the running process started at
(alias of ``startup_head``, named for clarity in reports).
- ``local_head`` -- the on-disk checkout HEAD (alias of ``current_head``).
- ``live_remote_head`` -- the live remote target commit, or ``None`` when it
could not be fetched.
- ``live_known`` -- whether the live remote target was resolved.
- ``live_stale`` -- the live remote master has advanced past the running
process (daemon is behind live master) even if local parity is green.
- ``mutation_safe`` -- the daemon code, local checkout, and live remote
target all agree; the only state in which a mutation may rely on parity.
"""
startup_head = (startup or {}).get("startup_head")
reasons: list[str] = []
@@ -91,32 +218,56 @@ def assess_master_parity(startup: dict | None, current_head: str | None) -> dict
if startup_head is None:
reasons.append(
"startup commit was not captured; code parity cannot be enforced")
return _result(True, False, False, startup_head, current_head, reasons)
return _result(True, False, False, startup_head, current_head,
live_remote_head, False, reasons)
if current_head is None:
reasons.append(
"current workspace HEAD could not be read; code parity cannot be "
"enforced")
return _result(True, False, False, startup_head, current_head, reasons)
if startup_head == current_head:
return _result(True, False, True, startup_head, current_head, reasons)
return _result(True, False, False, startup_head, current_head,
live_remote_head, False, reasons)
local_in_parity = startup_head == current_head
local_stale = not local_in_parity
if local_stale:
reasons.append(
f"MCP server started at commit {_short(startup_head)} but the workspace "
f"master is now {_short(current_head)}; restart the server to load the "
f"current capability gates")
return _result(False, True, True, startup_head, current_head, reasons)
f"MCP server started at commit {_short(startup_head)} but the "
f"workspace master is now {_short(current_head)}; restart the "
f"server to load the current capability gates")
live_known = live_remote_head is not None
live_stale = live_known and live_remote_head != startup_head
if live_stale:
reasons.append(
f"live remote master is {_short(live_remote_head)} but the MCP "
f"server started at {_short(startup_head)}; the daemon is stale "
f"relative to live master -- restart/reconnect before mutating")
return _result(
local_in_parity, local_stale, True, startup_head, current_head,
live_remote_head, live_stale, reasons)
def _result(in_parity, stale, determinable, startup_head, current_head, reasons):
def _result(in_parity, stale, determinable, startup_head, current_head,
live_remote_head, live_stale, reasons):
live_known = live_remote_head is not None
mutation_safe = (
determinable and in_parity and live_known and not live_stale)
return {
"in_parity": in_parity,
"stale": stale,
"restart_required": stale,
"restart_required": stale or live_stale,
"determinable": determinable,
"startup_head": startup_head,
"current_head": current_head,
# #610 distinguished signals:
"daemon_start_head": startup_head,
"local_head": current_head,
"live_remote_head": live_remote_head,
"live_known": live_known,
"live_stale": live_stale,
"mutation_safe": mutation_safe,
"reasons": list(reasons),
}
@@ -130,11 +281,13 @@ def parity_block_reasons(assessment: dict) -> list[str]:
"""Block reasons for a mutation gate (empty when the mutation may proceed).
A disabled gate or an in-parity / non-determinable assessment yields no
reasons; only a definitively stale server blocks.
reasons. A definitively stale server blocks, and (#610) a daemon that is
stale relative to the *live* remote master blocks even when the local
checkout HEAD still matches the daemon's startup commit.
"""
if gate_disabled():
return []
if assessment.get("stale"):
if assessment.get("stale") or assessment.get("live_stale"):
return list(assessment.get("reasons") or
["server code is stale relative to master (fail closed)"])
return []
@@ -147,6 +300,10 @@ def parity_report(assessment: dict) -> dict:
"restart_required": True,
"startup_head": assessment.get("startup_head"),
"current_head": assessment.get("current_head"),
# #610: name the live remote target so the report distinguishes a
# local-code stale from a daemon-behind-live-master stale.
"live_remote_head": assessment.get("live_remote_head"),
"live_stale": bool(assessment.get("live_stale")),
"reasons": list(assessment.get("reasons") or []),
"recovery": [
"The running MCP server is executing code older than the current "
@@ -157,6 +314,45 @@ def parity_report(assessment: dict) -> dict:
}
def parity_resolver_disagreement(
assessment: dict,
resolver_restart_required: bool,
) -> dict | None:
"""Typed blocker when the resolver requires restart but parity looks green.
The capability resolver (``gitea_resolve_task_capability``) detects stale
runtime authoritatively for mutation safety (#610). When it requires a
restart, local-only parity must never override it: this returns a typed,
fail-closed blocker that names the resolver as authoritative. Returns
``None`` when the resolver does not require a restart.
"""
if not resolver_restart_required:
return None
parity_optimistic = bool(assessment.get("in_parity")) and not (
assessment.get("stale") or assessment.get("live_stale"))
return {
"kind": "parity_resolver_disagreement",
"restart_required": True,
"resolver_authoritative": True,
"parity_optimistic": parity_optimistic,
"daemon_start_head": assessment.get("daemon_start_head"),
"local_head": assessment.get("local_head"),
"live_remote_head": assessment.get("live_remote_head"),
"reasons": [
"The capability resolver requires a restart/reconnect (stale "
"runtime) but master-parity reported local code as in-parity. "
"The resolver is authoritative for mutation safety; do not mutate "
"on local parity alone. Restart/reconnect the Gitea MCP server "
"and re-verify before mutating.",
],
"recovery": [
"Trust the resolver: treat this session as stale.",
"Restart or /mcp reconnect the Gitea MCP namespace so it reloads "
"current master and live target state, then re-run preflight.",
],
}
def format_parity(assessment: dict) -> str:
"""One-line human summary for logs / runtime context."""
if assessment.get("stale"):
+29
View File
@@ -167,6 +167,35 @@ def _reset_mutation_authority(monkeypatch):
import pytest
@pytest.fixture(autouse=True)
def _hermetic_live_remote_master_head():
"""#610 / PR #788 F1/F2: keep live-remote parity reads offline in tests.
``read_remote_master_head`` would otherwise ``git ls-remote`` whenever
``GITEA_TEST_LIVE_REMOTE_HEAD`` is unset. Feature worktrees under
``branches/`` always differ from live master, so legacy suites that assert
runtime-context ``safe_next_action`` flip to live_stale. Module-level
hermetic mode survives ``patch.dict(os.environ, …, clear=True)``.
Tests that exercise the real probe path call
``master_parity_gate.set_hermetic_test_mode(False)`` and/or set
``GITEA_TEST_ALLOW_LIVE_REMOTE_PROBE``.
"""
try:
import master_parity_gate as _mpg
_mpg.set_hermetic_test_mode(True)
except Exception:
_mpg = None
try:
yield
finally:
if _mpg is not None:
try:
_mpg.set_hermetic_test_mode(False)
except Exception:
pass
@pytest.fixture(autouse=True)
def _deterministic_workspace_remotes():
try:
+269
View File
@@ -78,6 +78,95 @@ class TestBlockReasonsAndReport(unittest.TestCase):
self.assertTrue(report["recovery"])
class TestLiveRemoteParity(unittest.TestCase):
"""#610: parity must account for the live remote master, not just local.
The daemon can be stale relative to the live remote target while the local
checkout HEAD still matches the daemon's startup commit, so local parity
reports green even though a mutation would run against outdated code.
"""
SHA_C = "c" * 40
def test_distinguishes_three_shas(self):
res = mp.assess_master_parity(
{"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
self.assertEqual(res["daemon_start_head"], SHA_A)
self.assertEqual(res["local_head"], SHA_A)
self.assertEqual(res["live_remote_head"], SHA_B)
def test_mutation_safe_only_when_all_three_match(self):
res = mp.assess_master_parity(
{"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_A)
self.assertTrue(res["mutation_safe"])
self.assertTrue(res["live_known"])
self.assertFalse(res["live_stale"])
def test_live_stale_when_remote_advanced_past_daemon(self):
# Local checkout still matches the daemon start (local parity green),
# but the live remote master has advanced -> daemon is live-stale.
res = mp.assess_master_parity(
{"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
self.assertTrue(res["in_parity"]) # local parity still green
self.assertTrue(res["live_stale"])
self.assertFalse(res["mutation_safe"])
self.assertTrue(any("live" in r.lower() for r in res["reasons"]))
def test_live_unknown_is_not_mutation_safe_but_not_stale(self):
# Non-goal: unfetchable live remote must not be treated as stale for
# read-only, but a mutation-safe claim fails closed.
res = mp.assess_master_parity(
{"startup_head": SHA_A}, SHA_A, live_remote_head=None)
self.assertFalse(res["live_known"])
self.assertFalse(res["mutation_safe"])
self.assertFalse(res["live_stale"])
self.assertTrue(res["in_parity"])
def test_default_live_remote_preserves_legacy_shape(self):
# Callers that do not supply a live head keep the pre-#610 behavior:
# in-parity, not live-stale, no live-derived block.
res = mp.assess_master_parity({"startup_head": SHA_A}, SHA_A)
self.assertFalse(res["live_stale"])
self.assertEqual(mp.parity_block_reasons(res), [])
class TestLiveStaleBlockAndReport(unittest.TestCase):
"""#610: live-staleness must block mutations and surface a typed blocker."""
def test_live_stale_produces_block_reasons(self):
res = mp.assess_master_parity(
{"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
self.assertTrue(mp.parity_block_reasons(res))
def test_disable_env_suppresses_live_stale_block(self):
res = mp.assess_master_parity(
{"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
with patch.dict(os.environ, {mp.ENV_DISABLE: "1"}):
self.assertEqual(mp.parity_block_reasons(res), [])
def test_resolver_disagreement_returns_typed_blocker(self):
# Parity says local-green, resolver says restart required -> disagreement
# is a typed, fail-closed blocker naming the resolver as authoritative.
res = mp.assess_master_parity({"startup_head": SHA_A}, SHA_A)
blocker = mp.parity_resolver_disagreement(res, resolver_restart_required=True)
self.assertIsNotNone(blocker)
self.assertEqual(blocker["kind"], "parity_resolver_disagreement")
self.assertTrue(blocker["restart_required"])
self.assertTrue(blocker["resolver_authoritative"])
def test_no_disagreement_when_resolver_agrees(self):
res = mp.assess_master_parity({"startup_head": SHA_A}, SHA_A)
self.assertIsNone(
mp.parity_resolver_disagreement(res, resolver_restart_required=False))
def test_live_stale_report_names_live_remote(self):
res = mp.assess_master_parity(
{"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
report = mp.parity_report(res)
self.assertEqual(report["live_remote_head"], SHA_B)
self.assertTrue(report["restart_required"])
class TestReadGitHead(unittest.TestCase):
def test_test_override_takes_precedence(self):
with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_B}):
@@ -95,6 +184,149 @@ class TestReadGitHead(unittest.TestCase):
self.assertIsNone(mp.read_git_head(""))
class TestReadRemoteMasterHead(unittest.TestCase):
"""#610: live remote master head reader (env-overridable, fails to None)."""
def test_test_override_takes_precedence(self):
with patch.dict(os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_B}):
self.assertEqual(mp.read_remote_master_head("/nonexistent"), SHA_B)
def test_blank_override_is_none(self):
with patch.dict(os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: " "}):
self.assertIsNone(mp.read_remote_master_head("/nonexistent"))
def test_unfetchable_remote_is_none(self):
# No override; a bogus root/remote must fail closed to None, never raise.
env = {k: v for k, v in os.environ.items()
if k != mp.ENV_TEST_LIVE_REMOTE_HEAD}
with patch.dict(os.environ, env, clear=True):
self.assertIsNone(
mp.read_remote_master_head("/nonexistent", remote="nope"))
class TestRemoteHeadCache(unittest.TestCase):
"""#610: live remote reads are cached with a TTL to stay off the network.
The parity gate runs on every mutation and every runtime-context read, so an
unbounded ``git ls-remote`` per call would be a latency/flakiness regression.
"""
def setUp(self):
# These cases intentionally exercise the subprocess/cache path, so they
# opt out of suite-wide hermetic mode (PR #788 F1).
self._saved_hermetic = mp.hermetic_test_mode()
mp.set_hermetic_test_mode(False)
mp._clear_remote_head_cache()
env = {
k: v for k, v in os.environ.items()
if k not in (mp.ENV_TEST_LIVE_REMOTE_HEAD,
mp.ENV_TEST_ALLOW_LIVE_REMOTE_PROBE,
"PYTEST_CURRENT_TEST")
}
# Allow the probe path under hermetic defenses while still mocking
# subprocess so no real network call runs.
env[mp.ENV_TEST_ALLOW_LIVE_REMOTE_PROBE] = "1"
self._env = patch.dict(os.environ, env, clear=True)
self._env.start()
self.addCleanup(self._env.stop)
self.addCleanup(mp._clear_remote_head_cache)
self.addCleanup(
lambda: mp.set_hermetic_test_mode(self._saved_hermetic)
)
def _fake_run(self, sha):
class _R:
returncode = 0
stdout = f"{sha}\trefs/heads/master\n"
calls = {"n": 0}
def run(*args, **kwargs):
calls["n"] += 1
return _R()
return run, calls
def test_second_call_within_ttl_uses_cache(self):
run, calls = self._fake_run(SHA_B)
with patch.object(mp.subprocess, "run", run):
a = mp.read_remote_master_head("/repo", remote="prgs", ttl=100)
b = mp.read_remote_master_head("/repo", remote="prgs", ttl=100)
self.assertEqual(a, SHA_B)
self.assertEqual(b, SHA_B)
self.assertEqual(calls["n"], 1)
def test_zero_ttl_bypasses_cache(self):
run, calls = self._fake_run(SHA_B)
with patch.object(mp.subprocess, "run", run):
mp.read_remote_master_head("/repo", remote="prgs", ttl=0)
mp.read_remote_master_head("/repo", remote="prgs", ttl=0)
self.assertEqual(calls["n"], 2)
def test_env_override_never_touches_subprocess(self):
run, calls = self._fake_run(SHA_B)
with patch.dict(os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_A}):
with patch.object(mp.subprocess, "run", run):
self.assertEqual(
mp.read_remote_master_head("/repo", remote="prgs"), SHA_A)
self.assertEqual(calls["n"], 0)
class TestHermeticLiveRemoteReads(unittest.TestCase):
"""#610 / PR #788 F1/F2: suite hermetic mode never hits the network."""
def setUp(self):
self._saved = mp.hermetic_test_mode()
mp.set_hermetic_test_mode(True)
mp._clear_remote_head_cache()
self.addCleanup(lambda: mp.set_hermetic_test_mode(self._saved))
self.addCleanup(mp._clear_remote_head_cache)
def test_hermetic_mode_returns_none_without_subprocess(self):
run_calls = {"n": 0}
def boom(*args, **kwargs):
run_calls["n"] += 1
raise AssertionError("ls-remote must not run under hermetic mode")
env = {
k: v for k, v in os.environ.items()
if k not in (mp.ENV_TEST_LIVE_REMOTE_HEAD,
mp.ENV_TEST_ALLOW_LIVE_REMOTE_PROBE)
}
with patch.dict(os.environ, env, clear=True):
with patch.object(mp.subprocess, "run", boom):
self.assertIsNone(
mp.read_remote_master_head("/repo", remote="prgs")
)
self.assertEqual(run_calls["n"], 0)
def test_hermetic_mode_survives_clear_true_env(self):
"""Module flag, not env pin: clear=True cannot re-enable the probe."""
run_calls = {"n": 0}
def boom(*args, **kwargs):
run_calls["n"] += 1
raise AssertionError("ls-remote must not run after clear=True")
with patch.dict(os.environ, {}, clear=True):
with patch.object(mp.subprocess, "run", boom):
self.assertIsNone(mp.read_remote_master_head("/repo"))
self.assertEqual(run_calls["n"], 0)
def test_explicit_override_still_wins_under_hermetic(self):
run_calls = {"n": 0}
def boom(*args, **kwargs):
run_calls["n"] += 1
raise AssertionError("override must bypass subprocess")
with patch.dict(os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_B}):
with patch.object(mp.subprocess, "run", boom):
self.assertEqual(
mp.read_remote_master_head("/repo"), SHA_B
)
self.assertEqual(run_calls["n"], 0)
class TestServerWiring(unittest.TestCase):
"""Integration with the gate choke point in the server namespace."""
@@ -105,6 +337,13 @@ class TestServerWiring(unittest.TestCase):
self._saved = self.srv._STARTUP_PARITY
self.srv._STARTUP_PARITY = {"root": self.srv.PROJECT_ROOT,
"startup_head": SHA_A}
# Keep the live-remote read hermetic (no real ls-remote network call):
# default the live master to the daemon start so parity is fully green
# unless a test overrides the live head explicitly (#610).
self._live_patch = patch.dict(
os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_A})
self._live_patch.start()
self.addCleanup(self._live_patch.stop)
def tearDown(self):
self.srv._STARTUP_PARITY = self._saved
@@ -147,6 +386,36 @@ class TestServerWiring(unittest.TestCase):
self.assertTrue(out["in_parity"])
self.assertNotIn("report", out)
# --- #610: live-remote wiring -------------------------------------------
def test_live_stale_blocks_mutation_though_local_green(self):
# Local checkout matches the daemon start (local parity green) but the
# live remote master has advanced -> mutations must fail closed.
with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_A,
mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_B}):
self.assertEqual(self.srv._master_parity_block("gitea.read"), [])
self.assertTrue(
self.srv._master_parity_block("gitea.pr.create"))
def test_assess_tool_exposes_three_distinct_shas(self):
with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_A,
mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_B}):
out = self.srv.gitea_assess_master_parity(remote="prgs")
self.assertEqual(out["daemon_start_head"], SHA_A)
self.assertEqual(out["local_head"], SHA_A)
self.assertEqual(out["live_remote_head"], SHA_B)
self.assertTrue(out["live_stale"])
self.assertFalse(out["mutation_safe"])
self.assertIn("report", out)
def test_assess_tool_mutation_safe_when_all_three_match(self):
with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_A,
mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_A}):
out = self.srv.gitea_assess_master_parity(remote="prgs")
self.assertTrue(out["mutation_safe"])
self.assertFalse(out["live_stale"])
self.assertNotIn("report", out)
if __name__ == "__main__":
unittest.main()
+703
View File
@@ -0,0 +1,703 @@
"""Console authorization, redaction, and audit model tests (#633).
Covers each acceptance criterion and each required test named in the issue:
* AC1 — RBAC matrix and privileged-action list.
* AC2 — redaction rules, unit-tested against sample payloads.
* AC3 — audit event schema with required fields and retention defaults.
* AC4 — Phase 2 integration points.
* AC5 — local-dev mode with explicit insecurity warnings.
Required tests: redaction units (token, keychain, password patterns),
default-deny for unauthenticated write stubs, and audit record creation for a
simulated privileged preview.
"""
from __future__ import annotations
import datetime
import json
import os
import pathlib
import sys
import tempfile
import unittest
from starlette.testclient import TestClient
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1]))
from task_capability_map import TASK_CAPABILITY_MAP # noqa: E402
from webui import console_audit, console_authz # noqa: E402
from webui.app import create_app # noqa: E402
from webui.console_redaction import ( # noqa: E402
REDACTED,
redact_payload,
redact_text,
redaction_policy,
scan_for_secrets,
)
DOCS = pathlib.Path(__file__).resolve().parents[1] / "docs"
AUTHZ_DOC = DOCS / "webui-authz-audit.md"
def _principal(role: str) -> console_authz.Principal:
return console_authz.Principal(
subject=f"{role}@example.com",
role=role,
identity_source=console_authz.IDENTITY_ACCESS_PROXY,
authenticated=True,
)
class TestRoleMatrix(unittest.TestCase):
"""AC1 — the written RBAC matrix and privileged-action list."""
def test_roles_are_ordered_least_to_most_authority(self):
self.assertEqual(
console_authz.ROLE_ORDER,
("viewer", "operator", "controller", "admin"),
)
def test_every_role_has_a_description(self):
for role in console_authz.ROLE_ORDER:
with self.subTest(role=role):
self.assertTrue(console_authz.ROLE_DESCRIPTIONS[role].strip())
def test_higher_roles_inherit_lower_role_actions(self):
matrix = {
entry["role"]: set(entry["permitted_actions"])
for entry in console_authz.rbac_matrix()["roles"]
}
for lower, higher in zip(
console_authz.ROLE_ORDER, console_authz.ROLE_ORDER[1:]
):
with self.subTest(lower=lower, higher=higher):
self.assertTrue(matrix[lower].issubset(matrix[higher]))
def test_viewer_holds_no_write_action(self):
matrix = {
entry["role"]: set(entry["permitted_actions"])
for entry in console_authz.rbac_matrix()["roles"]
}
self.assertEqual(matrix["viewer"], set())
def test_privileged_action_list_is_non_empty_and_classified(self):
privileged = console_authz.privileged_actions()
self.assertTrue(privileged)
ids = {action.action_id for action in privileged}
# Merge and branch deletion are the canonical privileged pair.
self.assertIn("merge_pr", ids)
self.assertIn("delete_branch", ids)
def test_merge_and_delete_require_dual_control_and_break_glass(self):
for action_id in ("merge_pr", "delete_branch"):
with self.subTest(action=action_id):
action = console_authz.get_action(action_id)
self.assertTrue(action.dual_control)
self.assertTrue(action.break_glass)
self.assertTrue(action.requires_confirmation)
def test_every_write_action_requires_confirmation(self):
for action in console_authz.ACTIONS.values():
with self.subTest(action=action.action_id):
self.assertTrue(action.requires_confirmation)
def test_delete_branch_is_admin_only(self):
self.assertEqual(
console_authz.get_action("delete_branch").minimum_role,
console_authz.ADMIN,
)
def test_actions_map_to_real_mcp_capability_vocabulary(self):
"""The console must not invent an authority the MCP layer lacks."""
for action in console_authz.ACTIONS.values():
with self.subTest(action=action.action_id):
self.assertIn(action.task_key, TASK_CAPABILITY_MAP)
self.assertEqual(
action.mcp_permission,
TASK_CAPABILITY_MAP[action.task_key]["permission"],
)
self.assertEqual(
action.mcp_role,
TASK_CAPABILITY_MAP[action.task_key]["role"],
)
def test_matrix_declares_deny_by_default_and_execution_disabled(self):
matrix = console_authz.rbac_matrix()
self.assertEqual(matrix["default_decision"], "deny")
self.assertFalse(matrix["execution_enabled"])
class TestAuthorizeDefaultDeny(unittest.TestCase):
"""Fail-closed behaviour of the authorization decision."""
def test_anonymous_is_denied_every_action(self):
for action_id in console_authz.ACTIONS:
with self.subTest(action=action_id):
decision = console_authz.authorize(action_id)
self.assertFalse(decision.allowed)
self.assertEqual(
decision.reason_code, console_authz.DENY_UNAUTHENTICATED
)
def test_unknown_action_is_denied(self):
decision = console_authz.authorize(
"not_a_real_action", _principal("admin")
)
self.assertFalse(decision.allowed)
self.assertEqual(decision.reason_code, console_authz.DENY_UNKNOWN_ACTION)
def test_unknown_role_is_denied(self):
rogue = console_authz.Principal(
subject="[email protected]",
role="superuser",
identity_source=console_authz.IDENTITY_ACCESS_PROXY,
authenticated=True,
)
decision = console_authz.authorize("comment_issue", rogue)
self.assertFalse(decision.allowed)
self.assertEqual(decision.reason_code, console_authz.DENY_UNKNOWN_ROLE)
def test_insufficient_role_is_denied(self):
decision = console_authz.authorize("merge_pr", _principal("operator"))
self.assertFalse(decision.allowed)
self.assertEqual(
decision.reason_code, console_authz.DENY_INSUFFICIENT_ROLE
)
def test_sufficient_role_allows_preview_only(self):
decision = console_authz.authorize("merge_pr", _principal("controller"))
self.assertTrue(decision.allowed)
self.assertFalse(decision.execution_enabled)
def test_execution_is_refused_while_phase_is_not_active(self):
decision = console_authz.authorize(
"merge_pr", _principal("controller"), for_execution=True
)
self.assertFalse(decision.allowed)
self.assertEqual(
decision.reason_code, console_authz.DENY_PHASE_NOT_ACTIVE
)
def test_allowed_decision_never_reports_execution_enabled(self):
for action_id in console_authz.ACTIONS:
with self.subTest(action=action_id):
decision = console_authz.authorize(
action_id, _principal("admin")
)
self.assertFalse(decision.execution_enabled)
class TestIdentityResolution(unittest.TestCase):
"""AC5 — identity sources, including the insecure local-dev mode."""
def test_no_auth_mode_yields_anonymous_viewer(self):
principal = console_authz.resolve_principal(env={})
self.assertFalse(principal.authenticated)
self.assertEqual(principal.role, console_authz.VIEWER)
self.assertEqual(principal.identity_source, console_authz.IDENTITY_NONE)
def test_local_dev_mode_warns_that_identity_is_unverified(self):
principal = console_authz.resolve_principal(
env={
console_authz.AUTH_MODE_ENV: "local-dev",
console_authz.DEV_SUBJECT_ENV: "[email protected]",
console_authz.DEV_ROLE_ENV: "admin",
}
)
self.assertTrue(principal.authenticated)
self.assertEqual(principal.role, "admin")
self.assertTrue(principal.warnings)
self.assertIn("asserted", " ".join(principal.warnings).lower())
def test_local_dev_without_subject_falls_back_to_anonymous(self):
principal = console_authz.resolve_principal(
env={console_authz.AUTH_MODE_ENV: "local-dev"}
)
self.assertFalse(principal.authenticated)
def test_local_dev_unknown_role_degrades_to_viewer(self):
principal = console_authz.resolve_principal(
env={
console_authz.AUTH_MODE_ENV: "local_dev",
console_authz.DEV_SUBJECT_ENV: "[email protected]",
console_authz.DEV_ROLE_ENV: "root",
}
)
self.assertEqual(principal.role, console_authz.VIEWER)
def test_access_proxy_without_header_fails_closed(self):
"""A proxy-mode request that did not traverse the proxy is anonymous."""
principal = console_authz.resolve_principal(
headers={},
env={console_authz.AUTH_MODE_ENV: "access_proxy"},
)
self.assertFalse(principal.authenticated)
def test_access_proxy_role_comes_from_server_config_not_client(self):
env = {
console_authz.AUTH_MODE_ENV: "access_proxy",
console_authz.ROLE_MAP_ENV: json.dumps(
{"[email protected]": "controller"}
),
}
principal = console_authz.resolve_principal(
headers={
console_authz.ACCESS_SUBJECT_HEADER: "[email protected]",
"x-role": "admin", # client-supplied role must be ignored
},
env=env,
)
self.assertEqual(principal.role, "controller")
def test_access_proxy_unmapped_subject_defaults_to_viewer(self):
principal = console_authz.resolve_principal(
headers={
console_authz.ACCESS_SUBJECT_HEADER: "[email protected]"
},
env={console_authz.AUTH_MODE_ENV: "access_proxy"},
)
self.assertEqual(principal.role, console_authz.VIEWER)
def test_malformed_role_map_does_not_raise_and_denies(self):
principal = console_authz.resolve_principal(
headers={console_authz.ACCESS_SUBJECT_HEADER: "[email protected]"},
env={
console_authz.AUTH_MODE_ENV: "access_proxy",
console_authz.ROLE_MAP_ENV: "{not json",
},
)
self.assertEqual(principal.role, console_authz.VIEWER)
def test_probe_auth_is_opt_in(self):
self.assertFalse(console_authz.probe_auth_required(env={}))
self.assertTrue(
console_authz.probe_auth_required(
env={console_authz.REQUIRE_PROBE_AUTH_ENV: "1"}
)
)
def test_probe_auth_is_declared_but_not_yet_enforced(self):
"""Phase 1 declares the probe-auth policy; no route enforces it yet.
The flag exists so the Phase 2 action framework has a declared policy
to honour instead of inventing a second one. Pinning the current
not-enforced status here means wiring it later is a deliberate change
that updates this test and the documentation together, rather than a
silent behaviour shift. The documentation must say so plainly, because
an operator who sets the variable believing it protects a probe is
worse off than one who knows it does not.
"""
import inspect
from webui import app as webui_app
source = inspect.getsource(webui_app)
self.assertNotIn(
"probe_auth_required",
source,
msg=(
"webui.app now consults probe_auth_required, so probe auth is "
"no longer merely declared. Update the 'Probe authentication' "
"section of docs/webui-authz-audit.md, which states it "
"enforces nothing, and replace this test with real "
"enforcement coverage."
),
)
self.assertIn(
"enforces nothing today",
AUTHZ_DOC.read_text(encoding="utf-8"),
)
class TestRedaction(unittest.TestCase):
"""AC2 — required redaction units: token, keychain, password patterns."""
def test_token_assignment_is_redacted(self):
out = redact_text("GITEA_TOKEN=abcd1234efgh5678ijkl")
self.assertIn(REDACTED, out)
self.assertNotIn("abcd1234efgh5678ijkl", out)
def test_password_assignment_is_redacted(self):
out = redact_text("password: hunter2supersecret")
self.assertIn(REDACTED, out)
self.assertNotIn("hunter2supersecret", out)
def test_keychain_reference_is_redacted(self):
out = redact_text("keychain:gitea-prgs-token")
self.assertIn(REDACTED, out)
self.assertNotIn("gitea-prgs-token", out)
def test_keychain_command_is_redacted(self):
out = redact_text("security find-generic-password -s gitea -w")
self.assertIn(REDACTED, out)
self.assertNotIn("find-generic-password -s gitea", out)
def test_bearer_credential_is_redacted(self):
out = redact_text("Authorization: Bearer abcdef1234567890abcdef")
self.assertNotIn("abcdef1234567890abcdef", out)
def test_jwt_is_redacted(self):
token = "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIxIn0.abcdefghijklmnop"
out = redact_text(f"session={token}")
self.assertNotIn(token, out)
def test_private_key_block_is_redacted(self):
pem = (
"-----BEGIN RSA PRIVATE KEY-----\n"
"MIIEowIBAAKCAQEAsecretmaterial\n"
"-----END RSA PRIVATE KEY-----"
)
out = redact_text(pem)
self.assertNotIn("MIIEowIBAAKCAQEAsecretmaterial", out)
def test_api_key_assignment_is_redacted(self):
out = redact_text('api_key = "sk-live-9f8e7d6c5b4a3210"')
self.assertNotIn("sk-live-9f8e7d6c5b4a3210", out)
def test_nested_payload_is_redacted_recursively(self):
payload = {
"token": "abc123456789",
"nested": {"note": "password=letmein12345"},
"list": ["keychain:some-entry"],
"safe": "plain text",
}
out = redact_payload(payload)
self.assertEqual(out["token"], REDACTED)
self.assertNotIn("letmein12345", json.dumps(out))
self.assertNotIn("some-entry", json.dumps(out))
self.assertEqual(out["safe"], "plain text")
def test_scan_reports_findings_before_and_none_after(self):
dirty = "password: hunter2supersecret"
self.assertTrue(scan_for_secrets(dirty))
self.assertEqual(scan_for_secrets(redact_text(dirty)), [])
def test_non_strings_pass_through_untouched(self):
self.assertEqual(redact_text(42), 42)
self.assertEqual(
redact_payload({"n": 1, "b": True}), {"n": 1, "b": True}
)
def test_policy_is_documented_and_declares_redact_before_persist(self):
policy = redaction_policy()
self.assertTrue(policy["redact_before_persist"])
self.assertIn("audit_records", policy["applies_to"])
self.assertTrue(policy["console_rules"])
def test_policy_statement_contains_no_secret_material(self):
self.assertEqual(scan_for_secrets(redaction_policy()), [])
class TestAuditSchema(unittest.TestCase):
"""AC3 — audit event schema, required fields, and retention defaults."""
def _event(self, action_id="merge_pr", **kwargs):
return console_audit.build_event(
action_id=action_id,
result=console_audit.RESULT_DENIED,
decision=console_authz.authorize(action_id, _principal("operator")),
target={"kind": "pr", "ref": "#123"},
request_id="req-test",
**kwargs,
)
def test_every_required_field_is_present(self):
event = self._event()
for field in console_audit.REQUIRED_FIELDS:
with self.subTest(field=field):
self.assertIn(field, event)
def test_actor_carries_who_and_how_they_were_identified(self):
event = self._event()
for field in console_audit.REQUIRED_ACTOR_FIELDS:
with self.subTest(field=field):
self.assertIn(field, event["actor"])
def test_correlation_ids_are_present(self):
event = self._event()
for field in console_audit.REQUIRED_CORRELATION_FIELDS:
with self.subTest(field=field):
self.assertIn(field, event["correlation"])
self.assertEqual(event["correlation"]["request_id"], "req-test")
self.assertEqual(event["correlation"]["mcp_task"], "merge_pr")
def test_timestamp_is_timezone_aware_utc_iso8601(self):
now = datetime.datetime(
2026, 7, 22, 10, 16, 42, tzinfo=datetime.timezone.utc
)
event = self._event(now=now)
self.assertEqual(event["timestamp"], "2026-07-22T10:16:42+00:00")
parsed = datetime.datetime.fromisoformat(event["timestamp"])
self.assertIsNotNone(parsed.tzinfo)
def test_retention_defaults_by_class(self):
self.assertEqual(
console_audit.RETENTION_DAYS[console_audit.RETENTION_STANDARD], 90
)
self.assertEqual(
console_audit.RETENTION_DAYS[console_audit.RETENTION_PRIVILEGED],
365,
)
self.assertEqual(
console_audit.RETENTION_DAYS[console_audit.RETENTION_BREAK_GLASS],
730,
)
def test_break_glass_action_retains_longest(self):
event = self._event("merge_pr")
self.assertEqual(
event["retention"]["class"], console_audit.RETENTION_BREAK_GLASS
)
def test_routine_write_uses_standard_retention(self):
event = self._event("comment_issue")
self.assertEqual(
event["retention"]["class"], console_audit.RETENTION_STANDARD
)
def test_unknown_action_retains_as_privileged_not_standard(self):
"""Conservative direction: keep an unclassifiable record longer."""
self.assertEqual(
console_audit.retention_class_for(None),
console_audit.RETENTION_PRIVILEGED,
)
def test_retention_expiry_matches_declared_days(self):
now = datetime.datetime(2026, 7, 22, tzinfo=datetime.timezone.utc)
event = self._event("comment_issue", now=now)
expires = datetime.datetime.fromisoformat(
event["retention"]["expires_at"]
)
self.assertEqual((expires - now).days, 90)
def test_invalid_result_degrades_to_failed(self):
event = console_audit.build_event(action_id="merge_pr", result="banana")
self.assertEqual(event["result"], console_audit.RESULT_FAILED)
def test_denied_result_is_representable(self):
"""An authorization denial has no MCP-side mutation record."""
self.assertIn(console_audit.RESULT_DENIED, console_audit.RESULTS)
def test_event_is_redacted_before_it_is_returned(self):
event = console_audit.build_event(
action_id="merge_pr",
result=console_audit.RESULT_DENIED,
detail="failed with token=abcdef1234567890",
metadata={"password": "hunter2supersecret"},
)
serialized = json.dumps(event)
self.assertNotIn("abcdef1234567890", serialized)
self.assertNotIn("hunter2supersecret", serialized)
self.assertTrue(event["redacted"])
def test_audit_policy_reports_schema_and_retention(self):
policy = console_audit.audit_policy()
self.assertTrue(policy["append_only"])
self.assertTrue(policy["redact_before_persist"])
self.assertEqual(
policy["retention_defaults_days"], console_audit.RETENTION_DAYS
)
class TestAuditSink(unittest.TestCase):
"""Append-only persistence behaviour."""
def test_write_is_a_noop_when_sink_is_unconfigured(self):
saved = os.environ.pop(console_audit.AUDIT_LOG_ENV, None)
try:
self.assertFalse(console_audit.audit_enabled())
self.assertFalse(console_audit.write_event({"schema_version": 1}))
finally:
if saved is not None:
os.environ[console_audit.AUDIT_LOG_ENV] = saved
def test_records_append_one_json_line_each(self):
with tempfile.TemporaryDirectory() as tmp:
sink = os.path.join(tmp, "console-audit.jsonl")
for _ in range(3):
event = console_audit.build_event(
action_id="merge_pr", result=console_audit.RESULT_DENIED
)
self.assertTrue(console_audit.write_event(event, path=sink))
with open(sink, encoding="utf-8") as handle:
lines = [json.loads(line) for line in handle if line.strip()]
self.assertEqual(len(lines), 3)
self.assertEqual(len({line["event_id"] for line in lines}), 3)
def test_a_record_that_still_carries_a_secret_is_not_persisted(self):
with tempfile.TemporaryDirectory() as tmp:
sink = os.path.join(tmp, "console-audit.jsonl")
leaky = {
"schema_version": 1,
"detail": "password: hunter2supersecret",
}
self.assertFalse(console_audit.write_event(leaky, path=sink))
self.assertFalse(os.path.exists(sink))
def test_write_never_raises_on_a_bad_path(self):
self.assertFalse(
console_audit.write_event(
{"schema_version": 1}, path="/nonexistent-dir/audit.jsonl"
)
)
def test_simulated_privileged_preview_creates_an_audit_record(self):
"""Required test: audit record creation for a privileged preview."""
with tempfile.TemporaryDirectory() as tmp:
sink = os.path.join(tmp, "console-audit.jsonl")
os.environ[console_audit.AUDIT_LOG_ENV] = sink
try:
decision = console_authz.authorize(
"merge_pr", _principal("controller")
)
outcome = console_audit.record_event(
action_id="merge_pr",
result=console_audit.RESULT_PREVIEWED,
decision=decision,
target={"kind": "pr", "ref": "#123"},
request_id="req-preview",
)
finally:
os.environ.pop(console_audit.AUDIT_LOG_ENV, None)
self.assertTrue(outcome["written"])
with open(sink, encoding="utf-8") as handle:
record = json.loads(handle.read().strip())
self.assertEqual(record["action"], "merge_pr")
self.assertEqual(record["result"], console_audit.RESULT_PREVIEWED)
self.assertEqual(record["action_class"], "privileged")
self.assertTrue(record["decision"]["allowed"])
self.assertFalse(record["decision"]["execution_enabled"])
self.assertEqual(record["actor"]["role"], "controller")
def test_decision_block_survives_redaction(self):
"""Regression: naming it 'authorization' collided with a secret hint.
``gitea_audit._SECRET_KEY_HINTS`` contains "authorization" (for the
HTTP header), so a block under that key was replaced wholesale by the
placeholder and the record lost its decision entirely.
"""
event = console_audit.build_event(
action_id="merge_pr",
result=console_audit.RESULT_DENIED,
decision=console_authz.authorize("merge_pr", _principal("admin")),
)
self.assertIsInstance(event["decision"], dict)
self.assertIn("allowed", event["decision"])
class TestConsoleRoutes(unittest.TestCase):
"""AC4 — the wired Phase 2 integration points, still fail-closed."""
def setUp(self):
self.client = TestClient(create_app(bind_host="127.0.0.1"))
def test_unauthenticated_write_stub_is_denied(self):
"""Required test: default-deny for unauthenticated write stubs."""
response = self.client.post(
"/api/actions/merge_pr/attempt", json={"pr_number": 99}
)
self.assertEqual(response.status_code, 403)
body = response.json()
self.assertFalse(body["success"])
authorization = body["authorization"]
self.assertFalse(authorization["allowed"])
self.assertEqual(
authorization["reason_code"], console_authz.DENY_UNAUTHENTICATED
)
self.assertFalse(authorization["execution_enabled"])
def test_preview_reports_an_authorization_decision(self):
response = self.client.get("/api/actions/merge_pr/preview?pr_number=7")
self.assertEqual(response.status_code, 200)
authorization = response.json()["authorization"]
self.assertFalse(authorization["allowed"])
self.assertTrue(authorization["dual_control"])
self.assertEqual(authorization["required_role"], "controller")
def test_unknown_action_preview_still_404s(self):
response = self.client.get("/api/actions/no_such_action/preview")
self.assertEqual(response.status_code, 404)
def test_security_model_endpoint_publishes_all_three_policies(self):
response = self.client.get("/api/console/security-model")
self.assertEqual(response.status_code, 200)
body = response.json()
self.assertIn("rbac", body)
self.assertIn("redaction", body)
self.assertIn("audit", body)
self.assertEqual(body["rbac"]["default_decision"], "deny")
def test_security_model_endpoint_leaks_no_secrets(self):
response = self.client.get("/api/console/security-model")
self.assertEqual(scan_for_secrets(response.json()), [])
def test_security_model_rejects_writes(self):
response = self.client.post("/api/console/security-model", json={})
self.assertEqual(response.status_code, 405)
def test_existing_read_routes_are_unaffected(self):
for path in ("/", "/health", "/actions", "/api/actions"):
with self.subTest(path=path):
self.assertEqual(self.client.get(path).status_code, 200)
class TestAuthzAuditDoc(unittest.TestCase):
"""The model must be written down, not only coded."""
@classmethod
def setUpClass(cls):
cls.text = (
AUTHZ_DOC.read_text(encoding="utf-8") if AUTHZ_DOC.exists() else ""
)
def test_doc_exists(self):
self.assertTrue(AUTHZ_DOC.exists(), f"missing {AUTHZ_DOC}")
def test_doc_covers_each_required_section(self):
for heading in (
"Identity sources",
"Role matrix",
"Privileged actions",
"Secret redaction",
"Audit event schema",
"Retention",
"Phase 2 integration",
"Local-dev mode",
):
with self.subTest(heading=heading):
self.assertIn(heading, self.text)
def test_doc_names_every_role(self):
for role in console_authz.ROLE_ORDER:
with self.subTest(role=role):
self.assertIn(role, self.text)
def test_doc_names_every_console_action(self):
for action_id in console_authz.ACTIONS:
with self.subTest(action=action_id):
self.assertIn(action_id, self.text)
def test_doc_states_retention_defaults(self):
for days in console_audit.RETENTION_DAYS.values():
with self.subTest(days=days):
self.assertIn(str(days), self.text)
def test_doc_warns_local_dev_is_insecure(self):
self.assertIn("INSECURE", self.text.upper())
def test_doc_states_default_deny(self):
self.assertIn("deny", self.text.lower())
def test_doc_contains_no_secret_material(self):
self.assertEqual(scan_for_secrets(self.text), [])
def test_deployment_doc_links_to_the_model(self):
deployment = (DOCS / "webui-deployment.md").read_text(encoding="utf-8")
self.assertIn("webui-authz-audit", deployment)
if __name__ == "__main__": # pragma: no cover
unittest.main()
+458
View File
@@ -0,0 +1,458 @@
"""Tests for the worker registry and configuration schema (#798, epic #797)."""
import json
import sys
import tempfile
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from webui.worker_registry import (
ALLOWED_ROLES,
SCHEMA_VERSION,
RegistryValidationError,
WorkerRegistry,
default_registry_path,
find_provider,
find_worker,
history_dir,
list_revisions,
load_registry,
registry_to_dict,
registry_to_document,
rollback_to_revision,
save_registry,
validate_payload,
worker_to_dict,
workers_for_provider,
)
_EXPECTED_PROVIDER_IDS = ("claude", "grok", "codex", "agy", "kimi-k")
def _provider(provider_id: str = "claude", **overrides) -> dict:
payload = {
"id": provider_id,
"display_name": "Claude",
"vendor": "Anthropic",
"executable": "claude",
"available": True,
"models": ["claude-opus-4-8"],
"notes": "",
}
payload.update(overrides)
return payload
def _worker(worker_id: str = "claude-author", **overrides) -> dict:
payload = {
"id": worker_id,
"display_name": "Claude author",
"provider": "claude",
"model": "claude-opus-4-8",
"project": "gitea-tools",
"role": "author",
"namespace": "gitea-author",
"profile": "prgs-author",
"workflow": "skills/llm-project-workflow/workflows/work-issue.md",
"schedule": {"kind": "cron", "expression": "0 * * * *"},
"timeout_seconds": 3600,
"enabled": True,
"scheduler": {"kind": "launchd", "label": "cc.prgs.claude.author"},
"notes": "",
}
payload.update(overrides)
return payload
def _document(providers=None, workers=None, **overrides) -> dict:
payload = {
"version": SCHEMA_VERSION,
"revision": 1,
"updated_at": "2026-07-22T00:00:00Z",
"providers": providers if providers is not None else [_provider()],
"workers": workers if workers is not None else [_worker()],
}
payload.update(overrides)
return payload
class _TempRegistryCase(unittest.TestCase):
"""Base case giving each test an isolated registry file."""
def setUp(self):
self._tmp = tempfile.TemporaryDirectory()
self.addCleanup(self._tmp.cleanup)
self.path = Path(self._tmp.name) / "workers.registry.json"
def write(self, document: dict) -> Path:
self.path.write_text(json.dumps(document, indent=2) + "\n", encoding="utf-8")
return self.path
def parse(self, document: dict) -> WorkerRegistry:
return validate_payload(document, source_path=self.path)
class TestPackagedRegistry(unittest.TestCase):
"""AC: the declarative registry is the source of truth and ships with the app."""
def test_default_path_points_at_packaged_data(self):
path = default_registry_path()
self.assertEqual(path.name, "workers.registry.json")
self.assertEqual(path.parent.name, "data")
def test_packaged_registry_loads_and_validates(self):
registry = load_registry()
self.assertEqual(registry.version, SCHEMA_VERSION)
self.assertGreaterEqual(registry.revision, 1)
def test_packaged_registry_declares_all_five_providers(self):
registry = load_registry()
self.assertEqual(
tuple(provider.id for provider in registry.providers),
_EXPECTED_PROVIDER_IDS,
)
def test_packaged_registry_carries_no_credentials(self):
raw = default_registry_path().read_text(encoding="utf-8").lower()
for marker in ("token", "password", "secret", "api_key", "credential"):
self.assertNotIn(marker, raw)
class TestSeparateEntities(_TempRegistryCase):
"""AC: providers and configured workers are separate entities."""
def test_provider_may_exist_with_no_workers(self):
registry = self.parse(
_document(providers=[_provider("grok", display_name="Grok")], workers=[])
)
self.assertEqual(len(registry.providers), 1)
self.assertEqual(registry.workers, ())
self.assertEqual(workers_for_provider(registry, "grok"), ())
def test_many_workers_may_share_one_provider(self):
registry = self.parse(
_document(
workers=[
_worker("claude-author"),
_worker(
"claude-reviewer",
role="reviewer",
namespace="gitea-reviewer",
profile="prgs-reviewer",
scheduler={"kind": "launchd", "label": "cc.prgs.claude.reviewer"},
),
]
)
)
self.assertEqual(len(workers_for_provider(registry, "claude")), 2)
self.assertEqual(len(registry.providers), 1)
def test_worker_referencing_unknown_provider_is_refused(self):
with self.assertRaises(RegistryValidationError) as ctx:
self.parse(_document(workers=[_worker(provider="mystery")]))
self.assertIn("unknown provider", str(ctx.exception))
def test_lookup_helpers(self):
registry = self.parse(_document())
self.assertIsNotNone(find_worker(registry, "claude-author"))
self.assertIsNone(find_worker(registry, "absent"))
self.assertIsNotNone(find_provider(registry, "claude"))
self.assertIsNone(find_provider(registry, "absent"))
class TestRecordedFields(_TempRegistryCase):
"""AC: records provider, model, project, role, namespace/profile, workflow,
schedule, timeout, enabled state, and scheduler metadata."""
def test_every_required_field_is_recorded(self):
registry = self.parse(_document())
worker = registry.workers[0]
self.assertEqual(worker.provider, "claude")
self.assertEqual(worker.model, "claude-opus-4-8")
self.assertEqual(worker.project, "gitea-tools")
self.assertEqual(worker.role, "author")
self.assertEqual(worker.namespace, "gitea-author")
self.assertEqual(worker.profile, "prgs-author")
self.assertEqual(worker.workflow, "skills/llm-project-workflow/workflows/work-issue.md")
self.assertEqual(worker.schedule.kind, "cron")
self.assertEqual(worker.schedule.expression, "0 * * * *")
self.assertEqual(worker.timeout_seconds, 3600)
self.assertTrue(worker.enabled)
self.assertEqual(worker.scheduler.kind, "launchd")
self.assertEqual(worker.scheduler.label, "cc.prgs.claude.author")
def test_each_required_field_is_individually_required(self):
for field in (
"provider", "model", "project", "role", "namespace",
"profile", "workflow", "schedule", "timeout_seconds",
"enabled", "scheduler", "id", "display_name",
):
with self.subTest(field=field):
worker = _worker()
worker.pop(field)
with self.assertRaises(RegistryValidationError):
self.parse(_document(workers=[worker]))
def test_all_sanctioned_roles_are_accepted(self):
for role in ALLOWED_ROLES:
with self.subTest(role=role):
registry = self.parse(_document(workers=[_worker(role=role)]))
self.assertEqual(registry.workers[0].role, role)
def test_unsanctioned_role_is_refused(self):
with self.assertRaises(RegistryValidationError) as ctx:
self.parse(_document(workers=[_worker(role="admin")]))
self.assertIn("role must be one of", str(ctx.exception))
def test_worker_dict_round_trips_every_field(self):
registry = self.parse(_document())
encoded = worker_to_dict(registry.workers[0])
self.assertEqual(encoded, _worker())
json.dumps(encoded) # must stay JSON-safe for the #799 API
class TestSchemaValidation(_TempRegistryCase):
"""AC: supports schema validation — and fails closed."""
def test_unsupported_version_is_refused(self):
with self.assertRaises(RegistryValidationError):
self.parse(_document(version=2))
def test_root_must_be_an_object(self):
with self.assertRaises(RegistryValidationError):
validate_payload([], source_path=self.path)
def test_providers_must_be_non_empty(self):
with self.assertRaises(RegistryValidationError):
self.parse(_document(providers=[]))
def test_unknown_top_level_field_is_refused(self):
with self.assertRaises(RegistryValidationError) as ctx:
self.parse(_document(fleet=[]))
self.assertIn("unknown fields", str(ctx.exception))
def test_unknown_worker_field_is_refused_not_ignored(self):
# A typo'd field must not be silently dropped: "timeout_second" would
# otherwise read as "no timeout declared".
worker = _worker()
worker["timeout_second"] = 30
with self.assertRaises(RegistryValidationError) as ctx:
self.parse(_document(workers=[worker]))
self.assertIn("timeout_second", str(ctx.exception))
def test_credentials_are_refused_anywhere_in_the_document(self):
for label, mutate in (
("provider.api_token", lambda doc: doc["providers"][0].__setitem__("api_token", "x")),
("worker.password", lambda doc: doc["workers"][0].__setitem__("password", "x")),
("root.secret", lambda doc: doc.__setitem__("secret", "x")),
):
with self.subTest(field=label):
document = _document()
mutate(document)
with self.assertRaises(ValueError) as ctx:
self.parse(document)
self.assertIn("credential", str(ctx.exception).lower())
def test_duplicate_worker_id_is_refused(self):
workers = [_worker("dup"), _worker("dup", scheduler={"kind": "manual"})]
with self.assertRaises(RegistryValidationError) as ctx:
self.parse(_document(workers=workers))
self.assertIn("duplicate worker id", str(ctx.exception))
def test_duplicate_provider_id_is_refused(self):
with self.assertRaises(RegistryValidationError) as ctx:
self.parse(_document(providers=[_provider("claude"), _provider("claude")], workers=[]))
self.assertIn("duplicate provider id", str(ctx.exception))
def test_duplicate_launchagent_label_is_refused(self):
# Two workers sharing a label would silently overwrite each other's agent.
workers = [
_worker("a", scheduler={"kind": "launchd", "label": "cc.prgs.same"}),
_worker("b", scheduler={"kind": "launchd", "label": "cc.prgs.same"}),
]
with self.assertRaises(RegistryValidationError) as ctx:
self.parse(_document(workers=workers))
self.assertIn("duplicate scheduler label", str(ctx.exception))
def test_manual_scheduler_needs_no_label_and_many_may_coexist(self):
workers = [
_worker("a", scheduler={"kind": "manual"}),
_worker("b", scheduler={"kind": "manual"}),
]
registry = self.parse(_document(workers=workers))
self.assertEqual([w.scheduler.label for w in registry.workers], [None, None])
def test_launchd_scheduler_requires_a_label(self):
with self.assertRaises(RegistryValidationError) as ctx:
self.parse(_document(workers=[_worker(scheduler={"kind": "launchd"})]))
self.assertIn("label is required", str(ctx.exception))
def test_unknown_scheduler_kind_is_refused(self):
with self.assertRaises(RegistryValidationError):
self.parse(_document(workers=[_worker(scheduler={"kind": "systemd", "label": "x"})]))
def test_timeout_must_be_a_positive_bounded_integer(self):
for bad in (0, -1, "3600", 1.5, True, 86_401):
with self.subTest(timeout=bad):
with self.assertRaises(RegistryValidationError):
self.parse(_document(workers=[_worker(timeout_seconds=bad)]))
def test_enabled_must_be_a_real_boolean(self):
for bad in ("true", 1, None):
with self.subTest(enabled=bad):
with self.assertRaises(RegistryValidationError):
self.parse(_document(workers=[_worker(enabled=bad)]))
def test_identifier_shape_is_enforced(self):
for bad in ("Claude Author", "-leading", "UPPER", ""):
with self.subTest(worker_id=bad):
with self.assertRaises(RegistryValidationError):
self.parse(_document(workers=[_worker(bad)]))
class TestScheduleValidation(_TempRegistryCase):
"""Schedules are declarations; next-run computation belongs to #803."""
def test_interval_schedule_requires_positive_seconds(self):
registry = self.parse(
_document(workers=[_worker(schedule={"kind": "interval", "seconds": 900})])
)
self.assertEqual(registry.workers[0].schedule.seconds, 900)
with self.assertRaises(RegistryValidationError):
self.parse(_document(workers=[_worker(schedule={"kind": "interval"})]))
with self.assertRaises(RegistryValidationError):
self.parse(_document(workers=[_worker(schedule={"kind": "interval", "seconds": 0})]))
def test_cron_schedule_requires_five_fields(self):
with self.assertRaises(RegistryValidationError) as ctx:
self.parse(_document(workers=[_worker(schedule={"kind": "cron", "expression": "0 *"})]))
self.assertIn("five crontab fields", str(ctx.exception))
def test_manual_schedule_needs_no_timing(self):
registry = self.parse(_document(workers=[_worker(schedule={"kind": "manual"})]))
schedule = registry.workers[0].schedule
self.assertEqual(schedule.kind, "manual")
self.assertIsNone(schedule.seconds)
self.assertIsNone(schedule.expression)
def test_fields_from_the_wrong_kind_are_refused(self):
with self.assertRaises(RegistryValidationError) as ctx:
self.parse(_document(workers=[_worker(schedule={"kind": "manual", "seconds": 60})]))
self.assertIn("not valid for kind", str(ctx.exception))
def test_unknown_schedule_kind_is_refused(self):
with self.assertRaises(RegistryValidationError):
self.parse(_document(workers=[_worker(schedule={"kind": "hourly"})]))
class TestAtomicPersistence(_TempRegistryCase):
"""AC: atomic persistence."""
def test_save_then_load_round_trips(self):
registry = self.parse(_document())
save_registry(registry, self.path)
reloaded = load_registry(self.path)
self.assertEqual(
[worker_to_dict(w) for w in reloaded.workers],
[worker_to_dict(w) for w in registry.workers],
)
def test_save_leaves_no_temp_files_behind(self):
registry = self.parse(_document())
save_registry(registry, self.path)
save_registry(registry, self.path)
leftovers = [p.name for p in self.path.parent.iterdir() if p.name.startswith(".")]
self.assertEqual(leftovers, [])
def test_save_refuses_to_persist_an_invalid_document(self):
registry = self.parse(_document())
broken = WorkerRegistry(
version=registry.version,
revision=registry.revision,
updated_at=registry.updated_at,
providers=registry.providers,
# A worker whose provider is not declared in the registry.
workers=tuple(
type(worker)(**{**worker.__dict__, "provider": "vanished"})
for worker in registry.workers
),
source_path=self.path,
)
with self.assertRaises(RegistryValidationError):
save_registry(broken, self.path)
self.assertFalse(self.path.exists(), "invalid save must not create the file")
def test_document_shape_excludes_local_paths_but_api_shape_includes_it(self):
registry = self.parse(_document())
self.assertNotIn("source_path", registry_to_document(registry))
self.assertEqual(registry_to_dict(registry)["source_path"], str(self.path))
class TestVersioningAndRollback(_TempRegistryCase):
"""AC: versioning and rollback."""
def _seed(self) -> WorkerRegistry:
self.write(_document())
return load_registry(self.path)
def test_revision_increments_on_each_save(self):
registry = self._seed()
self.assertEqual(registry.revision, 1)
second = save_registry(registry, self.path)
self.assertEqual(second.revision, 2)
third = save_registry(second, self.path)
self.assertEqual(third.revision, 3)
def test_updated_at_is_refreshed_and_utc(self):
registry = self._seed()
saved = save_registry(registry, self.path)
self.assertRegex(saved.updated_at, r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$")
def test_superseded_revisions_are_retained(self):
registry = self._seed()
second = save_registry(registry, self.path)
save_registry(second, self.path)
self.assertEqual(list_revisions(self.path), (1, 2))
self.assertTrue(history_dir(self.path).is_dir())
def test_rollback_restores_prior_content_as_a_new_revision(self):
self.write(_document(workers=[_worker("original")]))
registry = load_registry(self.path)
changed = WorkerRegistry(
version=registry.version,
revision=registry.revision,
updated_at=registry.updated_at,
providers=registry.providers,
workers=(), # operator deletes every worker
source_path=self.path,
)
save_registry(changed, self.path)
self.assertEqual(load_registry(self.path).workers, ())
restored = rollback_to_revision(1, self.path)
self.assertEqual([w.id for w in restored.workers], ["original"])
# Append-only: the rollback publishes a new head rather than rewinding.
self.assertGreater(restored.revision, 2)
self.assertEqual([w.id for w in load_registry(self.path).workers], ["original"])
def test_rollback_to_unknown_revision_fails_closed(self):
self._seed()
with self.assertRaises(RegistryValidationError) as ctx:
rollback_to_revision(99, self.path)
self.assertIn("not retained", str(ctx.exception))
def test_revision_must_be_a_positive_integer(self):
for bad in (0, -1, "1", None):
with self.subTest(revision=bad):
with self.assertRaises(RegistryValidationError):
self.parse(_document(revision=bad))
def test_history_is_empty_before_any_save(self):
self.write(_document())
self.assertEqual(list_revisions(self.path), ())
if __name__ == "__main__":
unittest.main()
+80
View File
@@ -2,6 +2,7 @@
from __future__ import annotations
import uuid
from datetime import datetime, timezone
from starlette.applications import Starlette
@@ -19,6 +20,9 @@ from final_report_validator import FINAL_REPORT_TASK_KINDS
from webui.gated_actions import attempt_action, load_action_registry, preview_action
from webui.gated_action_views import render_actions_page
from webui import console_audit
from webui.console_authz import authorize, rbac_matrix, resolve_principal
from webui.console_redaction import redaction_policy
from webui.audit_validator import audit_report, audit_to_dict
from webui.audit_views import render_audit_page
from webui.lease_loader import load_lease_snapshot, snapshot_to_dict as lease_snapshot_to_dict
@@ -215,6 +219,49 @@ async def api_actions(_request: Request) -> JSONResponse:
return JSONResponse(load_action_registry().to_dict())
def _request_id() -> str:
return f"req-{uuid.uuid4().hex}"
def _audit_target(action_id: str, params: dict[str, object]) -> dict[str, object]:
"""Describe the action target for the audit record (never secrets)."""
if "pr_number" in params:
return {"kind": "pr", "ref": f"#{params['pr_number']}"}
if "issue_number" in params:
return {"kind": "issue", "ref": f"#{params['issue_number']}"}
if "branch_name" in params:
return {"kind": "branch", "ref": str(params["branch_name"])}
return {"kind": "unspecified", "ref": action_id}
def _authorize_request(
request: Request,
action_id: str,
params: dict[str, object],
*,
for_execution: bool,
result: str,
) -> dict[str, object]:
"""Resolve principal, decide, and audit. Returns the decision payload.
Phase 1 records the decision rather than enforcing it as the terminal
outcome: ``webui.gated_actions`` already fails closed for every action, so
this layer cannot loosen anything. Phase 2 enforces on this same decision.
"""
principal = resolve_principal(headers=dict(request.headers))
decision = authorize(action_id, principal, for_execution=for_execution)
console_audit.record_event(
action_id=action_id,
result=result,
decision=decision,
principal=principal,
target=_audit_target(action_id, params),
request_id=_request_id(),
detail=decision.detail,
)
return decision.to_dict()
async def api_action_preview(request: Request) -> JSONResponse:
action_id = request.path_params["action_id"]
params = dict(request.query_params)
@@ -224,6 +271,13 @@ async def api_action_preview(request: Request) -> JSONResponse:
result = preview_action(action_id, **params)
if "error" in result:
return JSONResponse(result, status_code=404)
result["authorization"] = _authorize_request(
request,
action_id,
params,
for_execution=False,
result=console_audit.RESULT_PREVIEWED,
)
return JSONResponse(result)
@@ -237,10 +291,31 @@ async def api_action_attempt(request: Request) -> JSONResponse:
if not isinstance(body, dict):
body = {}
result = attempt_action(action_id, **body)
authorization = _authorize_request(
request,
action_id,
body,
for_execution=True,
result=(
console_audit.RESULT_DENIED
if not result.get("success")
else console_audit.RESULT_ALLOWED
),
)
result["authorization"] = authorization
status = 403 if not result.get("success") else 200
return JSONResponse(result, status_code=status)
async def api_console_security_model(_request: Request) -> JSONResponse:
"""Read-only publication of the #633 authorization/redaction/audit model."""
return JSONResponse({
"rbac": rbac_matrix(),
"redaction": redaction_policy(),
"audit": console_audit.audit_policy(),
})
async def method_not_allowed(request: Request, _exc: Exception) -> Response:
path = request.url.path
if path in _AUDIT_MUTATION_PATHS and request.method == "POST":
@@ -291,6 +366,11 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
methods=["POST"],
),
Route("/api/leases", api_leases, methods=["GET"]),
Route(
"/api/console/security-model",
api_console_security_model,
methods=["GET"],
),
],
exception_handlers={405: method_not_allowed},
)
+281
View File
@@ -0,0 +1,281 @@
"""Console audit event schema, retention, and append-only sink (#633).
``gitea_audit`` records MCP-side *mutations*: which profile and Gitea user
performed which tool call. It carries no console actor, no identity source, no
correlation identifier, and no retention class, so it cannot answer the
question #633 exists to answer — *who sat at the console, what did they
attempt, and was it authorized?* An authorization denial is not a mutation and
would never appear there at all.
This module adds the console-side record. It does not replace ``gitea_audit``:
when a Phase 2 action eventually reaches MCP, both fire, correlated by
``correlation.request_id``.
Design constraints:
- **Redact before persist.** Every record passes through
``webui.console_redaction.redact_payload`` before serialization, so an
unredacted field is never durable.
- **Append-only.** Records are appended as JSON lines. Nothing here updates or
deletes; retention is metadata on each record, enforced by an operator-run
policy, never by silent rewriting.
- **Never raises.** Auditing must not break the request it describes. A failed
write returns ``False``.
- **Off by default.** With ``WEBUI_CONSOLE_AUDIT_LOG`` unset, events are still
*built* (so callers and tests see the schema) but nothing is written.
A record looks like this (synthetic values):
{"schema_version": 1, "event_id": "evt-0001",
"timestamp": "2026-07-22T10:16:42+00:00",
"actor": {"subject": "[email protected]", "role": "operator",
"identity_source": "access_proxy", "authenticated": true},
"action": "merge_pr", "action_class": "privileged",
"target": {"kind": "pr", "ref": "#123"},
"result": "denied", "reason_code": "insufficient_role",
"correlation": {"request_id": "req-abc", "session_id": null,
"mcp_task": "merge_pr", "mcp_permission": "gitea.pr.merge"},
"retention": {"class": "privileged", "days": 365,
"expires_at": "2027-07-22T10:16:42+00:00"},
"redacted": true}
Timestamps are timezone-aware ISO-8601 in UTC.
"""
from __future__ import annotations
import datetime
import json
import os
import uuid
from typing import Any
from webui import console_authz
from webui.console_redaction import redact_payload, scan_for_secrets
SCHEMA_VERSION = 1
AUDIT_LOG_ENV = "WEBUI_CONSOLE_AUDIT_LOG"
# Result vocabulary. ``denied`` is the one ``gitea_audit`` has no equivalent
# for: an authorization refusal never reaches the MCP layer.
RESULT_ALLOWED = "allowed"
RESULT_DENIED = "denied"
RESULT_PREVIEWED = "previewed"
RESULT_FAILED = "failed"
RESULT_SUCCEEDED = "succeeded"
RESULTS = frozenset(
{
RESULT_ALLOWED,
RESULT_DENIED,
RESULT_PREVIEWED,
RESULT_FAILED,
RESULT_SUCCEEDED,
}
)
# Retention classes and default lifetimes in days. Privileged and break-glass
# records outlive routine ones because they are what an incident review needs.
RETENTION_STANDARD = "standard"
RETENTION_PRIVILEGED = "privileged"
RETENTION_BREAK_GLASS = "break_glass"
RETENTION_DAYS: dict[str, int] = {
RETENTION_STANDARD: 90,
RETENTION_PRIVILEGED: 365,
RETENTION_BREAK_GLASS: 730,
}
# Fields every record must carry. Asserted by the test suite so a future edit
# cannot quietly drop one.
REQUIRED_FIELDS: tuple[str, ...] = (
"schema_version",
"event_id",
"timestamp",
"actor",
"action",
"action_class",
"target",
"result",
"reason_code",
"correlation",
"retention",
"redacted",
)
REQUIRED_ACTOR_FIELDS: tuple[str, ...] = (
"subject",
"role",
"identity_source",
"authenticated",
)
REQUIRED_CORRELATION_FIELDS: tuple[str, ...] = (
"request_id",
"session_id",
"mcp_task",
"mcp_permission",
)
def audit_log_path() -> str | None:
"""Configured sink path, or ``None`` when console auditing is off."""
return (os.environ.get(AUDIT_LOG_ENV) or "").strip() or None
def audit_enabled() -> bool:
return audit_log_path() is not None
def retention_class_for(action: console_authz.ConsoleAction | None) -> str:
"""Classify retention from the action, defaulting to the longest-lived.
An unknown action is treated as privileged rather than standard: for a
safety control the conservative direction is to keep the record longer.
"""
if action is None:
return RETENTION_PRIVILEGED
if action.break_glass:
return RETENTION_BREAK_GLASS
if action.privileged:
return RETENTION_PRIVILEGED
return RETENTION_STANDARD
def _retention_block(
retention_class: str, now: datetime.datetime
) -> dict[str, Any]:
days = RETENTION_DAYS.get(
retention_class, RETENTION_DAYS[RETENTION_PRIVILEGED]
)
return {
"class": retention_class,
"days": days,
"expires_at": (now + datetime.timedelta(days=days)).isoformat(),
}
def build_event(
*,
action_id: str,
result: str,
decision: console_authz.AuthorizationDecision | None = None,
principal: console_authz.Principal | None = None,
target: dict[str, Any] | None = None,
reason_code: str | None = None,
request_id: str | None = None,
session_id: str | None = None,
detail: str | None = None,
metadata: dict[str, Any] | None = None,
now: datetime.datetime | None = None,
event_id: str | None = None,
) -> dict[str, Any]:
"""Build one redacted, JSON-able console audit record.
Redaction runs here rather than at write time so an in-memory record handed
to a template or an API response is already clean.
"""
ts = now or datetime.datetime.now(datetime.timezone.utc)
action = console_authz.get_action(action_id)
who = principal or (
decision.principal if decision else console_authz.ANONYMOUS
)
resolved_result = result if result in RESULTS else RESULT_FAILED
resolved_reason = reason_code or (
decision.reason_code if decision else "unspecified"
)
retention_class = retention_class_for(action)
event: dict[str, Any] = {
"schema_version": SCHEMA_VERSION,
"event_id": event_id or f"evt-{uuid.uuid4().hex}",
"timestamp": ts.isoformat(),
"actor": who.to_dict(),
"action": action_id,
"action_class": action.action_class if action else "unknown",
"target": dict(target or {}),
"result": resolved_result,
"reason_code": resolved_reason,
"correlation": {
"request_id": request_id,
"session_id": session_id,
"mcp_task": action.task_key if action else None,
"mcp_permission": action.mcp_permission if action else None,
},
"retention": _retention_block(retention_class, ts),
"redacted": True,
"detail": detail,
"metadata": dict(metadata or {}),
}
if decision is not None:
# Deliberately *not* named "authorization": ``gitea_audit`` treats that
# substring as a secret key hint (it matches the HTTP Authorization
# header) and would replace this whole block with the placeholder.
event["decision"] = {
"allowed": decision.allowed,
"required_role": decision.required_role,
"requires_confirmation": decision.requires_confirmation,
"dual_control": decision.dual_control,
"break_glass": decision.break_glass,
"execution_enabled": decision.execution_enabled,
}
redacted = redact_payload(event)
if not isinstance(redacted, dict): # pragma: no cover - defensive
return {"schema_version": SCHEMA_VERSION, "redacted": True}
return redacted
def write_event(event: dict[str, Any], path: str | None = None) -> bool:
"""Append *event* as one JSON line. Never raises.
Returns ``True`` when a line was written, ``False`` when auditing is off or
the write failed. A record that still trips a secret detector is dropped
rather than persisted.
"""
sink = path or audit_log_path()
if not sink:
return False
try:
if scan_for_secrets(event):
return False
line = json.dumps(event, default=str, sort_keys=True)
with open(sink, "a", encoding="utf-8") as handle:
handle.write(line + "\n")
return True
except Exception:
return False
def record_event(**kwargs: Any) -> dict[str, Any]:
"""Build and persist one record; return the record either way.
Callers get the record back so it can be surfaced in a response or a test
regardless of whether a sink is configured.
"""
event = build_event(**kwargs)
written = write_event(event)
return {"event": event, "written": written}
def audit_policy() -> dict[str, Any]:
"""Machine-readable audit schema and retention defaults (never secrets)."""
return {
"schema_version": SCHEMA_VERSION,
"required_fields": list(REQUIRED_FIELDS),
"required_actor_fields": list(REQUIRED_ACTOR_FIELDS),
"required_correlation_fields": list(REQUIRED_CORRELATION_FIELDS),
"results": sorted(RESULTS),
"retention_defaults_days": dict(RETENTION_DAYS),
"sink_env": AUDIT_LOG_ENV,
"enabled": audit_enabled(),
"append_only": True,
"redact_before_persist": True,
"timestamp_format": "ISO-8601, timezone-aware, UTC",
"relationship_to_mcp_audit": (
"webui.console_audit records console intent and authorization "
"outcomes; gitea_audit records MCP mutations. A Phase 2 action "
"emits both, correlated by correlation.request_id."
),
}
+537
View File
@@ -0,0 +1,537 @@
"""Console authorization and RBAC model (#633, Phase 1).
The read-only MVP (#426#436) ships with no authentication: protection comes
from network placement alone (#435). That is adequate while every route is a
GET, and inadequate the moment Phase 2 wires a gated write. This module is the
authorization model those writes must go through, landed *before* any of them
exists so no write can be added without an authority to check against.
Phase 1 scope is the model itself: identity resolution, the role matrix, the
privileged-action list, and a fail-closed :func:`authorize`. It deliberately
does **not** enable any write. ``webui.gated_actions`` stays globally disabled,
so an allow decision here is necessary but never sufficient.
Two invariants hold for every caller:
- **Default deny.** An unrecognised action, an unknown role, or an absent
principal denies. There is no implicit allow branch and no "unless" clause.
- **Authorization is not execution.** :func:`authorize` returns a decision
record. It never calls MCP, never mutates, and never consults credentials.
"""
from __future__ import annotations
import json
import os
from dataclasses import asdict, dataclass, field
from typing import Any
from task_capability_map import required_permission, required_role
# --- Roles ------------------------------------------------------------------
# Ordered least to most authority. Higher ranks inherit every lower rank's
# permitted actions; the matrix below is expressed as a minimum required rank.
VIEWER = "viewer"
OPERATOR = "operator"
CONTROLLER = "controller"
ADMIN = "admin"
ROLE_ORDER: tuple[str, ...] = (VIEWER, OPERATOR, CONTROLLER, ADMIN)
_ROLE_RANK: dict[str, int] = {role: idx for idx, role in enumerate(ROLE_ORDER)}
ROLE_DESCRIPTIONS: dict[str, str] = {
VIEWER: "Read every console view. No write, ever, in any phase.",
OPERATOR: "Viewer, plus author-class work: claim, comment, open a PR.",
CONTROLLER: "Operator, plus reviewer/merger-class decisions on a PR.",
ADMIN: "Controller, plus destructive and policy-editing actions.",
}
# --- Identity sources -------------------------------------------------------
IDENTITY_NONE = "none"
IDENTITY_LOCAL_DEV = "local_dev"
IDENTITY_ACCESS_PROXY = "access_proxy"
IDENTITY_SOURCES: dict[str, dict[str, Any]] = {
IDENTITY_NONE: {
"description": (
"No authentication configured. Every request is anonymous and "
"capped at viewer. This is the MVP default and the only mode "
"whose safety rests entirely on network placement (#435)."
),
"authenticated": False,
"safe_for_shared_host": False,
"phase_available": 1,
},
IDENTITY_LOCAL_DEV: {
"description": (
"Developer-supplied principal read from the environment. INSECURE: "
"the subject and role are asserted, never verified. Loopback only."
),
"authenticated": True,
"safe_for_shared_host": False,
"phase_available": 1,
},
IDENTITY_ACCESS_PROXY: {
"description": (
"Subject asserted by a trusted access proxy (Cloudflare Access, "
"WARP, or an org VPN portal) via a verified request header. The "
"proxy performs authentication; the console performs authorization."
),
"authenticated": True,
"safe_for_shared_host": True,
"phase_available": 2,
},
}
# Environment configuration. All are read server-side and never rendered.
AUTH_MODE_ENV = "WEBUI_AUTH_MODE"
DEV_SUBJECT_ENV = "WEBUI_DEV_SUBJECT"
DEV_ROLE_ENV = "WEBUI_DEV_ROLE"
ROLE_MAP_ENV = "WEBUI_ROLE_MAP"
REQUIRE_PROBE_AUTH_ENV = "WEBUI_REQUIRE_PROBE_AUTH"
ACCESS_SUBJECT_HEADER = "cf-access-authenticated-user-email"
# --- Action classes ---------------------------------------------------------
CLASS_READ = "read"
CLASS_WRITE = "gated_write"
CLASS_PRIVILEGED = "privileged"
CLASS_DESTRUCTIVE = "destructive"
# --- Privileged action list -------------------------------------------------
# ``task_key`` ties each console action back to ``task_capability_map``, so the
# console cannot invent an authority the MCP layer does not already define.
@dataclass(frozen=True)
class ConsoleAction:
"""One console action and the authority required to invoke it."""
action_id: str
task_key: str
action_class: str
minimum_role: str
requires_confirmation: bool
dual_control: bool
break_glass: bool
phase: int
summary: str
@property
def mcp_permission(self) -> str:
return required_permission(self.task_key)
@property
def mcp_role(self) -> str:
return required_role(self.task_key)
@property
def privileged(self) -> bool:
return self.action_class in {CLASS_PRIVILEGED, CLASS_DESTRUCTIVE}
def to_dict(self) -> dict[str, Any]:
data = asdict(self)
data["mcp_permission"] = self.mcp_permission
data["mcp_role"] = self.mcp_role
data["privileged"] = self.privileged
return data
_ACTION_SPECS: tuple[ConsoleAction, ...] = (
ConsoleAction(
action_id="claim_issue",
task_key="claim_issue",
action_class=CLASS_WRITE,
minimum_role=OPERATOR,
requires_confirmation=True,
dual_control=False,
break_glass=False,
phase=2,
summary="Apply status:in-progress to an issue.",
),
ConsoleAction(
action_id="comment_issue",
task_key="comment_issue",
action_class=CLASS_WRITE,
minimum_role=OPERATOR,
requires_confirmation=True,
dual_control=False,
break_glass=False,
phase=2,
summary="Post an issue comment.",
),
ConsoleAction(
action_id="create_issue",
task_key="create_issue",
action_class=CLASS_WRITE,
minimum_role=OPERATOR,
requires_confirmation=True,
dual_control=False,
break_glass=False,
phase=2,
summary="Open a new tracking issue.",
),
ConsoleAction(
action_id="comment_pr",
task_key="comment_pr",
action_class=CLASS_WRITE,
minimum_role=OPERATOR,
requires_confirmation=True,
dual_control=False,
break_glass=False,
phase=2,
summary="Post a PR thread comment.",
),
ConsoleAction(
action_id="create_pr",
task_key="create_pr",
action_class=CLASS_WRITE,
minimum_role=OPERATOR,
requires_confirmation=True,
dual_control=False,
break_glass=False,
phase=2,
summary="Open a PR from a locked feature branch.",
),
ConsoleAction(
action_id="review_pr",
task_key="review_pr",
action_class=CLASS_PRIVILEGED,
minimum_role=CONTROLLER,
requires_confirmation=True,
dual_control=False,
break_glass=False,
phase=3,
summary="Submit an approve / request-changes verdict.",
),
ConsoleAction(
action_id="close_pr",
task_key="close_pr",
action_class=CLASS_PRIVILEGED,
minimum_role=CONTROLLER,
requires_confirmation=True,
dual_control=False,
break_glass=False,
phase=3,
summary="Close a pull request without merging.",
),
ConsoleAction(
action_id="merge_pr",
task_key="merge_pr",
action_class=CLASS_PRIVILEGED,
minimum_role=CONTROLLER,
requires_confirmation=True,
dual_control=True,
break_glass=True,
phase=3,
summary="Merge an approved pull request.",
),
ConsoleAction(
action_id="delete_branch",
task_key="delete_branch",
action_class=CLASS_DESTRUCTIVE,
minimum_role=ADMIN,
requires_confirmation=True,
dual_control=True,
break_glass=True,
phase=3,
summary="Remove a remote feature branch.",
),
)
ACTIONS: dict[str, ConsoleAction] = {a.action_id: a for a in _ACTION_SPECS}
def privileged_actions() -> tuple[ConsoleAction, ...]:
"""Actions requiring dual control, break-glass, or controller+ authority."""
return tuple(a for a in _ACTION_SPECS if a.privileged)
def get_action(action_id: str) -> ConsoleAction | None:
return ACTIONS.get(action_id)
# --- Principals -------------------------------------------------------------
@dataclass(frozen=True)
class Principal:
"""Who is making a request, and how strongly that is known."""
subject: str
role: str
identity_source: str
authenticated: bool
warnings: tuple[str, ...] = field(default_factory=tuple)
@property
def rank(self) -> int:
return _ROLE_RANK.get(self.role, -1)
def to_dict(self) -> dict[str, Any]:
return {
"subject": self.subject,
"role": self.role,
"identity_source": self.identity_source,
"authenticated": self.authenticated,
"warnings": list(self.warnings),
}
ANONYMOUS = Principal(
subject="anonymous",
role=VIEWER,
identity_source=IDENTITY_NONE,
authenticated=False,
warnings=("No authentication configured; capped at viewer.",),
)
def auth_mode(env: dict[str, str] | None = None) -> str:
"""Resolve the configured identity source, defaulting to ``none``."""
source = env if env is not None else os.environ
raw = (source.get(AUTH_MODE_ENV) or "").strip().lower().replace("-", "_")
if raw in IDENTITY_SOURCES:
return raw
return IDENTITY_NONE
def _role_map(env: dict[str, str]) -> dict[str, str]:
"""Parse ``WEBUI_ROLE_MAP`` (JSON subject→role). Invalid config yields {}."""
raw = (env.get(ROLE_MAP_ENV) or "").strip()
if not raw:
return {}
try:
parsed = json.loads(raw)
except Exception:
return {}
if not isinstance(parsed, dict):
return {}
return {
str(k): str(v).strip().lower()
for k, v in parsed.items()
if str(v).strip().lower() in _ROLE_RANK
}
def resolve_principal(
headers: dict[str, str] | None = None,
env: dict[str, str] | None = None,
) -> Principal:
"""Resolve the requesting principal. Unknown or unconfigured → anonymous.
Never raises and never trusts a client-supplied role: the role always comes
from server-side configuration keyed by the resolved subject.
"""
source_env = dict(env) if env is not None else dict(os.environ)
lowered = {str(k).lower(): str(v) for k, v in (headers or {}).items()}
mode = auth_mode(source_env)
if mode == IDENTITY_LOCAL_DEV:
subject = (source_env.get(DEV_SUBJECT_ENV) or "").strip()
if not subject:
return ANONYMOUS
role = (source_env.get(DEV_ROLE_ENV) or VIEWER).strip().lower()
if role not in _ROLE_RANK:
role = VIEWER
return Principal(
subject=subject,
role=role,
identity_source=IDENTITY_LOCAL_DEV,
authenticated=True,
warnings=(
"local-dev identity is asserted, not verified; never use "
"outside loopback.",
),
)
if mode == IDENTITY_ACCESS_PROXY:
subject = (lowered.get(ACCESS_SUBJECT_HEADER) or "").strip()
if not subject:
# Proxy mode with no proxy header means the request did not
# traverse the proxy. Fail closed rather than trust it.
return ANONYMOUS
role = _role_map(source_env).get(subject, VIEWER)
return Principal(
subject=subject,
role=role,
identity_source=IDENTITY_ACCESS_PROXY,
authenticated=True,
)
return ANONYMOUS
def probe_auth_required(env: dict[str, str] | None = None) -> bool:
"""Whether non-public probes must be authenticated. Default False.
#633 requires the console to *fail closed on missing auth for non-public
health probes if configured*. The default stays off so the MVP ``/health``
contract is unchanged; an operator opts in explicitly.
"""
source = env if env is not None else os.environ
return (source.get(REQUIRE_PROBE_AUTH_ENV) or "").strip().lower() in {
"1",
"true",
"yes",
}
# --- Authorization ----------------------------------------------------------
DENY_UNKNOWN_ACTION = "unknown_action"
DENY_UNAUTHENTICATED = "unauthenticated"
DENY_INSUFFICIENT_ROLE = "insufficient_role"
DENY_UNKNOWN_ROLE = "unknown_role"
DENY_PHASE_NOT_ACTIVE = "phase_not_active"
ALLOW_PREVIEW = "allowed_preview_only"
# Phase 1 is the only active console phase. Phase 2 opens gated writes and is
# gated on this model landing; nothing here enables it.
ACTIVE_PHASE = 1
@dataclass(frozen=True)
class AuthorizationDecision:
"""Result of an authorization check. Never an execution grant."""
allowed: bool
reason_code: str
detail: str
action_id: str
principal: Principal
required_role: str | None = None
action_class: str | None = None
requires_confirmation: bool = False
dual_control: bool = False
break_glass: bool = False
execution_enabled: bool = False
def to_dict(self) -> dict[str, Any]:
return {
"allowed": self.allowed,
"reason_code": self.reason_code,
"detail": self.detail,
"action_id": self.action_id,
"principal": self.principal.to_dict(),
"required_role": self.required_role,
"action_class": self.action_class,
"requires_confirmation": self.requires_confirmation,
"dual_control": self.dual_control,
"break_glass": self.break_glass,
"execution_enabled": self.execution_enabled,
"active_phase": ACTIVE_PHASE,
}
def authorize(
action_id: str,
principal: Principal | None = None,
*,
for_execution: bool = False,
) -> AuthorizationDecision:
"""Decide whether *principal* may invoke *action_id*. Deny by default.
``for_execution`` distinguishes a read-only preview from a real invocation.
Even an allowed decision reports ``execution_enabled=False`` while the
console is in Phase 1, so no caller can read an allow as permission to
mutate.
"""
who = principal if principal is not None else ANONYMOUS
action = get_action(action_id)
if action is None:
return AuthorizationDecision(
allowed=False,
reason_code=DENY_UNKNOWN_ACTION,
detail=f"No console action registered as {action_id!r}.",
action_id=action_id,
principal=who,
)
base: dict[str, Any] = {
"action_id": action_id,
"principal": who,
"required_role": action.minimum_role,
"action_class": action.action_class,
"requires_confirmation": action.requires_confirmation,
"dual_control": action.dual_control,
"break_glass": action.break_glass,
"execution_enabled": False,
}
if not who.authenticated:
return AuthorizationDecision(
allowed=False,
reason_code=DENY_UNAUTHENTICATED,
detail=(
"Write actions require an authenticated principal; this "
"request is anonymous."
),
**base,
)
if who.rank < 0:
return AuthorizationDecision(
allowed=False,
reason_code=DENY_UNKNOWN_ROLE,
detail=f"Role {who.role!r} is not in the console role matrix.",
**base,
)
if who.rank < _ROLE_RANK[action.minimum_role]:
return AuthorizationDecision(
allowed=False,
reason_code=DENY_INSUFFICIENT_ROLE,
detail=(
f"Action {action_id!r} requires {action.minimum_role!r}; "
f"principal holds {who.role!r}."
),
**base,
)
if for_execution and action.phase > ACTIVE_PHASE:
return AuthorizationDecision(
allowed=False,
reason_code=DENY_PHASE_NOT_ACTIVE,
detail=(
f"Action {action_id!r} belongs to phase {action.phase}; the "
f"console is in phase {ACTIVE_PHASE}. Execution is not wired."
),
**base,
)
return AuthorizationDecision(
allowed=True,
reason_code=ALLOW_PREVIEW,
detail=(
"Principal holds the required role. Preview only — execution "
"remains disabled until the Phase 2 action framework ships."
),
**base,
)
def rbac_matrix() -> dict[str, Any]:
"""Machine-readable RBAC matrix and privileged-action list."""
return {
"model_version": 1,
"active_phase": ACTIVE_PHASE,
"roles": [
{
"role": role,
"rank": _ROLE_RANK[role],
"description": ROLE_DESCRIPTIONS[role],
"permitted_actions": sorted(
a.action_id
for a in _ACTION_SPECS
if _ROLE_RANK[role] >= _ROLE_RANK[a.minimum_role]
),
}
for role in ROLE_ORDER
],
"identity_sources": IDENTITY_SOURCES,
"actions": [a.to_dict() for a in _ACTION_SPECS],
"privileged_actions": [a.action_id for a in privileged_actions()],
"default_decision": "deny",
"execution_enabled": False,
}
+169
View File
@@ -0,0 +1,169 @@
"""Secret redaction policy for every console surface (#633).
The MVP already redacts MCP-side mutation records through ``gitea_audit``.
This module is the console-facing policy: one redaction pass applied to API
payloads, rendered HTML, log lines, and audit records *before* they leave the
server or reach persistent storage.
Design constraints:
- **Reuse, never fork.** ``gitea_audit.redact`` remains the authority for
secret-looking dict keys, ``Authorization`` material, and raw URLs. This
module runs that pass first and then applies console-specific patterns for
keychain references, key/value assignments, private-key blocks, and JWTs.
- **Never raises.** Redaction is a safety control; a malformed payload must
degrade to a redacted placeholder rather than propagate an exception.
- **Redact before persist.** ``webui.console_audit`` calls this module before
writing, so an unredacted record is never durable.
"""
from __future__ import annotations
import json
import re
from typing import Any
import gitea_audit
REDACTED = gitea_audit.REDACTED
# Console-specific patterns applied after the shared ``gitea_audit`` pass.
# Each keeps the identifying key so an operator can still tell *what* was
# removed, and replaces only the secret run itself.
_KEYCHAIN_REF = re.compile(r"(?i)\bkeychain:[\w.\-/@]+")
_KEYCHAIN_CMD = re.compile(
r"(?i)\bsecurity\s+find-(?:generic|internet)-password\b[^\n]*"
)
_ASSIGNMENT = re.compile(
r"(?i)\b(token|password|passwd|secret|api[_-]?key|access[_-]?key|"
r"client[_-]?secret|private[_-]?key)\b(\s*[:=]\s*)"
r"(\"[^\"]*\"|'[^']*'|\S+)"
)
_ENV_ASSIGNMENT = re.compile(
r"(?i)\b(GITEA_(?:TOKEN|PASS|PASSWORD)[A-Z0-9_]*)(\s*=\s*)"
r"(\"[^\"]*\"|'[^']*'|\S+)"
)
_PRIVATE_KEY_BLOCK = re.compile(
r"-----BEGIN [A-Z ]*PRIVATE KEY-----.*?-----END [A-Z ]*PRIVATE KEY-----",
re.S,
)
_JWT = re.compile(
r"\beyJ[A-Za-z0-9_\-]{8,}\.[A-Za-z0-9_\-]{8,}\.[A-Za-z0-9_\-]{8,}\b"
)
# Shapes that mean a payload still carries a secret. ``scan_for_secrets`` uses
# these to assert a surface is clean.
_DETECTORS: tuple[tuple[str, re.Pattern[str]], ...] = (
("keychain_reference", _KEYCHAIN_REF),
("keychain_command", _KEYCHAIN_CMD),
("credential_assignment", _ASSIGNMENT),
("credential_env_assignment", _ENV_ASSIGNMENT),
("private_key_block", _PRIVATE_KEY_BLOCK),
("json_web_token", _JWT),
("bearer_credential", re.compile(r"(?i)\b(?:bearer|basic)\s+\S{8,}")),
)
def _mask_assignment(match: re.Match[str]) -> str:
"""Keep the key and separator, replace the value."""
return f"{match.group(1)}{match.group(2)}{REDACTED}"
def redact_text(text: Any) -> Any:
"""Redact secret material from a single string.
Non-strings are returned unchanged so this is safe to map over mixed
payloads. Runs the shared ``gitea_audit`` pass first, then the
console-specific patterns.
"""
if not isinstance(text, str) or not text:
return text
try:
out = gitea_audit.redact(text)
if not isinstance(out, str): # defensive; redact() returns str for str
return REDACTED
out = _PRIVATE_KEY_BLOCK.sub(f"{REDACTED}_PRIVATE_KEY", out)
out = _ENV_ASSIGNMENT.sub(_mask_assignment, out)
out = _ASSIGNMENT.sub(_mask_assignment, out)
out = _KEYCHAIN_CMD.sub(f"{REDACTED}_KEYCHAIN_COMMAND", out)
out = _KEYCHAIN_REF.sub(f"{REDACTED}_KEYCHAIN_REF", out)
out = _JWT.sub(f"{REDACTED}_JWT", out)
return out
except Exception:
# Fail closed: an unredactable string is dropped rather than emitted raw.
return REDACTED
def redact_payload(value: Any) -> Any:
"""Recursively redact a JSON-able payload for any console surface.
Secret-looking dict keys are replaced wholesale by the shared
``gitea_audit`` policy; every remaining string is run through
:func:`redact_text`.
"""
try:
shared = gitea_audit.redact(value)
except Exception:
return REDACTED
return _walk(shared)
def _walk(value: Any) -> Any:
if isinstance(value, dict):
return {k: _walk(v) for k, v in value.items()}
if isinstance(value, (list, tuple)):
return [_walk(v) for v in value]
if isinstance(value, str):
return redact_text(value)
return value
def scan_for_secrets(value: Any) -> list[str]:
"""Return detector names that still match *value* after serialization.
Used to assert an outbound payload or rendered page is clean. An empty
list means no known secret shape was found. Already-redacted hits are not
findings.
"""
if isinstance(value, str):
text = value
else:
try:
text = json.dumps(value, default=str)
except Exception:
text = str(value)
findings: list[str] = []
for name, pattern in _DETECTORS:
for match in pattern.finditer(text):
if REDACTED in match.group(0):
continue
findings.append(name)
break
return findings
def redaction_policy() -> dict[str, Any]:
"""Machine-readable statement of the redaction rules (never secrets)."""
return {
"policy_version": 1,
"applies_to": [
"json_api_responses",
"rendered_html",
"server_logs",
"audit_records",
],
"ordering": "shared gitea_audit pass, then console patterns",
"redact_before_persist": True,
"shared_rules": {
"source": "gitea_audit.redact",
"secret_key_hints": list(gitea_audit._SECRET_KEY_HINTS),
"secret_value_prefixes": list(gitea_audit._SECRET_VALUE_PREFIXES),
"urls": "credentials, secret query parameters, and real hosts redacted",
},
"console_rules": [
{"name": name, "pattern": pattern.pattern}
for name, pattern in _DETECTORS
],
"placeholder": REDACTED,
"failure_mode": "fail closed — unredactable values become the placeholder",
}
+57
View File
@@ -0,0 +1,57 @@
{
"version": 1,
"revision": 1,
"updated_at": "2026-07-22T00:00:00Z",
"providers": [
{
"id": "claude",
"display_name": "Claude",
"vendor": "Anthropic",
"executable": "claude",
"available": true,
"models": [
"claude-opus-4-8",
"claude-sonnet-5",
"claude-haiku-4-5-20251001"
],
"notes": "Model list is a declaration. Live enumeration and version inspection belong to the provider adapter framework (#800)."
},
{
"id": "grok",
"display_name": "Grok",
"vendor": "xAI",
"executable": "grok",
"available": true,
"models": [],
"notes": "Models enumerated by the provider adapter (#800); not declared here."
},
{
"id": "codex",
"display_name": "Codex",
"vendor": "OpenAI",
"executable": "codex",
"available": true,
"models": [],
"notes": "Models enumerated by the provider adapter (#800); not declared here."
},
{
"id": "agy",
"display_name": "AGY",
"vendor": "Antigravity",
"executable": "agy",
"available": true,
"models": [],
"notes": "MCP allowlist gating applies to this provider; confirm server-side allowlist before configuring a worker."
},
{
"id": "kimi-k",
"display_name": "Kimi K",
"vendor": "Moonshot AI",
"executable": "kimi",
"available": true,
"models": [],
"notes": "Provider id is kimi-k; the executable on PATH is kimi. Models enumerated by the provider adapter (#800)."
}
],
"workers": []
}
+1 -33
View File
@@ -8,27 +8,7 @@ from dataclasses import dataclass
from pathlib import Path
from typing import Any
_FORBIDDEN_EXACT_KEYS = frozenset({
"token",
"password",
"secret",
"credential",
"auth",
"api_key",
"api-key",
})
_FORBIDDEN_KEY_PREFIXES = ("auth_", "api_key_", "api-key_")
_FORBIDDEN_KEY_SUFFIXES = ("_token", "_secret", "_password", "_credential", "_auth")
def _is_forbidden_key(key: str) -> bool:
lowered = key.lower()
if lowered in _FORBIDDEN_EXACT_KEYS:
return True
return (
lowered.startswith(_FORBIDDEN_KEY_PREFIXES)
or lowered.endswith(_FORBIDDEN_KEY_SUFFIXES)
)
from webui.registry_safety import reject_credential_keys as _reject_credential_keys
_REQUIRED_PROJECT_FIELDS = (
"id",
@@ -79,18 +59,6 @@ def default_registry_path() -> Path:
return (Path(__file__).resolve().parent / "data" / "projects.registry.json").resolve()
def _reject_credential_keys(obj: Any, *, path: str = "") -> None:
if isinstance(obj, dict):
for key, value in obj.items():
key_path = f"{path}.{key}" if path else key
if _is_forbidden_key(key):
raise ValueError(f"registry must not store credentials ({key_path})")
_reject_credential_keys(value, path=key_path)
elif isinstance(obj, list):
for index, item in enumerate(obj):
_reject_credential_keys(item, path=f"{path}[{index}]")
def _parse_onboarding(raw: list[dict[str, Any]] | None) -> tuple[OnboardingStep, ...]:
if not raw:
return ()
+52
View File
@@ -0,0 +1,52 @@
"""Shared credential-rejection guard for web UI registries (#427, #798).
Registries are operator-editable declarative files that the web UI loads and,
for the worker registry, writes back. None of them may ever carry a secret:
credentials belong in the keychain and reach worker processes through
environment injection, never through a file the browser layer can read.
The check is structural rather than value-based on purpose. A value scanner has
to guess what a secret looks like; a key scanner refuses the *shape* of a
credential field, so an operator cannot introduce one by accident and a later
loader cannot silently pass one through.
"""
from __future__ import annotations
from typing import Any
_FORBIDDEN_EXACT_KEYS = frozenset({
"token",
"password",
"secret",
"credential",
"auth",
"api_key",
"api-key",
})
_FORBIDDEN_KEY_PREFIXES = ("auth_", "api_key_", "api-key_")
_FORBIDDEN_KEY_SUFFIXES = ("_token", "_secret", "_password", "_credential", "_auth")
def is_forbidden_key(key: str) -> bool:
"""Return True when *key* names a credential field."""
lowered = key.lower()
if lowered in _FORBIDDEN_EXACT_KEYS:
return True
return (
lowered.startswith(_FORBIDDEN_KEY_PREFIXES)
or lowered.endswith(_FORBIDDEN_KEY_SUFFIXES)
)
def reject_credential_keys(obj: Any, *, path: str = "", subject: str = "registry") -> None:
"""Raise ValueError when *obj* carries a credential-shaped key at any depth."""
if isinstance(obj, dict):
for key, value in obj.items():
key_path = f"{path}.{key}" if path else key
if is_forbidden_key(key):
raise ValueError(f"{subject} must not store credentials ({key_path})")
reject_credential_keys(value, path=key_path, subject=subject)
elif isinstance(obj, list):
for index, item in enumerate(obj):
reject_credential_keys(item, path=f"{path}[{index}]", subject=subject)
+647
View File
@@ -0,0 +1,647 @@
"""Declarative worker registry and configuration schema (#798, epic #797).
The registry is the single source of truth for the scheduled multi-LLM worker
fleet. It is a versioned JSON document holding two *separate* entity kinds:
* **Providers** — the LLM runtimes a worker can be built on (Claude, Grok,
Codex, AGY, Kimi K). A provider describes the runtime itself: vendor,
executable name, models it can serve, and whether it is available on this
machine. Providers exist whether or not any worker uses them.
* **Workers** — a configured *instance*: one provider, one model, one project,
one role, one MCP namespace/profile, one workflow, one schedule. Several
workers may share a provider; a worker naming an undeclared provider is
refused.
Keeping them separate is what lets #799 list all five providers even when a
provider currently has no configured worker, and it stops provider facts from
being copied into (and drifting across) every worker record.
Scope boundary. This module owns the data model, its validation, and its
persistence. It does **not** schedule anything, launch anything, probe provider
executables, or serve HTTP. Loading a registry never touches a process; the
live fields a dashboard wants (PID, elapsed time, next run) are derived
elsewhere (#799, #801, #803, #804) from these declarations.
Safety invariants:
* No credential may be stored (:mod:`webui.registry_safety`), so the registry
stays safe to render and to hand to a browser layer.
* Validation fails closed. Unknown fields are refused rather than ignored, so a
typo cannot silently disable a timeout or a role binding.
* Writes are atomic and every superseded document is retained as a numbered
revision, so a bad edit is recoverable by rollback rather than hand-repair.
"""
from __future__ import annotations
import json
import os
import re
import tempfile
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from webui.registry_safety import reject_credential_keys
SCHEMA_VERSION = 1
#: Roles a worker may hold. These mirror the sanctioned MCP role kinds; a
#: worker may not invent one, because the role selects the namespace/profile
#: whose capability gates constrain it.
ALLOWED_ROLES = ("author", "reviewer", "merger", "reconciler", "cleanup")
#: Scheduler backends the registry can describe. ``manual`` means the worker is
#: only ever started on request and has no recurring trigger.
ALLOWED_SCHEDULER_KINDS = ("launchd", "manual")
#: Schedule kinds. Next-run computation belongs to #803; this module only
#: guarantees the declaration is well formed.
ALLOWED_SCHEDULE_KINDS = ("interval", "cron", "manual")
_REQUIRED_PROVIDER_FIELDS = ("id", "display_name", "vendor", "executable", "available")
_OPTIONAL_PROVIDER_FIELDS = ("models", "notes")
_REQUIRED_WORKER_FIELDS = (
"id",
"display_name",
"provider",
"model",
"project",
"role",
"namespace",
"profile",
"workflow",
"schedule",
"timeout_seconds",
"enabled",
"scheduler",
)
_OPTIONAL_WORKER_FIELDS = ("notes",)
_ID_RE = re.compile(r"^[a-z0-9][a-z0-9._-]*$")
#: Guards against an operator writing a timeout that would let a worker hold a
#: lease effectively forever. 24h is far above any sanctioned cycle.
_MAX_TIMEOUT_SECONDS = 86_400
#: How many superseded revisions to retain beside the live file.
_HISTORY_LIMIT = 20
_TOP_LEVEL_FIELDS = frozenset({"version", "revision", "updated_at", "providers", "workers"})
class RegistryValidationError(ValueError):
"""Raised when a registry document violates the schema."""
@dataclass(frozen=True)
class ProviderRecord:
id: str
display_name: str
vendor: str
executable: str
available: bool
models: tuple[str, ...]
notes: str
@dataclass(frozen=True)
class ScheduleSpec:
kind: str
#: Set for ``interval`` schedules.
seconds: int | None
#: Set for ``cron`` schedules — a five-field crontab expression.
expression: str | None
@dataclass(frozen=True)
class SchedulerSpec:
kind: str
#: LaunchAgent label; required for ``launchd``, absent for ``manual``.
label: str | None
@dataclass(frozen=True)
class WorkerRecord:
id: str
display_name: str
provider: str
model: str
project: str
role: str
namespace: str
profile: str
workflow: str
schedule: ScheduleSpec
timeout_seconds: int
enabled: bool
scheduler: SchedulerSpec
notes: str
@dataclass(frozen=True)
class WorkerRegistry:
version: int
revision: int
updated_at: str
providers: tuple[ProviderRecord, ...]
workers: tuple[WorkerRecord, ...]
source_path: Path
# ── paths ────────────────────────────────────────────────────────────────────
def default_registry_path() -> Path:
"""Location of the packaged worker registry, overridable for tests/deploys."""
override = os.environ.get("WEBUI_WORKER_REGISTRY", "").strip()
if override:
return Path(override).expanduser().resolve()
return (Path(__file__).resolve().parent / "data" / "workers.registry.json").resolve()
def history_dir(path: Path | None = None) -> Path:
"""Directory holding superseded revisions of *path*."""
source = (path or default_registry_path()).resolve()
return source.parent / f"{source.name}.history"
# ── field helpers ────────────────────────────────────────────────────────────
def _require_exact_fields(
raw: Any,
*,
required: tuple[str, ...],
optional: tuple[str, ...],
subject: str,
) -> dict[str, Any]:
if not isinstance(raw, dict):
raise RegistryValidationError(f"{subject} must be an object")
missing = [field for field in required if field not in raw]
if missing:
raise RegistryValidationError(
f"{subject} missing required fields: {', '.join(sorted(missing))}"
)
unknown = sorted(set(raw) - set(required) - set(optional))
if unknown:
# Fail closed: silently dropping an unrecognized key is how a typo'd
# "timeout_second" ends up meaning "no timeout".
raise RegistryValidationError(f"{subject} has unknown fields: {', '.join(unknown)}")
return raw
def _require_identifier(value: Any, *, subject: str) -> str:
text = str(value).strip()
if not _ID_RE.match(text):
raise RegistryValidationError(
f"{subject} must be lowercase alphanumeric with '.', '_', or '-' (got {value!r})"
)
return text
def _require_text(value: Any, *, subject: str) -> str:
if not isinstance(value, str):
raise RegistryValidationError(f"{subject} must be a string (got {value!r})")
text = value.strip()
if not text:
raise RegistryValidationError(f"{subject} must be a non-empty string")
return text
def _require_bool(value: Any, *, subject: str) -> bool:
if not isinstance(value, bool):
raise RegistryValidationError(f"{subject} must be a boolean (got {value!r})")
return value
def _require_positive_int(value: Any, *, subject: str, maximum: int | None = None) -> int:
if isinstance(value, bool) or not isinstance(value, int):
raise RegistryValidationError(f"{subject} must be an integer (got {value!r})")
if value <= 0:
raise RegistryValidationError(f"{subject} must be greater than zero (got {value})")
if maximum is not None and value > maximum:
raise RegistryValidationError(f"{subject} must not exceed {maximum} (got {value})")
return value
# ── parsing ──────────────────────────────────────────────────────────────────
def _parse_provider(raw: Any) -> ProviderRecord:
data = _require_exact_fields(
raw,
required=_REQUIRED_PROVIDER_FIELDS,
optional=_OPTIONAL_PROVIDER_FIELDS,
subject="provider",
)
provider_id = _require_identifier(data["id"], subject="provider.id")
models_raw = data.get("models") or []
if not isinstance(models_raw, list):
raise RegistryValidationError(f"provider[{provider_id}].models must be an array")
models = tuple(
_require_text(item, subject=f"provider[{provider_id}].models[]") for item in models_raw
)
return ProviderRecord(
id=provider_id,
display_name=_require_text(
data["display_name"], subject=f"provider[{provider_id}].display_name"
),
vendor=_require_text(data["vendor"], subject=f"provider[{provider_id}].vendor"),
executable=_require_text(data["executable"], subject=f"provider[{provider_id}].executable"),
available=_require_bool(data["available"], subject=f"provider[{provider_id}].available"),
models=models,
notes=str(data.get("notes") or "").strip(),
)
def _parse_schedule(raw: Any, *, subject: str) -> ScheduleSpec:
if not isinstance(raw, dict):
raise RegistryValidationError(f"{subject} must be an object")
kind = _require_text(raw.get("kind"), subject=f"{subject}.kind")
if kind not in ALLOWED_SCHEDULE_KINDS:
raise RegistryValidationError(
f"{subject}.kind must be one of {', '.join(ALLOWED_SCHEDULE_KINDS)} (got {kind!r})"
)
seconds: int | None = None
expression: str | None = None
if kind == "interval":
if "seconds" not in raw:
raise RegistryValidationError(f"{subject}.seconds is required for interval schedules")
seconds = _require_positive_int(raw["seconds"], subject=f"{subject}.seconds")
elif kind == "cron":
if "expression" not in raw:
raise RegistryValidationError(f"{subject}.expression is required for cron schedules")
expression = _require_text(raw["expression"], subject=f"{subject}.expression")
if len(expression.split()) != 5:
raise RegistryValidationError(
f"{subject}.expression must have five crontab fields (got {expression!r})"
)
allowed = {"kind"}
if kind == "interval":
allowed.add("seconds")
elif kind == "cron":
allowed.add("expression")
unknown = sorted(set(raw) - allowed)
if unknown:
raise RegistryValidationError(
f"{subject} has fields not valid for kind {kind!r}: {', '.join(unknown)}"
)
return ScheduleSpec(kind=kind, seconds=seconds, expression=expression)
def _parse_scheduler(raw: Any, *, subject: str) -> SchedulerSpec:
if not isinstance(raw, dict):
raise RegistryValidationError(f"{subject} must be an object")
kind = _require_text(raw.get("kind"), subject=f"{subject}.kind")
if kind not in ALLOWED_SCHEDULER_KINDS:
raise RegistryValidationError(
f"{subject}.kind must be one of {', '.join(ALLOWED_SCHEDULER_KINDS)} (got {kind!r})"
)
label: str | None = None
if kind == "launchd":
if "label" not in raw:
raise RegistryValidationError(f"{subject}.label is required for launchd schedulers")
label = _require_text(raw["label"], subject=f"{subject}.label")
allowed = {"kind"}
if kind == "launchd":
allowed.add("label")
unknown = sorted(set(raw) - allowed)
if unknown:
raise RegistryValidationError(
f"{subject} has fields not valid for kind {kind!r}: {', '.join(unknown)}"
)
return SchedulerSpec(kind=kind, label=label)
def _parse_worker(raw: Any) -> WorkerRecord:
data = _require_exact_fields(
raw,
required=_REQUIRED_WORKER_FIELDS,
optional=_OPTIONAL_WORKER_FIELDS,
subject="worker",
)
worker_id = _require_identifier(data["id"], subject="worker.id")
role = _require_text(data["role"], subject=f"worker[{worker_id}].role")
if role not in ALLOWED_ROLES:
raise RegistryValidationError(
f"worker[{worker_id}].role must be one of {', '.join(ALLOWED_ROLES)} (got {role!r})"
)
return WorkerRecord(
id=worker_id,
display_name=_require_text(
data["display_name"], subject=f"worker[{worker_id}].display_name"
),
provider=_require_identifier(data["provider"], subject=f"worker[{worker_id}].provider"),
model=_require_text(data["model"], subject=f"worker[{worker_id}].model"),
project=_require_text(data["project"], subject=f"worker[{worker_id}].project"),
role=role,
namespace=_require_text(data["namespace"], subject=f"worker[{worker_id}].namespace"),
profile=_require_text(data["profile"], subject=f"worker[{worker_id}].profile"),
workflow=_require_text(data["workflow"], subject=f"worker[{worker_id}].workflow"),
schedule=_parse_schedule(data["schedule"], subject=f"worker[{worker_id}].schedule"),
timeout_seconds=_require_positive_int(
data["timeout_seconds"],
subject=f"worker[{worker_id}].timeout_seconds",
maximum=_MAX_TIMEOUT_SECONDS,
),
enabled=_require_bool(data["enabled"], subject=f"worker[{worker_id}].enabled"),
scheduler=_parse_scheduler(data["scheduler"], subject=f"worker[{worker_id}].scheduler"),
notes=str(data.get("notes") or "").strip(),
)
def _require_unique(values: list[str], *, subject: str) -> None:
seen: set[str] = set()
for value in values:
if value in seen:
raise RegistryValidationError(f"duplicate {subject}: {value}")
seen.add(value)
def validate_payload(payload: Any, *, source_path: Path) -> WorkerRegistry:
"""Validate a decoded registry document and return the typed registry.
Raises :class:`RegistryValidationError` on any violation; never partially
accepts a document.
"""
if not isinstance(payload, dict):
raise RegistryValidationError("registry root must be an object")
version = payload.get("version")
if version != SCHEMA_VERSION:
raise RegistryValidationError(f"unsupported registry version: {version!r}")
reject_credential_keys(payload, subject="worker registry")
unknown = sorted(set(payload) - _TOP_LEVEL_FIELDS)
if unknown:
raise RegistryValidationError(f"registry has unknown fields: {', '.join(unknown)}")
revision = _require_positive_int(payload.get("revision"), subject="revision")
updated_at = _require_text(payload.get("updated_at"), subject="updated_at")
providers_raw = payload.get("providers")
if not isinstance(providers_raw, list) or not providers_raw:
raise RegistryValidationError("providers must be a non-empty array")
providers = tuple(_parse_provider(item) for item in providers_raw)
_require_unique([provider.id for provider in providers], subject="provider id")
workers_raw = payload.get("workers")
if not isinstance(workers_raw, list):
raise RegistryValidationError("workers must be an array")
workers = tuple(_parse_worker(item) for item in workers_raw)
_require_unique([worker.id for worker in workers], subject="worker id")
# Referential integrity: a worker naming an undeclared provider would look
# configured while being unrunnable, which is exactly the ambiguous
# ownership the epic requires to fail closed.
known_providers = {provider.id for provider in providers}
for worker in workers:
if worker.provider not in known_providers:
raise RegistryValidationError(
f"worker[{worker.id}].provider references unknown provider {worker.provider!r}"
)
# A LaunchAgent label identifies a job to launchd; two workers sharing one
# would silently overwrite each other's agent.
_require_unique(
[worker.scheduler.label for worker in workers if worker.scheduler.label],
subject="scheduler label",
)
return WorkerRegistry(
version=version,
revision=revision,
updated_at=updated_at,
providers=providers,
workers=workers,
source_path=source_path,
)
def load_registry(path: Path | None = None) -> WorkerRegistry:
"""Load and validate the worker registry from disk."""
source = (path or default_registry_path()).resolve()
payload = json.loads(source.read_text(encoding="utf-8"))
return validate_payload(payload, source_path=source)
# ── serialization ────────────────────────────────────────────────────────────
def provider_to_dict(provider: ProviderRecord) -> dict[str, Any]:
return {
"id": provider.id,
"display_name": provider.display_name,
"vendor": provider.vendor,
"executable": provider.executable,
"available": provider.available,
"models": list(provider.models),
"notes": provider.notes,
}
def _schedule_to_dict(schedule: ScheduleSpec) -> dict[str, Any]:
payload: dict[str, Any] = {"kind": schedule.kind}
if schedule.kind == "interval":
payload["seconds"] = schedule.seconds
elif schedule.kind == "cron":
payload["expression"] = schedule.expression
return payload
def _scheduler_to_dict(scheduler: SchedulerSpec) -> dict[str, Any]:
payload: dict[str, Any] = {"kind": scheduler.kind}
if scheduler.kind == "launchd":
payload["label"] = scheduler.label
return payload
def worker_to_dict(worker: WorkerRecord) -> dict[str, Any]:
return {
"id": worker.id,
"display_name": worker.display_name,
"provider": worker.provider,
"model": worker.model,
"project": worker.project,
"role": worker.role,
"namespace": worker.namespace,
"profile": worker.profile,
"workflow": worker.workflow,
"schedule": _schedule_to_dict(worker.schedule),
"timeout_seconds": worker.timeout_seconds,
"enabled": worker.enabled,
"scheduler": _scheduler_to_dict(worker.scheduler),
"notes": worker.notes,
}
def registry_to_document(registry: WorkerRegistry) -> dict[str, Any]:
"""Serialize to the on-disk document shape (no local paths embedded)."""
return {
"version": registry.version,
"revision": registry.revision,
"updated_at": registry.updated_at,
"providers": [provider_to_dict(provider) for provider in registry.providers],
"workers": [worker_to_dict(worker) for worker in registry.workers],
}
def registry_to_dict(registry: WorkerRegistry) -> dict[str, Any]:
"""Serialize for JSON API responses (adds the resolved source path)."""
document = registry_to_document(registry)
document["source_path"] = str(registry.source_path)
return document
def find_worker(registry: WorkerRegistry, worker_id: str) -> WorkerRecord | None:
for worker in registry.workers:
if worker.id == worker_id:
return worker
return None
def find_provider(registry: WorkerRegistry, provider_id: str) -> ProviderRecord | None:
for provider in registry.providers:
if provider.id == provider_id:
return provider
return None
def workers_for_provider(registry: WorkerRegistry, provider_id: str) -> tuple[WorkerRecord, ...]:
return tuple(worker for worker in registry.workers if worker.provider == provider_id)
# ── persistence ──────────────────────────────────────────────────────────────
def _utc_now() -> str:
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
def _atomic_write(path: Path, payload: str) -> None:
"""Write *payload* to *path* atomically: temp file in the same dir, fsync, replace."""
parent = path.parent
parent.mkdir(parents=True, exist_ok=True)
fd, temp_path = tempfile.mkstemp(prefix=f".{path.name}-", suffix=".tmp", dir=parent)
try:
with os.fdopen(fd, "w", encoding="utf-8") as handle:
handle.write(payload)
handle.flush()
os.fsync(handle.fileno())
os.replace(temp_path, path)
finally:
if os.path.exists(temp_path):
try:
os.remove(temp_path)
except OSError:
pass
def _revision_path(directory: Path, revision: int) -> Path:
return directory / f"rev-{revision:06d}.json"
def _prune_history(path: Path) -> None:
directory = history_dir(path)
revisions = list_revisions(path)
excess = len(revisions) - _HISTORY_LIMIT
for revision in revisions[: max(0, excess)]:
_revision_path(directory, revision).unlink(missing_ok=True)
def _archive_current(path: Path) -> int | None:
"""Copy the live document into the history dir under its own revision number."""
if not path.exists():
return None
try:
existing = json.loads(path.read_text(encoding="utf-8"))
revision = int(existing.get("revision", 0))
except (json.JSONDecodeError, TypeError, ValueError, AttributeError):
# An unreadable live file has no trustworthy revision number to file it
# under, so it cannot join the history chain.
return None
if revision <= 0:
return None
_atomic_write(
_revision_path(history_dir(path), revision),
json.dumps(existing, indent=2, sort_keys=True) + "\n",
)
_prune_history(path)
return revision
def list_revisions(path: Path | None = None) -> tuple[int, ...]:
"""Revision numbers retained in history for *path*, oldest first."""
directory = history_dir(path)
if not directory.is_dir():
return ()
revisions: list[int] = []
for entry in directory.glob("rev-*.json"):
try:
revisions.append(int(entry.stem.split("-", 1)[1]))
except (IndexError, ValueError):
continue
return tuple(sorted(revisions))
def save_registry(
registry: WorkerRegistry,
path: Path | None = None,
*,
updated_at: str | None = None,
) -> WorkerRegistry:
"""Validate, archive the superseded revision, then atomically persist a new one.
The stored revision is always the previous revision plus one, so a reader
can tell two documents apart even when their content is otherwise equal.
Returns the registry exactly as persisted.
"""
target = (path or registry.source_path or default_registry_path()).resolve()
document = registry_to_document(registry)
# Re-validate before writing: a registry assembled in memory has not
# necessarily been through the loader.
validate_payload(document, source_path=target)
archived = _archive_current(target)
document["revision"] = (archived + 1) if archived is not None else registry.revision
document["updated_at"] = updated_at or _utc_now()
persisted = validate_payload(document, source_path=target)
_atomic_write(target, json.dumps(document, indent=2, sort_keys=True) + "\n")
return persisted
def rollback_to_revision(revision: int, path: Path | None = None) -> WorkerRegistry:
"""Restore a retained *revision* as a new head revision.
History is append-only: rolling back does not delete the revisions in
between, it republishes the chosen one under the next revision number, so a
rollback is itself reversible.
"""
target = (path or default_registry_path()).resolve()
snapshot_path = _revision_path(history_dir(target), revision)
if not snapshot_path.exists():
available = ", ".join(str(item) for item in list_revisions(target)) or "(none)"
raise RegistryValidationError(
f"revision {revision} is not retained for {target.name}; available: {available}"
)
payload = json.loads(snapshot_path.read_text(encoding="utf-8"))
restored = validate_payload(payload, source_path=target)
return save_registry(restored, target)