Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
  sysadminandClaude Opus 4.8
|
b2f6e9a6dc |
@@ -1,295 +0,0 @@
|
||||
# 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.
|
||||
@@ -7,10 +7,7 @@ only.
|
||||
## MVP deployment model
|
||||
|
||||
- **Default bind:** `127.0.0.1:8765` (`WEBUI_HOST` / `WEBUI_PORT`)
|
||||
- **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).
|
||||
- **Authentication:** none in MVP — protection comes from network placement
|
||||
- **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
|
||||
|
||||
@@ -43,6 +43,10 @@ for the console architecture: layer and authority boundaries, the redaction
|
||||
boundary, `/api/v1/...` versioning, the target page map, and the phase gates
|
||||
that govern when a write path may open (#632, epic #631).
|
||||
|
||||
See [webui-project-registry-api.md](webui-project-registry-api.md) for the
|
||||
versioned project registry contract: registry schema versions 1 and 2, project
|
||||
status, onboarding checklist state, and the fail-closed error payloads (#635).
|
||||
|
||||
## Routes (MVP)
|
||||
|
||||
| Path | Description |
|
||||
@@ -51,9 +55,11 @@ that govern when a write path may open (#632, epic #631).
|
||||
| `/health` | JSON liveness (`status`, `service`, `mode`, `timestamp`) |
|
||||
| `/queue` | Live PR and issue queue dashboard (#429) |
|
||||
| `/api/queue` | JSON queue export with pagination metadata |
|
||||
| `/projects` | Project registry list (#427) |
|
||||
| `/projects` | Project registry list with status and onboarding progress (#427, #635) |
|
||||
| `/projects/{id}` | Project detail + onboarding checklist |
|
||||
| `/api/projects` | JSON registry export |
|
||||
| `/api/v1/projects` | Versioned JSON registry export (#635) |
|
||||
| `/api/v1/projects/{id}` | Versioned JSON project detail (#635) |
|
||||
| `/api/projects` | JSON registry export — unversioned Phase 1 alias of `/api/v1/projects` |
|
||||
| `/prompts` | Prompt library with per-prompt copy buttons (#428) |
|
||||
| `/api/prompts` | JSON prompt export with workflow hashes |
|
||||
| `/runtime` | MCP runtime health and stale detection (#430) |
|
||||
|
||||
@@ -0,0 +1,213 @@
|
||||
# Project registry API (#635)
|
||||
|
||||
Phase 1 of the [console architecture ADR](architecture/webui-control-plane-console-architecture-adr.md)
|
||||
gives the project registry a versioned, read-only API. This document is the
|
||||
field-by-field contract for that API and for the registry file behind it.
|
||||
|
||||
Everything here is **read-only**. The console never writes the registry; an
|
||||
operator edits the JSON file, and an invalid file fails closed rather than
|
||||
rendering a partial inventory.
|
||||
|
||||
## Routes
|
||||
|
||||
| Route | Method | Description |
|
||||
|-------|--------|-------------|
|
||||
| `/api/v1/projects` | GET | Versioned registry export: all projects, with provenance |
|
||||
| `/api/v1/projects/{project_id}` | GET | Single project; `404` with `project_not_found` when unknown |
|
||||
| `/api/projects` | GET | Unversioned MVP alias (#427), retained for all of Phase 1 |
|
||||
| `/projects` | GET | HTML list — status and onboarding progress per project |
|
||||
| `/projects/{project_id}` | GET | HTML detail — identity, profiles, paths, checklist |
|
||||
|
||||
Per ADR section 6 the unversioned alias may be retired no earlier than Phase 2,
|
||||
and only after this document and `webui-local-dev.md` record the swap. The alias
|
||||
returns the same payload as `/api/v1/projects`, including the legacy `version`
|
||||
and `source_path` keys #427 consumers already read.
|
||||
|
||||
The HTML views render from the same DTO the JSON routes serialize
|
||||
(`project_to_dict`), so the console and the API cannot disagree about a
|
||||
project's status or onboarding progress.
|
||||
|
||||
## Registry file
|
||||
|
||||
Default location: `webui/data/projects.registry.json`. Override with the
|
||||
`WEBUI_PROJECT_REGISTRY` environment variable.
|
||||
|
||||
Schema versions: **1** and **2** are accepted; **2** is current. A version 1
|
||||
file loads unchanged and is normalized with the documented defaults, so an
|
||||
existing operator registry keeps working without edits.
|
||||
|
||||
### Root
|
||||
|
||||
| Field | Type | Required | Notes |
|
||||
|-------|------|----------|-------|
|
||||
| `version` | int | yes | `1` or `2`. Anything else fails closed |
|
||||
| `projects` | array | yes | Must be non-empty |
|
||||
|
||||
### Project
|
||||
|
||||
| Field | Type | Required | Default | Notes |
|
||||
|-------|------|----------|---------|-------|
|
||||
| `id` | string | yes | — | Stable registry id used in URLs |
|
||||
| `repo_name` | string | yes | — | Gitea repository name |
|
||||
| `gitea_owner` | string | yes | — | Owning org or user |
|
||||
| `remote_host` | string | yes | — | Instance base URL, no credentials |
|
||||
| `remote_name` | string | no | `null` | Logical remote label, e.g. `prgs` (v2) |
|
||||
| `default_branch` | string | yes | — | Stable branch name |
|
||||
| `local_checkout_path` | string | yes | — | Control checkout path |
|
||||
| `status` | string | no | `active` | `active`, `onboarding`, `paused`, `archived` (v2) |
|
||||
| `profiles` | object | yes | — | Must map `author`, `reviewer`, `reconciler` |
|
||||
| `workflow_paths` | object | yes | — | Non-empty; label to repo-relative path |
|
||||
| `schema_paths` | object | no | `{}` | Label to repo-relative path |
|
||||
| `onboarding_checklist` | array | no | `[]` | See below |
|
||||
| `last_seen_health` | object | no | `null` | Redacted health only (v2) |
|
||||
|
||||
### Onboarding step
|
||||
|
||||
| Field | Type | Required | Default | Notes |
|
||||
|-------|------|----------|---------|-------|
|
||||
| `id` | string | yes | — | Stable step id |
|
||||
| `title` | string | yes | — | Short operator-facing label |
|
||||
| `description` | string | yes | — | Self-contained; assumes no chat history |
|
||||
| `state` | string | no | `pending` | `complete`, `pending`, `blocked`, `not_applicable` (v2) |
|
||||
| `required` | bool | no | `true` | Optional steps never block readiness (v2) |
|
||||
|
||||
### Last-seen health
|
||||
|
||||
| Field | Type | Required | Notes |
|
||||
|-------|------|----------|-------|
|
||||
| `status` | string | no (default `unknown`) | `healthy`, `degraded`, `unreachable`, `unknown` |
|
||||
| `checked_at` | string | no | ISO-8601 UTC timestamp, e.g. `2026-01-01T00:00:00Z` |
|
||||
| `detail` | string | no | Short redacted note |
|
||||
|
||||
Health is recorded metadata, not a live probe: Phase 1 performs no outbound
|
||||
health checks. Endpoints, tokens, and keychain identifiers must never appear
|
||||
here.
|
||||
|
||||
## Response shape
|
||||
|
||||
`GET /api/v1/projects`:
|
||||
|
||||
```json
|
||||
{
|
||||
"api_version": "v1",
|
||||
"schema_version": 2,
|
||||
"version": 2,
|
||||
"source_path": "/path/to/webui/data/projects.registry.json",
|
||||
"source": {
|
||||
"kind": "file",
|
||||
"path": "/path/to/webui/data/projects.registry.json",
|
||||
"inventory_complete": true
|
||||
},
|
||||
"project_count": 1,
|
||||
"projects": [
|
||||
{
|
||||
"id": "example",
|
||||
"repo_name": "Example",
|
||||
"gitea_owner": "Org",
|
||||
"repo_full_name": "Org/Example",
|
||||
"remote_host": "https://gitea.example.invalid",
|
||||
"remote_name": "example-remote",
|
||||
"default_branch": "main",
|
||||
"local_checkout_path": ".",
|
||||
"status": "active",
|
||||
"profiles": {"author": "...", "reviewer": "...", "reconciler": "..."},
|
||||
"workflow_paths": {"skill": "skills/..."},
|
||||
"schema_paths": {},
|
||||
"onboarding_checklist": [
|
||||
{
|
||||
"id": "profiles",
|
||||
"title": "Configure execution profiles",
|
||||
"description": "...",
|
||||
"state": "complete",
|
||||
"required": true
|
||||
}
|
||||
],
|
||||
"onboarding_summary": {
|
||||
"total": 1,
|
||||
"complete": 1,
|
||||
"pending": 0,
|
||||
"blocked": 0,
|
||||
"not_applicable": 0,
|
||||
"required_outstanding": 0,
|
||||
"onboarding_complete": true
|
||||
},
|
||||
"last_seen_health": null
|
||||
}
|
||||
]
|
||||
}
|
||||
```
|
||||
|
||||
`GET /api/v1/projects/{project_id}` returns `api_version`, `schema_version`,
|
||||
`source`, and a single `project` object with the same fields.
|
||||
|
||||
The `source` block satisfies the ADR section 6 provenance rule: every payload
|
||||
states where the data came from and whether the inventory is complete. A
|
||||
file-backed registry is always complete — there is no pagination to truncate it.
|
||||
|
||||
`onboarding_summary` is derived, never stored. `required_outstanding` counts
|
||||
steps that are `required` **and** in state `pending` or `blocked`;
|
||||
`onboarding_complete` is true when that count is zero.
|
||||
|
||||
## Fail-closed errors
|
||||
|
||||
Validation failures raise `RegistryError`, which routes render instead of a
|
||||
traceback.
|
||||
|
||||
`404` — unknown project id on `/api/v1/projects/{project_id}`:
|
||||
|
||||
```json
|
||||
{
|
||||
"error": "project_not_found",
|
||||
"project_id": "not-registered",
|
||||
"known_project_ids": ["example"],
|
||||
"remediation": "Request one of the known project ids, or add the project ...",
|
||||
"source": {"kind": "file", "path": "...", "inventory_complete": true}
|
||||
}
|
||||
```
|
||||
|
||||
`500` — invalid registry, on both the versioned route and the alias:
|
||||
|
||||
```json
|
||||
{
|
||||
"error": "registry_invalid",
|
||||
"detail": "unsupported registry version: 42",
|
||||
"remediation": "Set 'version' to one of 1, 2 (current schema is 2) ...",
|
||||
"field_path": "version",
|
||||
"source_path": "/path/to/registry.json"
|
||||
}
|
||||
```
|
||||
|
||||
`field_path` points at the offending location (`projects[0].profiles.reconciler`,
|
||||
`projects[0].onboarding_checklist[2].state`, and so on). The HTML routes render
|
||||
the same detail, field, source, and remediation on a "Project registry
|
||||
unavailable" page.
|
||||
|
||||
Conditions that fail closed:
|
||||
|
||||
* file missing or unreadable;
|
||||
* invalid JSON (the remediation names line and column);
|
||||
* root not an object, or `projects` missing/empty;
|
||||
* unsupported `version`;
|
||||
* a credential-shaped key anywhere in the file (`token`, `*_secret`, `auth_*`, and similar);
|
||||
* a project missing a required field, or missing an `author`/`reviewer`/`reconciler` profile;
|
||||
* an unknown `status`, onboarding `state`, or health `status`.
|
||||
|
||||
## Credential rule
|
||||
|
||||
The registry stores redacted metadata only. Credential-shaped keys are
|
||||
rejected at load time, before any DTO is built, consistent with
|
||||
[safety-model.md](safety-model.md) and
|
||||
[credential-isolation.md](credential-isolation.md). Tokens live in the keychain
|
||||
and are resolved server-side by `gitea_auth`.
|
||||
|
||||
## Migrating a version 1 registry
|
||||
|
||||
1. Set `"version": 2`.
|
||||
2. Optionally add `"status"` per project (omitted means `active`).
|
||||
3. Optionally add `"remote_name"` per project.
|
||||
4. Optionally add `"state"` and `"required"` to each onboarding step (omitted
|
||||
means `pending` and `true`).
|
||||
5. Optionally add `"last_seen_health"`.
|
||||
|
||||
No step is mandatory: a version 1 file keeps loading. Bumping the version only
|
||||
declares that the file may use the v2 fields.
|
||||
@@ -1,703 +0,0 @@
|
||||
"""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()
|
||||
@@ -1,4 +1,4 @@
|
||||
"""Tests for web UI project registry (#427)."""
|
||||
"""Tests for web UI project registry (#427) and its API evolution (#635)."""
|
||||
import json
|
||||
import sys
|
||||
import tempfile
|
||||
@@ -11,57 +11,246 @@ from starlette.testclient import TestClient
|
||||
|
||||
from webui.app import create_app
|
||||
from webui.project_registry import (
|
||||
CURRENT_SCHEMA_VERSION,
|
||||
REGISTRY_API_VERSION,
|
||||
SUPPORTED_SCHEMA_VERSIONS,
|
||||
RegistryError,
|
||||
default_registry_path,
|
||||
load_registry,
|
||||
onboarding_summary,
|
||||
project_to_dict,
|
||||
)
|
||||
from webui.registry_safety import is_forbidden_key
|
||||
|
||||
_REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
_API_DOC = _REPO_ROOT / "docs" / "webui-project-registry-api.md"
|
||||
|
||||
|
||||
class TestProjectRegistryLoader(unittest.TestCase):
|
||||
def _valid_project(**overrides):
|
||||
project = {
|
||||
"id": "example",
|
||||
"repo_name": "Example",
|
||||
"gitea_owner": "Org",
|
||||
"remote_host": "https://gitea.example.invalid",
|
||||
"default_branch": "main",
|
||||
"local_checkout_path": ".",
|
||||
"profiles": {"author": "a", "reviewer": "r", "reconciler": "c"},
|
||||
"workflow_paths": {"skill": "skills/x.md"},
|
||||
}
|
||||
project.update(overrides)
|
||||
return project
|
||||
|
||||
|
||||
def _write_registry(payload) -> Path:
|
||||
with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as handle:
|
||||
json.dump(payload, handle)
|
||||
return Path(handle.name)
|
||||
|
||||
|
||||
class RegistryFileCase(unittest.TestCase):
|
||||
"""Base class that cleans up temporary registry files."""
|
||||
|
||||
def setUp(self):
|
||||
self._temp_paths: list[Path] = []
|
||||
|
||||
def tearDown(self):
|
||||
for path in self._temp_paths:
|
||||
path.unlink(missing_ok=True)
|
||||
|
||||
def write_registry(self, payload) -> Path:
|
||||
path = _write_registry(payload)
|
||||
self._temp_paths.append(path)
|
||||
return path
|
||||
|
||||
|
||||
class TestProjectRegistryLoader(RegistryFileCase):
|
||||
def test_default_registry_loads_gitea_tools(self):
|
||||
registry = load_registry()
|
||||
self.assertEqual(registry.version, 1)
|
||||
self.assertEqual(registry.version, CURRENT_SCHEMA_VERSION)
|
||||
self.assertEqual(registry.schema_version, CURRENT_SCHEMA_VERSION)
|
||||
self.assertEqual(registry.api_version, REGISTRY_API_VERSION)
|
||||
self.assertEqual(len(registry.projects), 1)
|
||||
project = registry.projects[0]
|
||||
self.assertEqual(project.id, "gitea-tools")
|
||||
self.assertEqual(project.repo_name, "Gitea-Tools")
|
||||
self.assertEqual(project.gitea_owner, "Scaled-Tech-Consulting")
|
||||
self.assertEqual(project.repo_full_name, "Scaled-Tech-Consulting/Gitea-Tools")
|
||||
self.assertEqual(project.remote_host, "https://gitea.prgs.cc")
|
||||
self.assertEqual(project.remote_name, "prgs")
|
||||
self.assertEqual(project.status, "active")
|
||||
self.assertEqual(project.profiles["author"], "prgs-author")
|
||||
self.assertEqual(project.profiles["reviewer"], "prgs-reviewer")
|
||||
self.assertEqual(project.profiles["reconciler"], "prgs-reconciler")
|
||||
self.assertIn("skill", project.workflow_paths)
|
||||
self.assertGreaterEqual(len(project.onboarding_checklist), 4)
|
||||
|
||||
def test_registry_rejects_credential_keys(self):
|
||||
payload = {
|
||||
def test_default_registry_onboarding_summary_is_complete(self):
|
||||
summary = onboarding_summary(load_registry().projects[0])
|
||||
self.assertEqual(summary.total, summary.complete)
|
||||
self.assertEqual(summary.required_outstanding, 0)
|
||||
self.assertTrue(summary.onboarding_complete)
|
||||
|
||||
def test_version_1_registry_still_loads_with_defaults(self):
|
||||
path = self.write_registry({
|
||||
"version": 1,
|
||||
"projects": [
|
||||
{
|
||||
"id": "bad",
|
||||
"repo_name": "Bad",
|
||||
"gitea_owner": "Org",
|
||||
"remote_host": "https://gitea.example.invalid",
|
||||
"default_branch": "main",
|
||||
"local_checkout_path": ".",
|
||||
"profiles": {
|
||||
"author": "a",
|
||||
"reviewer": "r",
|
||||
"reconciler": "c",
|
||||
},
|
||||
"workflow_paths": {"skill": "skills/x.md"},
|
||||
"api_token": "secret",
|
||||
}
|
||||
_valid_project(
|
||||
onboarding_checklist=[
|
||||
{"id": "step", "title": "Step", "description": "Do it"}
|
||||
]
|
||||
)
|
||||
],
|
||||
}
|
||||
})
|
||||
registry = load_registry(path)
|
||||
self.assertEqual(registry.schema_version, 1)
|
||||
self.assertIn(1, SUPPORTED_SCHEMA_VERSIONS)
|
||||
project = registry.projects[0]
|
||||
self.assertEqual(project.status, "active")
|
||||
self.assertIsNone(project.remote_name)
|
||||
self.assertIsNone(project.last_seen_health)
|
||||
step = project.onboarding_checklist[0]
|
||||
self.assertEqual(step.state, "pending")
|
||||
self.assertTrue(step.required)
|
||||
self.assertFalse(onboarding_summary(project).onboarding_complete)
|
||||
|
||||
def test_onboarding_summary_counts_states(self):
|
||||
path = self.write_registry({
|
||||
"version": 2,
|
||||
"projects": [
|
||||
_valid_project(
|
||||
onboarding_checklist=[
|
||||
{"id": "a", "title": "A", "description": "d", "state": "complete"},
|
||||
{"id": "b", "title": "B", "description": "d", "state": "blocked"},
|
||||
{
|
||||
"id": "c",
|
||||
"title": "C",
|
||||
"description": "d",
|
||||
"state": "pending",
|
||||
"required": False,
|
||||
},
|
||||
{
|
||||
"id": "d",
|
||||
"title": "D",
|
||||
"description": "d",
|
||||
"state": "not_applicable",
|
||||
},
|
||||
]
|
||||
)
|
||||
],
|
||||
})
|
||||
summary = onboarding_summary(load_registry(path).projects[0])
|
||||
self.assertEqual(summary.total, 4)
|
||||
self.assertEqual(summary.complete, 1)
|
||||
self.assertEqual(summary.blocked, 1)
|
||||
self.assertEqual(summary.pending, 1)
|
||||
self.assertEqual(summary.not_applicable, 1)
|
||||
# Only the blocked step is both required and outstanding.
|
||||
self.assertEqual(summary.required_outstanding, 1)
|
||||
self.assertFalse(summary.onboarding_complete)
|
||||
|
||||
def test_last_seen_health_is_parsed_when_present(self):
|
||||
path = self.write_registry({
|
||||
"version": 2,
|
||||
"projects": [
|
||||
_valid_project(
|
||||
last_seen_health={
|
||||
"status": "degraded",
|
||||
"checked_at": "2026-01-01T00:00:00Z",
|
||||
"detail": "daemon restart pending",
|
||||
}
|
||||
)
|
||||
],
|
||||
})
|
||||
health = load_registry(path).projects[0].last_seen_health
|
||||
self.assertIsNotNone(health)
|
||||
self.assertEqual(health.status, "degraded")
|
||||
self.assertEqual(health.checked_at, "2026-01-01T00:00:00Z")
|
||||
|
||||
def test_registry_rejects_credential_keys(self):
|
||||
path = self.write_registry({
|
||||
"version": 1,
|
||||
"projects": [_valid_project(id="bad", api_token="redacted-placeholder")],
|
||||
})
|
||||
with self.assertRaises(RegistryError) as ctx:
|
||||
load_registry(path)
|
||||
self.assertIn("credential", ctx.exception.remediation.lower())
|
||||
self.assertEqual(ctx.exception.field_path, "projects[0].api_token")
|
||||
|
||||
def test_unsupported_version_fails_closed_with_remediation(self):
|
||||
path = self.write_registry({"version": 99, "projects": [_valid_project()]})
|
||||
with self.assertRaises(RegistryError) as ctx:
|
||||
load_registry(path)
|
||||
self.assertIn("unsupported registry version", ctx.exception.message)
|
||||
self.assertIn(str(CURRENT_SCHEMA_VERSION), ctx.exception.remediation)
|
||||
self.assertEqual(ctx.exception.field_path, "version")
|
||||
|
||||
def test_missing_required_field_fails_closed(self):
|
||||
broken = _valid_project()
|
||||
del broken["default_branch"]
|
||||
path = self.write_registry({"version": 2, "projects": [broken]})
|
||||
with self.assertRaises(RegistryError) as ctx:
|
||||
load_registry(path)
|
||||
self.assertIn("default_branch", ctx.exception.message)
|
||||
self.assertEqual(ctx.exception.field_path, "projects[0]")
|
||||
|
||||
def test_unknown_status_fails_closed(self):
|
||||
path = self.write_registry({
|
||||
"version": 2,
|
||||
"projects": [_valid_project(status="mystery")],
|
||||
})
|
||||
with self.assertRaises(RegistryError) as ctx:
|
||||
load_registry(path)
|
||||
self.assertEqual(ctx.exception.field_path, "projects[0].status")
|
||||
self.assertIn("active", ctx.exception.remediation)
|
||||
|
||||
def test_unknown_onboarding_state_fails_closed(self):
|
||||
path = self.write_registry({
|
||||
"version": 2,
|
||||
"projects": [
|
||||
_valid_project(
|
||||
onboarding_checklist=[
|
||||
{"id": "a", "title": "A", "description": "d", "state": "almost"}
|
||||
]
|
||||
)
|
||||
],
|
||||
})
|
||||
with self.assertRaises(RegistryError) as ctx:
|
||||
load_registry(path)
|
||||
self.assertEqual(
|
||||
ctx.exception.field_path,
|
||||
"projects[0].onboarding_checklist[0].state",
|
||||
)
|
||||
|
||||
def test_missing_profile_role_fails_closed(self):
|
||||
path = self.write_registry({
|
||||
"version": 2,
|
||||
"projects": [_valid_project(profiles={"author": "a", "reviewer": "r"})],
|
||||
})
|
||||
with self.assertRaises(RegistryError) as ctx:
|
||||
load_registry(path)
|
||||
self.assertEqual(ctx.exception.field_path, "projects[0].profiles.reconciler")
|
||||
|
||||
def test_empty_projects_fails_closed(self):
|
||||
path = self.write_registry({"version": 2, "projects": []})
|
||||
with self.assertRaises(RegistryError) as ctx:
|
||||
load_registry(path)
|
||||
self.assertEqual(ctx.exception.field_path, "projects")
|
||||
|
||||
def test_invalid_json_fails_closed_with_location(self):
|
||||
with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as handle:
|
||||
json.dump(payload, handle)
|
||||
handle.write("{not json")
|
||||
path = Path(handle.name)
|
||||
try:
|
||||
with self.assertRaises(ValueError):
|
||||
load_registry(path)
|
||||
finally:
|
||||
path.unlink(missing_ok=True)
|
||||
self._temp_paths.append(path)
|
||||
with self.assertRaises(RegistryError) as ctx:
|
||||
load_registry(path)
|
||||
self.assertIn("not valid JSON", ctx.exception.message)
|
||||
self.assertIn("line", ctx.exception.remediation)
|
||||
|
||||
def test_missing_file_fails_closed(self):
|
||||
missing = Path(tempfile.gettempdir()) / "webui-registry-does-not-exist.json"
|
||||
with self.assertRaises(RegistryError) as ctx:
|
||||
load_registry(missing)
|
||||
self.assertIn("could not be read", ctx.exception.message)
|
||||
|
||||
def test_default_registry_path_points_at_packaged_data(self):
|
||||
path = default_registry_path()
|
||||
@@ -81,31 +270,147 @@ class TestProjectRegistryRoutes(unittest.TestCase):
|
||||
self.assertIn("prgs-author", response.text)
|
||||
self.assertNotIn("child issue", response.text.lower())
|
||||
|
||||
def test_projects_page_shows_status_and_progress(self):
|
||||
response = self.client.get("/projects")
|
||||
self.assertIn("Status", response.text)
|
||||
self.assertIn("Onboarding", response.text)
|
||||
self.assertIn("4/4 complete", response.text)
|
||||
|
||||
def test_project_detail_renders_checklist(self):
|
||||
response = self.client.get("/projects/gitea-tools")
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertIn("Onboarding checklist", response.text)
|
||||
self.assertIn("Configure execution profiles", response.text)
|
||||
self.assertIn("branches/", response.text)
|
||||
self.assertIn("Complete", response.text)
|
||||
self.assertIn("required outstanding 0", response.text)
|
||||
|
||||
def test_project_detail_404(self):
|
||||
response = self.client.get("/projects/unknown-repo")
|
||||
self.assertEqual(response.status_code, 404)
|
||||
|
||||
def test_api_projects_json(self):
|
||||
def test_api_projects_alias_stays_compatible(self):
|
||||
response = self.client.get("/api/projects")
|
||||
self.assertEqual(response.status_code, 200)
|
||||
data = response.json()
|
||||
self.assertEqual(data["version"], 1)
|
||||
# #427 consumers keep these keys.
|
||||
self.assertEqual(data["version"], CURRENT_SCHEMA_VERSION)
|
||||
self.assertIn("source_path", data)
|
||||
self.assertEqual(len(data["projects"]), 1)
|
||||
self.assertEqual(data["projects"][0]["id"], "gitea-tools")
|
||||
self.assertIn("onboarding_checklist", data["projects"][0])
|
||||
|
||||
def test_api_v1_projects_payload(self):
|
||||
response = self.client.get("/api/v1/projects")
|
||||
self.assertEqual(response.status_code, 200)
|
||||
data = response.json()
|
||||
self.assertEqual(data["api_version"], REGISTRY_API_VERSION)
|
||||
self.assertEqual(data["schema_version"], CURRENT_SCHEMA_VERSION)
|
||||
self.assertEqual(data["project_count"], 1)
|
||||
self.assertEqual(data["source"]["kind"], "file")
|
||||
self.assertTrue(data["source"]["inventory_complete"])
|
||||
project = data["projects"][0]
|
||||
self.assertEqual(project["status"], "active")
|
||||
self.assertEqual(project["remote_name"], "prgs")
|
||||
self.assertEqual(
|
||||
project["repo_full_name"], "Scaled-Tech-Consulting/Gitea-Tools"
|
||||
)
|
||||
self.assertTrue(project["onboarding_summary"]["onboarding_complete"])
|
||||
self.assertEqual(project["onboarding_checklist"][0]["state"], "complete")
|
||||
self.assertIsNone(project["last_seen_health"])
|
||||
|
||||
def test_api_v1_project_detail(self):
|
||||
response = self.client.get("/api/v1/projects/gitea-tools")
|
||||
self.assertEqual(response.status_code, 200)
|
||||
data = response.json()
|
||||
self.assertEqual(data["api_version"], REGISTRY_API_VERSION)
|
||||
self.assertEqual(data["project"]["id"], "gitea-tools")
|
||||
self.assertEqual(data["source"]["kind"], "file")
|
||||
|
||||
def test_api_v1_project_detail_missing_fails_closed(self):
|
||||
response = self.client.get("/api/v1/projects/not-registered")
|
||||
self.assertEqual(response.status_code, 404)
|
||||
data = response.json()
|
||||
self.assertEqual(data["error"], "project_not_found")
|
||||
self.assertEqual(data["project_id"], "not-registered")
|
||||
self.assertIn("gitea-tools", data["known_project_ids"])
|
||||
self.assertIn("remediation", data)
|
||||
|
||||
def test_api_v1_projects_is_read_only(self):
|
||||
response = self.client.post("/api/v1/projects", json={})
|
||||
self.assertEqual(response.status_code, 405)
|
||||
self.assertEqual(response.json()["error"], "read-only-mvp")
|
||||
|
||||
def test_project_to_dict_is_json_safe(self):
|
||||
registry = load_registry()
|
||||
encoded = json.dumps(project_to_dict(registry.projects[0]))
|
||||
dto = project_to_dict(registry.projects[0])
|
||||
encoded = json.dumps(dto)
|
||||
self.assertIn("gitea-tools", encoded)
|
||||
# Prose may mention tokens; no serialized *key* may look like a secret.
|
||||
for key in dto:
|
||||
with self.subTest(key=key):
|
||||
self.assertFalse(is_forbidden_key(key))
|
||||
|
||||
|
||||
class TestInvalidRegistryFailsClosedOverHttp(RegistryFileCase):
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
self.path = self.write_registry({"version": 42, "projects": []})
|
||||
self.client = TestClient(create_app())
|
||||
|
||||
def _with_bad_registry(self, url: str):
|
||||
import os
|
||||
from unittest import mock
|
||||
|
||||
with mock.patch.dict(
|
||||
os.environ, {"WEBUI_PROJECT_REGISTRY": str(self.path)}, clear=False
|
||||
):
|
||||
return self.client.get(url)
|
||||
|
||||
def test_api_v1_reports_actionable_error(self):
|
||||
response = self._with_bad_registry("/api/v1/projects")
|
||||
self.assertEqual(response.status_code, 500)
|
||||
data = response.json()
|
||||
self.assertEqual(data["error"], "registry_invalid")
|
||||
self.assertIn("unsupported registry version", data["detail"])
|
||||
self.assertTrue(data["remediation"])
|
||||
self.assertEqual(data["field_path"], "version")
|
||||
|
||||
def test_unversioned_alias_reports_actionable_error(self):
|
||||
response = self._with_bad_registry("/api/projects")
|
||||
self.assertEqual(response.status_code, 500)
|
||||
self.assertEqual(response.json()["error"], "registry_invalid")
|
||||
|
||||
def test_html_page_reports_actionable_error(self):
|
||||
response = self._with_bad_registry("/projects")
|
||||
self.assertEqual(response.status_code, 500)
|
||||
self.assertIn("Project registry unavailable", response.text)
|
||||
self.assertIn("Remediation", response.text)
|
||||
|
||||
|
||||
class TestProjectRegistryApiDocs(unittest.TestCase):
|
||||
def test_api_contract_is_documented(self):
|
||||
self.assertTrue(_API_DOC.is_file(), f"missing {_API_DOC}")
|
||||
text = _API_DOC.read_text(encoding="utf-8")
|
||||
for token in (
|
||||
"/api/v1/projects",
|
||||
"/api/v1/projects/{project_id}",
|
||||
"/api/projects",
|
||||
"onboarding_summary",
|
||||
"last_seen_health",
|
||||
"registry_invalid",
|
||||
"#635",
|
||||
):
|
||||
with self.subTest(token=token):
|
||||
self.assertIn(token, text)
|
||||
|
||||
def test_route_table_lists_versioned_routes(self):
|
||||
local_dev = (_REPO_ROOT / "docs" / "webui-local-dev.md").read_text(
|
||||
encoding="utf-8"
|
||||
)
|
||||
self.assertIn("/api/v1/projects", local_dev)
|
||||
self.assertIn("webui-project-registry-api.md", local_dev)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
unittest.main()
|
||||
|
||||
+72
-85
@@ -2,7 +2,6 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from starlette.applications import Starlette
|
||||
@@ -12,17 +11,26 @@ from starlette.routing import Route
|
||||
|
||||
from webui.deployment_boundary import deployment_snapshot
|
||||
from webui.layout import render_page
|
||||
from webui.project_registry import find_project, load_registry, registry_to_dict
|
||||
from webui.project_views import render_project_detail, render_projects_list
|
||||
from webui.project_registry import (
|
||||
ProjectRegistry,
|
||||
RegistryError,
|
||||
find_project,
|
||||
known_project_ids,
|
||||
load_registry,
|
||||
project_detail_to_dict,
|
||||
registry_to_dict,
|
||||
)
|
||||
from webui.project_views import (
|
||||
render_project_detail,
|
||||
render_projects_list,
|
||||
render_registry_error,
|
||||
)
|
||||
from webui.prompt_library import find_prompt, library_to_dict
|
||||
from webui.prompt_views import render_prompt_detail, render_prompts_page
|
||||
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
|
||||
@@ -85,14 +93,26 @@ async def api_queue(_request: Request) -> JSONResponse:
|
||||
return JSONResponse(queue_snapshot_to_dict(load_queue_snapshot()))
|
||||
|
||||
|
||||
def _load_project_registry() -> tuple[ProjectRegistry | None, RegistryError | None]:
|
||||
"""Load the registry, converting validation failure into a fail-closed pair."""
|
||||
try:
|
||||
return load_registry(), None
|
||||
except RegistryError as exc:
|
||||
return None, exc
|
||||
|
||||
|
||||
async def projects(_request: Request) -> HTMLResponse:
|
||||
registry = load_registry()
|
||||
registry, error = _load_project_registry()
|
||||
if error is not None:
|
||||
return HTMLResponse(render_registry_error(error), status_code=500)
|
||||
return HTMLResponse(render_projects_list(registry))
|
||||
|
||||
|
||||
async def project_detail(request: Request) -> HTMLResponse:
|
||||
project_id = request.path_params["project_id"]
|
||||
registry = load_registry()
|
||||
registry, error = _load_project_registry()
|
||||
if error is not None:
|
||||
return HTMLResponse(render_registry_error(error), status_code=500)
|
||||
project = find_project(registry, project_id)
|
||||
if project is None:
|
||||
return HTMLResponse(
|
||||
@@ -110,10 +130,47 @@ async def project_detail(request: Request) -> HTMLResponse:
|
||||
|
||||
|
||||
async def api_projects(_request: Request) -> JSONResponse:
|
||||
registry = load_registry()
|
||||
"""Unversioned MVP alias, retained through Phase 1 (#632 section 6)."""
|
||||
registry, error = _load_project_registry()
|
||||
if error is not None:
|
||||
return JSONResponse(error.to_dict(), status_code=500)
|
||||
return JSONResponse(registry_to_dict(registry))
|
||||
|
||||
|
||||
async def api_v1_projects(_request: Request) -> JSONResponse:
|
||||
registry, error = _load_project_registry()
|
||||
if error is not None:
|
||||
return JSONResponse(error.to_dict(), status_code=500)
|
||||
return JSONResponse(registry_to_dict(registry))
|
||||
|
||||
|
||||
async def api_v1_project_detail(request: Request) -> JSONResponse:
|
||||
project_id = request.path_params["project_id"]
|
||||
registry, error = _load_project_registry()
|
||||
if error is not None:
|
||||
return JSONResponse(error.to_dict(), status_code=500)
|
||||
project = find_project(registry, project_id)
|
||||
if project is None:
|
||||
return JSONResponse(
|
||||
{
|
||||
"error": "project_not_found",
|
||||
"project_id": project_id,
|
||||
"known_project_ids": known_project_ids(registry),
|
||||
"remediation": (
|
||||
"Request one of the known project ids, or add the project to the "
|
||||
"registry file named in 'source'."
|
||||
),
|
||||
"source": {
|
||||
"kind": "file",
|
||||
"path": str(registry.source_path),
|
||||
"inventory_complete": True,
|
||||
},
|
||||
},
|
||||
status_code=404,
|
||||
)
|
||||
return JSONResponse(project_detail_to_dict(registry, project))
|
||||
|
||||
|
||||
async def prompts(_request: Request) -> HTMLResponse:
|
||||
return HTMLResponse(render_prompts_page())
|
||||
|
||||
@@ -219,49 +276,6 @@ 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)
|
||||
@@ -271,13 +285,6 @@ 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)
|
||||
|
||||
|
||||
@@ -291,31 +298,10 @@ 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":
|
||||
@@ -343,6 +329,12 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
|
||||
Route("/projects", projects, methods=["GET"]),
|
||||
Route("/projects/{project_id}", project_detail, methods=["GET"]),
|
||||
Route("/api/projects", api_projects, methods=["GET"]),
|
||||
Route("/api/v1/projects", api_v1_projects, methods=["GET"]),
|
||||
Route(
|
||||
"/api/v1/projects/{project_id}",
|
||||
api_v1_project_detail,
|
||||
methods=["GET"],
|
||||
),
|
||||
Route("/prompts", prompts, methods=["GET"]),
|
||||
Route("/prompts/{prompt_id}", prompt_detail, methods=["GET"]),
|
||||
Route("/api/prompts", api_prompts, methods=["GET"]),
|
||||
@@ -366,11 +358,6 @@ 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},
|
||||
)
|
||||
|
||||
@@ -1,281 +0,0 @@
|
||||
"""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."
|
||||
),
|
||||
}
|
||||
@@ -1,537 +0,0 @@
|
||||
"""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,
|
||||
}
|
||||
@@ -1,169 +0,0 @@
|
||||
"""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",
|
||||
}
|
||||
@@ -1,13 +1,15 @@
|
||||
{
|
||||
"version": 1,
|
||||
"version": 2,
|
||||
"projects": [
|
||||
{
|
||||
"id": "gitea-tools",
|
||||
"repo_name": "Gitea-Tools",
|
||||
"gitea_owner": "Scaled-Tech-Consulting",
|
||||
"remote_name": "prgs",
|
||||
"remote_host": "https://gitea.prgs.cc",
|
||||
"default_branch": "master",
|
||||
"local_checkout_path": ".",
|
||||
"status": "active",
|
||||
"profiles": {
|
||||
"author": "prgs-author",
|
||||
"reviewer": "prgs-reviewer",
|
||||
@@ -26,24 +28,32 @@
|
||||
{
|
||||
"id": "profiles",
|
||||
"title": "Configure execution profiles",
|
||||
"description": "Install author, reviewer, and reconciler MCP profiles (prgs-author, prgs-reviewer, prgs-reconciler) in separate namespaces. Tokens stay in keychain — never in this registry."
|
||||
"description": "Install author, reviewer, and reconciler MCP profiles (prgs-author, prgs-reviewer, prgs-reconciler) in separate namespaces. Tokens stay in keychain — never in this registry.",
|
||||
"state": "complete",
|
||||
"required": true
|
||||
},
|
||||
{
|
||||
"id": "mcp_config",
|
||||
"title": "Wire MCP v2 contexts",
|
||||
"description": "Copy and customize gitea-mcp.v2-contexts.example.json for your machine. Map this repo path under projects with default_owner Scaled-Tech-Consulting and default_repo Gitea-Tools."
|
||||
"description": "Copy and customize gitea-mcp.v2-contexts.example.json for your machine. Map this repo path under projects with default_owner Scaled-Tech-Consulting and default_repo Gitea-Tools.",
|
||||
"state": "complete",
|
||||
"required": true
|
||||
},
|
||||
{
|
||||
"id": "wiki_gate",
|
||||
"title": "Wiki publication readiness",
|
||||
"description": "For wiki-tracked work, satisfy the live Gitea Wiki proof gate (#224) before closing issues. See docs/wiki/Safety-and-Gates.md."
|
||||
"description": "For wiki-tracked work, satisfy the live Gitea Wiki proof gate (#224) before closing issues. See docs/wiki/Safety-and-Gates.md.",
|
||||
"state": "complete",
|
||||
"required": true
|
||||
},
|
||||
{
|
||||
"id": "branches_layout",
|
||||
"title": "Isolate work under branches/",
|
||||
"description": "All LLM task edits happen in worktrees under branches/. Main checkout stays clean; use skills/llm-project-workflow templates for start-issue and review flows."
|
||||
"description": "All LLM task edits happen in worktrees under branches/. Main checkout stays clean; use skills/llm-project-workflow templates for start-issue and review flows.",
|
||||
"state": "complete",
|
||||
"required": true
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
+460
-25
@@ -1,4 +1,16 @@
|
||||
"""Load and validate the web UI project registry (#427)."""
|
||||
"""Load and validate the web UI project registry (#427, evolved for #635).
|
||||
|
||||
Phase 1 of the console architecture ADR keeps this loader read-only. It owns
|
||||
the versioned project registry contract served at ``/api/v1/projects``:
|
||||
|
||||
* the on-disk file carries a ``version`` (schema version 1 or 2);
|
||||
* version 1 files stay loadable and are normalized with explicit defaults, so
|
||||
an operator registry written for #427 keeps working;
|
||||
* every validation failure raises :class:`RegistryError`, which carries an
|
||||
actionable ``remediation`` string instead of leaking a traceback;
|
||||
* serialization never emits credentials — credential-shaped keys are rejected
|
||||
at load time, before any DTO is built.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
@@ -8,7 +20,59 @@ from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from webui.registry_safety import reject_credential_keys as _reject_credential_keys
|
||||
from webui.registry_safety import is_forbidden_key
|
||||
|
||||
#: Version of the JSON contract served under ``/api/v1/...``.
|
||||
REGISTRY_API_VERSION = "v1"
|
||||
|
||||
#: Schema version written by this repository's packaged registry.
|
||||
CURRENT_SCHEMA_VERSION = 2
|
||||
|
||||
#: Schema versions this loader accepts. Version 1 is normalized on load.
|
||||
SUPPORTED_SCHEMA_VERSIONS = (1, 2)
|
||||
|
||||
#: Lifecycle state of a registered project.
|
||||
PROJECT_STATUSES = ("active", "onboarding", "paused", "archived")
|
||||
_DEFAULT_PROJECT_STATUS = "active"
|
||||
|
||||
#: Completion state of a single onboarding step.
|
||||
ONBOARDING_STATES = ("complete", "pending", "blocked", "not_applicable")
|
||||
_DEFAULT_ONBOARDING_STATE = "pending"
|
||||
|
||||
#: Redacted, last-seen health of a project's control plane.
|
||||
HEALTH_STATUSES = ("healthy", "degraded", "unreachable", "unknown")
|
||||
|
||||
class RegistryError(ValueError):
|
||||
"""A registry file could not be loaded or failed validation.
|
||||
|
||||
Carries an operator-facing ``remediation`` so routes can fail closed with
|
||||
an actionable message rather than a stack trace.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
message: str,
|
||||
*,
|
||||
remediation: str,
|
||||
source_path: Path | None = None,
|
||||
field_path: str | None = None,
|
||||
) -> None:
|
||||
super().__init__(message)
|
||||
self.message = message
|
||||
self.remediation = remediation
|
||||
self.source_path = source_path
|
||||
self.field_path = field_path
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
"""Serialize for a fail-closed JSON error response."""
|
||||
return {
|
||||
"error": "registry_invalid",
|
||||
"detail": self.message,
|
||||
"remediation": self.remediation,
|
||||
"field_path": self.field_path,
|
||||
"source_path": str(self.source_path) if self.source_path else None,
|
||||
}
|
||||
|
||||
|
||||
_REQUIRED_PROJECT_FIELDS = (
|
||||
"id",
|
||||
@@ -29,6 +93,30 @@ class OnboardingStep:
|
||||
id: str
|
||||
title: str
|
||||
description: str
|
||||
state: str = _DEFAULT_ONBOARDING_STATE
|
||||
required: bool = True
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class OnboardingSummary:
|
||||
"""Aggregate onboarding progress for a single project."""
|
||||
|
||||
total: int
|
||||
complete: int
|
||||
pending: int
|
||||
blocked: int
|
||||
not_applicable: int
|
||||
required_outstanding: int
|
||||
onboarding_complete: bool
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ProjectHealth:
|
||||
"""Redacted last-seen health. Never carries endpoints or credentials."""
|
||||
|
||||
status: str
|
||||
checked_at: str | None
|
||||
detail: str | None
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
@@ -43,6 +131,13 @@ class ProjectRecord:
|
||||
workflow_paths: dict[str, str]
|
||||
schema_paths: dict[str, str]
|
||||
onboarding_checklist: tuple[OnboardingStep, ...]
|
||||
status: str = _DEFAULT_PROJECT_STATUS
|
||||
remote_name: str | None = None
|
||||
last_seen_health: ProjectHealth | None = None
|
||||
|
||||
@property
|
||||
def repo_full_name(self) -> str:
|
||||
return f"{self.gitea_owner}/{self.repo_name}"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
@@ -51,6 +146,15 @@ class ProjectRegistry:
|
||||
projects: tuple[ProjectRecord, ...]
|
||||
source_path: Path
|
||||
|
||||
@property
|
||||
def schema_version(self) -> int:
|
||||
"""Alias of :attr:`version` — the schema version read from disk."""
|
||||
return self.version
|
||||
|
||||
@property
|
||||
def api_version(self) -> str:
|
||||
return REGISTRY_API_VERSION
|
||||
|
||||
|
||||
def default_registry_path() -> Path:
|
||||
override = os.environ.get("WEBUI_PROJECT_REGISTRY", "").strip()
|
||||
@@ -59,40 +163,221 @@ def default_registry_path() -> Path:
|
||||
return (Path(__file__).resolve().parent / "data" / "projects.registry.json").resolve()
|
||||
|
||||
|
||||
def _parse_onboarding(raw: list[dict[str, Any]] | None) -> tuple[OnboardingStep, ...]:
|
||||
if not raw:
|
||||
def _reject_credential_keys(obj: Any, *, path: str = "", source: Path | None = None) -> None:
|
||||
"""Recursive credential-key guard that reports an actionable ``field_path``.
|
||||
|
||||
Key *shape* is decided by :func:`webui.registry_safety.is_forbidden_key`, the
|
||||
single source of truth shared with the worker registry (#798).
|
||||
"""
|
||||
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 RegistryError(
|
||||
f"registry must not store credentials ({key_path})",
|
||||
remediation=(
|
||||
f"Remove the credential-shaped key '{key_path}' from the registry. "
|
||||
"Tokens live in the keychain and are resolved server-side by "
|
||||
"gitea_auth; the registry is redacted metadata only."
|
||||
),
|
||||
source_path=source,
|
||||
field_path=key_path,
|
||||
)
|
||||
_reject_credential_keys(value, path=key_path, source=source)
|
||||
elif isinstance(obj, list):
|
||||
for index, item in enumerate(obj):
|
||||
_reject_credential_keys(item, path=f"{path}[{index}]", source=source)
|
||||
|
||||
|
||||
def _require_enum(
|
||||
value: Any,
|
||||
*,
|
||||
allowed: tuple[str, ...],
|
||||
field_path: str,
|
||||
source: Path | None,
|
||||
) -> str:
|
||||
text = str(value)
|
||||
if text not in allowed:
|
||||
raise RegistryError(
|
||||
f"{field_path} must be one of {', '.join(allowed)} (got {text!r})",
|
||||
remediation=(
|
||||
f"Set {field_path} to one of: {', '.join(allowed)}. "
|
||||
"Unknown values fail closed so the console never renders an "
|
||||
"unverified state."
|
||||
),
|
||||
source_path=source,
|
||||
field_path=field_path,
|
||||
)
|
||||
return text
|
||||
|
||||
|
||||
def _parse_onboarding(
|
||||
raw: Any,
|
||||
*,
|
||||
project_path: str,
|
||||
source: Path | None,
|
||||
) -> tuple[OnboardingStep, ...]:
|
||||
if raw is None:
|
||||
return ()
|
||||
if not isinstance(raw, list):
|
||||
raise RegistryError(
|
||||
f"{project_path}.onboarding_checklist must be an array",
|
||||
remediation=(
|
||||
f"Rewrite {project_path}.onboarding_checklist as a JSON array of "
|
||||
"steps with id, title, description, and optional state."
|
||||
),
|
||||
source_path=source,
|
||||
field_path=f"{project_path}.onboarding_checklist",
|
||||
)
|
||||
steps: list[OnboardingStep] = []
|
||||
for item in raw:
|
||||
for index, item in enumerate(raw):
|
||||
step_path = f"{project_path}.onboarding_checklist[{index}]"
|
||||
if not isinstance(item, dict):
|
||||
raise RegistryError(
|
||||
f"{step_path} must be an object",
|
||||
remediation=f"Rewrite {step_path} as an object with id, title, description.",
|
||||
source_path=source,
|
||||
field_path=step_path,
|
||||
)
|
||||
missing = [field for field in ("id", "title", "description") if field not in item]
|
||||
if missing:
|
||||
raise RegistryError(
|
||||
f"{step_path} missing required fields: {', '.join(missing)}",
|
||||
remediation=(
|
||||
f"Add {', '.join(missing)} to {step_path}. Every onboarding step "
|
||||
"must be self-describing for an operator who has no chat history."
|
||||
),
|
||||
source_path=source,
|
||||
field_path=step_path,
|
||||
)
|
||||
state = _require_enum(
|
||||
item.get("state", _DEFAULT_ONBOARDING_STATE),
|
||||
allowed=ONBOARDING_STATES,
|
||||
field_path=f"{step_path}.state",
|
||||
source=source,
|
||||
)
|
||||
steps.append(
|
||||
OnboardingStep(
|
||||
id=str(item["id"]),
|
||||
title=str(item["title"]),
|
||||
description=str(item["description"]),
|
||||
state=state,
|
||||
required=bool(item.get("required", True)),
|
||||
)
|
||||
)
|
||||
return tuple(steps)
|
||||
|
||||
|
||||
def _parse_project(raw: dict[str, Any]) -> ProjectRecord:
|
||||
def _parse_health(
|
||||
raw: Any,
|
||||
*,
|
||||
project_path: str,
|
||||
source: Path | None,
|
||||
) -> ProjectHealth | None:
|
||||
if raw is None:
|
||||
return None
|
||||
if not isinstance(raw, dict):
|
||||
raise RegistryError(
|
||||
f"{project_path}.last_seen_health must be an object when present",
|
||||
remediation=(
|
||||
f"Rewrite {project_path}.last_seen_health as an object with status "
|
||||
f"(one of {', '.join(HEALTH_STATUSES)}), optional checked_at and detail, "
|
||||
"or remove it. Never store endpoints or credentials here."
|
||||
),
|
||||
source_path=source,
|
||||
field_path=f"{project_path}.last_seen_health",
|
||||
)
|
||||
status = _require_enum(
|
||||
raw.get("status", "unknown"),
|
||||
allowed=HEALTH_STATUSES,
|
||||
field_path=f"{project_path}.last_seen_health.status",
|
||||
source=source,
|
||||
)
|
||||
checked_at = raw.get("checked_at")
|
||||
detail = raw.get("detail")
|
||||
return ProjectHealth(
|
||||
status=status,
|
||||
checked_at=str(checked_at) if checked_at is not None else None,
|
||||
detail=str(detail) if detail is not None else None,
|
||||
)
|
||||
|
||||
|
||||
def _parse_project(raw: Any, *, index: int, source: Path | None) -> ProjectRecord:
|
||||
project_path = f"projects[{index}]"
|
||||
if not isinstance(raw, dict):
|
||||
raise RegistryError(
|
||||
f"{project_path} must be an object",
|
||||
remediation=f"Rewrite {project_path} as a JSON object describing one project.",
|
||||
source_path=source,
|
||||
field_path=project_path,
|
||||
)
|
||||
|
||||
missing = [field for field in _REQUIRED_PROJECT_FIELDS if field not in raw]
|
||||
if missing:
|
||||
raise ValueError(f"project missing required fields: {', '.join(missing)}")
|
||||
raise RegistryError(
|
||||
f"{project_path} missing required fields: {', '.join(missing)}",
|
||||
remediation=(
|
||||
f"Add {', '.join(missing)} to {project_path}. See "
|
||||
"docs/webui-project-registry-api.md for the field-by-field contract."
|
||||
),
|
||||
source_path=source,
|
||||
field_path=project_path,
|
||||
)
|
||||
|
||||
profiles = raw["profiles"]
|
||||
if not isinstance(profiles, dict):
|
||||
raise ValueError("profiles must be an object")
|
||||
raise RegistryError(
|
||||
f"{project_path}.profiles must be an object",
|
||||
remediation=(
|
||||
f"Rewrite {project_path}.profiles as an object mapping "
|
||||
f"{', '.join(_REQUIRED_PROFILE_ROLES)} to MCP profile names."
|
||||
),
|
||||
source_path=source,
|
||||
field_path=f"{project_path}.profiles",
|
||||
)
|
||||
for role in _REQUIRED_PROFILE_ROLES:
|
||||
if role not in profiles or not profiles[role]:
|
||||
raise ValueError(f"profiles.{role} is required")
|
||||
raise RegistryError(
|
||||
f"{project_path}.profiles.{role} is required",
|
||||
remediation=(
|
||||
f"Set {project_path}.profiles.{role} to the configured MCP profile "
|
||||
"name for that role. Role separation is a workflow-safety invariant."
|
||||
),
|
||||
source_path=source,
|
||||
field_path=f"{project_path}.profiles.{role}",
|
||||
)
|
||||
|
||||
workflow_paths = raw["workflow_paths"]
|
||||
if not isinstance(workflow_paths, dict) or not workflow_paths:
|
||||
raise ValueError("workflow_paths must be a non-empty object")
|
||||
raise RegistryError(
|
||||
f"{project_path}.workflow_paths must be a non-empty object",
|
||||
remediation=(
|
||||
f"Add at least a 'skill' entry to {project_path}.workflow_paths pointing "
|
||||
"at the project's canonical workflow skill."
|
||||
),
|
||||
source_path=source,
|
||||
field_path=f"{project_path}.workflow_paths",
|
||||
)
|
||||
|
||||
schema_paths = raw.get("schema_paths") or {}
|
||||
if not isinstance(schema_paths, dict):
|
||||
raise ValueError("schema_paths must be an object when present")
|
||||
raise RegistryError(
|
||||
f"{project_path}.schema_paths must be an object when present",
|
||||
remediation=(
|
||||
f"Rewrite {project_path}.schema_paths as an object of label to repo path, "
|
||||
"or remove it."
|
||||
),
|
||||
source_path=source,
|
||||
field_path=f"{project_path}.schema_paths",
|
||||
)
|
||||
|
||||
status = _require_enum(
|
||||
raw.get("status", _DEFAULT_PROJECT_STATUS),
|
||||
allowed=PROJECT_STATUSES,
|
||||
field_path=f"{project_path}.status",
|
||||
source=source,
|
||||
)
|
||||
remote_name = raw.get("remote_name")
|
||||
|
||||
return ProjectRecord(
|
||||
id=str(raw["id"]),
|
||||
@@ -104,61 +389,211 @@ def _parse_project(raw: dict[str, Any]) -> ProjectRecord:
|
||||
profiles={role: str(profiles[role]) for role in _REQUIRED_PROFILE_ROLES},
|
||||
workflow_paths={key: str(value) for key, value in workflow_paths.items()},
|
||||
schema_paths={key: str(value) for key, value in schema_paths.items()},
|
||||
onboarding_checklist=_parse_onboarding(raw.get("onboarding_checklist")),
|
||||
onboarding_checklist=_parse_onboarding(
|
||||
raw.get("onboarding_checklist"),
|
||||
project_path=project_path,
|
||||
source=source,
|
||||
),
|
||||
status=status,
|
||||
remote_name=str(remote_name) if remote_name else None,
|
||||
last_seen_health=_parse_health(
|
||||
raw.get("last_seen_health"),
|
||||
project_path=project_path,
|
||||
source=source,
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def load_registry(path: Path | None = None) -> ProjectRegistry:
|
||||
"""Load the versioned project registry from disk."""
|
||||
"""Load the versioned project registry from disk.
|
||||
|
||||
Raises:
|
||||
RegistryError: whenever the file is unreadable, is not valid JSON, or
|
||||
fails schema validation. The error carries an operator remediation.
|
||||
"""
|
||||
source = (path or default_registry_path()).resolve()
|
||||
raw_text = source.read_text(encoding="utf-8")
|
||||
payload = json.loads(raw_text)
|
||||
try:
|
||||
raw_text = source.read_text(encoding="utf-8")
|
||||
except OSError as exc:
|
||||
raise RegistryError(
|
||||
f"registry file could not be read: {exc.strerror or exc}",
|
||||
remediation=(
|
||||
f"Create a readable registry at {source}, or point "
|
||||
"WEBUI_PROJECT_REGISTRY at an existing file."
|
||||
),
|
||||
source_path=source,
|
||||
) from exc
|
||||
|
||||
try:
|
||||
payload = json.loads(raw_text)
|
||||
except json.JSONDecodeError as exc:
|
||||
raise RegistryError(
|
||||
f"registry is not valid JSON: {exc.msg} (line {exc.lineno}, column {exc.colno})",
|
||||
remediation=(
|
||||
f"Fix the JSON syntax in {source} at line {exc.lineno}, column {exc.colno}."
|
||||
),
|
||||
source_path=source,
|
||||
) from exc
|
||||
|
||||
if not isinstance(payload, dict):
|
||||
raise ValueError("registry root must be an object")
|
||||
raise RegistryError(
|
||||
"registry root must be an object",
|
||||
remediation=(
|
||||
"Wrap the registry in a JSON object with 'version' and 'projects' keys."
|
||||
),
|
||||
source_path=source,
|
||||
)
|
||||
|
||||
version = payload.get("version")
|
||||
if version != 1:
|
||||
raise ValueError(f"unsupported registry version: {version!r}")
|
||||
if version not in SUPPORTED_SCHEMA_VERSIONS:
|
||||
supported = ", ".join(str(item) for item in SUPPORTED_SCHEMA_VERSIONS)
|
||||
raise RegistryError(
|
||||
f"unsupported registry version: {version!r}",
|
||||
remediation=(
|
||||
f"Set 'version' to one of {supported} (current schema is "
|
||||
f"{CURRENT_SCHEMA_VERSION}). Migration notes live in "
|
||||
"docs/webui-project-registry-api.md."
|
||||
),
|
||||
source_path=source,
|
||||
field_path="version",
|
||||
)
|
||||
|
||||
_reject_credential_keys(payload)
|
||||
_reject_credential_keys(payload, source=source)
|
||||
|
||||
projects_raw = payload.get("projects")
|
||||
if not isinstance(projects_raw, list) or not projects_raw:
|
||||
raise ValueError("projects must be a non-empty array")
|
||||
raise RegistryError(
|
||||
"projects must be a non-empty array",
|
||||
remediation=(
|
||||
"Add at least one project object to 'projects'. An empty console "
|
||||
"registry fails closed rather than rendering a blank inventory."
|
||||
),
|
||||
source_path=source,
|
||||
field_path="projects",
|
||||
)
|
||||
|
||||
projects = tuple(_parse_project(item) for item in projects_raw)
|
||||
return ProjectRegistry(version=version, projects=projects, source_path=source)
|
||||
projects = tuple(
|
||||
_parse_project(item, index=index, source=source)
|
||||
for index, item in enumerate(projects_raw)
|
||||
)
|
||||
return ProjectRegistry(version=int(version), projects=projects, source_path=source)
|
||||
|
||||
|
||||
def onboarding_summary(project: ProjectRecord) -> OnboardingSummary:
|
||||
"""Aggregate a project's onboarding checklist state."""
|
||||
steps = project.onboarding_checklist
|
||||
counts = {state: 0 for state in ONBOARDING_STATES}
|
||||
for step in steps:
|
||||
counts[step.state] += 1
|
||||
required_outstanding = sum(
|
||||
1
|
||||
for step in steps
|
||||
if step.required and step.state in ("pending", "blocked")
|
||||
)
|
||||
return OnboardingSummary(
|
||||
total=len(steps),
|
||||
complete=counts["complete"],
|
||||
pending=counts["pending"],
|
||||
blocked=counts["blocked"],
|
||||
not_applicable=counts["not_applicable"],
|
||||
required_outstanding=required_outstanding,
|
||||
onboarding_complete=required_outstanding == 0,
|
||||
)
|
||||
|
||||
|
||||
def project_to_dict(project: ProjectRecord) -> dict[str, Any]:
|
||||
"""Serialize a project for JSON API responses."""
|
||||
"""Serialize a project for JSON API responses and HTML views.
|
||||
|
||||
The HTML views render from this same DTO, so the console and the API can
|
||||
never disagree about a project's status or onboarding progress.
|
||||
"""
|
||||
summary = onboarding_summary(project)
|
||||
health = project.last_seen_health
|
||||
return {
|
||||
"id": project.id,
|
||||
"repo_name": project.repo_name,
|
||||
"gitea_owner": project.gitea_owner,
|
||||
"repo_full_name": project.repo_full_name,
|
||||
"remote_host": project.remote_host,
|
||||
"remote_name": project.remote_name,
|
||||
"default_branch": project.default_branch,
|
||||
"local_checkout_path": project.local_checkout_path,
|
||||
"status": project.status,
|
||||
"profiles": dict(project.profiles),
|
||||
"workflow_paths": dict(project.workflow_paths),
|
||||
"schema_paths": dict(project.schema_paths),
|
||||
"onboarding_checklist": [
|
||||
{"id": step.id, "title": step.title, "description": step.description}
|
||||
{
|
||||
"id": step.id,
|
||||
"title": step.title,
|
||||
"description": step.description,
|
||||
"state": step.state,
|
||||
"required": step.required,
|
||||
}
|
||||
for step in project.onboarding_checklist
|
||||
],
|
||||
"onboarding_summary": {
|
||||
"total": summary.total,
|
||||
"complete": summary.complete,
|
||||
"pending": summary.pending,
|
||||
"blocked": summary.blocked,
|
||||
"not_applicable": summary.not_applicable,
|
||||
"required_outstanding": summary.required_outstanding,
|
||||
"onboarding_complete": summary.onboarding_complete,
|
||||
},
|
||||
"last_seen_health": (
|
||||
None
|
||||
if health is None
|
||||
else {
|
||||
"status": health.status,
|
||||
"checked_at": health.checked_at,
|
||||
"detail": health.detail,
|
||||
}
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def registry_to_dict(registry: ProjectRegistry) -> dict[str, Any]:
|
||||
"""Serialize the whole registry, including API provenance (#632 section 6)."""
|
||||
return {
|
||||
"api_version": registry.api_version,
|
||||
"schema_version": registry.schema_version,
|
||||
# Retained for the unversioned MVP alias consumers (#427).
|
||||
"version": registry.version,
|
||||
"source_path": str(registry.source_path),
|
||||
"source": {
|
||||
"kind": "file",
|
||||
"path": str(registry.source_path),
|
||||
"inventory_complete": True,
|
||||
},
|
||||
"project_count": len(registry.projects),
|
||||
"projects": [project_to_dict(project) for project in registry.projects],
|
||||
}
|
||||
|
||||
|
||||
def project_detail_to_dict(
|
||||
registry: ProjectRegistry,
|
||||
project: ProjectRecord,
|
||||
) -> dict[str, Any]:
|
||||
"""Serialize a single project for ``/api/v1/projects/{project_id}``."""
|
||||
return {
|
||||
"api_version": registry.api_version,
|
||||
"schema_version": registry.schema_version,
|
||||
"source": {
|
||||
"kind": "file",
|
||||
"path": str(registry.source_path),
|
||||
"inventory_complete": True,
|
||||
},
|
||||
"project": project_to_dict(project),
|
||||
}
|
||||
|
||||
|
||||
def find_project(registry: ProjectRegistry, project_id: str) -> ProjectRecord | None:
|
||||
for project in registry.projects:
|
||||
if project.id == project_id:
|
||||
return project
|
||||
return None
|
||||
return None
|
||||
|
||||
|
||||
def known_project_ids(registry: ProjectRegistry) -> list[str]:
|
||||
return [project.id for project in registry.projects]
|
||||
|
||||
+107
-25
@@ -1,34 +1,65 @@
|
||||
"""HTML views for project registry pages (#427)."""
|
||||
"""HTML views for project registry pages (#427, evolved for #635).
|
||||
|
||||
Every view renders from :func:`webui.project_registry.project_to_dict`, the
|
||||
same DTO the ``/api/v1/projects`` JSON responses use, so the HTML console and
|
||||
the API can never disagree about status or onboarding progress.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import html
|
||||
from typing import Any
|
||||
|
||||
from webui.layout import render_page
|
||||
from webui.project_registry import ProjectRecord, ProjectRegistry
|
||||
from webui.project_registry import (
|
||||
ProjectRecord,
|
||||
ProjectRegistry,
|
||||
RegistryError,
|
||||
project_to_dict,
|
||||
)
|
||||
|
||||
_STATE_LABELS = {
|
||||
"complete": "Complete",
|
||||
"pending": "Pending",
|
||||
"blocked": "Blocked",
|
||||
"not_applicable": "Not applicable",
|
||||
}
|
||||
|
||||
|
||||
def _escape(text: str) -> str:
|
||||
return html.escape(text, quote=True)
|
||||
|
||||
|
||||
def _progress_label(summary: dict[str, Any]) -> str:
|
||||
total = summary["total"]
|
||||
if not total:
|
||||
return "no steps"
|
||||
label = f"{summary['complete']}/{total} complete"
|
||||
if summary["blocked"]:
|
||||
label += f", {summary['blocked']} blocked"
|
||||
return label
|
||||
|
||||
|
||||
def render_projects_list(registry: ProjectRegistry) -> str:
|
||||
rows = []
|
||||
for project in registry.projects:
|
||||
dto = project_to_dict(project)
|
||||
rows.append(
|
||||
"<tr>"
|
||||
f"<td><a href=\"/projects/{_escape(project.id)}\">{_escape(project.repo_name)}</a></td>"
|
||||
f"<td>{_escape(project.gitea_owner)}</td>"
|
||||
f"<td>{_escape(project.remote_host)}</td>"
|
||||
f"<td>{_escape(project.default_branch)}</td>"
|
||||
f"<td><code>{_escape(project.profiles['author'])}</code></td>"
|
||||
f"<td><a href=\"/projects/{_escape(dto['id'])}\">{_escape(dto['repo_name'])}</a></td>"
|
||||
f"<td>{_escape(dto['gitea_owner'])}</td>"
|
||||
f"<td>{_escape(dto['remote_host'])}</td>"
|
||||
f"<td>{_escape(dto['default_branch'])}</td>"
|
||||
f"<td><code>{_escape(dto['status'])}</code></td>"
|
||||
f"<td>{_escape(_progress_label(dto['onboarding_summary']))}</td>"
|
||||
f"<td><code>{_escape(dto['profiles']['author'])}</code></td>"
|
||||
"</tr>"
|
||||
)
|
||||
table = (
|
||||
"<table class=\"registry\">"
|
||||
"<thead><tr>"
|
||||
"<th>Repository</th><th>Owner</th><th>Remote</th>"
|
||||
"<th>Branch</th><th>Author profile</th>"
|
||||
"<th>Branch</th><th>Status</th><th>Onboarding</th><th>Author profile</th>"
|
||||
"</tr></thead>"
|
||||
f"<tbody>{''.join(rows)}</tbody></table>"
|
||||
)
|
||||
@@ -36,32 +67,39 @@ def render_projects_list(registry: ProjectRegistry) -> str:
|
||||
"<h2>Projects</h2>"
|
||||
"<p>Configured repositories managed by the MCP Control Plane.</p>"
|
||||
f"<p class=\"meta\">Registry: <code>{_escape(str(registry.source_path))}</code> "
|
||||
f"(version {registry.version})</p>"
|
||||
f"(schema version {registry.schema_version}, "
|
||||
f"API {_escape(registry.api_version)})</p>"
|
||||
f"{table}"
|
||||
"<p><a href=\"/api/projects\">JSON API</a></p>"
|
||||
"<p><a href=\"/api/v1/projects\">JSON API</a> "
|
||||
"(<a href=\"/api/projects\">unversioned alias</a>)</p>"
|
||||
)
|
||||
return render_page(title="Projects", body_html=body)
|
||||
|
||||
|
||||
def render_project_detail(project: ProjectRecord) -> str:
|
||||
dto = project_to_dict(project)
|
||||
profile_rows = "".join(
|
||||
f"<tr><th>{_escape(role)}</th><td><code>{_escape(name)}</code></td></tr>"
|
||||
for role, name in project.profiles.items()
|
||||
for role, name in dto["profiles"].items()
|
||||
)
|
||||
workflow_rows = "".join(
|
||||
f"<tr><th>{_escape(key)}</th><td><code>{_escape(path)}</code></td></tr>"
|
||||
for key, path in project.workflow_paths.items()
|
||||
for key, path in dto["workflow_paths"].items()
|
||||
)
|
||||
schema_rows = "".join(
|
||||
f"<tr><th>{_escape(key)}</th><td><code>{_escape(path)}</code></td></tr>"
|
||||
for key, path in project.schema_paths.items()
|
||||
for key, path in dto["schema_paths"].items()
|
||||
)
|
||||
checklist_items = []
|
||||
for index, step in enumerate(project.onboarding_checklist, start=1):
|
||||
for index, step in enumerate(dto["onboarding_checklist"], start=1):
|
||||
state_label = _STATE_LABELS.get(step["state"], step["state"])
|
||||
requirement = "required" if step["required"] else "optional"
|
||||
checklist_items.append(
|
||||
"<li>"
|
||||
f"<strong>{index}. {_escape(step.title)}</strong>"
|
||||
f"<p>{_escape(step.description)}</p>"
|
||||
f"<li class=\"step-{_escape(step['state'])}\">"
|
||||
f"<strong>{index}. {_escape(step['title'])}</strong>"
|
||||
f" <span class=\"badge\">{_escape(state_label)}</span>"
|
||||
f" <span class=\"meta\">({_escape(requirement)})</span>"
|
||||
f"<p>{_escape(step['description'])}</p>"
|
||||
"</li>"
|
||||
)
|
||||
checklist_html = (
|
||||
@@ -69,16 +107,38 @@ def render_project_detail(project: ProjectRecord) -> str:
|
||||
if checklist_items
|
||||
else "<p>No onboarding steps defined.</p>"
|
||||
)
|
||||
summary = dto["onboarding_summary"]
|
||||
summary_html = (
|
||||
"<p class=\"meta\">Onboarding: "
|
||||
f"{_escape(_progress_label(summary))}; required outstanding "
|
||||
f"{summary['required_outstanding']}.</p>"
|
||||
)
|
||||
health = dto["last_seen_health"]
|
||||
health_html = (
|
||||
"<p class=\"meta\">No health probe recorded (Phase 1 is read-only).</p>"
|
||||
if health is None
|
||||
else (
|
||||
"<table class=\"detail\">"
|
||||
f"<tr><th>Status</th><td><code>{_escape(health['status'])}</code></td></tr>"
|
||||
f"<tr><th>Checked at</th><td>{_escape(str(health['checked_at'] or 'unknown'))}</td></tr>"
|
||||
f"<tr><th>Detail</th><td>{_escape(str(health['detail'] or ''))}</td></tr>"
|
||||
"</table>"
|
||||
)
|
||||
)
|
||||
remote_name = dto["remote_name"] or "unset"
|
||||
body = (
|
||||
f"<h2>{_escape(project.repo_name)}</h2>"
|
||||
f"<h2>{_escape(dto['repo_name'])}</h2>"
|
||||
"<p><a href=\"/projects\">← All projects</a></p>"
|
||||
"<h3>Identity</h3>"
|
||||
"<table class=\"detail\">"
|
||||
f"<tr><th>Registry id</th><td><code>{_escape(project.id)}</code></td></tr>"
|
||||
f"<tr><th>Gitea owner</th><td>{_escape(project.gitea_owner)}</td></tr>"
|
||||
f"<tr><th>Remote host</th><td>{_escape(project.remote_host)}</td></tr>"
|
||||
f"<tr><th>Default branch</th><td><code>{_escape(project.default_branch)}</code></td></tr>"
|
||||
f"<tr><th>Local checkout</th><td><code>{_escape(project.local_checkout_path)}</code></td></tr>"
|
||||
f"<tr><th>Registry id</th><td><code>{_escape(dto['id'])}</code></td></tr>"
|
||||
f"<tr><th>Status</th><td><code>{_escape(dto['status'])}</code></td></tr>"
|
||||
f"<tr><th>Gitea owner</th><td>{_escape(dto['gitea_owner'])}</td></tr>"
|
||||
f"<tr><th>Repository</th><td><code>{_escape(dto['repo_full_name'])}</code></td></tr>"
|
||||
f"<tr><th>Remote name</th><td><code>{_escape(remote_name)}</code></td></tr>"
|
||||
f"<tr><th>Remote host</th><td>{_escape(dto['remote_host'])}</td></tr>"
|
||||
f"<tr><th>Default branch</th><td><code>{_escape(dto['default_branch'])}</code></td></tr>"
|
||||
f"<tr><th>Local checkout</th><td><code>{_escape(dto['local_checkout_path'])}</code></td></tr>"
|
||||
"</table>"
|
||||
"<h3>Profiles</h3>"
|
||||
f"<table class=\"detail\">{profile_rows}</table>"
|
||||
@@ -86,8 +146,30 @@ def render_project_detail(project: ProjectRecord) -> str:
|
||||
f"<table class=\"detail\">{workflow_rows}</table>"
|
||||
"<h3>Schema paths</h3>"
|
||||
f"<table class=\"detail\">{schema_rows}</table>"
|
||||
"<h3>Last seen health</h3>"
|
||||
f"{health_html}"
|
||||
"<h3>Onboarding checklist</h3>"
|
||||
"<p class=\"meta\">Read-only MVP — complete these steps outside the UI.</p>"
|
||||
"<p class=\"meta\">Read-only — complete these steps outside the UI.</p>"
|
||||
f"{summary_html}"
|
||||
f"{checklist_html}"
|
||||
f"<p><a href=\"/api/v1/projects/{_escape(dto['id'])}\">JSON detail</a></p>"
|
||||
)
|
||||
return render_page(title=project.repo_name, body_html=body)
|
||||
return render_page(title=dto["repo_name"], body_html=body)
|
||||
|
||||
|
||||
def render_registry_error(error: RegistryError) -> str:
|
||||
"""Render a fail-closed page for an invalid registry."""
|
||||
source = str(error.source_path) if error.source_path else "unknown"
|
||||
field = error.field_path or "n/a"
|
||||
body = (
|
||||
"<h2>Project registry unavailable</h2>"
|
||||
"<p>The registry failed validation, so the console refuses to render a "
|
||||
"partial inventory.</p>"
|
||||
"<table class=\"detail\">"
|
||||
f"<tr><th>Detail</th><td>{_escape(error.message)}</td></tr>"
|
||||
f"<tr><th>Field</th><td><code>{_escape(field)}</code></td></tr>"
|
||||
f"<tr><th>Source</th><td><code>{_escape(source)}</code></td></tr>"
|
||||
f"<tr><th>Remediation</th><td>{_escape(error.remediation)}</td></tr>"
|
||||
"</table>"
|
||||
)
|
||||
return render_page(title="Project registry unavailable", body_html=body)
|
||||
|
||||
Reference in New Issue
Block a user