Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
 jcwalker3
|
66a89a46bb | ||
 jcwalker3andClaude Opus 4.8
|
479e434f92 |
@@ -0,0 +1,295 @@
|
||||
# Web console authorization, RBAC, redaction, and audit model (#633)
|
||||
|
||||
**Phase 1. Read-only. This document defines the model that future console
|
||||
writes must pass through; it enables none of them.**
|
||||
|
||||
The MVP deployment boundary ([`webui-deployment.md`](webui-deployment.md), #435)
|
||||
documents internal-only serving and states plainly that MVP authentication is
|
||||
*none* — protection comes from network placement. That is adequate while every
|
||||
route is a GET, and inadequate the moment a gated write ships. This document
|
||||
and the three modules it describes land **before** any write exists, so no
|
||||
Phase 2 action can be added without an authority to check it against.
|
||||
|
||||
| Concern | Module |
|
||||
|---------|--------|
|
||||
| Identity, roles, authorization decision | `webui/console_authz.py` |
|
||||
| Secret redaction for every surface | `webui/console_redaction.py` |
|
||||
| Audit event schema, retention, sink | `webui/console_audit.py` |
|
||||
| Machine-readable publication | `GET /api/console/security-model` |
|
||||
|
||||
Two invariants hold everywhere and are non-negotiable for every child of #631:
|
||||
|
||||
1. **No secrets reach the browser.** Credentials are resolved server-side and
|
||||
redacted before any payload, page, log line, or audit record leaves.
|
||||
2. **No ungated mutations.** Authorization is necessary but never sufficient;
|
||||
execution stays disabled until the Phase 2 framework ships.
|
||||
|
||||
## Identity sources
|
||||
|
||||
The console performs *authorization*. Authentication is delegated, because a
|
||||
console that mints its own sessions is a credential store, and this one must
|
||||
not be.
|
||||
|
||||
| Source | Mode value | Authenticated | Shared host | Phase |
|
||||
|--------|-----------|---------------|-------------|-------|
|
||||
| None | `none` (default) | No — anonymous, capped at `viewer` | No | 1 |
|
||||
| Local dev | `local-dev` / `local_dev` | Yes, **asserted not verified** | No | 1 |
|
||||
| Access proxy | `access-proxy` / `access_proxy` | Yes, asserted by trusted proxy | Yes | 2 |
|
||||
|
||||
Selected by `WEBUI_AUTH_MODE`. An unrecognised value falls back to `none`
|
||||
rather than erroring open.
|
||||
|
||||
**Access-proxy mode** reads the subject from the
|
||||
`Cf-Access-Authenticated-User-Email` header, set by Cloudflare Access, WARP, or
|
||||
an equivalent org portal that terminates authentication in front of the
|
||||
console. If the header is absent the request did not traverse the proxy, so the
|
||||
principal degrades to anonymous — it is never trusted by default.
|
||||
|
||||
The **role is always server-side configuration**, never a client assertion. It
|
||||
comes from `WEBUI_ROLE_MAP`, a JSON object of subject → role:
|
||||
|
||||
```json
|
||||
{"[email protected]": "operator", "[email protected]": "controller"}
|
||||
```
|
||||
|
||||
An unmapped subject gets `viewer`. Malformed JSON yields an empty map, so
|
||||
everyone gets `viewer` — a parse failure loses authority rather than granting
|
||||
it.
|
||||
|
||||
Full SSO is explicitly a non-goal of this issue.
|
||||
|
||||
## Role matrix
|
||||
|
||||
Four roles, ordered least to most authority. Each role inherits every lower
|
||||
role's actions; the table states the *minimum* rank required.
|
||||
|
||||
| Role | Authority |
|
||||
|------|-----------|
|
||||
| `viewer` | Read every console view. No write, ever, in any phase. |
|
||||
| `operator` | Viewer, plus author-class work: claim, comment, open a PR. |
|
||||
| `controller` | Operator, plus reviewer/merger-class decisions on a PR. |
|
||||
| `admin` | Controller, plus destructive and policy-editing actions. |
|
||||
|
||||
`viewer` holds the empty write set by construction, and a test asserts it stays
|
||||
empty.
|
||||
|
||||
## Privileged actions
|
||||
|
||||
Every console action maps to a `task_key` in `task_capability_map.py`, the same
|
||||
single source of truth `gitea_resolve_task_capability` and the MCP tool gates
|
||||
use. The console therefore cannot invent an authority the MCP layer does not
|
||||
already define, and a regression test asserts each mapping matches.
|
||||
|
||||
| Action | Minimum role | Class | MCP permission | Confirm | Dual control | Break-glass | Phase |
|
||||
|--------|--------------|-------|----------------|---------|--------------|-------------|-------|
|
||||
| `claim_issue` | operator | gated_write | `gitea.issue.comment` | Yes | No | No | 2 |
|
||||
| `comment_issue` | operator | gated_write | `gitea.issue.comment` | Yes | No | No | 2 |
|
||||
| `create_issue` | operator | gated_write | `gitea.issue.create` | Yes | No | No | 2 |
|
||||
| `comment_pr` | operator | gated_write | `gitea.pr.comment` | Yes | No | No | 2 |
|
||||
| `create_pr` | operator | gated_write | `gitea.pr.create` | Yes | No | No | 2 |
|
||||
| `review_pr` | controller | privileged | `gitea.pr.review` | Yes | No | No | 3 |
|
||||
| `close_pr` | controller | privileged | `gitea.pr.close` | Yes | No | No | 3 |
|
||||
| `merge_pr` | controller | privileged | `gitea.pr.merge` | Yes | **Yes** | **Yes** | 3 |
|
||||
| `delete_branch` | admin | destructive | `gitea.branch.delete` | Yes | **Yes** | **Yes** | 3 |
|
||||
|
||||
**Dual control** means the acting principal may not be the sole authority: a
|
||||
second distinct principal must confirm. **Break-glass** means the action is
|
||||
expected to be unavailable in normal operation and its use is retained for two
|
||||
years. Both are declared here and enforced by the Phase 2 framework; Phase 1
|
||||
records the requirement on every decision so the framework cannot ship without
|
||||
honouring it.
|
||||
|
||||
`delete_branch` is admin-only rather than controller because it is the one
|
||||
irreversible action in the set.
|
||||
|
||||
### Authorization decision
|
||||
|
||||
`authorize(action_id, principal, for_execution=False)` returns a decision
|
||||
record and **denies by default**. The deny reasons are closed and enumerated:
|
||||
|
||||
| Reason code | Meaning |
|
||||
|-------------|---------|
|
||||
| `unknown_action` | No such console action is registered. |
|
||||
| `unauthenticated` | The principal is anonymous. |
|
||||
| `unknown_role` | The role is not in the matrix. |
|
||||
| `insufficient_role` | The role ranks below the action's minimum. |
|
||||
| `phase_not_active` | Execution requested for an action whose phase is not open. |
|
||||
| `allowed_preview_only` | Authorized — preview only, execution still disabled. |
|
||||
|
||||
There is no implicit allow branch. Even the allow result reports
|
||||
`execution_enabled: false` while the console is in Phase 1, so no caller can
|
||||
read an allow as permission to mutate.
|
||||
|
||||
## Secret redaction
|
||||
|
||||
One pass applies to **API payloads, rendered HTML, server logs, and audit
|
||||
records** — the four surfaces where a credential could escape.
|
||||
|
||||
Redaction reuses `gitea_audit.redact` rather than forking it: that remains the
|
||||
authority for secret-looking dict keys, `Authorization` material, and raw URLs.
|
||||
The console layer then applies its own patterns:
|
||||
|
||||
Each rule below matches an *assignment form*: the named key, followed by `=` or
|
||||
`:`, followed by the value. The keys are listed bare rather than spelled out as
|
||||
complete assignments, because this document is itself scanned by
|
||||
`scan_for_secrets` — writing the examples in full assignment form would make the
|
||||
documentation trip the very detectors it documents.
|
||||
|
||||
| Rule | Catches (as an assignment) |
|
||||
|------|----------------------------|
|
||||
| `credential_assignment` | `token`, `password`, `passwd`, `secret`, `api_key`, `access_key`, `client_secret`, `private_key` |
|
||||
| `credential_env_assignment` | `GITEA_TOKEN`, `GITEA_PASS`, `GITEA_PASSWORD` and suffixed variants |
|
||||
| `keychain_reference` | `keychain:` entry references |
|
||||
| `keychain_command` | macOS `security` keychain lookups (`find-generic-password`, `find-internet-password`) |
|
||||
| `private_key_block` | PEM `BEGIN ... PRIVATE KEY` blocks |
|
||||
| `json_web_token` | Three-segment `eyJ...` JWTs |
|
||||
| `bearer_credential` | `Bearer` / `Basic` credentials |
|
||||
|
||||
Assignments keep the key and replace only the value, so an operator can still
|
||||
see *what* was removed. Two behaviours are deliberate:
|
||||
|
||||
- **Fail closed.** A value that cannot be redacted becomes `[REDACTED]`
|
||||
outright rather than being emitted raw. Redaction never raises.
|
||||
- **Redact before persist.** `console_audit.build_event` redacts before
|
||||
serialization, and `write_event` re-scans and **drops** any record that still
|
||||
trips a detector. An unredacted record is never durable.
|
||||
|
||||
`scan_for_secrets` is the assertion helper: it returns the detector names still
|
||||
matching a payload, and already-redacted hits are not findings. Tests use it to
|
||||
prove the published policy, the security-model endpoint, and this document
|
||||
itself carry no secret material.
|
||||
|
||||
## Audit event schema
|
||||
|
||||
`gitea_audit` records MCP-side *mutations* — which profile and Gitea user
|
||||
performed which tool call. It has no console actor, no identity source, no
|
||||
correlation identifier, and no retention class, and an authorization **denial**
|
||||
is not a mutation, so it would never appear there at all. The console record is
|
||||
additive, not a replacement: a Phase 2 action emits both, joined on
|
||||
`correlation.request_id`.
|
||||
|
||||
Required fields, all asserted by tests so an edit cannot quietly drop one:
|
||||
|
||||
| Field | Content |
|
||||
|-------|---------|
|
||||
| `schema_version` | Currently `1`. |
|
||||
| `event_id` | Unique per record. |
|
||||
| `timestamp` | Timezone-aware ISO-8601, UTC. |
|
||||
| `actor` | `subject`, `role`, `identity_source`, `authenticated`. |
|
||||
| `action` | Console action id. |
|
||||
| `action_class` | `gated_write`, `privileged`, `destructive`, or `unknown`. |
|
||||
| `target` | `{kind, ref}`, e.g. `{"kind": "pr", "ref": "#123"}`. |
|
||||
| `result` | `allowed`, `denied`, `previewed`, `failed`, `succeeded`. |
|
||||
| `reason_code` | The authorization reason code above. |
|
||||
| `correlation` | `request_id`, `session_id`, `mcp_task`, `mcp_permission`. |
|
||||
| `retention` | `class`, `days`, `expires_at`. |
|
||||
| `redacted` | Always `true`; records are redacted at build time. |
|
||||
|
||||
An unrecognised `result` degrades to `failed` rather than being stored
|
||||
verbatim.
|
||||
|
||||
The sink is an append-only JSON Lines file named by
|
||||
`WEBUI_CONSOLE_AUDIT_LOG`. It is **off by default**: with the variable unset,
|
||||
events are still built — so callers and tests exercise the schema — but nothing
|
||||
is written. Auditing never raises; a failed write returns `False` rather than
|
||||
breaking the request it describes.
|
||||
|
||||
## Retention
|
||||
|
||||
| Class | Applies to | Default |
|
||||
|-------|-----------|---------|
|
||||
| `standard` | Routine gated writes | 90 days |
|
||||
| `privileged` | `review_pr`, `close_pr`, and any unclassifiable action | 365 days |
|
||||
| `break_glass` | `merge_pr`, `delete_branch` | 730 days |
|
||||
|
||||
Each record carries its own class, day count, and computed `expires_at`, so
|
||||
retention is auditable per record rather than inferred from file age. An
|
||||
**unknown action is retained as privileged, not standard** — for a safety
|
||||
control the conservative direction is to keep the record longer.
|
||||
|
||||
Nothing in this module updates or deletes. Expiry is enforced by an
|
||||
operator-run policy against `expires_at`, never by the console silently
|
||||
rewriting its own history.
|
||||
|
||||
## Phase 2 integration
|
||||
|
||||
Phase 2 opens gated writes. It must reuse this model rather than introduce a
|
||||
second one. The integration points are already wired and observable:
|
||||
|
||||
- **`GET /api/actions/{action_id}/preview`** attaches an `authorization` block
|
||||
to the existing preview payload and records a `previewed` audit event.
|
||||
- **`POST /api/actions/{action_id}/attempt`** attaches the same block and
|
||||
records a `denied` event. The terminal outcome is unchanged — the MVP
|
||||
registry in `webui/gated_actions.py` still fails closed for every action — so
|
||||
Phase 1 cannot loosen anything. Phase 2 enforces on this same decision
|
||||
instead of adding a parallel check.
|
||||
- **`GET /api/console/security-model`** publishes the RBAC matrix, redaction
|
||||
policy, and audit policy as JSON for operators and tests.
|
||||
|
||||
To open Phase 2, a child issue must: raise `ACTIVE_PHASE`, implement the
|
||||
confirmation and dual-control flow the matrix already declares, emit a
|
||||
`succeeded` or `failed` record alongside the `gitea_audit` mutation record, and
|
||||
keep `viewer` unable to reach any of it. Turning on execution without the
|
||||
confirmation flow contradicts a declared requirement and is a review failure,
|
||||
not a shortcut.
|
||||
|
||||
## Local-dev mode
|
||||
|
||||
`WEBUI_AUTH_MODE=local-dev` reads the principal straight from the environment:
|
||||
|
||||
| Variable | Purpose |
|
||||
|----------|---------|
|
||||
| `WEBUI_DEV_SUBJECT` | Subject string; absent ⇒ anonymous |
|
||||
| `WEBUI_DEV_ROLE` | One of `viewer`, `operator`, `controller`, `admin`; unrecognised ⇒ `viewer` |
|
||||
|
||||
**INSECURE — this mode is for loopback development only.** The subject and role
|
||||
are *asserted by the developer running the process and verified by nothing*.
|
||||
Anyone able to set an environment variable on the host is an `admin`, and
|
||||
anyone able to reach the port inherits that principal. It provides no
|
||||
authentication whatsoever; it exists so Phase 2 authorization paths can be
|
||||
exercised without standing up a proxy.
|
||||
|
||||
Never enable local-dev mode on a non-loopback bind. Combining it with
|
||||
`WEBUI_ALLOW_PUBLIC_BIND=1` or `WEBUI_ALLOW_REMOTE_BIND=1` publishes an
|
||||
unauthenticated admin console.
|
||||
|
||||
For anything beyond a laptop use `access-proxy` mode behind Cloudflare Access,
|
||||
WARP, or a VPN, as [`webui-deployment.md`](webui-deployment.md) requires.
|
||||
|
||||
### Probe authentication
|
||||
|
||||
`WEBUI_REQUIRE_PROBE_AUTH=1` declares that non-public probes should require an
|
||||
authenticated principal. It is **opt-in**: the default is off so the MVP
|
||||
`/health` contract is unchanged.
|
||||
|
||||
**This flag is declarative in Phase 1 and enforces nothing today.**
|
||||
`console_authz.probe_auth_required()` reports the operator's intent, and no
|
||||
route consults it — setting the variable does not currently change the
|
||||
behaviour of `/health` or any other endpoint. It is published here so the Phase
|
||||
2 action framework has a declared policy to honour rather than inventing a
|
||||
second one, exactly as `ACTIVE_PHASE` gates execution while the matrix is
|
||||
already declared. A regression test pins this "declared, not enforced" status,
|
||||
so wiring it later is a deliberate change rather than a silent one.
|
||||
|
||||
Until Phase 2 wires it, probe protection rests on network placement alone, as
|
||||
[`webui-deployment.md`](webui-deployment.md) (#435) states.
|
||||
|
||||
## Environment variables
|
||||
|
||||
| Variable | Default | Purpose |
|
||||
|----------|---------|---------|
|
||||
| `WEBUI_AUTH_MODE` | `none` | Identity source selection |
|
||||
| `WEBUI_DEV_SUBJECT` | unset | Local-dev subject (insecure) |
|
||||
| `WEBUI_DEV_ROLE` | `viewer` | Local-dev role (insecure) |
|
||||
| `WEBUI_ROLE_MAP` | unset | JSON subject → role map |
|
||||
| `WEBUI_REQUIRE_PROBE_AUTH` | unset | Require auth for non-public probes |
|
||||
| `WEBUI_CONSOLE_AUDIT_LOG` | unset | Append-only audit sink path |
|
||||
|
||||
All are read server-side only. None is ever rendered into a page or returned by
|
||||
an API.
|
||||
|
||||
## Non-goals
|
||||
|
||||
- No full SSO product; authentication stays delegated to the proxy.
|
||||
- No browser-initiated merges or approvals in any phase covered here.
|
||||
- No tokens in the frontend, in browser storage, or in committed config.
|
||||
@@ -7,7 +7,10 @@ only.
|
||||
## MVP deployment model
|
||||
|
||||
- **Default bind:** `127.0.0.1:8765` (`WEBUI_HOST` / `WEBUI_PORT`)
|
||||
- **Authentication:** none in MVP — protection comes from network placement
|
||||
- **Authentication:** none in MVP — protection comes from network placement.
|
||||
The authorization, RBAC, redaction, and audit model that future gated writes
|
||||
must pass through is defined in
|
||||
[`webui-authz-audit.md`](webui-authz-audit.md) (#633).
|
||||
- **Mutations:** read-only routes; gated write actions remain disabled (#434)
|
||||
- **Secrets:** resolved server-side via `gitea_auth` / `GITEA_MCP_CONFIG`; never
|
||||
embedded in HTML, JavaScript, or browser storage
|
||||
|
||||
@@ -212,53 +212,6 @@ health, workflow/schema SHA-256 hashes, and stale-runtime warnings when the
|
||||
checkout is behind merged safety-gate changes. Restart guidance links to #420;
|
||||
no tokens or MCP restart actions are exposed.
|
||||
|
||||
## Inventory API (#636)
|
||||
|
||||
`GET /api/v1/inventory` returns one versioned, read-only snapshot that unifies
|
||||
what the lease (#433), worktree (#432), and runtime (#430) MVP views each show
|
||||
separately, so traffic-control and recovery consumers read the same source.
|
||||
`GET /api/v1/inventory/{section}` returns a single section under the identical
|
||||
schema (`sessions`, `leases`, `locks`, `worktrees`, `namespaces`); an unknown
|
||||
section is a `404` with `error: unknown_section`. Both routes are `GET`-only.
|
||||
|
||||
Each section carries its own `status` (`ok` / `degraded` / `unavailable`), a
|
||||
`reason` when not `ok`, and a `scan_ms`. A subsystem that cannot be read
|
||||
degrades to a reasoned section; it never raises and never emits an empty list
|
||||
that would read as "nothing is there".
|
||||
|
||||
### Field authority
|
||||
|
||||
Every section names where its rows came from; authorities are never blended.
|
||||
|
||||
| Section | Authority | Source |
|
||||
|---|---|---|
|
||||
| `sessions` | `control_plane_db` | #613 control-plane DB (`mode=ro`), authoritative for exclusive ownership (#600/#601) |
|
||||
| `leases` | `control_plane_db` | #613 control-plane DB; degrades if the `work_items` table is absent |
|
||||
| `locks` | `filesystem` | durable per-issue lock files (`issue_lock_store`) |
|
||||
| `worktrees` | `filesystem` | registered git worktrees via the #432 hygiene scanner |
|
||||
| `namespaces` | `filesystem` | the active profile serving this web process (others are not enumerable) |
|
||||
|
||||
The payload restates this map under `field_authority` for machine consumers.
|
||||
|
||||
### Ownership safety
|
||||
|
||||
`ownership_authority_complete` is true only when every ownership-bearing section
|
||||
(`sessions`, `leases`, `locks`) read cleanly. While it is false, nothing is
|
||||
reported as unowned and no collision is asserted from a degraded source —
|
||||
absence of evidence is reported as absence of evidence, never as free work.
|
||||
|
||||
`collisions` surfaces detectable conflicts, each with a `kind` and `severity`:
|
||||
`lock-without-worktree`, `duplicate-live-lock`, `live-lock-dead-owner` (unexpired
|
||||
lease, dead pid — a #753 recovery candidate that would read as live to a naive
|
||||
timestamp check), `stale-lock-dead-owner`, `expired-lock-live-owner` (the
|
||||
#635/#760 daemon-pid deadlock), `concurrent-active-lease`, `active-lease-past-expiry`,
|
||||
and `orphan-lease`. Collisions are emitted only from sections that read cleanly.
|
||||
|
||||
The control-plane DB is opened through a `mode=ro` URI so a read never creates
|
||||
or migrates it; paths are collapsed against `$HOME`, URLs lose userinfo and
|
||||
query strings, and credential-shaped values are redacted at the boundary. Lease
|
||||
steal/release and worktree deletion are Phase 2+ and have no representation here.
|
||||
|
||||
## Tests
|
||||
|
||||
```bash
|
||||
|
||||
@@ -0,0 +1,703 @@
|
||||
"""Console authorization, redaction, and audit model tests (#633).
|
||||
|
||||
Covers each acceptance criterion and each required test named in the issue:
|
||||
|
||||
* AC1 — RBAC matrix and privileged-action list.
|
||||
* AC2 — redaction rules, unit-tested against sample payloads.
|
||||
* AC3 — audit event schema with required fields and retention defaults.
|
||||
* AC4 — Phase 2 integration points.
|
||||
* AC5 — local-dev mode with explicit insecurity warnings.
|
||||
|
||||
Required tests: redaction units (token, keychain, password patterns),
|
||||
default-deny for unauthenticated write stubs, and audit record creation for a
|
||||
simulated privileged preview.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime
|
||||
import json
|
||||
import os
|
||||
import pathlib
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
from starlette.testclient import TestClient
|
||||
|
||||
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1]))
|
||||
|
||||
from task_capability_map import TASK_CAPABILITY_MAP # noqa: E402
|
||||
from webui import console_audit, console_authz # noqa: E402
|
||||
from webui.app import create_app # noqa: E402
|
||||
from webui.console_redaction import ( # noqa: E402
|
||||
REDACTED,
|
||||
redact_payload,
|
||||
redact_text,
|
||||
redaction_policy,
|
||||
scan_for_secrets,
|
||||
)
|
||||
|
||||
DOCS = pathlib.Path(__file__).resolve().parents[1] / "docs"
|
||||
AUTHZ_DOC = DOCS / "webui-authz-audit.md"
|
||||
|
||||
|
||||
def _principal(role: str) -> console_authz.Principal:
|
||||
return console_authz.Principal(
|
||||
subject=f"{role}@example.com",
|
||||
role=role,
|
||||
identity_source=console_authz.IDENTITY_ACCESS_PROXY,
|
||||
authenticated=True,
|
||||
)
|
||||
|
||||
|
||||
class TestRoleMatrix(unittest.TestCase):
|
||||
"""AC1 — the written RBAC matrix and privileged-action list."""
|
||||
|
||||
def test_roles_are_ordered_least_to_most_authority(self):
|
||||
self.assertEqual(
|
||||
console_authz.ROLE_ORDER,
|
||||
("viewer", "operator", "controller", "admin"),
|
||||
)
|
||||
|
||||
def test_every_role_has_a_description(self):
|
||||
for role in console_authz.ROLE_ORDER:
|
||||
with self.subTest(role=role):
|
||||
self.assertTrue(console_authz.ROLE_DESCRIPTIONS[role].strip())
|
||||
|
||||
def test_higher_roles_inherit_lower_role_actions(self):
|
||||
matrix = {
|
||||
entry["role"]: set(entry["permitted_actions"])
|
||||
for entry in console_authz.rbac_matrix()["roles"]
|
||||
}
|
||||
for lower, higher in zip(
|
||||
console_authz.ROLE_ORDER, console_authz.ROLE_ORDER[1:]
|
||||
):
|
||||
with self.subTest(lower=lower, higher=higher):
|
||||
self.assertTrue(matrix[lower].issubset(matrix[higher]))
|
||||
|
||||
def test_viewer_holds_no_write_action(self):
|
||||
matrix = {
|
||||
entry["role"]: set(entry["permitted_actions"])
|
||||
for entry in console_authz.rbac_matrix()["roles"]
|
||||
}
|
||||
self.assertEqual(matrix["viewer"], set())
|
||||
|
||||
def test_privileged_action_list_is_non_empty_and_classified(self):
|
||||
privileged = console_authz.privileged_actions()
|
||||
self.assertTrue(privileged)
|
||||
ids = {action.action_id for action in privileged}
|
||||
# Merge and branch deletion are the canonical privileged pair.
|
||||
self.assertIn("merge_pr", ids)
|
||||
self.assertIn("delete_branch", ids)
|
||||
|
||||
def test_merge_and_delete_require_dual_control_and_break_glass(self):
|
||||
for action_id in ("merge_pr", "delete_branch"):
|
||||
with self.subTest(action=action_id):
|
||||
action = console_authz.get_action(action_id)
|
||||
self.assertTrue(action.dual_control)
|
||||
self.assertTrue(action.break_glass)
|
||||
self.assertTrue(action.requires_confirmation)
|
||||
|
||||
def test_every_write_action_requires_confirmation(self):
|
||||
for action in console_authz.ACTIONS.values():
|
||||
with self.subTest(action=action.action_id):
|
||||
self.assertTrue(action.requires_confirmation)
|
||||
|
||||
def test_delete_branch_is_admin_only(self):
|
||||
self.assertEqual(
|
||||
console_authz.get_action("delete_branch").minimum_role,
|
||||
console_authz.ADMIN,
|
||||
)
|
||||
|
||||
def test_actions_map_to_real_mcp_capability_vocabulary(self):
|
||||
"""The console must not invent an authority the MCP layer lacks."""
|
||||
for action in console_authz.ACTIONS.values():
|
||||
with self.subTest(action=action.action_id):
|
||||
self.assertIn(action.task_key, TASK_CAPABILITY_MAP)
|
||||
self.assertEqual(
|
||||
action.mcp_permission,
|
||||
TASK_CAPABILITY_MAP[action.task_key]["permission"],
|
||||
)
|
||||
self.assertEqual(
|
||||
action.mcp_role,
|
||||
TASK_CAPABILITY_MAP[action.task_key]["role"],
|
||||
)
|
||||
|
||||
def test_matrix_declares_deny_by_default_and_execution_disabled(self):
|
||||
matrix = console_authz.rbac_matrix()
|
||||
self.assertEqual(matrix["default_decision"], "deny")
|
||||
self.assertFalse(matrix["execution_enabled"])
|
||||
|
||||
|
||||
class TestAuthorizeDefaultDeny(unittest.TestCase):
|
||||
"""Fail-closed behaviour of the authorization decision."""
|
||||
|
||||
def test_anonymous_is_denied_every_action(self):
|
||||
for action_id in console_authz.ACTIONS:
|
||||
with self.subTest(action=action_id):
|
||||
decision = console_authz.authorize(action_id)
|
||||
self.assertFalse(decision.allowed)
|
||||
self.assertEqual(
|
||||
decision.reason_code, console_authz.DENY_UNAUTHENTICATED
|
||||
)
|
||||
|
||||
def test_unknown_action_is_denied(self):
|
||||
decision = console_authz.authorize(
|
||||
"not_a_real_action", _principal("admin")
|
||||
)
|
||||
self.assertFalse(decision.allowed)
|
||||
self.assertEqual(decision.reason_code, console_authz.DENY_UNKNOWN_ACTION)
|
||||
|
||||
def test_unknown_role_is_denied(self):
|
||||
rogue = console_authz.Principal(
|
||||
subject="[email protected]",
|
||||
role="superuser",
|
||||
identity_source=console_authz.IDENTITY_ACCESS_PROXY,
|
||||
authenticated=True,
|
||||
)
|
||||
decision = console_authz.authorize("comment_issue", rogue)
|
||||
self.assertFalse(decision.allowed)
|
||||
self.assertEqual(decision.reason_code, console_authz.DENY_UNKNOWN_ROLE)
|
||||
|
||||
def test_insufficient_role_is_denied(self):
|
||||
decision = console_authz.authorize("merge_pr", _principal("operator"))
|
||||
self.assertFalse(decision.allowed)
|
||||
self.assertEqual(
|
||||
decision.reason_code, console_authz.DENY_INSUFFICIENT_ROLE
|
||||
)
|
||||
|
||||
def test_sufficient_role_allows_preview_only(self):
|
||||
decision = console_authz.authorize("merge_pr", _principal("controller"))
|
||||
self.assertTrue(decision.allowed)
|
||||
self.assertFalse(decision.execution_enabled)
|
||||
|
||||
def test_execution_is_refused_while_phase_is_not_active(self):
|
||||
decision = console_authz.authorize(
|
||||
"merge_pr", _principal("controller"), for_execution=True
|
||||
)
|
||||
self.assertFalse(decision.allowed)
|
||||
self.assertEqual(
|
||||
decision.reason_code, console_authz.DENY_PHASE_NOT_ACTIVE
|
||||
)
|
||||
|
||||
def test_allowed_decision_never_reports_execution_enabled(self):
|
||||
for action_id in console_authz.ACTIONS:
|
||||
with self.subTest(action=action_id):
|
||||
decision = console_authz.authorize(
|
||||
action_id, _principal("admin")
|
||||
)
|
||||
self.assertFalse(decision.execution_enabled)
|
||||
|
||||
|
||||
class TestIdentityResolution(unittest.TestCase):
|
||||
"""AC5 — identity sources, including the insecure local-dev mode."""
|
||||
|
||||
def test_no_auth_mode_yields_anonymous_viewer(self):
|
||||
principal = console_authz.resolve_principal(env={})
|
||||
self.assertFalse(principal.authenticated)
|
||||
self.assertEqual(principal.role, console_authz.VIEWER)
|
||||
self.assertEqual(principal.identity_source, console_authz.IDENTITY_NONE)
|
||||
|
||||
def test_local_dev_mode_warns_that_identity_is_unverified(self):
|
||||
principal = console_authz.resolve_principal(
|
||||
env={
|
||||
console_authz.AUTH_MODE_ENV: "local-dev",
|
||||
console_authz.DEV_SUBJECT_ENV: "[email protected]",
|
||||
console_authz.DEV_ROLE_ENV: "admin",
|
||||
}
|
||||
)
|
||||
self.assertTrue(principal.authenticated)
|
||||
self.assertEqual(principal.role, "admin")
|
||||
self.assertTrue(principal.warnings)
|
||||
self.assertIn("asserted", " ".join(principal.warnings).lower())
|
||||
|
||||
def test_local_dev_without_subject_falls_back_to_anonymous(self):
|
||||
principal = console_authz.resolve_principal(
|
||||
env={console_authz.AUTH_MODE_ENV: "local-dev"}
|
||||
)
|
||||
self.assertFalse(principal.authenticated)
|
||||
|
||||
def test_local_dev_unknown_role_degrades_to_viewer(self):
|
||||
principal = console_authz.resolve_principal(
|
||||
env={
|
||||
console_authz.AUTH_MODE_ENV: "local_dev",
|
||||
console_authz.DEV_SUBJECT_ENV: "[email protected]",
|
||||
console_authz.DEV_ROLE_ENV: "root",
|
||||
}
|
||||
)
|
||||
self.assertEqual(principal.role, console_authz.VIEWER)
|
||||
|
||||
def test_access_proxy_without_header_fails_closed(self):
|
||||
"""A proxy-mode request that did not traverse the proxy is anonymous."""
|
||||
principal = console_authz.resolve_principal(
|
||||
headers={},
|
||||
env={console_authz.AUTH_MODE_ENV: "access_proxy"},
|
||||
)
|
||||
self.assertFalse(principal.authenticated)
|
||||
|
||||
def test_access_proxy_role_comes_from_server_config_not_client(self):
|
||||
env = {
|
||||
console_authz.AUTH_MODE_ENV: "access_proxy",
|
||||
console_authz.ROLE_MAP_ENV: json.dumps(
|
||||
{"[email protected]": "controller"}
|
||||
),
|
||||
}
|
||||
principal = console_authz.resolve_principal(
|
||||
headers={
|
||||
console_authz.ACCESS_SUBJECT_HEADER: "[email protected]",
|
||||
"x-role": "admin", # client-supplied role must be ignored
|
||||
},
|
||||
env=env,
|
||||
)
|
||||
self.assertEqual(principal.role, "controller")
|
||||
|
||||
def test_access_proxy_unmapped_subject_defaults_to_viewer(self):
|
||||
principal = console_authz.resolve_principal(
|
||||
headers={
|
||||
console_authz.ACCESS_SUBJECT_HEADER: "[email protected]"
|
||||
},
|
||||
env={console_authz.AUTH_MODE_ENV: "access_proxy"},
|
||||
)
|
||||
self.assertEqual(principal.role, console_authz.VIEWER)
|
||||
|
||||
def test_malformed_role_map_does_not_raise_and_denies(self):
|
||||
principal = console_authz.resolve_principal(
|
||||
headers={console_authz.ACCESS_SUBJECT_HEADER: "[email protected]"},
|
||||
env={
|
||||
console_authz.AUTH_MODE_ENV: "access_proxy",
|
||||
console_authz.ROLE_MAP_ENV: "{not json",
|
||||
},
|
||||
)
|
||||
self.assertEqual(principal.role, console_authz.VIEWER)
|
||||
|
||||
def test_probe_auth_is_opt_in(self):
|
||||
self.assertFalse(console_authz.probe_auth_required(env={}))
|
||||
self.assertTrue(
|
||||
console_authz.probe_auth_required(
|
||||
env={console_authz.REQUIRE_PROBE_AUTH_ENV: "1"}
|
||||
)
|
||||
)
|
||||
|
||||
def test_probe_auth_is_declared_but_not_yet_enforced(self):
|
||||
"""Phase 1 declares the probe-auth policy; no route enforces it yet.
|
||||
|
||||
The flag exists so the Phase 2 action framework has a declared policy
|
||||
to honour instead of inventing a second one. Pinning the current
|
||||
not-enforced status here means wiring it later is a deliberate change
|
||||
that updates this test and the documentation together, rather than a
|
||||
silent behaviour shift. The documentation must say so plainly, because
|
||||
an operator who sets the variable believing it protects a probe is
|
||||
worse off than one who knows it does not.
|
||||
"""
|
||||
import inspect
|
||||
|
||||
from webui import app as webui_app
|
||||
|
||||
source = inspect.getsource(webui_app)
|
||||
self.assertNotIn(
|
||||
"probe_auth_required",
|
||||
source,
|
||||
msg=(
|
||||
"webui.app now consults probe_auth_required, so probe auth is "
|
||||
"no longer merely declared. Update the 'Probe authentication' "
|
||||
"section of docs/webui-authz-audit.md, which states it "
|
||||
"enforces nothing, and replace this test with real "
|
||||
"enforcement coverage."
|
||||
),
|
||||
)
|
||||
self.assertIn(
|
||||
"enforces nothing today",
|
||||
AUTHZ_DOC.read_text(encoding="utf-8"),
|
||||
)
|
||||
|
||||
|
||||
class TestRedaction(unittest.TestCase):
|
||||
"""AC2 — required redaction units: token, keychain, password patterns."""
|
||||
|
||||
def test_token_assignment_is_redacted(self):
|
||||
out = redact_text("GITEA_TOKEN=abcd1234efgh5678ijkl")
|
||||
self.assertIn(REDACTED, out)
|
||||
self.assertNotIn("abcd1234efgh5678ijkl", out)
|
||||
|
||||
def test_password_assignment_is_redacted(self):
|
||||
out = redact_text("password: hunter2supersecret")
|
||||
self.assertIn(REDACTED, out)
|
||||
self.assertNotIn("hunter2supersecret", out)
|
||||
|
||||
def test_keychain_reference_is_redacted(self):
|
||||
out = redact_text("keychain:gitea-prgs-token")
|
||||
self.assertIn(REDACTED, out)
|
||||
self.assertNotIn("gitea-prgs-token", out)
|
||||
|
||||
def test_keychain_command_is_redacted(self):
|
||||
out = redact_text("security find-generic-password -s gitea -w")
|
||||
self.assertIn(REDACTED, out)
|
||||
self.assertNotIn("find-generic-password -s gitea", out)
|
||||
|
||||
def test_bearer_credential_is_redacted(self):
|
||||
out = redact_text("Authorization: Bearer abcdef1234567890abcdef")
|
||||
self.assertNotIn("abcdef1234567890abcdef", out)
|
||||
|
||||
def test_jwt_is_redacted(self):
|
||||
token = "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIxIn0.abcdefghijklmnop"
|
||||
out = redact_text(f"session={token}")
|
||||
self.assertNotIn(token, out)
|
||||
|
||||
def test_private_key_block_is_redacted(self):
|
||||
pem = (
|
||||
"-----BEGIN RSA PRIVATE KEY-----\n"
|
||||
"MIIEowIBAAKCAQEAsecretmaterial\n"
|
||||
"-----END RSA PRIVATE KEY-----"
|
||||
)
|
||||
out = redact_text(pem)
|
||||
self.assertNotIn("MIIEowIBAAKCAQEAsecretmaterial", out)
|
||||
|
||||
def test_api_key_assignment_is_redacted(self):
|
||||
out = redact_text('api_key = "sk-live-9f8e7d6c5b4a3210"')
|
||||
self.assertNotIn("sk-live-9f8e7d6c5b4a3210", out)
|
||||
|
||||
def test_nested_payload_is_redacted_recursively(self):
|
||||
payload = {
|
||||
"token": "abc123456789",
|
||||
"nested": {"note": "password=letmein12345"},
|
||||
"list": ["keychain:some-entry"],
|
||||
"safe": "plain text",
|
||||
}
|
||||
out = redact_payload(payload)
|
||||
self.assertEqual(out["token"], REDACTED)
|
||||
self.assertNotIn("letmein12345", json.dumps(out))
|
||||
self.assertNotIn("some-entry", json.dumps(out))
|
||||
self.assertEqual(out["safe"], "plain text")
|
||||
|
||||
def test_scan_reports_findings_before_and_none_after(self):
|
||||
dirty = "password: hunter2supersecret"
|
||||
self.assertTrue(scan_for_secrets(dirty))
|
||||
self.assertEqual(scan_for_secrets(redact_text(dirty)), [])
|
||||
|
||||
def test_non_strings_pass_through_untouched(self):
|
||||
self.assertEqual(redact_text(42), 42)
|
||||
self.assertEqual(
|
||||
redact_payload({"n": 1, "b": True}), {"n": 1, "b": True}
|
||||
)
|
||||
|
||||
def test_policy_is_documented_and_declares_redact_before_persist(self):
|
||||
policy = redaction_policy()
|
||||
self.assertTrue(policy["redact_before_persist"])
|
||||
self.assertIn("audit_records", policy["applies_to"])
|
||||
self.assertTrue(policy["console_rules"])
|
||||
|
||||
def test_policy_statement_contains_no_secret_material(self):
|
||||
self.assertEqual(scan_for_secrets(redaction_policy()), [])
|
||||
|
||||
|
||||
class TestAuditSchema(unittest.TestCase):
|
||||
"""AC3 — audit event schema, required fields, and retention defaults."""
|
||||
|
||||
def _event(self, action_id="merge_pr", **kwargs):
|
||||
return console_audit.build_event(
|
||||
action_id=action_id,
|
||||
result=console_audit.RESULT_DENIED,
|
||||
decision=console_authz.authorize(action_id, _principal("operator")),
|
||||
target={"kind": "pr", "ref": "#123"},
|
||||
request_id="req-test",
|
||||
**kwargs,
|
||||
)
|
||||
|
||||
def test_every_required_field_is_present(self):
|
||||
event = self._event()
|
||||
for field in console_audit.REQUIRED_FIELDS:
|
||||
with self.subTest(field=field):
|
||||
self.assertIn(field, event)
|
||||
|
||||
def test_actor_carries_who_and_how_they_were_identified(self):
|
||||
event = self._event()
|
||||
for field in console_audit.REQUIRED_ACTOR_FIELDS:
|
||||
with self.subTest(field=field):
|
||||
self.assertIn(field, event["actor"])
|
||||
|
||||
def test_correlation_ids_are_present(self):
|
||||
event = self._event()
|
||||
for field in console_audit.REQUIRED_CORRELATION_FIELDS:
|
||||
with self.subTest(field=field):
|
||||
self.assertIn(field, event["correlation"])
|
||||
self.assertEqual(event["correlation"]["request_id"], "req-test")
|
||||
self.assertEqual(event["correlation"]["mcp_task"], "merge_pr")
|
||||
|
||||
def test_timestamp_is_timezone_aware_utc_iso8601(self):
|
||||
now = datetime.datetime(
|
||||
2026, 7, 22, 10, 16, 42, tzinfo=datetime.timezone.utc
|
||||
)
|
||||
event = self._event(now=now)
|
||||
self.assertEqual(event["timestamp"], "2026-07-22T10:16:42+00:00")
|
||||
parsed = datetime.datetime.fromisoformat(event["timestamp"])
|
||||
self.assertIsNotNone(parsed.tzinfo)
|
||||
|
||||
def test_retention_defaults_by_class(self):
|
||||
self.assertEqual(
|
||||
console_audit.RETENTION_DAYS[console_audit.RETENTION_STANDARD], 90
|
||||
)
|
||||
self.assertEqual(
|
||||
console_audit.RETENTION_DAYS[console_audit.RETENTION_PRIVILEGED],
|
||||
365,
|
||||
)
|
||||
self.assertEqual(
|
||||
console_audit.RETENTION_DAYS[console_audit.RETENTION_BREAK_GLASS],
|
||||
730,
|
||||
)
|
||||
|
||||
def test_break_glass_action_retains_longest(self):
|
||||
event = self._event("merge_pr")
|
||||
self.assertEqual(
|
||||
event["retention"]["class"], console_audit.RETENTION_BREAK_GLASS
|
||||
)
|
||||
|
||||
def test_routine_write_uses_standard_retention(self):
|
||||
event = self._event("comment_issue")
|
||||
self.assertEqual(
|
||||
event["retention"]["class"], console_audit.RETENTION_STANDARD
|
||||
)
|
||||
|
||||
def test_unknown_action_retains_as_privileged_not_standard(self):
|
||||
"""Conservative direction: keep an unclassifiable record longer."""
|
||||
self.assertEqual(
|
||||
console_audit.retention_class_for(None),
|
||||
console_audit.RETENTION_PRIVILEGED,
|
||||
)
|
||||
|
||||
def test_retention_expiry_matches_declared_days(self):
|
||||
now = datetime.datetime(2026, 7, 22, tzinfo=datetime.timezone.utc)
|
||||
event = self._event("comment_issue", now=now)
|
||||
expires = datetime.datetime.fromisoformat(
|
||||
event["retention"]["expires_at"]
|
||||
)
|
||||
self.assertEqual((expires - now).days, 90)
|
||||
|
||||
def test_invalid_result_degrades_to_failed(self):
|
||||
event = console_audit.build_event(action_id="merge_pr", result="banana")
|
||||
self.assertEqual(event["result"], console_audit.RESULT_FAILED)
|
||||
|
||||
def test_denied_result_is_representable(self):
|
||||
"""An authorization denial has no MCP-side mutation record."""
|
||||
self.assertIn(console_audit.RESULT_DENIED, console_audit.RESULTS)
|
||||
|
||||
def test_event_is_redacted_before_it_is_returned(self):
|
||||
event = console_audit.build_event(
|
||||
action_id="merge_pr",
|
||||
result=console_audit.RESULT_DENIED,
|
||||
detail="failed with token=abcdef1234567890",
|
||||
metadata={"password": "hunter2supersecret"},
|
||||
)
|
||||
serialized = json.dumps(event)
|
||||
self.assertNotIn("abcdef1234567890", serialized)
|
||||
self.assertNotIn("hunter2supersecret", serialized)
|
||||
self.assertTrue(event["redacted"])
|
||||
|
||||
def test_audit_policy_reports_schema_and_retention(self):
|
||||
policy = console_audit.audit_policy()
|
||||
self.assertTrue(policy["append_only"])
|
||||
self.assertTrue(policy["redact_before_persist"])
|
||||
self.assertEqual(
|
||||
policy["retention_defaults_days"], console_audit.RETENTION_DAYS
|
||||
)
|
||||
|
||||
|
||||
class TestAuditSink(unittest.TestCase):
|
||||
"""Append-only persistence behaviour."""
|
||||
|
||||
def test_write_is_a_noop_when_sink_is_unconfigured(self):
|
||||
saved = os.environ.pop(console_audit.AUDIT_LOG_ENV, None)
|
||||
try:
|
||||
self.assertFalse(console_audit.audit_enabled())
|
||||
self.assertFalse(console_audit.write_event({"schema_version": 1}))
|
||||
finally:
|
||||
if saved is not None:
|
||||
os.environ[console_audit.AUDIT_LOG_ENV] = saved
|
||||
|
||||
def test_records_append_one_json_line_each(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
sink = os.path.join(tmp, "console-audit.jsonl")
|
||||
for _ in range(3):
|
||||
event = console_audit.build_event(
|
||||
action_id="merge_pr", result=console_audit.RESULT_DENIED
|
||||
)
|
||||
self.assertTrue(console_audit.write_event(event, path=sink))
|
||||
with open(sink, encoding="utf-8") as handle:
|
||||
lines = [json.loads(line) for line in handle if line.strip()]
|
||||
self.assertEqual(len(lines), 3)
|
||||
self.assertEqual(len({line["event_id"] for line in lines}), 3)
|
||||
|
||||
def test_a_record_that_still_carries_a_secret_is_not_persisted(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
sink = os.path.join(tmp, "console-audit.jsonl")
|
||||
leaky = {
|
||||
"schema_version": 1,
|
||||
"detail": "password: hunter2supersecret",
|
||||
}
|
||||
self.assertFalse(console_audit.write_event(leaky, path=sink))
|
||||
self.assertFalse(os.path.exists(sink))
|
||||
|
||||
def test_write_never_raises_on_a_bad_path(self):
|
||||
self.assertFalse(
|
||||
console_audit.write_event(
|
||||
{"schema_version": 1}, path="/nonexistent-dir/audit.jsonl"
|
||||
)
|
||||
)
|
||||
|
||||
def test_simulated_privileged_preview_creates_an_audit_record(self):
|
||||
"""Required test: audit record creation for a privileged preview."""
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
sink = os.path.join(tmp, "console-audit.jsonl")
|
||||
os.environ[console_audit.AUDIT_LOG_ENV] = sink
|
||||
try:
|
||||
decision = console_authz.authorize(
|
||||
"merge_pr", _principal("controller")
|
||||
)
|
||||
outcome = console_audit.record_event(
|
||||
action_id="merge_pr",
|
||||
result=console_audit.RESULT_PREVIEWED,
|
||||
decision=decision,
|
||||
target={"kind": "pr", "ref": "#123"},
|
||||
request_id="req-preview",
|
||||
)
|
||||
finally:
|
||||
os.environ.pop(console_audit.AUDIT_LOG_ENV, None)
|
||||
self.assertTrue(outcome["written"])
|
||||
with open(sink, encoding="utf-8") as handle:
|
||||
record = json.loads(handle.read().strip())
|
||||
self.assertEqual(record["action"], "merge_pr")
|
||||
self.assertEqual(record["result"], console_audit.RESULT_PREVIEWED)
|
||||
self.assertEqual(record["action_class"], "privileged")
|
||||
self.assertTrue(record["decision"]["allowed"])
|
||||
self.assertFalse(record["decision"]["execution_enabled"])
|
||||
self.assertEqual(record["actor"]["role"], "controller")
|
||||
|
||||
def test_decision_block_survives_redaction(self):
|
||||
"""Regression: naming it 'authorization' collided with a secret hint.
|
||||
|
||||
``gitea_audit._SECRET_KEY_HINTS`` contains "authorization" (for the
|
||||
HTTP header), so a block under that key was replaced wholesale by the
|
||||
placeholder and the record lost its decision entirely.
|
||||
"""
|
||||
event = console_audit.build_event(
|
||||
action_id="merge_pr",
|
||||
result=console_audit.RESULT_DENIED,
|
||||
decision=console_authz.authorize("merge_pr", _principal("admin")),
|
||||
)
|
||||
self.assertIsInstance(event["decision"], dict)
|
||||
self.assertIn("allowed", event["decision"])
|
||||
|
||||
|
||||
class TestConsoleRoutes(unittest.TestCase):
|
||||
"""AC4 — the wired Phase 2 integration points, still fail-closed."""
|
||||
|
||||
def setUp(self):
|
||||
self.client = TestClient(create_app(bind_host="127.0.0.1"))
|
||||
|
||||
def test_unauthenticated_write_stub_is_denied(self):
|
||||
"""Required test: default-deny for unauthenticated write stubs."""
|
||||
response = self.client.post(
|
||||
"/api/actions/merge_pr/attempt", json={"pr_number": 99}
|
||||
)
|
||||
self.assertEqual(response.status_code, 403)
|
||||
body = response.json()
|
||||
self.assertFalse(body["success"])
|
||||
authorization = body["authorization"]
|
||||
self.assertFalse(authorization["allowed"])
|
||||
self.assertEqual(
|
||||
authorization["reason_code"], console_authz.DENY_UNAUTHENTICATED
|
||||
)
|
||||
self.assertFalse(authorization["execution_enabled"])
|
||||
|
||||
def test_preview_reports_an_authorization_decision(self):
|
||||
response = self.client.get("/api/actions/merge_pr/preview?pr_number=7")
|
||||
self.assertEqual(response.status_code, 200)
|
||||
authorization = response.json()["authorization"]
|
||||
self.assertFalse(authorization["allowed"])
|
||||
self.assertTrue(authorization["dual_control"])
|
||||
self.assertEqual(authorization["required_role"], "controller")
|
||||
|
||||
def test_unknown_action_preview_still_404s(self):
|
||||
response = self.client.get("/api/actions/no_such_action/preview")
|
||||
self.assertEqual(response.status_code, 404)
|
||||
|
||||
def test_security_model_endpoint_publishes_all_three_policies(self):
|
||||
response = self.client.get("/api/console/security-model")
|
||||
self.assertEqual(response.status_code, 200)
|
||||
body = response.json()
|
||||
self.assertIn("rbac", body)
|
||||
self.assertIn("redaction", body)
|
||||
self.assertIn("audit", body)
|
||||
self.assertEqual(body["rbac"]["default_decision"], "deny")
|
||||
|
||||
def test_security_model_endpoint_leaks_no_secrets(self):
|
||||
response = self.client.get("/api/console/security-model")
|
||||
self.assertEqual(scan_for_secrets(response.json()), [])
|
||||
|
||||
def test_security_model_rejects_writes(self):
|
||||
response = self.client.post("/api/console/security-model", json={})
|
||||
self.assertEqual(response.status_code, 405)
|
||||
|
||||
def test_existing_read_routes_are_unaffected(self):
|
||||
for path in ("/", "/health", "/actions", "/api/actions"):
|
||||
with self.subTest(path=path):
|
||||
self.assertEqual(self.client.get(path).status_code, 200)
|
||||
|
||||
|
||||
class TestAuthzAuditDoc(unittest.TestCase):
|
||||
"""The model must be written down, not only coded."""
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
cls.text = (
|
||||
AUTHZ_DOC.read_text(encoding="utf-8") if AUTHZ_DOC.exists() else ""
|
||||
)
|
||||
|
||||
def test_doc_exists(self):
|
||||
self.assertTrue(AUTHZ_DOC.exists(), f"missing {AUTHZ_DOC}")
|
||||
|
||||
def test_doc_covers_each_required_section(self):
|
||||
for heading in (
|
||||
"Identity sources",
|
||||
"Role matrix",
|
||||
"Privileged actions",
|
||||
"Secret redaction",
|
||||
"Audit event schema",
|
||||
"Retention",
|
||||
"Phase 2 integration",
|
||||
"Local-dev mode",
|
||||
):
|
||||
with self.subTest(heading=heading):
|
||||
self.assertIn(heading, self.text)
|
||||
|
||||
def test_doc_names_every_role(self):
|
||||
for role in console_authz.ROLE_ORDER:
|
||||
with self.subTest(role=role):
|
||||
self.assertIn(role, self.text)
|
||||
|
||||
def test_doc_names_every_console_action(self):
|
||||
for action_id in console_authz.ACTIONS:
|
||||
with self.subTest(action=action_id):
|
||||
self.assertIn(action_id, self.text)
|
||||
|
||||
def test_doc_states_retention_defaults(self):
|
||||
for days in console_audit.RETENTION_DAYS.values():
|
||||
with self.subTest(days=days):
|
||||
self.assertIn(str(days), self.text)
|
||||
|
||||
def test_doc_warns_local_dev_is_insecure(self):
|
||||
self.assertIn("INSECURE", self.text.upper())
|
||||
|
||||
def test_doc_states_default_deny(self):
|
||||
self.assertIn("deny", self.text.lower())
|
||||
|
||||
def test_doc_contains_no_secret_material(self):
|
||||
self.assertEqual(scan_for_secrets(self.text), [])
|
||||
|
||||
def test_deployment_doc_links_to_the_model(self):
|
||||
deployment = (DOCS / "webui-deployment.md").read_text(encoding="utf-8")
|
||||
self.assertIn("webui-authz-audit", deployment)
|
||||
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover
|
||||
unittest.main()
|
||||
@@ -1,468 +0,0 @@
|
||||
"""Tests for the unified web-console inventory API (#636).
|
||||
|
||||
Covers the four cases the issue names — empty, populated, partial failure, and
|
||||
the no-false-unowned invariant — plus redaction, collision detection, the
|
||||
resource-split routes, and read-only guarantees against a real control-plane
|
||||
database and real durable lock files.
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from starlette.testclient import TestClient
|
||||
|
||||
import control_plane_db
|
||||
from webui.app import create_app
|
||||
from webui import inventory
|
||||
|
||||
|
||||
def _iso(dt: datetime) -> str:
|
||||
return dt.astimezone(timezone.utc).isoformat()
|
||||
|
||||
|
||||
def _write_lock(lock_dir: str, name: str, payload: dict) -> str:
|
||||
path = os.path.join(lock_dir, name)
|
||||
with open(path, "w", encoding="utf-8") as handle:
|
||||
json.dump(payload, handle)
|
||||
return path
|
||||
|
||||
|
||||
def _live_lock_payload(
|
||||
*,
|
||||
issue_number: int,
|
||||
branch: str,
|
||||
worktree_path: str,
|
||||
pid: int,
|
||||
username: str = "jcwalker3",
|
||||
profile: str = "prgs-author",
|
||||
) -> dict:
|
||||
now = datetime.now(timezone.utc)
|
||||
future = now + timedelta(hours=2)
|
||||
return {
|
||||
"branch_name": branch,
|
||||
"issue_number": issue_number,
|
||||
"org": "Scaled-Tech-Consulting",
|
||||
"repo": "Gitea-Tools",
|
||||
"remote": "prgs",
|
||||
"pid": pid,
|
||||
"session_pid": pid,
|
||||
"lock_generation": 1,
|
||||
"worktree_path": worktree_path,
|
||||
"claimant": {"username": username, "profile": profile},
|
||||
"work_lease": {
|
||||
"branch": branch,
|
||||
"issue_number": issue_number,
|
||||
"operation_type": "author_issue_work",
|
||||
"created_at": _iso(now),
|
||||
"expires_at": _iso(future),
|
||||
"last_heartbeat_at": _iso(now),
|
||||
"claimant": {"username": username, "profile": profile},
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
class _FixtureMixin(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self._tmp = tempfile.TemporaryDirectory()
|
||||
self.tmp = self._tmp.name
|
||||
self.lock_dir = os.path.join(self.tmp, "locks")
|
||||
os.makedirs(self.lock_dir, mode=0o700)
|
||||
self.db_path = os.path.join(self.tmp, "control_plane.db")
|
||||
self.addCleanup(self._tmp.cleanup)
|
||||
|
||||
def _seed_db(self) -> control_plane_db.ControlPlaneDB:
|
||||
db = control_plane_db.ControlPlaneDB(self.db_path)
|
||||
db.upsert_session(
|
||||
session_id="prgs-author-1",
|
||||
role="author",
|
||||
profile="prgs-author",
|
||||
namespace="gitea-author",
|
||||
pid=os.getpid(),
|
||||
)
|
||||
db.upsert_work_item(
|
||||
remote="prgs",
|
||||
org="Scaled-Tech-Consulting",
|
||||
repo="Gitea-Tools",
|
||||
kind="issue",
|
||||
number=636,
|
||||
)
|
||||
db.assign_and_lease(
|
||||
session_id="prgs-author-1",
|
||||
role="author",
|
||||
remote="prgs",
|
||||
org="Scaled-Tech-Consulting",
|
||||
repo="Gitea-Tools",
|
||||
kind="issue",
|
||||
number=636,
|
||||
)
|
||||
return db
|
||||
|
||||
|
||||
class TestRedaction(unittest.TestCase):
|
||||
def test_redact_path_collapses_home(self):
|
||||
home = os.path.expanduser("~")
|
||||
self.assertEqual(
|
||||
inventory.redact_path(f"{home}/Development/Gitea-Tools"),
|
||||
"~/Development/Gitea-Tools",
|
||||
)
|
||||
|
||||
def test_redact_url_strips_userinfo_and_query(self):
|
||||
self.assertEqual(
|
||||
inventory.redact_url("https://user:[email protected]/api?token=abc"),
|
||||
"https://gitea.prgs.cc/api",
|
||||
)
|
||||
|
||||
def test_scrub_drops_credential_keys(self):
|
||||
scrubbed = inventory.scrub(
|
||||
{"token": "abc123", "api_key": "k", "profile": "prgs-author"}
|
||||
)
|
||||
self.assertEqual(scrubbed["token"], "[redacted]")
|
||||
self.assertEqual(scrubbed["api_key"], "[redacted]")
|
||||
self.assertEqual(scrubbed["profile"], "prgs-author")
|
||||
|
||||
def test_scrub_is_recursive_and_never_raises(self):
|
||||
class Weird:
|
||||
def __repr__(self) -> str:
|
||||
return "weird-obj"
|
||||
|
||||
out = inventory.scrub({"nested": [{"password": "p", "obj": Weird()}]})
|
||||
self.assertEqual(out["nested"][0]["password"], "[redacted]")
|
||||
self.assertEqual(out["nested"][0]["obj"], "weird-obj")
|
||||
|
||||
|
||||
class TestEmptyInventory(_FixtureMixin):
|
||||
def test_empty_db_and_locks_degrade_without_raising(self):
|
||||
# No DB file, no locks: sessions/leases unavailable, locks ok+empty.
|
||||
snap = inventory.load_inventory_snapshot(
|
||||
db_path=self.db_path,
|
||||
lock_dir=self.lock_dir,
|
||||
load_hygiene=lambda: _StubHygiene(entries=()),
|
||||
)
|
||||
sessions = snap.section("sessions")
|
||||
leases = snap.section("leases")
|
||||
locks = snap.section("locks")
|
||||
self.assertEqual(sessions.status, inventory.STATUS_UNAVAILABLE)
|
||||
self.assertEqual(leases.status, inventory.STATUS_UNAVAILABLE)
|
||||
self.assertEqual(locks.status, inventory.STATUS_OK)
|
||||
self.assertEqual(len(locks.items), 0)
|
||||
# Ownership authority is incomplete because the DB is missing.
|
||||
self.assertFalse(snap.ownership_authority_complete)
|
||||
self.assertEqual(snap.collisions, ())
|
||||
|
||||
def test_empty_db_present_but_unpopulated(self):
|
||||
control_plane_db.ControlPlaneDB(self.db_path) # creates schema, no rows
|
||||
snap = inventory.load_inventory_snapshot(
|
||||
db_path=self.db_path,
|
||||
lock_dir=self.lock_dir,
|
||||
load_hygiene=lambda: _StubHygiene(entries=()),
|
||||
)
|
||||
self.assertEqual(snap.section("sessions").status, inventory.STATUS_OK)
|
||||
self.assertEqual(len(snap.section("sessions").items), 0)
|
||||
self.assertEqual(snap.section("leases").status, inventory.STATUS_OK)
|
||||
self.assertTrue(snap.ownership_authority_complete)
|
||||
|
||||
|
||||
class TestPopulatedInventory(_FixtureMixin):
|
||||
def test_sections_populated_and_correlated(self):
|
||||
self._seed_db()
|
||||
wt = f"{self.tmp}/branches/issue-636-inventory-api"
|
||||
_write_lock(
|
||||
self.lock_dir,
|
||||
"prgs-Scaled-Tech-Consulting-Gitea-Tools-636.json",
|
||||
_live_lock_payload(
|
||||
issue_number=636,
|
||||
branch="feat/issue-636-inventory-api",
|
||||
worktree_path=wt,
|
||||
pid=os.getpid(),
|
||||
),
|
||||
)
|
||||
hygiene = _StubHygiene(
|
||||
entries=(
|
||||
_StubEntry(
|
||||
rel_path="branches/issue-636-inventory-api",
|
||||
branch="feat/issue-636-inventory-api",
|
||||
classification="active-issue",
|
||||
),
|
||||
)
|
||||
)
|
||||
snap = inventory.load_inventory_snapshot(
|
||||
db_path=self.db_path,
|
||||
lock_dir=self.lock_dir,
|
||||
load_hygiene=lambda: hygiene,
|
||||
)
|
||||
self.assertTrue(snap.ownership_authority_complete)
|
||||
self.assertEqual(len(snap.section("sessions").items), 1)
|
||||
self.assertEqual(len(snap.section("leases").items), 1)
|
||||
self.assertEqual(len(snap.section("locks").items), 1)
|
||||
self.assertEqual(len(snap.section("worktrees").items), 1)
|
||||
|
||||
# The lease, lock, and worktree for #636 correlate onto one row.
|
||||
row = next(r for r in snap.correlations if r["issue_number"] == 636)
|
||||
self.assertEqual(row["branch"], "feat/issue-636-inventory-api")
|
||||
self.assertTrue(row["lock_live"])
|
||||
self.assertEqual(row["worktree_classification"], "active-issue")
|
||||
self.assertEqual(len(row["lease_ids"]), 1)
|
||||
# No collision: live lock, live pid, matching worktree.
|
||||
self.assertEqual(snap.collisions, ())
|
||||
|
||||
def test_serialized_payload_declares_field_authority(self):
|
||||
self._seed_db()
|
||||
snap = inventory.load_inventory_snapshot(
|
||||
db_path=self.db_path,
|
||||
lock_dir=self.lock_dir,
|
||||
load_hygiene=lambda: _StubHygiene(entries=()),
|
||||
)
|
||||
payload = inventory.snapshot_to_dict(snap)
|
||||
self.assertEqual(payload["api_version"], "v1")
|
||||
self.assertEqual(payload["schema_version"], 1)
|
||||
self.assertEqual(payload["field_authority"]["sessions"], "control_plane_db")
|
||||
self.assertEqual(payload["field_authority"]["locks"], "filesystem")
|
||||
self.assertIn("sessions", payload["sections"])
|
||||
|
||||
|
||||
class TestPartialFailure(_FixtureMixin):
|
||||
def test_worktree_scan_failure_degrades_only_that_section(self):
|
||||
self._seed_db()
|
||||
|
||||
def _boom():
|
||||
raise RuntimeError("git worktree list exploded")
|
||||
|
||||
snap = inventory.load_inventory_snapshot(
|
||||
db_path=self.db_path,
|
||||
lock_dir=self.lock_dir,
|
||||
load_hygiene=_boom,
|
||||
)
|
||||
self.assertEqual(
|
||||
snap.section("worktrees").status, inventory.STATUS_UNAVAILABLE
|
||||
)
|
||||
self.assertIn("exploded", snap.section("worktrees").reason)
|
||||
# DB-backed sections still healthy.
|
||||
self.assertEqual(snap.section("sessions").status, inventory.STATUS_OK)
|
||||
self.assertIn("worktrees", snap.degraded_sections)
|
||||
|
||||
def test_degraded_ownership_suppresses_unowned_claim(self):
|
||||
# DB absent → sessions/leases unavailable → ownership incomplete even
|
||||
# though a lock exists and could look "unclaimed" by the DB alone.
|
||||
_write_lock(
|
||||
self.lock_dir,
|
||||
"prgs-Scaled-Tech-Consulting-Gitea-Tools-636.json",
|
||||
_live_lock_payload(
|
||||
issue_number=636,
|
||||
branch="feat/issue-636-inventory-api",
|
||||
worktree_path=f"{self.tmp}/wt",
|
||||
pid=os.getpid(),
|
||||
),
|
||||
)
|
||||
snap = inventory.load_inventory_snapshot(
|
||||
db_path=self.db_path,
|
||||
lock_dir=self.lock_dir,
|
||||
load_hygiene=lambda: _StubHygiene(entries=()),
|
||||
)
|
||||
self.assertFalse(snap.ownership_authority_complete)
|
||||
payload = inventory.snapshot_to_dict(snap)
|
||||
self.assertIn("may be treated as unowned", payload["ownership_note"])
|
||||
|
||||
|
||||
class TestCollisionDetection(_FixtureMixin):
|
||||
def test_live_lock_dead_owner_flagged(self):
|
||||
_write_lock(
|
||||
self.lock_dir,
|
||||
"prgs-Scaled-Tech-Consulting-Gitea-Tools-700.json",
|
||||
_live_lock_payload(
|
||||
issue_number=700,
|
||||
branch="feat/issue-700-x",
|
||||
worktree_path=f"{self.tmp}/wt700",
|
||||
pid=999_999_999, # not a running pid
|
||||
),
|
||||
)
|
||||
control_plane_db.ControlPlaneDB(self.db_path) # empty but present
|
||||
snap = inventory.load_inventory_snapshot(
|
||||
db_path=self.db_path,
|
||||
lock_dir=self.lock_dir,
|
||||
load_hygiene=lambda: _StubHygiene(entries=()),
|
||||
)
|
||||
kinds = {c.kind for c in snap.collisions}
|
||||
self.assertIn("live-lock-dead-owner", kinds)
|
||||
# Also lock-without-worktree, since no worktree carries the branch.
|
||||
self.assertIn("lock-without-worktree", kinds)
|
||||
|
||||
def test_duplicate_live_lock_on_same_branch(self):
|
||||
for issue in (800, 801):
|
||||
_write_lock(
|
||||
self.lock_dir,
|
||||
f"prgs-Scaled-Tech-Consulting-Gitea-Tools-{issue}.json",
|
||||
_live_lock_payload(
|
||||
issue_number=issue,
|
||||
branch="feat/issue-800-shared",
|
||||
worktree_path=f"{self.tmp}/wt{issue}",
|
||||
pid=os.getpid(),
|
||||
),
|
||||
)
|
||||
control_plane_db.ControlPlaneDB(self.db_path)
|
||||
snap = inventory.load_inventory_snapshot(
|
||||
db_path=self.db_path,
|
||||
lock_dir=self.lock_dir,
|
||||
load_hygiene=lambda: _StubHygiene(entries=()),
|
||||
)
|
||||
self.assertIn(
|
||||
"duplicate-live-lock", {c.kind for c in snap.collisions}
|
||||
)
|
||||
|
||||
def test_no_collision_when_sections_degraded(self):
|
||||
# locks ok but worktrees unavailable → lock-without-worktree must NOT
|
||||
# be asserted (a missing scan is not a missing worktree).
|
||||
_write_lock(
|
||||
self.lock_dir,
|
||||
"prgs-Scaled-Tech-Consulting-Gitea-Tools-636.json",
|
||||
_live_lock_payload(
|
||||
issue_number=636,
|
||||
branch="feat/issue-636-inventory-api",
|
||||
worktree_path=f"{self.tmp}/wt",
|
||||
pid=os.getpid(),
|
||||
),
|
||||
)
|
||||
control_plane_db.ControlPlaneDB(self.db_path)
|
||||
|
||||
def _boom():
|
||||
raise RuntimeError("scan down")
|
||||
|
||||
snap = inventory.load_inventory_snapshot(
|
||||
db_path=self.db_path,
|
||||
lock_dir=self.lock_dir,
|
||||
load_hygiene=_boom,
|
||||
)
|
||||
self.assertNotIn(
|
||||
"lock-without-worktree", {c.kind for c in snap.collisions}
|
||||
)
|
||||
|
||||
|
||||
class TestSectionInclude(_FixtureMixin):
|
||||
def test_include_restricts_scanned_sections(self):
|
||||
self._seed_db()
|
||||
snap = inventory.load_inventory_snapshot(
|
||||
db_path=self.db_path,
|
||||
lock_dir=self.lock_dir,
|
||||
include=("locks",),
|
||||
)
|
||||
self.assertIsNotNone(snap.section("locks"))
|
||||
self.assertIsNone(snap.section("sessions"))
|
||||
self.assertIsNone(snap.section("worktrees"))
|
||||
|
||||
|
||||
class TestRoutes(_FixtureMixin):
|
||||
def setUp(self) -> None:
|
||||
super().setUp()
|
||||
# Point the loaders at the fixture DB and lock dir via env, and stub
|
||||
# the worktree scan so the route does not shell out to git.
|
||||
self._prev_env = {
|
||||
"GITEA_CONTROL_PLANE_DB": os.environ.get("GITEA_CONTROL_PLANE_DB"),
|
||||
"GITEA_ISSUE_LOCK_DIR": os.environ.get("GITEA_ISSUE_LOCK_DIR"),
|
||||
"WEBUI_TEST_OFFLINE": os.environ.get("WEBUI_TEST_OFFLINE"),
|
||||
}
|
||||
os.environ["GITEA_CONTROL_PLANE_DB"] = self.db_path
|
||||
os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir
|
||||
os.environ["WEBUI_TEST_OFFLINE"] = "1"
|
||||
self._seed_db()
|
||||
self.client = TestClient(create_app())
|
||||
|
||||
def tearDown(self) -> None:
|
||||
for key, value in self._prev_env.items():
|
||||
if value is None:
|
||||
os.environ.pop(key, None)
|
||||
else:
|
||||
os.environ[key] = value
|
||||
|
||||
def test_inventory_route_returns_versioned_payload(self):
|
||||
resp = self.client.get("/api/v1/inventory")
|
||||
self.assertEqual(resp.status_code, 200)
|
||||
body = resp.json()
|
||||
self.assertEqual(body["api_version"], "v1")
|
||||
self.assertIn("sessions", body["sections"])
|
||||
self.assertIn("field_authority", body)
|
||||
|
||||
def test_section_route_restricts_and_labels(self):
|
||||
resp = self.client.get("/api/v1/inventory/locks")
|
||||
self.assertEqual(resp.status_code, 200)
|
||||
body = resp.json()
|
||||
self.assertEqual(body["requested_section"], "locks")
|
||||
self.assertIn("locks", body["sections"])
|
||||
self.assertNotIn("sessions", body["sections"])
|
||||
|
||||
def test_unknown_section_is_404(self):
|
||||
resp = self.client.get("/api/v1/inventory/bogus")
|
||||
self.assertEqual(resp.status_code, 404)
|
||||
self.assertEqual(resp.json()["error"], "unknown_section")
|
||||
|
||||
def test_inventory_route_rejects_post(self):
|
||||
resp = self.client.post("/api/v1/inventory")
|
||||
self.assertEqual(resp.status_code, 405)
|
||||
|
||||
|
||||
class TestReadOnly(_FixtureMixin):
|
||||
def test_snapshot_does_not_create_db_file(self):
|
||||
missing = os.path.join(self.tmp, "does-not-exist.db")
|
||||
inventory.load_inventory_snapshot(
|
||||
db_path=missing,
|
||||
lock_dir=self.lock_dir,
|
||||
load_hygiene=lambda: _StubHygiene(entries=()),
|
||||
)
|
||||
self.assertFalse(os.path.exists(missing))
|
||||
|
||||
def test_readonly_connection_refuses_write(self):
|
||||
self._seed_db()
|
||||
conn = inventory._open_readonly(self.db_path)
|
||||
try:
|
||||
with self.assertRaises(Exception):
|
||||
conn.execute(
|
||||
"INSERT INTO sessions(session_id, role, started_at, "
|
||||
"last_heartbeat_at, status) VALUES ('x','author',"
|
||||
"'t','t','active')"
|
||||
)
|
||||
conn.commit()
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
# ── lightweight stand-ins for the #432 hygiene snapshot ──────────────────────
|
||||
|
||||
|
||||
class _StubEntry:
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
rel_path: str,
|
||||
branch: str | None = None,
|
||||
classification: str = "stale-clean",
|
||||
head_sha: str | None = "abc123",
|
||||
dirty_tracked: int = 0,
|
||||
dirty_untracked: bool = False,
|
||||
detached: bool = False,
|
||||
registered_worktree: bool = True,
|
||||
notes: str = "",
|
||||
) -> None:
|
||||
self.rel_path = rel_path
|
||||
self.folder_name = rel_path.split("/", 1)[-1]
|
||||
self.branch = branch
|
||||
self.classification = classification
|
||||
self.head_sha = head_sha
|
||||
self.dirty_tracked = dirty_tracked
|
||||
self.dirty_untracked = dirty_untracked
|
||||
self.detached = detached
|
||||
self.registered_worktree = registered_worktree
|
||||
self.notes = notes
|
||||
|
||||
|
||||
class _StubHygiene:
|
||||
def __init__(self, *, entries=(), scan_error=None) -> None:
|
||||
self.entries = tuple(entries)
|
||||
self.scan_error = scan_error
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
+75
-30
@@ -2,6 +2,7 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from starlette.applications import Starlette
|
||||
@@ -31,6 +32,9 @@ from final_report_validator import FINAL_REPORT_TASK_KINDS
|
||||
|
||||
from webui.gated_actions import attempt_action, load_action_registry, preview_action
|
||||
from webui.gated_action_views import render_actions_page
|
||||
from webui import console_audit
|
||||
from webui.console_authz import authorize, rbac_matrix, resolve_principal
|
||||
from webui.console_redaction import redaction_policy
|
||||
from webui.audit_validator import audit_report, audit_to_dict
|
||||
from webui.audit_views import render_audit_page
|
||||
from webui.lease_loader import load_lease_snapshot, snapshot_to_dict as lease_snapshot_to_dict
|
||||
@@ -41,11 +45,6 @@ from webui.worktree_scanner import load_hygiene_snapshot, snapshot_to_dict as wo
|
||||
from webui.worktree_views import render_worktrees_page
|
||||
from webui.runtime_health import load_runtime_snapshot, snapshot_to_dict as runtime_snapshot_to_dict
|
||||
from webui.runtime_views import render_runtime_page
|
||||
from webui.inventory import (
|
||||
SECTION_NAMES as _INVENTORY_SECTIONS,
|
||||
load_inventory_snapshot,
|
||||
snapshot_to_dict as inventory_snapshot_to_dict,
|
||||
)
|
||||
|
||||
_READ_ONLY_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})
|
||||
_AUDIT_MUTATION_PATHS = frozenset({"/audit", "/api/audit"})
|
||||
@@ -281,6 +280,49 @@ async def api_actions(_request: Request) -> JSONResponse:
|
||||
return JSONResponse(load_action_registry().to_dict())
|
||||
|
||||
|
||||
def _request_id() -> str:
|
||||
return f"req-{uuid.uuid4().hex}"
|
||||
|
||||
|
||||
def _audit_target(action_id: str, params: dict[str, object]) -> dict[str, object]:
|
||||
"""Describe the action target for the audit record (never secrets)."""
|
||||
if "pr_number" in params:
|
||||
return {"kind": "pr", "ref": f"#{params['pr_number']}"}
|
||||
if "issue_number" in params:
|
||||
return {"kind": "issue", "ref": f"#{params['issue_number']}"}
|
||||
if "branch_name" in params:
|
||||
return {"kind": "branch", "ref": str(params["branch_name"])}
|
||||
return {"kind": "unspecified", "ref": action_id}
|
||||
|
||||
|
||||
def _authorize_request(
|
||||
request: Request,
|
||||
action_id: str,
|
||||
params: dict[str, object],
|
||||
*,
|
||||
for_execution: bool,
|
||||
result: str,
|
||||
) -> dict[str, object]:
|
||||
"""Resolve principal, decide, and audit. Returns the decision payload.
|
||||
|
||||
Phase 1 records the decision rather than enforcing it as the terminal
|
||||
outcome: ``webui.gated_actions`` already fails closed for every action, so
|
||||
this layer cannot loosen anything. Phase 2 enforces on this same decision.
|
||||
"""
|
||||
principal = resolve_principal(headers=dict(request.headers))
|
||||
decision = authorize(action_id, principal, for_execution=for_execution)
|
||||
console_audit.record_event(
|
||||
action_id=action_id,
|
||||
result=result,
|
||||
decision=decision,
|
||||
principal=principal,
|
||||
target=_audit_target(action_id, params),
|
||||
request_id=_request_id(),
|
||||
detail=decision.detail,
|
||||
)
|
||||
return decision.to_dict()
|
||||
|
||||
|
||||
async def api_action_preview(request: Request) -> JSONResponse:
|
||||
action_id = request.path_params["action_id"]
|
||||
params = dict(request.query_params)
|
||||
@@ -290,6 +332,13 @@ async def api_action_preview(request: Request) -> JSONResponse:
|
||||
result = preview_action(action_id, **params)
|
||||
if "error" in result:
|
||||
return JSONResponse(result, status_code=404)
|
||||
result["authorization"] = _authorize_request(
|
||||
request,
|
||||
action_id,
|
||||
params,
|
||||
for_execution=False,
|
||||
result=console_audit.RESULT_PREVIEWED,
|
||||
)
|
||||
return JSONResponse(result)
|
||||
|
||||
|
||||
@@ -303,32 +352,29 @@ 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_inventory(_request: Request) -> JSONResponse:
|
||||
"""Unified read-only session/lease/lock/worktree inventory (#636)."""
|
||||
snapshot = load_inventory_snapshot()
|
||||
return JSONResponse(inventory_snapshot_to_dict(snapshot))
|
||||
|
||||
|
||||
async def api_inventory_section(request: Request) -> JSONResponse:
|
||||
"""Resource-split view: one inventory section under the shared schema."""
|
||||
section = request.path_params["section"]
|
||||
if section not in _INVENTORY_SECTIONS:
|
||||
return JSONResponse(
|
||||
{
|
||||
"error": "unknown_section",
|
||||
"detail": f"no inventory section named {section!r}",
|
||||
"available": sorted(_INVENTORY_SECTIONS),
|
||||
},
|
||||
status_code=404,
|
||||
)
|
||||
snapshot = load_inventory_snapshot(include=(section,))
|
||||
payload = inventory_snapshot_to_dict(snapshot)
|
||||
payload["requested_section"] = section
|
||||
return JSONResponse(payload)
|
||||
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:
|
||||
@@ -387,10 +433,9 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
|
||||
methods=["POST"],
|
||||
),
|
||||
Route("/api/leases", api_leases, methods=["GET"]),
|
||||
Route("/api/v1/inventory", api_inventory, methods=["GET"]),
|
||||
Route(
|
||||
"/api/v1/inventory/{section}",
|
||||
api_inventory_section,
|
||||
"/api/console/security-model",
|
||||
api_console_security_model,
|
||||
methods=["GET"],
|
||||
),
|
||||
],
|
||||
|
||||
@@ -0,0 +1,281 @@
|
||||
"""Console audit event schema, retention, and append-only sink (#633).
|
||||
|
||||
``gitea_audit`` records MCP-side *mutations*: which profile and Gitea user
|
||||
performed which tool call. It carries no console actor, no identity source, no
|
||||
correlation identifier, and no retention class, so it cannot answer the
|
||||
question #633 exists to answer — *who sat at the console, what did they
|
||||
attempt, and was it authorized?* An authorization denial is not a mutation and
|
||||
would never appear there at all.
|
||||
|
||||
This module adds the console-side record. It does not replace ``gitea_audit``:
|
||||
when a Phase 2 action eventually reaches MCP, both fire, correlated by
|
||||
``correlation.request_id``.
|
||||
|
||||
Design constraints:
|
||||
|
||||
- **Redact before persist.** Every record passes through
|
||||
``webui.console_redaction.redact_payload`` before serialization, so an
|
||||
unredacted field is never durable.
|
||||
- **Append-only.** Records are appended as JSON lines. Nothing here updates or
|
||||
deletes; retention is metadata on each record, enforced by an operator-run
|
||||
policy, never by silent rewriting.
|
||||
- **Never raises.** Auditing must not break the request it describes. A failed
|
||||
write returns ``False``.
|
||||
- **Off by default.** With ``WEBUI_CONSOLE_AUDIT_LOG`` unset, events are still
|
||||
*built* (so callers and tests see the schema) but nothing is written.
|
||||
|
||||
A record looks like this (synthetic values):
|
||||
|
||||
{"schema_version": 1, "event_id": "evt-0001",
|
||||
"timestamp": "2026-07-22T10:16:42+00:00",
|
||||
"actor": {"subject": "[email protected]", "role": "operator",
|
||||
"identity_source": "access_proxy", "authenticated": true},
|
||||
"action": "merge_pr", "action_class": "privileged",
|
||||
"target": {"kind": "pr", "ref": "#123"},
|
||||
"result": "denied", "reason_code": "insufficient_role",
|
||||
"correlation": {"request_id": "req-abc", "session_id": null,
|
||||
"mcp_task": "merge_pr", "mcp_permission": "gitea.pr.merge"},
|
||||
"retention": {"class": "privileged", "days": 365,
|
||||
"expires_at": "2027-07-22T10:16:42+00:00"},
|
||||
"redacted": true}
|
||||
|
||||
Timestamps are timezone-aware ISO-8601 in UTC.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime
|
||||
import json
|
||||
import os
|
||||
import uuid
|
||||
from typing import Any
|
||||
|
||||
from webui import console_authz
|
||||
from webui.console_redaction import redact_payload, scan_for_secrets
|
||||
|
||||
SCHEMA_VERSION = 1
|
||||
|
||||
AUDIT_LOG_ENV = "WEBUI_CONSOLE_AUDIT_LOG"
|
||||
|
||||
# Result vocabulary. ``denied`` is the one ``gitea_audit`` has no equivalent
|
||||
# for: an authorization refusal never reaches the MCP layer.
|
||||
RESULT_ALLOWED = "allowed"
|
||||
RESULT_DENIED = "denied"
|
||||
RESULT_PREVIEWED = "previewed"
|
||||
RESULT_FAILED = "failed"
|
||||
RESULT_SUCCEEDED = "succeeded"
|
||||
|
||||
RESULTS = frozenset(
|
||||
{
|
||||
RESULT_ALLOWED,
|
||||
RESULT_DENIED,
|
||||
RESULT_PREVIEWED,
|
||||
RESULT_FAILED,
|
||||
RESULT_SUCCEEDED,
|
||||
}
|
||||
)
|
||||
|
||||
# Retention classes and default lifetimes in days. Privileged and break-glass
|
||||
# records outlive routine ones because they are what an incident review needs.
|
||||
RETENTION_STANDARD = "standard"
|
||||
RETENTION_PRIVILEGED = "privileged"
|
||||
RETENTION_BREAK_GLASS = "break_glass"
|
||||
|
||||
RETENTION_DAYS: dict[str, int] = {
|
||||
RETENTION_STANDARD: 90,
|
||||
RETENTION_PRIVILEGED: 365,
|
||||
RETENTION_BREAK_GLASS: 730,
|
||||
}
|
||||
|
||||
# Fields every record must carry. Asserted by the test suite so a future edit
|
||||
# cannot quietly drop one.
|
||||
REQUIRED_FIELDS: tuple[str, ...] = (
|
||||
"schema_version",
|
||||
"event_id",
|
||||
"timestamp",
|
||||
"actor",
|
||||
"action",
|
||||
"action_class",
|
||||
"target",
|
||||
"result",
|
||||
"reason_code",
|
||||
"correlation",
|
||||
"retention",
|
||||
"redacted",
|
||||
)
|
||||
|
||||
REQUIRED_ACTOR_FIELDS: tuple[str, ...] = (
|
||||
"subject",
|
||||
"role",
|
||||
"identity_source",
|
||||
"authenticated",
|
||||
)
|
||||
|
||||
REQUIRED_CORRELATION_FIELDS: tuple[str, ...] = (
|
||||
"request_id",
|
||||
"session_id",
|
||||
"mcp_task",
|
||||
"mcp_permission",
|
||||
)
|
||||
|
||||
|
||||
def audit_log_path() -> str | None:
|
||||
"""Configured sink path, or ``None`` when console auditing is off."""
|
||||
return (os.environ.get(AUDIT_LOG_ENV) or "").strip() or None
|
||||
|
||||
|
||||
def audit_enabled() -> bool:
|
||||
return audit_log_path() is not None
|
||||
|
||||
|
||||
def retention_class_for(action: console_authz.ConsoleAction | None) -> str:
|
||||
"""Classify retention from the action, defaulting to the longest-lived.
|
||||
|
||||
An unknown action is treated as privileged rather than standard: for a
|
||||
safety control the conservative direction is to keep the record longer.
|
||||
"""
|
||||
if action is None:
|
||||
return RETENTION_PRIVILEGED
|
||||
if action.break_glass:
|
||||
return RETENTION_BREAK_GLASS
|
||||
if action.privileged:
|
||||
return RETENTION_PRIVILEGED
|
||||
return RETENTION_STANDARD
|
||||
|
||||
|
||||
def _retention_block(
|
||||
retention_class: str, now: datetime.datetime
|
||||
) -> dict[str, Any]:
|
||||
days = RETENTION_DAYS.get(
|
||||
retention_class, RETENTION_DAYS[RETENTION_PRIVILEGED]
|
||||
)
|
||||
return {
|
||||
"class": retention_class,
|
||||
"days": days,
|
||||
"expires_at": (now + datetime.timedelta(days=days)).isoformat(),
|
||||
}
|
||||
|
||||
|
||||
def build_event(
|
||||
*,
|
||||
action_id: str,
|
||||
result: str,
|
||||
decision: console_authz.AuthorizationDecision | None = None,
|
||||
principal: console_authz.Principal | None = None,
|
||||
target: dict[str, Any] | None = None,
|
||||
reason_code: str | None = None,
|
||||
request_id: str | None = None,
|
||||
session_id: str | None = None,
|
||||
detail: str | None = None,
|
||||
metadata: dict[str, Any] | None = None,
|
||||
now: datetime.datetime | None = None,
|
||||
event_id: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Build one redacted, JSON-able console audit record.
|
||||
|
||||
Redaction runs here rather than at write time so an in-memory record handed
|
||||
to a template or an API response is already clean.
|
||||
"""
|
||||
ts = now or datetime.datetime.now(datetime.timezone.utc)
|
||||
action = console_authz.get_action(action_id)
|
||||
who = principal or (
|
||||
decision.principal if decision else console_authz.ANONYMOUS
|
||||
)
|
||||
resolved_result = result if result in RESULTS else RESULT_FAILED
|
||||
resolved_reason = reason_code or (
|
||||
decision.reason_code if decision else "unspecified"
|
||||
)
|
||||
retention_class = retention_class_for(action)
|
||||
|
||||
event: dict[str, Any] = {
|
||||
"schema_version": SCHEMA_VERSION,
|
||||
"event_id": event_id or f"evt-{uuid.uuid4().hex}",
|
||||
"timestamp": ts.isoformat(),
|
||||
"actor": who.to_dict(),
|
||||
"action": action_id,
|
||||
"action_class": action.action_class if action else "unknown",
|
||||
"target": dict(target or {}),
|
||||
"result": resolved_result,
|
||||
"reason_code": resolved_reason,
|
||||
"correlation": {
|
||||
"request_id": request_id,
|
||||
"session_id": session_id,
|
||||
"mcp_task": action.task_key if action else None,
|
||||
"mcp_permission": action.mcp_permission if action else None,
|
||||
},
|
||||
"retention": _retention_block(retention_class, ts),
|
||||
"redacted": True,
|
||||
"detail": detail,
|
||||
"metadata": dict(metadata or {}),
|
||||
}
|
||||
if decision is not None:
|
||||
# Deliberately *not* named "authorization": ``gitea_audit`` treats that
|
||||
# substring as a secret key hint (it matches the HTTP Authorization
|
||||
# header) and would replace this whole block with the placeholder.
|
||||
event["decision"] = {
|
||||
"allowed": decision.allowed,
|
||||
"required_role": decision.required_role,
|
||||
"requires_confirmation": decision.requires_confirmation,
|
||||
"dual_control": decision.dual_control,
|
||||
"break_glass": decision.break_glass,
|
||||
"execution_enabled": decision.execution_enabled,
|
||||
}
|
||||
|
||||
redacted = redact_payload(event)
|
||||
if not isinstance(redacted, dict): # pragma: no cover - defensive
|
||||
return {"schema_version": SCHEMA_VERSION, "redacted": True}
|
||||
return redacted
|
||||
|
||||
|
||||
def write_event(event: dict[str, Any], path: str | None = None) -> bool:
|
||||
"""Append *event* as one JSON line. Never raises.
|
||||
|
||||
Returns ``True`` when a line was written, ``False`` when auditing is off or
|
||||
the write failed. A record that still trips a secret detector is dropped
|
||||
rather than persisted.
|
||||
"""
|
||||
sink = path or audit_log_path()
|
||||
if not sink:
|
||||
return False
|
||||
try:
|
||||
if scan_for_secrets(event):
|
||||
return False
|
||||
line = json.dumps(event, default=str, sort_keys=True)
|
||||
with open(sink, "a", encoding="utf-8") as handle:
|
||||
handle.write(line + "\n")
|
||||
return True
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
def record_event(**kwargs: Any) -> dict[str, Any]:
|
||||
"""Build and persist one record; return the record either way.
|
||||
|
||||
Callers get the record back so it can be surfaced in a response or a test
|
||||
regardless of whether a sink is configured.
|
||||
"""
|
||||
event = build_event(**kwargs)
|
||||
written = write_event(event)
|
||||
return {"event": event, "written": written}
|
||||
|
||||
|
||||
def audit_policy() -> dict[str, Any]:
|
||||
"""Machine-readable audit schema and retention defaults (never secrets)."""
|
||||
return {
|
||||
"schema_version": SCHEMA_VERSION,
|
||||
"required_fields": list(REQUIRED_FIELDS),
|
||||
"required_actor_fields": list(REQUIRED_ACTOR_FIELDS),
|
||||
"required_correlation_fields": list(REQUIRED_CORRELATION_FIELDS),
|
||||
"results": sorted(RESULTS),
|
||||
"retention_defaults_days": dict(RETENTION_DAYS),
|
||||
"sink_env": AUDIT_LOG_ENV,
|
||||
"enabled": audit_enabled(),
|
||||
"append_only": True,
|
||||
"redact_before_persist": True,
|
||||
"timestamp_format": "ISO-8601, timezone-aware, UTC",
|
||||
"relationship_to_mcp_audit": (
|
||||
"webui.console_audit records console intent and authorization "
|
||||
"outcomes; gitea_audit records MCP mutations. A Phase 2 action "
|
||||
"emits both, correlated by correlation.request_id."
|
||||
),
|
||||
}
|
||||
@@ -0,0 +1,537 @@
|
||||
"""Console authorization and RBAC model (#633, Phase 1).
|
||||
|
||||
The read-only MVP (#426–#436) ships with no authentication: protection comes
|
||||
from network placement alone (#435). That is adequate while every route is a
|
||||
GET, and inadequate the moment Phase 2 wires a gated write. This module is the
|
||||
authorization model those writes must go through, landed *before* any of them
|
||||
exists so no write can be added without an authority to check against.
|
||||
|
||||
Phase 1 scope is the model itself: identity resolution, the role matrix, the
|
||||
privileged-action list, and a fail-closed :func:`authorize`. It deliberately
|
||||
does **not** enable any write. ``webui.gated_actions`` stays globally disabled,
|
||||
so an allow decision here is necessary but never sufficient.
|
||||
|
||||
Two invariants hold for every caller:
|
||||
|
||||
- **Default deny.** An unrecognised action, an unknown role, or an absent
|
||||
principal denies. There is no implicit allow branch and no "unless" clause.
|
||||
- **Authorization is not execution.** :func:`authorize` returns a decision
|
||||
record. It never calls MCP, never mutates, and never consults credentials.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
from dataclasses import asdict, dataclass, field
|
||||
from typing import Any
|
||||
|
||||
from task_capability_map import required_permission, required_role
|
||||
|
||||
# --- Roles ------------------------------------------------------------------
|
||||
# Ordered least to most authority. Higher ranks inherit every lower rank's
|
||||
# permitted actions; the matrix below is expressed as a minimum required rank.
|
||||
VIEWER = "viewer"
|
||||
OPERATOR = "operator"
|
||||
CONTROLLER = "controller"
|
||||
ADMIN = "admin"
|
||||
|
||||
ROLE_ORDER: tuple[str, ...] = (VIEWER, OPERATOR, CONTROLLER, ADMIN)
|
||||
_ROLE_RANK: dict[str, int] = {role: idx for idx, role in enumerate(ROLE_ORDER)}
|
||||
|
||||
ROLE_DESCRIPTIONS: dict[str, str] = {
|
||||
VIEWER: "Read every console view. No write, ever, in any phase.",
|
||||
OPERATOR: "Viewer, plus author-class work: claim, comment, open a PR.",
|
||||
CONTROLLER: "Operator, plus reviewer/merger-class decisions on a PR.",
|
||||
ADMIN: "Controller, plus destructive and policy-editing actions.",
|
||||
}
|
||||
|
||||
# --- Identity sources -------------------------------------------------------
|
||||
IDENTITY_NONE = "none"
|
||||
IDENTITY_LOCAL_DEV = "local_dev"
|
||||
IDENTITY_ACCESS_PROXY = "access_proxy"
|
||||
|
||||
IDENTITY_SOURCES: dict[str, dict[str, Any]] = {
|
||||
IDENTITY_NONE: {
|
||||
"description": (
|
||||
"No authentication configured. Every request is anonymous and "
|
||||
"capped at viewer. This is the MVP default and the only mode "
|
||||
"whose safety rests entirely on network placement (#435)."
|
||||
),
|
||||
"authenticated": False,
|
||||
"safe_for_shared_host": False,
|
||||
"phase_available": 1,
|
||||
},
|
||||
IDENTITY_LOCAL_DEV: {
|
||||
"description": (
|
||||
"Developer-supplied principal read from the environment. INSECURE: "
|
||||
"the subject and role are asserted, never verified. Loopback only."
|
||||
),
|
||||
"authenticated": True,
|
||||
"safe_for_shared_host": False,
|
||||
"phase_available": 1,
|
||||
},
|
||||
IDENTITY_ACCESS_PROXY: {
|
||||
"description": (
|
||||
"Subject asserted by a trusted access proxy (Cloudflare Access, "
|
||||
"WARP, or an org VPN portal) via a verified request header. The "
|
||||
"proxy performs authentication; the console performs authorization."
|
||||
),
|
||||
"authenticated": True,
|
||||
"safe_for_shared_host": True,
|
||||
"phase_available": 2,
|
||||
},
|
||||
}
|
||||
|
||||
# Environment configuration. All are read server-side and never rendered.
|
||||
AUTH_MODE_ENV = "WEBUI_AUTH_MODE"
|
||||
DEV_SUBJECT_ENV = "WEBUI_DEV_SUBJECT"
|
||||
DEV_ROLE_ENV = "WEBUI_DEV_ROLE"
|
||||
ROLE_MAP_ENV = "WEBUI_ROLE_MAP"
|
||||
REQUIRE_PROBE_AUTH_ENV = "WEBUI_REQUIRE_PROBE_AUTH"
|
||||
ACCESS_SUBJECT_HEADER = "cf-access-authenticated-user-email"
|
||||
|
||||
# --- Action classes ---------------------------------------------------------
|
||||
CLASS_READ = "read"
|
||||
CLASS_WRITE = "gated_write"
|
||||
CLASS_PRIVILEGED = "privileged"
|
||||
CLASS_DESTRUCTIVE = "destructive"
|
||||
|
||||
# --- Privileged action list -------------------------------------------------
|
||||
# ``task_key`` ties each console action back to ``task_capability_map``, so the
|
||||
# console cannot invent an authority the MCP layer does not already define.
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ConsoleAction:
|
||||
"""One console action and the authority required to invoke it."""
|
||||
|
||||
action_id: str
|
||||
task_key: str
|
||||
action_class: str
|
||||
minimum_role: str
|
||||
requires_confirmation: bool
|
||||
dual_control: bool
|
||||
break_glass: bool
|
||||
phase: int
|
||||
summary: str
|
||||
|
||||
@property
|
||||
def mcp_permission(self) -> str:
|
||||
return required_permission(self.task_key)
|
||||
|
||||
@property
|
||||
def mcp_role(self) -> str:
|
||||
return required_role(self.task_key)
|
||||
|
||||
@property
|
||||
def privileged(self) -> bool:
|
||||
return self.action_class in {CLASS_PRIVILEGED, CLASS_DESTRUCTIVE}
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
data = asdict(self)
|
||||
data["mcp_permission"] = self.mcp_permission
|
||||
data["mcp_role"] = self.mcp_role
|
||||
data["privileged"] = self.privileged
|
||||
return data
|
||||
|
||||
|
||||
_ACTION_SPECS: tuple[ConsoleAction, ...] = (
|
||||
ConsoleAction(
|
||||
action_id="claim_issue",
|
||||
task_key="claim_issue",
|
||||
action_class=CLASS_WRITE,
|
||||
minimum_role=OPERATOR,
|
||||
requires_confirmation=True,
|
||||
dual_control=False,
|
||||
break_glass=False,
|
||||
phase=2,
|
||||
summary="Apply status:in-progress to an issue.",
|
||||
),
|
||||
ConsoleAction(
|
||||
action_id="comment_issue",
|
||||
task_key="comment_issue",
|
||||
action_class=CLASS_WRITE,
|
||||
minimum_role=OPERATOR,
|
||||
requires_confirmation=True,
|
||||
dual_control=False,
|
||||
break_glass=False,
|
||||
phase=2,
|
||||
summary="Post an issue comment.",
|
||||
),
|
||||
ConsoleAction(
|
||||
action_id="create_issue",
|
||||
task_key="create_issue",
|
||||
action_class=CLASS_WRITE,
|
||||
minimum_role=OPERATOR,
|
||||
requires_confirmation=True,
|
||||
dual_control=False,
|
||||
break_glass=False,
|
||||
phase=2,
|
||||
summary="Open a new tracking issue.",
|
||||
),
|
||||
ConsoleAction(
|
||||
action_id="comment_pr",
|
||||
task_key="comment_pr",
|
||||
action_class=CLASS_WRITE,
|
||||
minimum_role=OPERATOR,
|
||||
requires_confirmation=True,
|
||||
dual_control=False,
|
||||
break_glass=False,
|
||||
phase=2,
|
||||
summary="Post a PR thread comment.",
|
||||
),
|
||||
ConsoleAction(
|
||||
action_id="create_pr",
|
||||
task_key="create_pr",
|
||||
action_class=CLASS_WRITE,
|
||||
minimum_role=OPERATOR,
|
||||
requires_confirmation=True,
|
||||
dual_control=False,
|
||||
break_glass=False,
|
||||
phase=2,
|
||||
summary="Open a PR from a locked feature branch.",
|
||||
),
|
||||
ConsoleAction(
|
||||
action_id="review_pr",
|
||||
task_key="review_pr",
|
||||
action_class=CLASS_PRIVILEGED,
|
||||
minimum_role=CONTROLLER,
|
||||
requires_confirmation=True,
|
||||
dual_control=False,
|
||||
break_glass=False,
|
||||
phase=3,
|
||||
summary="Submit an approve / request-changes verdict.",
|
||||
),
|
||||
ConsoleAction(
|
||||
action_id="close_pr",
|
||||
task_key="close_pr",
|
||||
action_class=CLASS_PRIVILEGED,
|
||||
minimum_role=CONTROLLER,
|
||||
requires_confirmation=True,
|
||||
dual_control=False,
|
||||
break_glass=False,
|
||||
phase=3,
|
||||
summary="Close a pull request without merging.",
|
||||
),
|
||||
ConsoleAction(
|
||||
action_id="merge_pr",
|
||||
task_key="merge_pr",
|
||||
action_class=CLASS_PRIVILEGED,
|
||||
minimum_role=CONTROLLER,
|
||||
requires_confirmation=True,
|
||||
dual_control=True,
|
||||
break_glass=True,
|
||||
phase=3,
|
||||
summary="Merge an approved pull request.",
|
||||
),
|
||||
ConsoleAction(
|
||||
action_id="delete_branch",
|
||||
task_key="delete_branch",
|
||||
action_class=CLASS_DESTRUCTIVE,
|
||||
minimum_role=ADMIN,
|
||||
requires_confirmation=True,
|
||||
dual_control=True,
|
||||
break_glass=True,
|
||||
phase=3,
|
||||
summary="Remove a remote feature branch.",
|
||||
),
|
||||
)
|
||||
|
||||
ACTIONS: dict[str, ConsoleAction] = {a.action_id: a for a in _ACTION_SPECS}
|
||||
|
||||
|
||||
def privileged_actions() -> tuple[ConsoleAction, ...]:
|
||||
"""Actions requiring dual control, break-glass, or controller+ authority."""
|
||||
return tuple(a for a in _ACTION_SPECS if a.privileged)
|
||||
|
||||
|
||||
def get_action(action_id: str) -> ConsoleAction | None:
|
||||
return ACTIONS.get(action_id)
|
||||
|
||||
|
||||
# --- Principals -------------------------------------------------------------
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Principal:
|
||||
"""Who is making a request, and how strongly that is known."""
|
||||
|
||||
subject: str
|
||||
role: str
|
||||
identity_source: str
|
||||
authenticated: bool
|
||||
warnings: tuple[str, ...] = field(default_factory=tuple)
|
||||
|
||||
@property
|
||||
def rank(self) -> int:
|
||||
return _ROLE_RANK.get(self.role, -1)
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
return {
|
||||
"subject": self.subject,
|
||||
"role": self.role,
|
||||
"identity_source": self.identity_source,
|
||||
"authenticated": self.authenticated,
|
||||
"warnings": list(self.warnings),
|
||||
}
|
||||
|
||||
|
||||
ANONYMOUS = Principal(
|
||||
subject="anonymous",
|
||||
role=VIEWER,
|
||||
identity_source=IDENTITY_NONE,
|
||||
authenticated=False,
|
||||
warnings=("No authentication configured; capped at viewer.",),
|
||||
)
|
||||
|
||||
|
||||
def auth_mode(env: dict[str, str] | None = None) -> str:
|
||||
"""Resolve the configured identity source, defaulting to ``none``."""
|
||||
source = env if env is not None else os.environ
|
||||
raw = (source.get(AUTH_MODE_ENV) or "").strip().lower().replace("-", "_")
|
||||
if raw in IDENTITY_SOURCES:
|
||||
return raw
|
||||
return IDENTITY_NONE
|
||||
|
||||
|
||||
def _role_map(env: dict[str, str]) -> dict[str, str]:
|
||||
"""Parse ``WEBUI_ROLE_MAP`` (JSON subject→role). Invalid config yields {}."""
|
||||
raw = (env.get(ROLE_MAP_ENV) or "").strip()
|
||||
if not raw:
|
||||
return {}
|
||||
try:
|
||||
parsed = json.loads(raw)
|
||||
except Exception:
|
||||
return {}
|
||||
if not isinstance(parsed, dict):
|
||||
return {}
|
||||
return {
|
||||
str(k): str(v).strip().lower()
|
||||
for k, v in parsed.items()
|
||||
if str(v).strip().lower() in _ROLE_RANK
|
||||
}
|
||||
|
||||
|
||||
def resolve_principal(
|
||||
headers: dict[str, str] | None = None,
|
||||
env: dict[str, str] | None = None,
|
||||
) -> Principal:
|
||||
"""Resolve the requesting principal. Unknown or unconfigured → anonymous.
|
||||
|
||||
Never raises and never trusts a client-supplied role: the role always comes
|
||||
from server-side configuration keyed by the resolved subject.
|
||||
"""
|
||||
source_env = dict(env) if env is not None else dict(os.environ)
|
||||
lowered = {str(k).lower(): str(v) for k, v in (headers or {}).items()}
|
||||
mode = auth_mode(source_env)
|
||||
|
||||
if mode == IDENTITY_LOCAL_DEV:
|
||||
subject = (source_env.get(DEV_SUBJECT_ENV) or "").strip()
|
||||
if not subject:
|
||||
return ANONYMOUS
|
||||
role = (source_env.get(DEV_ROLE_ENV) or VIEWER).strip().lower()
|
||||
if role not in _ROLE_RANK:
|
||||
role = VIEWER
|
||||
return Principal(
|
||||
subject=subject,
|
||||
role=role,
|
||||
identity_source=IDENTITY_LOCAL_DEV,
|
||||
authenticated=True,
|
||||
warnings=(
|
||||
"local-dev identity is asserted, not verified; never use "
|
||||
"outside loopback.",
|
||||
),
|
||||
)
|
||||
|
||||
if mode == IDENTITY_ACCESS_PROXY:
|
||||
subject = (lowered.get(ACCESS_SUBJECT_HEADER) or "").strip()
|
||||
if not subject:
|
||||
# Proxy mode with no proxy header means the request did not
|
||||
# traverse the proxy. Fail closed rather than trust it.
|
||||
return ANONYMOUS
|
||||
role = _role_map(source_env).get(subject, VIEWER)
|
||||
return Principal(
|
||||
subject=subject,
|
||||
role=role,
|
||||
identity_source=IDENTITY_ACCESS_PROXY,
|
||||
authenticated=True,
|
||||
)
|
||||
|
||||
return ANONYMOUS
|
||||
|
||||
|
||||
def probe_auth_required(env: dict[str, str] | None = None) -> bool:
|
||||
"""Whether non-public probes must be authenticated. Default False.
|
||||
|
||||
#633 requires the console to *fail closed on missing auth for non-public
|
||||
health probes if configured*. The default stays off so the MVP ``/health``
|
||||
contract is unchanged; an operator opts in explicitly.
|
||||
"""
|
||||
source = env if env is not None else os.environ
|
||||
return (source.get(REQUIRE_PROBE_AUTH_ENV) or "").strip().lower() in {
|
||||
"1",
|
||||
"true",
|
||||
"yes",
|
||||
}
|
||||
|
||||
|
||||
# --- Authorization ----------------------------------------------------------
|
||||
|
||||
DENY_UNKNOWN_ACTION = "unknown_action"
|
||||
DENY_UNAUTHENTICATED = "unauthenticated"
|
||||
DENY_INSUFFICIENT_ROLE = "insufficient_role"
|
||||
DENY_UNKNOWN_ROLE = "unknown_role"
|
||||
DENY_PHASE_NOT_ACTIVE = "phase_not_active"
|
||||
ALLOW_PREVIEW = "allowed_preview_only"
|
||||
|
||||
# Phase 1 is the only active console phase. Phase 2 opens gated writes and is
|
||||
# gated on this model landing; nothing here enables it.
|
||||
ACTIVE_PHASE = 1
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class AuthorizationDecision:
|
||||
"""Result of an authorization check. Never an execution grant."""
|
||||
|
||||
allowed: bool
|
||||
reason_code: str
|
||||
detail: str
|
||||
action_id: str
|
||||
principal: Principal
|
||||
required_role: str | None = None
|
||||
action_class: str | None = None
|
||||
requires_confirmation: bool = False
|
||||
dual_control: bool = False
|
||||
break_glass: bool = False
|
||||
execution_enabled: bool = False
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
return {
|
||||
"allowed": self.allowed,
|
||||
"reason_code": self.reason_code,
|
||||
"detail": self.detail,
|
||||
"action_id": self.action_id,
|
||||
"principal": self.principal.to_dict(),
|
||||
"required_role": self.required_role,
|
||||
"action_class": self.action_class,
|
||||
"requires_confirmation": self.requires_confirmation,
|
||||
"dual_control": self.dual_control,
|
||||
"break_glass": self.break_glass,
|
||||
"execution_enabled": self.execution_enabled,
|
||||
"active_phase": ACTIVE_PHASE,
|
||||
}
|
||||
|
||||
|
||||
def authorize(
|
||||
action_id: str,
|
||||
principal: Principal | None = None,
|
||||
*,
|
||||
for_execution: bool = False,
|
||||
) -> AuthorizationDecision:
|
||||
"""Decide whether *principal* may invoke *action_id*. Deny by default.
|
||||
|
||||
``for_execution`` distinguishes a read-only preview from a real invocation.
|
||||
Even an allowed decision reports ``execution_enabled=False`` while the
|
||||
console is in Phase 1, so no caller can read an allow as permission to
|
||||
mutate.
|
||||
"""
|
||||
who = principal if principal is not None else ANONYMOUS
|
||||
action = get_action(action_id)
|
||||
|
||||
if action is None:
|
||||
return AuthorizationDecision(
|
||||
allowed=False,
|
||||
reason_code=DENY_UNKNOWN_ACTION,
|
||||
detail=f"No console action registered as {action_id!r}.",
|
||||
action_id=action_id,
|
||||
principal=who,
|
||||
)
|
||||
|
||||
base: dict[str, Any] = {
|
||||
"action_id": action_id,
|
||||
"principal": who,
|
||||
"required_role": action.minimum_role,
|
||||
"action_class": action.action_class,
|
||||
"requires_confirmation": action.requires_confirmation,
|
||||
"dual_control": action.dual_control,
|
||||
"break_glass": action.break_glass,
|
||||
"execution_enabled": False,
|
||||
}
|
||||
|
||||
if not who.authenticated:
|
||||
return AuthorizationDecision(
|
||||
allowed=False,
|
||||
reason_code=DENY_UNAUTHENTICATED,
|
||||
detail=(
|
||||
"Write actions require an authenticated principal; this "
|
||||
"request is anonymous."
|
||||
),
|
||||
**base,
|
||||
)
|
||||
|
||||
if who.rank < 0:
|
||||
return AuthorizationDecision(
|
||||
allowed=False,
|
||||
reason_code=DENY_UNKNOWN_ROLE,
|
||||
detail=f"Role {who.role!r} is not in the console role matrix.",
|
||||
**base,
|
||||
)
|
||||
|
||||
if who.rank < _ROLE_RANK[action.minimum_role]:
|
||||
return AuthorizationDecision(
|
||||
allowed=False,
|
||||
reason_code=DENY_INSUFFICIENT_ROLE,
|
||||
detail=(
|
||||
f"Action {action_id!r} requires {action.minimum_role!r}; "
|
||||
f"principal holds {who.role!r}."
|
||||
),
|
||||
**base,
|
||||
)
|
||||
|
||||
if for_execution and action.phase > ACTIVE_PHASE:
|
||||
return AuthorizationDecision(
|
||||
allowed=False,
|
||||
reason_code=DENY_PHASE_NOT_ACTIVE,
|
||||
detail=(
|
||||
f"Action {action_id!r} belongs to phase {action.phase}; the "
|
||||
f"console is in phase {ACTIVE_PHASE}. Execution is not wired."
|
||||
),
|
||||
**base,
|
||||
)
|
||||
|
||||
return AuthorizationDecision(
|
||||
allowed=True,
|
||||
reason_code=ALLOW_PREVIEW,
|
||||
detail=(
|
||||
"Principal holds the required role. Preview only — execution "
|
||||
"remains disabled until the Phase 2 action framework ships."
|
||||
),
|
||||
**base,
|
||||
)
|
||||
|
||||
|
||||
def rbac_matrix() -> dict[str, Any]:
|
||||
"""Machine-readable RBAC matrix and privileged-action list."""
|
||||
return {
|
||||
"model_version": 1,
|
||||
"active_phase": ACTIVE_PHASE,
|
||||
"roles": [
|
||||
{
|
||||
"role": role,
|
||||
"rank": _ROLE_RANK[role],
|
||||
"description": ROLE_DESCRIPTIONS[role],
|
||||
"permitted_actions": sorted(
|
||||
a.action_id
|
||||
for a in _ACTION_SPECS
|
||||
if _ROLE_RANK[role] >= _ROLE_RANK[a.minimum_role]
|
||||
),
|
||||
}
|
||||
for role in ROLE_ORDER
|
||||
],
|
||||
"identity_sources": IDENTITY_SOURCES,
|
||||
"actions": [a.to_dict() for a in _ACTION_SPECS],
|
||||
"privileged_actions": [a.action_id for a in privileged_actions()],
|
||||
"default_decision": "deny",
|
||||
"execution_enabled": False,
|
||||
}
|
||||
@@ -0,0 +1,169 @@
|
||||
"""Secret redaction policy for every console surface (#633).
|
||||
|
||||
The MVP already redacts MCP-side mutation records through ``gitea_audit``.
|
||||
This module is the console-facing policy: one redaction pass applied to API
|
||||
payloads, rendered HTML, log lines, and audit records *before* they leave the
|
||||
server or reach persistent storage.
|
||||
|
||||
Design constraints:
|
||||
|
||||
- **Reuse, never fork.** ``gitea_audit.redact`` remains the authority for
|
||||
secret-looking dict keys, ``Authorization`` material, and raw URLs. This
|
||||
module runs that pass first and then applies console-specific patterns for
|
||||
keychain references, key/value assignments, private-key blocks, and JWTs.
|
||||
- **Never raises.** Redaction is a safety control; a malformed payload must
|
||||
degrade to a redacted placeholder rather than propagate an exception.
|
||||
- **Redact before persist.** ``webui.console_audit`` calls this module before
|
||||
writing, so an unredacted record is never durable.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
import gitea_audit
|
||||
|
||||
REDACTED = gitea_audit.REDACTED
|
||||
|
||||
# Console-specific patterns applied after the shared ``gitea_audit`` pass.
|
||||
# Each keeps the identifying key so an operator can still tell *what* was
|
||||
# removed, and replaces only the secret run itself.
|
||||
_KEYCHAIN_REF = re.compile(r"(?i)\bkeychain:[\w.\-/@]+")
|
||||
_KEYCHAIN_CMD = re.compile(
|
||||
r"(?i)\bsecurity\s+find-(?:generic|internet)-password\b[^\n]*"
|
||||
)
|
||||
_ASSIGNMENT = re.compile(
|
||||
r"(?i)\b(token|password|passwd|secret|api[_-]?key|access[_-]?key|"
|
||||
r"client[_-]?secret|private[_-]?key)\b(\s*[:=]\s*)"
|
||||
r"(\"[^\"]*\"|'[^']*'|\S+)"
|
||||
)
|
||||
_ENV_ASSIGNMENT = re.compile(
|
||||
r"(?i)\b(GITEA_(?:TOKEN|PASS|PASSWORD)[A-Z0-9_]*)(\s*=\s*)"
|
||||
r"(\"[^\"]*\"|'[^']*'|\S+)"
|
||||
)
|
||||
_PRIVATE_KEY_BLOCK = re.compile(
|
||||
r"-----BEGIN [A-Z ]*PRIVATE KEY-----.*?-----END [A-Z ]*PRIVATE KEY-----",
|
||||
re.S,
|
||||
)
|
||||
_JWT = re.compile(
|
||||
r"\beyJ[A-Za-z0-9_\-]{8,}\.[A-Za-z0-9_\-]{8,}\.[A-Za-z0-9_\-]{8,}\b"
|
||||
)
|
||||
|
||||
# Shapes that mean a payload still carries a secret. ``scan_for_secrets`` uses
|
||||
# these to assert a surface is clean.
|
||||
_DETECTORS: tuple[tuple[str, re.Pattern[str]], ...] = (
|
||||
("keychain_reference", _KEYCHAIN_REF),
|
||||
("keychain_command", _KEYCHAIN_CMD),
|
||||
("credential_assignment", _ASSIGNMENT),
|
||||
("credential_env_assignment", _ENV_ASSIGNMENT),
|
||||
("private_key_block", _PRIVATE_KEY_BLOCK),
|
||||
("json_web_token", _JWT),
|
||||
("bearer_credential", re.compile(r"(?i)\b(?:bearer|basic)\s+\S{8,}")),
|
||||
)
|
||||
|
||||
|
||||
def _mask_assignment(match: re.Match[str]) -> str:
|
||||
"""Keep the key and separator, replace the value."""
|
||||
return f"{match.group(1)}{match.group(2)}{REDACTED}"
|
||||
|
||||
|
||||
def redact_text(text: Any) -> Any:
|
||||
"""Redact secret material from a single string.
|
||||
|
||||
Non-strings are returned unchanged so this is safe to map over mixed
|
||||
payloads. Runs the shared ``gitea_audit`` pass first, then the
|
||||
console-specific patterns.
|
||||
"""
|
||||
if not isinstance(text, str) or not text:
|
||||
return text
|
||||
try:
|
||||
out = gitea_audit.redact(text)
|
||||
if not isinstance(out, str): # defensive; redact() returns str for str
|
||||
return REDACTED
|
||||
out = _PRIVATE_KEY_BLOCK.sub(f"{REDACTED}_PRIVATE_KEY", out)
|
||||
out = _ENV_ASSIGNMENT.sub(_mask_assignment, out)
|
||||
out = _ASSIGNMENT.sub(_mask_assignment, out)
|
||||
out = _KEYCHAIN_CMD.sub(f"{REDACTED}_KEYCHAIN_COMMAND", out)
|
||||
out = _KEYCHAIN_REF.sub(f"{REDACTED}_KEYCHAIN_REF", out)
|
||||
out = _JWT.sub(f"{REDACTED}_JWT", out)
|
||||
return out
|
||||
except Exception:
|
||||
# Fail closed: an unredactable string is dropped rather than emitted raw.
|
||||
return REDACTED
|
||||
|
||||
|
||||
def redact_payload(value: Any) -> Any:
|
||||
"""Recursively redact a JSON-able payload for any console surface.
|
||||
|
||||
Secret-looking dict keys are replaced wholesale by the shared
|
||||
``gitea_audit`` policy; every remaining string is run through
|
||||
:func:`redact_text`.
|
||||
"""
|
||||
try:
|
||||
shared = gitea_audit.redact(value)
|
||||
except Exception:
|
||||
return REDACTED
|
||||
return _walk(shared)
|
||||
|
||||
|
||||
def _walk(value: Any) -> Any:
|
||||
if isinstance(value, dict):
|
||||
return {k: _walk(v) for k, v in value.items()}
|
||||
if isinstance(value, (list, tuple)):
|
||||
return [_walk(v) for v in value]
|
||||
if isinstance(value, str):
|
||||
return redact_text(value)
|
||||
return value
|
||||
|
||||
|
||||
def scan_for_secrets(value: Any) -> list[str]:
|
||||
"""Return detector names that still match *value* after serialization.
|
||||
|
||||
Used to assert an outbound payload or rendered page is clean. An empty
|
||||
list means no known secret shape was found. Already-redacted hits are not
|
||||
findings.
|
||||
"""
|
||||
if isinstance(value, str):
|
||||
text = value
|
||||
else:
|
||||
try:
|
||||
text = json.dumps(value, default=str)
|
||||
except Exception:
|
||||
text = str(value)
|
||||
findings: list[str] = []
|
||||
for name, pattern in _DETECTORS:
|
||||
for match in pattern.finditer(text):
|
||||
if REDACTED in match.group(0):
|
||||
continue
|
||||
findings.append(name)
|
||||
break
|
||||
return findings
|
||||
|
||||
|
||||
def redaction_policy() -> dict[str, Any]:
|
||||
"""Machine-readable statement of the redaction rules (never secrets)."""
|
||||
return {
|
||||
"policy_version": 1,
|
||||
"applies_to": [
|
||||
"json_api_responses",
|
||||
"rendered_html",
|
||||
"server_logs",
|
||||
"audit_records",
|
||||
],
|
||||
"ordering": "shared gitea_audit pass, then console patterns",
|
||||
"redact_before_persist": True,
|
||||
"shared_rules": {
|
||||
"source": "gitea_audit.redact",
|
||||
"secret_key_hints": list(gitea_audit._SECRET_KEY_HINTS),
|
||||
"secret_value_prefixes": list(gitea_audit._SECRET_VALUE_PREFIXES),
|
||||
"urls": "credentials, secret query parameters, and real hosts redacted",
|
||||
},
|
||||
"console_rules": [
|
||||
{"name": name, "pattern": pattern.pattern}
|
||||
for name, pattern in _DETECTORS
|
||||
],
|
||||
"placeholder": REDACTED,
|
||||
"failure_mode": "fail closed — unredactable values become the placeholder",
|
||||
}
|
||||
@@ -1,952 +0,0 @@
|
||||
"""Unified session/lease/lock/worktree inventory for the web console (#636).
|
||||
|
||||
Leases (#433), worktrees (#432), and runtime (#430) each ship their own MVP
|
||||
view, each with its own shape and its own idea of what "owned" means. A
|
||||
traffic-control or recovery operator has to read all three and correlate them
|
||||
by hand, which is exactly the step that goes wrong under collision pressure.
|
||||
|
||||
This module aggregates them into one versioned, read-only snapshot so the
|
||||
console, and any worker asking "what is safe to do next", read the same
|
||||
inventory from the same authority.
|
||||
|
||||
Field authority is explicit and never blended. Every section declares where its
|
||||
rows came from:
|
||||
|
||||
* ``control_plane_db`` — the #613 substrate: sessions, leases, assignments.
|
||||
Authoritative for *exclusive ownership* (#600/#601).
|
||||
* ``filesystem`` — durable per-issue lock files (:mod:`issue_lock_store`) and
|
||||
registered git worktrees. Authoritative for *what exists on this machine*.
|
||||
* ``gitea`` — remote issue/PR state, reached only through existing loaders.
|
||||
|
||||
Safety invariants:
|
||||
|
||||
* **Read-only.** The control-plane database is opened through a ``mode=ro``
|
||||
URI. :class:`control_plane_db.ControlPlaneDB` creates directories and runs
|
||||
migrations in its constructor, which an inventory read must never do, so this
|
||||
module talks to sqlite directly rather than through that class.
|
||||
* **Fail-soft, never fail-silent.** A subsystem that cannot be read degrades to
|
||||
a section carrying ``status`` and ``reason``. It never raises, and it never
|
||||
produces an empty list that reads like "nothing is there".
|
||||
* **Never invent active ownership.** This is the invariant that matters most.
|
||||
A degraded ownership source sets ``ownership_authority_complete`` false, and
|
||||
while that flag is false no work item is reported unowned and no collision is
|
||||
asserted. Absence of evidence is reported as absence of evidence.
|
||||
* **Redaction at the boundary.** Absolute paths are collapsed against the home
|
||||
directory, URLs lose userinfo and query strings, and no credential-shaped
|
||||
value is emitted. No session token exists in these sources and none is read.
|
||||
|
||||
Phase 1 is read-only. Lease steal/release and worktree deletion are Phase 2+
|
||||
and deliberately have no representation here, not even a disabled one.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import re
|
||||
import sqlite3
|
||||
import time
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any, Callable
|
||||
from urllib.parse import urlparse
|
||||
|
||||
import control_plane_db
|
||||
import issue_lock_store
|
||||
|
||||
SCHEMA_VERSION = 1
|
||||
API_VERSION = "v1"
|
||||
|
||||
#: Sections whose absence would make an ownership claim unprovable. If any of
|
||||
#: these is not ``ok``, the snapshot refuses to describe anything as unowned.
|
||||
OWNERSHIP_SECTIONS = ("sessions", "leases", "locks")
|
||||
|
||||
SECTION_NAMES = ("sessions", "leases", "locks", "worktrees", "namespaces")
|
||||
|
||||
STATUS_OK = "ok"
|
||||
STATUS_DEGRADED = "degraded"
|
||||
STATUS_UNAVAILABLE = "unavailable"
|
||||
|
||||
AUTHORITY_CONTROL_PLANE_DB = "control_plane_db"
|
||||
AUTHORITY_FILESYSTEM = "filesystem"
|
||||
AUTHORITY_GITEA = "gitea"
|
||||
|
||||
_CREDENTIAL_KEY_RE = re.compile(
|
||||
r"(token|secret|password|passwd|api[_-]?key|authorization|bearer|credential)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_REDACTED = "[redacted]"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class InventorySection:
|
||||
"""One subsystem's contribution, with its authority and health."""
|
||||
|
||||
name: str
|
||||
authority: str
|
||||
status: str
|
||||
items: tuple[dict[str, Any], ...] = ()
|
||||
reason: str | None = None
|
||||
scan_ms: float | None = None
|
||||
|
||||
@property
|
||||
def ok(self) -> bool:
|
||||
return self.status == STATUS_OK
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
return {
|
||||
"name": self.name,
|
||||
"authority": self.authority,
|
||||
"status": self.status,
|
||||
"count": len(self.items),
|
||||
"reason": self.reason,
|
||||
"scan_ms": self.scan_ms,
|
||||
"items": [dict(item) for item in self.items],
|
||||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CollisionSignal:
|
||||
"""A detected conflict between two ownership records."""
|
||||
|
||||
kind: str
|
||||
message: str
|
||||
severity: str = "warning"
|
||||
issue_number: int | None = None
|
||||
branch: str | None = None
|
||||
worktree_path: str | None = None
|
||||
session_ids: tuple[str, ...] = ()
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
return {
|
||||
"kind": self.kind,
|
||||
"severity": self.severity,
|
||||
"message": self.message,
|
||||
"issue_number": self.issue_number,
|
||||
"branch": self.branch,
|
||||
"worktree_path": self.worktree_path,
|
||||
"session_ids": list(self.session_ids),
|
||||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class InventorySnapshot:
|
||||
"""Versioned aggregate of every inventory section."""
|
||||
|
||||
generated_at: str
|
||||
sections: tuple[InventorySection, ...]
|
||||
collisions: tuple[CollisionSignal, ...] = ()
|
||||
correlations: tuple[dict[str, Any], ...] = ()
|
||||
schema_version: int = SCHEMA_VERSION
|
||||
api_version: str = API_VERSION
|
||||
scan_ms: float | None = None
|
||||
_section_index: dict[str, InventorySection] = field(
|
||||
default_factory=dict, repr=False, compare=False
|
||||
)
|
||||
|
||||
def section(self, name: str) -> InventorySection | None:
|
||||
return self._section_index.get(name)
|
||||
|
||||
@property
|
||||
def degraded_sections(self) -> tuple[str, ...]:
|
||||
return tuple(s.name for s in self.sections if not s.ok)
|
||||
|
||||
@property
|
||||
def ownership_authority_complete(self) -> bool:
|
||||
"""True only when every ownership-bearing section read cleanly.
|
||||
|
||||
While this is false the snapshot must not describe any work item as
|
||||
unowned: a lease the reader could not load is not an absent lease.
|
||||
"""
|
||||
for name in OWNERSHIP_SECTIONS:
|
||||
section = self._section_index.get(name)
|
||||
if section is None or not section.ok:
|
||||
return False
|
||||
return True
|
||||
|
||||
@property
|
||||
def status(self) -> str:
|
||||
if all(s.ok for s in self.sections):
|
||||
return STATUS_OK
|
||||
return STATUS_DEGRADED
|
||||
|
||||
|
||||
# ── redaction ────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def redact_path(path: str | None) -> str | None:
|
||||
"""Collapse an absolute path against ``$HOME`` for browser display."""
|
||||
if not path:
|
||||
return path
|
||||
text = str(path)
|
||||
home = os.path.expanduser("~")
|
||||
if home and home != "/" and text.startswith(home):
|
||||
return "~" + text[len(home) :]
|
||||
return text
|
||||
|
||||
|
||||
def redact_url(value: str | None) -> str | None:
|
||||
"""Strip userinfo and query string from a URL."""
|
||||
if not value:
|
||||
return value
|
||||
text = str(value)
|
||||
try:
|
||||
parsed = urlparse(text)
|
||||
except ValueError:
|
||||
return _REDACTED
|
||||
if not parsed.scheme or not parsed.netloc:
|
||||
return text
|
||||
netloc = parsed.hostname or ""
|
||||
if parsed.port:
|
||||
netloc = f"{netloc}:{parsed.port}"
|
||||
rebuilt = f"{parsed.scheme}://{netloc}{parsed.path}"
|
||||
return rebuilt.rstrip("/") or rebuilt
|
||||
|
||||
|
||||
def scrub(value: Any, *, key: str | None = None) -> Any:
|
||||
"""Recursively drop credential-shaped values and redact paths/URLs.
|
||||
|
||||
Never raises: an unexpected object degrades to its ``repr`` rather than
|
||||
propagating out of a read-only view.
|
||||
"""
|
||||
if key and _CREDENTIAL_KEY_RE.search(key):
|
||||
return _REDACTED
|
||||
if isinstance(value, dict):
|
||||
return {str(k): scrub(v, key=str(k)) for k, v in value.items()}
|
||||
if isinstance(value, (list, tuple)):
|
||||
return [scrub(v, key=key) for v in value]
|
||||
if isinstance(value, str):
|
||||
if value.startswith(("http://", "https://")):
|
||||
return redact_url(value)
|
||||
if value.startswith("/") or value.startswith("~"):
|
||||
return redact_path(value)
|
||||
return value
|
||||
if isinstance(value, (int, float, bool)) or value is None:
|
||||
return value
|
||||
return repr(value)
|
||||
|
||||
|
||||
# ── control-plane database (read-only) ───────────────────────────────────────
|
||||
|
||||
|
||||
def _open_readonly(db_path: str) -> sqlite3.Connection:
|
||||
"""Open the control-plane DB without creating or migrating anything."""
|
||||
conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True, timeout=5)
|
||||
conn.row_factory = sqlite3.Row
|
||||
return conn
|
||||
|
||||
|
||||
def _table_names(conn: sqlite3.Connection) -> set[str]:
|
||||
rows = conn.execute(
|
||||
"SELECT name FROM sqlite_master WHERE type = 'table'"
|
||||
).fetchall()
|
||||
return {str(row[0]) for row in rows}
|
||||
|
||||
|
||||
def _load_cp_db_sections(
|
||||
*,
|
||||
db_path: str | None = None,
|
||||
limit: int = 200,
|
||||
) -> tuple[InventorySection, InventorySection]:
|
||||
"""Return the ``sessions`` and ``leases`` sections from the #613 DB."""
|
||||
path = (db_path or control_plane_db.default_db_path()).strip()
|
||||
|
||||
def _both_unavailable(reason: str) -> tuple[InventorySection, InventorySection]:
|
||||
return (
|
||||
InventorySection(
|
||||
name="sessions",
|
||||
authority=AUTHORITY_CONTROL_PLANE_DB,
|
||||
status=STATUS_UNAVAILABLE,
|
||||
reason=reason,
|
||||
),
|
||||
InventorySection(
|
||||
name="leases",
|
||||
authority=AUTHORITY_CONTROL_PLANE_DB,
|
||||
status=STATUS_UNAVAILABLE,
|
||||
reason=reason,
|
||||
),
|
||||
)
|
||||
|
||||
if not path:
|
||||
return _both_unavailable("control-plane database path is not configured")
|
||||
if not os.path.exists(path):
|
||||
return _both_unavailable(
|
||||
f"control-plane database not present at {redact_path(path)}; "
|
||||
"no session or lease authority available"
|
||||
)
|
||||
|
||||
started = time.perf_counter()
|
||||
try:
|
||||
conn = _open_readonly(path)
|
||||
except sqlite3.Error as exc:
|
||||
return _both_unavailable(f"control-plane database could not be opened: {exc}")
|
||||
|
||||
try:
|
||||
tables = _table_names(conn)
|
||||
if "sessions" not in tables or "leases" not in tables:
|
||||
missing = sorted({"sessions", "leases"} - tables)
|
||||
return _both_unavailable(
|
||||
"control-plane database is missing required tables: "
|
||||
+ ", ".join(missing)
|
||||
)
|
||||
|
||||
session_rows = [
|
||||
dict(row)
|
||||
for row in conn.execute(
|
||||
"SELECT session_id, role, profile, namespace, pid, started_at,"
|
||||
" last_heartbeat_at, status FROM sessions"
|
||||
" ORDER BY last_heartbeat_at DESC LIMIT ?",
|
||||
(max(1, int(limit)),),
|
||||
).fetchall()
|
||||
]
|
||||
|
||||
has_work_items = "work_items" in tables
|
||||
if has_work_items:
|
||||
lease_sql = (
|
||||
"SELECT l.lease_id, l.session_id, l.role, l.phase, l.status,"
|
||||
" l.expires_at, w.remote, w.org, w.repo, w.kind AS work_kind,"
|
||||
" w.number AS work_number, w.state AS work_state,"
|
||||
" s.pid AS session_pid, s.profile AS session_profile,"
|
||||
" s.namespace AS session_namespace, s.status AS session_status"
|
||||
" FROM leases l"
|
||||
" JOIN work_items w ON w.work_item_id = l.work_item_id"
|
||||
" LEFT JOIN sessions s ON s.session_id = l.session_id"
|
||||
" ORDER BY l.expires_at DESC LIMIT ?"
|
||||
)
|
||||
else:
|
||||
lease_sql = (
|
||||
"SELECT l.lease_id, l.session_id, l.role, l.phase, l.status,"
|
||||
" l.expires_at FROM leases l"
|
||||
" ORDER BY l.expires_at DESC LIMIT ?"
|
||||
)
|
||||
lease_rows = [
|
||||
dict(row)
|
||||
for row in conn.execute(lease_sql, (max(1, int(limit)),)).fetchall()
|
||||
]
|
||||
except sqlite3.Error as exc:
|
||||
return _both_unavailable(f"control-plane database read failed: {exc}")
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
elapsed = (time.perf_counter() - started) * 1000.0
|
||||
now = datetime.now(timezone.utc)
|
||||
|
||||
sessions = tuple(
|
||||
scrub(
|
||||
{
|
||||
"session_id": row.get("session_id"),
|
||||
"role": row.get("role"),
|
||||
"profile": row.get("profile"),
|
||||
"namespace": row.get("namespace"),
|
||||
"pid": row.get("pid"),
|
||||
"pid_alive": issue_lock_store.is_process_alive(row.get("pid")),
|
||||
"started_at": row.get("started_at"),
|
||||
"last_heartbeat_at": row.get("last_heartbeat_at"),
|
||||
"status": row.get("status"),
|
||||
}
|
||||
)
|
||||
for row in session_rows
|
||||
)
|
||||
|
||||
leases = tuple(
|
||||
scrub(
|
||||
{
|
||||
"lease_id": row.get("lease_id"),
|
||||
"session_id": row.get("session_id"),
|
||||
"role": row.get("role"),
|
||||
"phase": row.get("phase"),
|
||||
"status": row.get("status"),
|
||||
"expires_at": row.get("expires_at"),
|
||||
"expired": _is_expired(row.get("expires_at"), now=now),
|
||||
"remote": row.get("remote"),
|
||||
"org": row.get("org"),
|
||||
"repo": row.get("repo"),
|
||||
"work_kind": row.get("work_kind"),
|
||||
"work_number": row.get("work_number"),
|
||||
"work_state": row.get("work_state"),
|
||||
"session_pid": row.get("session_pid"),
|
||||
"session_profile": row.get("session_profile"),
|
||||
"session_namespace": row.get("session_namespace"),
|
||||
"session_status": row.get("session_status"),
|
||||
}
|
||||
)
|
||||
for row in lease_rows
|
||||
)
|
||||
|
||||
degraded_reason = (
|
||||
None
|
||||
if has_work_items
|
||||
else "work_items table absent; lease rows carry no work linkage"
|
||||
)
|
||||
lease_status = STATUS_OK if has_work_items else STATUS_DEGRADED
|
||||
|
||||
return (
|
||||
InventorySection(
|
||||
name="sessions",
|
||||
authority=AUTHORITY_CONTROL_PLANE_DB,
|
||||
status=STATUS_OK,
|
||||
items=sessions,
|
||||
scan_ms=round(elapsed, 3),
|
||||
),
|
||||
InventorySection(
|
||||
name="leases",
|
||||
authority=AUTHORITY_CONTROL_PLANE_DB,
|
||||
status=lease_status,
|
||||
items=leases,
|
||||
reason=degraded_reason,
|
||||
scan_ms=round(elapsed, 3),
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def _is_expired(expires_at: str | None, *, now: datetime) -> bool | None:
|
||||
if not expires_at:
|
||||
return None
|
||||
text = str(expires_at).strip().replace("Z", "+00:00")
|
||||
try:
|
||||
parsed = datetime.fromisoformat(text)
|
||||
except ValueError:
|
||||
return None
|
||||
if parsed.tzinfo is None:
|
||||
parsed = parsed.replace(tzinfo=timezone.utc)
|
||||
return parsed <= now
|
||||
|
||||
|
||||
# ── durable issue locks (filesystem) ─────────────────────────────────────────
|
||||
|
||||
|
||||
def _load_locks_section(*, lock_dir: str | None = None) -> InventorySection:
|
||||
started = time.perf_counter()
|
||||
try:
|
||||
paths = issue_lock_store.iter_lock_files(lock_dir)
|
||||
except OSError as exc:
|
||||
return InventorySection(
|
||||
name="locks",
|
||||
authority=AUTHORITY_FILESYSTEM,
|
||||
status=STATUS_UNAVAILABLE,
|
||||
reason=f"issue lock directory could not be listed: {exc}",
|
||||
)
|
||||
|
||||
items: list[dict[str, Any]] = []
|
||||
unreadable = 0
|
||||
for path in paths:
|
||||
try:
|
||||
record = issue_lock_store.read_lock_file(path)
|
||||
except (OSError, ValueError):
|
||||
unreadable += 1
|
||||
continue
|
||||
if not record:
|
||||
unreadable += 1
|
||||
continue
|
||||
try:
|
||||
freshness = issue_lock_store.assess_lock_freshness(record)
|
||||
except Exception: # noqa: BLE001 — a read-only view never raises
|
||||
freshness = {"status": "unknown", "live": False, "stale": False}
|
||||
claimant = record.get("claimant") or (
|
||||
(record.get("work_lease") or {}).get("claimant") or {}
|
||||
)
|
||||
items.append(
|
||||
scrub(
|
||||
{
|
||||
"issue_number": record.get("issue_number"),
|
||||
"branch_name": record.get("branch_name"),
|
||||
"remote": record.get("remote"),
|
||||
"org": record.get("org"),
|
||||
"repo": record.get("repo"),
|
||||
"worktree_path": record.get("worktree_path"),
|
||||
"pid": record.get("session_pid") or record.get("pid"),
|
||||
"pid_alive": issue_lock_store.is_process_alive(
|
||||
record.get("session_pid") or record.get("pid")
|
||||
),
|
||||
"claimant_username": (claimant or {}).get("username"),
|
||||
"claimant_profile": (claimant or {}).get("profile"),
|
||||
"lock_generation": record.get("lock_generation"),
|
||||
"freshness_status": freshness.get("status"),
|
||||
"live": bool(freshness.get("live")),
|
||||
"stale": bool(freshness.get("stale")),
|
||||
"freshness_reason": freshness.get("reason"),
|
||||
"lock_path": record.get("lock_file_path") or path,
|
||||
}
|
||||
)
|
||||
)
|
||||
|
||||
elapsed = (time.perf_counter() - started) * 1000.0
|
||||
reason = (
|
||||
f"{unreadable} lock file(s) were unreadable and are not represented"
|
||||
if unreadable
|
||||
else None
|
||||
)
|
||||
return InventorySection(
|
||||
name="locks",
|
||||
authority=AUTHORITY_FILESYSTEM,
|
||||
status=STATUS_DEGRADED if unreadable else STATUS_OK,
|
||||
items=tuple(items),
|
||||
reason=reason,
|
||||
scan_ms=round(elapsed, 3),
|
||||
)
|
||||
|
||||
|
||||
# ── worktrees (filesystem, via the #432 scanner) ─────────────────────────────
|
||||
|
||||
|
||||
def _load_worktrees_section(
|
||||
*, load_hygiene: Callable[[], Any] | None = None
|
||||
) -> InventorySection:
|
||||
started = time.perf_counter()
|
||||
try:
|
||||
loader = load_hygiene
|
||||
if loader is None:
|
||||
from webui.worktree_scanner import load_hygiene_snapshot
|
||||
|
||||
loader = load_hygiene_snapshot
|
||||
snapshot = loader()
|
||||
except Exception as exc: # noqa: BLE001 — fail soft, never fail the request
|
||||
return InventorySection(
|
||||
name="worktrees",
|
||||
authority=AUTHORITY_FILESYSTEM,
|
||||
status=STATUS_UNAVAILABLE,
|
||||
reason=f"worktree scan failed: {exc}",
|
||||
)
|
||||
|
||||
items = tuple(
|
||||
scrub(
|
||||
{
|
||||
"rel_path": entry.rel_path,
|
||||
"folder_name": entry.folder_name,
|
||||
"classification": entry.classification,
|
||||
"branch": entry.branch,
|
||||
"head_sha": entry.head_sha,
|
||||
"dirty_tracked": entry.dirty_tracked,
|
||||
"dirty_untracked": entry.dirty_untracked,
|
||||
"detached": entry.detached,
|
||||
"registered_worktree": entry.registered_worktree,
|
||||
"notes": entry.notes,
|
||||
}
|
||||
)
|
||||
for entry in snapshot.entries
|
||||
)
|
||||
scan_error = getattr(snapshot, "scan_error", None)
|
||||
elapsed = (time.perf_counter() - started) * 1000.0
|
||||
return InventorySection(
|
||||
name="worktrees",
|
||||
authority=AUTHORITY_FILESYSTEM,
|
||||
status=STATUS_DEGRADED if scan_error else STATUS_OK,
|
||||
items=items,
|
||||
reason=scan_error,
|
||||
scan_ms=round(elapsed, 3),
|
||||
)
|
||||
|
||||
|
||||
# ── namespaces / capability summary ──────────────────────────────────────────
|
||||
|
||||
|
||||
def _load_namespaces_section() -> InventorySection:
|
||||
started = time.perf_counter()
|
||||
try:
|
||||
from gitea_auth import get_profile
|
||||
|
||||
profile = get_profile() or {}
|
||||
except Exception as exc: # noqa: BLE001
|
||||
return InventorySection(
|
||||
name="namespaces",
|
||||
authority=AUTHORITY_FILESYSTEM,
|
||||
status=STATUS_UNAVAILABLE,
|
||||
reason=f"active profile could not be resolved: {exc}",
|
||||
)
|
||||
|
||||
allowed = list(profile.get("allowed_operations") or [])
|
||||
forbidden = list(profile.get("forbidden_operations") or [])
|
||||
profile_name = str(profile.get("profile_name") or "")
|
||||
|
||||
namespace = None
|
||||
try:
|
||||
import role_namespace_gate
|
||||
|
||||
namespace = role_namespace_gate.infer_mcp_namespace(profile_name)
|
||||
except Exception: # noqa: BLE001 — namespace inference is advisory
|
||||
namespace = None
|
||||
|
||||
item = scrub(
|
||||
{
|
||||
"profile_name": profile_name,
|
||||
"role": profile.get("role"),
|
||||
"mcp_namespace": namespace,
|
||||
"allowed_operations": sorted(allowed),
|
||||
"forbidden_operations": sorted(forbidden),
|
||||
"capability_summary": {
|
||||
"can_author": "gitea.pr.create" in allowed,
|
||||
"can_review": "gitea.pr.approve" in allowed,
|
||||
"can_merge": "gitea.pr.merge" in allowed,
|
||||
"can_close_pr": "gitea.pr.close" in allowed,
|
||||
},
|
||||
"active": True,
|
||||
}
|
||||
)
|
||||
elapsed = (time.perf_counter() - started) * 1000.0
|
||||
return InventorySection(
|
||||
name="namespaces",
|
||||
authority=AUTHORITY_FILESYSTEM,
|
||||
status=STATUS_OK,
|
||||
items=(item,),
|
||||
reason=(
|
||||
"only the profile serving this web process is observable; other "
|
||||
"namespaces are not enumerable from here"
|
||||
),
|
||||
scan_ms=round(elapsed, 3),
|
||||
)
|
||||
|
||||
|
||||
# ── correlation and collision detection ──────────────────────────────────────
|
||||
|
||||
|
||||
def _issue_from_branch(branch: str | None) -> int | None:
|
||||
match = re.search(r"issue-(\d+)", str(branch or ""), re.IGNORECASE)
|
||||
return int(match.group(1)) if match else None
|
||||
|
||||
|
||||
def correlate(
|
||||
*,
|
||||
leases: InventorySection,
|
||||
locks: InventorySection,
|
||||
worktrees: InventorySection,
|
||||
sessions: InventorySection,
|
||||
) -> tuple[tuple[dict[str, Any], ...], tuple[CollisionSignal, ...]]:
|
||||
"""Join lease owner ↔ lock ↔ worktree ↔ namespace where evidence allows.
|
||||
|
||||
Correlation rows are emitted from whatever sections did load. Collision
|
||||
signals are only emitted from sections that are ``ok``: a conflict inferred
|
||||
from a partially-read source would be a false accusation.
|
||||
"""
|
||||
correlations: list[dict[str, Any]] = []
|
||||
collisions: list[CollisionSignal] = []
|
||||
|
||||
worktree_by_branch: dict[str, dict[str, Any]] = {}
|
||||
for entry in worktrees.items:
|
||||
branch = (entry.get("branch") or "").strip()
|
||||
if branch:
|
||||
worktree_by_branch.setdefault(branch, entry)
|
||||
|
||||
session_by_id = {
|
||||
str(s.get("session_id")): s for s in sessions.items if s.get("session_id")
|
||||
}
|
||||
|
||||
# Lock-centred rows: a durable lock names an issue, a branch, and a worktree.
|
||||
for lock in locks.items:
|
||||
branch = (lock.get("branch_name") or "").strip()
|
||||
worktree = worktree_by_branch.get(branch)
|
||||
matching_leases = [
|
||||
lease
|
||||
for lease in leases.items
|
||||
if lease.get("work_kind") == "issue"
|
||||
and lease.get("work_number") == lock.get("issue_number")
|
||||
]
|
||||
correlations.append(
|
||||
{
|
||||
"issue_number": lock.get("issue_number"),
|
||||
"branch": branch or None,
|
||||
"lock_live": bool(lock.get("live")),
|
||||
"lock_claimant": lock.get("claimant_profile"),
|
||||
"lock_pid": lock.get("pid"),
|
||||
"lock_pid_alive": lock.get("pid_alive"),
|
||||
"worktree_rel_path": (worktree or {}).get("rel_path"),
|
||||
"worktree_classification": (worktree or {}).get("classification"),
|
||||
"worktree_registered": (worktree or {}).get("registered_worktree"),
|
||||
"lease_ids": [
|
||||
lease.get("lease_id")
|
||||
for lease in matching_leases
|
||||
if lease.get("lease_id")
|
||||
],
|
||||
"lease_sessions": [
|
||||
lease.get("session_id")
|
||||
for lease in matching_leases
|
||||
if lease.get("session_id")
|
||||
],
|
||||
}
|
||||
)
|
||||
|
||||
if locks.ok and worktrees.ok:
|
||||
# A claim whose lease window is still open but has no registered
|
||||
# worktree is an anomaly regardless of whether its pid is alive; a
|
||||
# fully time-expired lease is on its way out and is not flagged.
|
||||
if (
|
||||
lock.get("freshness_status") != "expired"
|
||||
and branch
|
||||
and worktree is None
|
||||
):
|
||||
collisions.append(
|
||||
CollisionSignal(
|
||||
kind="lock-without-worktree",
|
||||
severity="warning",
|
||||
issue_number=lock.get("issue_number"),
|
||||
branch=branch,
|
||||
worktree_path=lock.get("worktree_path"),
|
||||
message=(
|
||||
f"Live lock on issue #{lock.get('issue_number')} names "
|
||||
f"branch {branch!r} but no registered worktree carries "
|
||||
"that branch (#404)"
|
||||
),
|
||||
)
|
||||
)
|
||||
|
||||
if locks.ok:
|
||||
# A lock whose recorded pid is gone is held by nobody: a clean #753
|
||||
# dead-session recovery candidate. Subclassify by the lease window,
|
||||
# because the two cases need different operator urgency. When the
|
||||
# window is still open the lock would read as live to a naive
|
||||
# timestamp check even though the owner is dead — the more dangerous
|
||||
# case — so it is flagged distinctly from a fully time-expired lease.
|
||||
if lock.get("pid_alive") is False and lock.get("stale"):
|
||||
if lock.get("freshness_status") == "expired":
|
||||
collisions.append(
|
||||
CollisionSignal(
|
||||
kind="stale-lock-dead-owner",
|
||||
severity="warning",
|
||||
issue_number=lock.get("issue_number"),
|
||||
branch=branch or None,
|
||||
message=(
|
||||
f"Lock on issue #{lock.get('issue_number')} is stale "
|
||||
f"and its recorded pid {lock.get('pid')} is not running "
|
||||
"(#753 dead-session recovery candidate)"
|
||||
),
|
||||
)
|
||||
)
|
||||
else:
|
||||
collisions.append(
|
||||
CollisionSignal(
|
||||
kind="live-lock-dead-owner",
|
||||
severity="warning",
|
||||
issue_number=lock.get("issue_number"),
|
||||
branch=branch or None,
|
||||
message=(
|
||||
f"Lock on issue #{lock.get('issue_number')} has an "
|
||||
"unexpired lease but its recorded pid "
|
||||
f"{lock.get('pid')} is not running; it would read as "
|
||||
"live to a timestamp check (#753 dead-session recovery "
|
||||
"candidate)"
|
||||
),
|
||||
)
|
||||
)
|
||||
# The #635 trap: the lease has expired but the recorded pid is a
|
||||
# still-running daemon, so neither dead-pid reclaim nor exact-owner
|
||||
# renewal applies. This is the collision an operator must see.
|
||||
elif (
|
||||
lock.get("freshness_status") == "expired"
|
||||
and lock.get("pid_alive") is True
|
||||
):
|
||||
collisions.append(
|
||||
CollisionSignal(
|
||||
kind="expired-lock-live-owner",
|
||||
severity="error",
|
||||
issue_number=lock.get("issue_number"),
|
||||
branch=branch or None,
|
||||
message=(
|
||||
f"Lock on issue #{lock.get('issue_number')} has an expired "
|
||||
f"lease but its recorded pid {lock.get('pid')} is still "
|
||||
"running (daemon-pid deadlock; needs an operator decision, "
|
||||
"#635/#760)"
|
||||
),
|
||||
)
|
||||
)
|
||||
|
||||
# Two live locks on one branch, or two active leases on one work item.
|
||||
if locks.ok:
|
||||
by_branch: dict[str, list[dict[str, Any]]] = {}
|
||||
for lock in locks.items:
|
||||
if not lock.get("live"):
|
||||
continue
|
||||
branch = (lock.get("branch_name") or "").strip()
|
||||
if branch:
|
||||
by_branch.setdefault(branch, []).append(lock)
|
||||
for branch, entries in sorted(by_branch.items()):
|
||||
if len(entries) > 1:
|
||||
collisions.append(
|
||||
CollisionSignal(
|
||||
kind="duplicate-live-lock",
|
||||
severity="error",
|
||||
branch=branch,
|
||||
message=(
|
||||
f"{len(entries)} live locks name branch {branch!r}: "
|
||||
"issues "
|
||||
+ ", ".join(
|
||||
f"#{e.get('issue_number')}" for e in entries
|
||||
)
|
||||
),
|
||||
)
|
||||
)
|
||||
|
||||
if leases.ok:
|
||||
by_work: dict[tuple[str, int], list[dict[str, Any]]] = {}
|
||||
for lease in leases.items:
|
||||
if str(lease.get("status") or "").lower() != "active":
|
||||
continue
|
||||
kind = str(lease.get("work_kind") or "").strip().lower()
|
||||
number = lease.get("work_number")
|
||||
if not kind or number is None:
|
||||
continue
|
||||
by_work.setdefault((kind, int(number)), []).append(lease)
|
||||
for (kind, number), entries in sorted(by_work.items()):
|
||||
sessions_held = {
|
||||
str(e.get("session_id")) for e in entries if e.get("session_id")
|
||||
}
|
||||
if len(sessions_held) > 1:
|
||||
collisions.append(
|
||||
CollisionSignal(
|
||||
kind="concurrent-active-lease",
|
||||
severity="error",
|
||||
issue_number=number if kind == "issue" else None,
|
||||
session_ids=tuple(sorted(sessions_held)),
|
||||
message=(
|
||||
f"{len(sessions_held)} sessions hold an active lease on "
|
||||
f"{kind} #{number}"
|
||||
),
|
||||
)
|
||||
)
|
||||
for entry in entries:
|
||||
if entry.get("expired") is True:
|
||||
collisions.append(
|
||||
CollisionSignal(
|
||||
kind="active-lease-past-expiry",
|
||||
severity="warning",
|
||||
issue_number=number if kind == "issue" else None,
|
||||
session_ids=(
|
||||
(str(entry.get("session_id")),)
|
||||
if entry.get("session_id")
|
||||
else ()
|
||||
),
|
||||
message=(
|
||||
f"Lease {entry.get('lease_id')} on {kind} #{number} "
|
||||
"is still marked active past its expiry"
|
||||
),
|
||||
)
|
||||
)
|
||||
|
||||
# A lease whose owning session is gone is an orphan, not free work.
|
||||
if leases.ok and sessions.ok:
|
||||
for lease in leases.items:
|
||||
if str(lease.get("status") or "").lower() != "active":
|
||||
continue
|
||||
session_id = str(lease.get("session_id") or "")
|
||||
if session_id and session_id not in session_by_id:
|
||||
collisions.append(
|
||||
CollisionSignal(
|
||||
kind="orphan-lease",
|
||||
severity="error",
|
||||
session_ids=(session_id,),
|
||||
message=(
|
||||
f"Active lease {lease.get('lease_id')} names session "
|
||||
f"{session_id}, which has no session record"
|
||||
),
|
||||
)
|
||||
)
|
||||
|
||||
return tuple(correlations), tuple(collisions)
|
||||
|
||||
|
||||
# ── snapshot assembly ────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def load_inventory_snapshot(
|
||||
*,
|
||||
db_path: str | None = None,
|
||||
lock_dir: str | None = None,
|
||||
load_hygiene: Callable[[], Any] | None = None,
|
||||
include: tuple[str, ...] | None = None,
|
||||
) -> InventorySnapshot:
|
||||
"""Build the unified inventory snapshot.
|
||||
|
||||
Every section is loaded independently and fails soft. *include* restricts
|
||||
the sections that are scanned; omitted sections are simply absent rather
|
||||
than reported as empty, so a resource-split request cannot be mistaken for
|
||||
a whole-inventory answer.
|
||||
"""
|
||||
started = time.perf_counter()
|
||||
wanted = tuple(include) if include else SECTION_NAMES
|
||||
|
||||
sections: list[InventorySection] = []
|
||||
sessions_section: InventorySection | None = None
|
||||
leases_section: InventorySection | None = None
|
||||
|
||||
if "sessions" in wanted or "leases" in wanted:
|
||||
sessions_section, leases_section = _load_cp_db_sections(db_path=db_path)
|
||||
if "sessions" in wanted:
|
||||
sections.append(sessions_section)
|
||||
if "leases" in wanted:
|
||||
sections.append(leases_section)
|
||||
|
||||
locks_section = (
|
||||
_load_locks_section(lock_dir=lock_dir)
|
||||
if "locks" in wanted
|
||||
else _empty_section("locks", AUTHORITY_FILESYSTEM)
|
||||
)
|
||||
if "locks" in wanted:
|
||||
sections.append(locks_section)
|
||||
|
||||
worktrees_section = (
|
||||
_load_worktrees_section(load_hygiene=load_hygiene)
|
||||
if "worktrees" in wanted
|
||||
else _empty_section("worktrees", AUTHORITY_FILESYSTEM)
|
||||
)
|
||||
if "worktrees" in wanted:
|
||||
sections.append(worktrees_section)
|
||||
|
||||
if "namespaces" in wanted:
|
||||
sections.append(_load_namespaces_section())
|
||||
|
||||
correlations, collisions = correlate(
|
||||
leases=leases_section or _empty_section("leases", AUTHORITY_CONTROL_PLANE_DB),
|
||||
locks=locks_section,
|
||||
worktrees=worktrees_section,
|
||||
sessions=sessions_section
|
||||
or _empty_section("sessions", AUTHORITY_CONTROL_PLANE_DB),
|
||||
)
|
||||
|
||||
elapsed = (time.perf_counter() - started) * 1000.0
|
||||
index = {section.name: section for section in sections}
|
||||
return InventorySnapshot(
|
||||
generated_at=datetime.now(timezone.utc).isoformat(),
|
||||
sections=tuple(sections),
|
||||
collisions=collisions,
|
||||
correlations=correlations,
|
||||
scan_ms=round(elapsed, 3),
|
||||
_section_index=index,
|
||||
)
|
||||
|
||||
|
||||
def _empty_section(name: str, authority: str) -> InventorySection:
|
||||
"""A section that was not requested — never a claim that it is empty."""
|
||||
return InventorySection(
|
||||
name=name,
|
||||
authority=authority,
|
||||
status=STATUS_UNAVAILABLE,
|
||||
reason="section not requested in this scan",
|
||||
)
|
||||
|
||||
|
||||
def snapshot_to_dict(snapshot: InventorySnapshot) -> dict[str, Any]:
|
||||
"""Serialize the snapshot for the versioned API."""
|
||||
return {
|
||||
"api_version": snapshot.api_version,
|
||||
"schema_version": snapshot.schema_version,
|
||||
"generated_at": snapshot.generated_at,
|
||||
"status": snapshot.status,
|
||||
"scan_ms": snapshot.scan_ms,
|
||||
"ownership_authority_complete": snapshot.ownership_authority_complete,
|
||||
"ownership_note": (
|
||||
"Every ownership source read cleanly; an item absent from leases "
|
||||
"and locks is genuinely unclaimed."
|
||||
if snapshot.ownership_authority_complete
|
||||
else "One or more ownership sources are degraded; nothing in this "
|
||||
"snapshot may be treated as unowned. Collisions are reported only "
|
||||
"from sections that read cleanly."
|
||||
),
|
||||
"degraded_sections": list(snapshot.degraded_sections),
|
||||
"field_authority": {
|
||||
"sessions": AUTHORITY_CONTROL_PLANE_DB,
|
||||
"leases": AUTHORITY_CONTROL_PLANE_DB,
|
||||
"locks": AUTHORITY_FILESYSTEM,
|
||||
"worktrees": AUTHORITY_FILESYSTEM,
|
||||
"namespaces": AUTHORITY_FILESYSTEM,
|
||||
},
|
||||
"sections": {section.name: section.to_dict() for section in snapshot.sections},
|
||||
"correlations": [dict(row) for row in snapshot.correlations],
|
||||
"collisions": [signal.to_dict() for signal in snapshot.collisions],
|
||||
}
|
||||
Reference in New Issue
Block a user