From 22e0a41bd5dfe4f47d2417b1f020e7d17f18c2fb Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Sat, 25 Jul 2026 17:09:50 -0400 Subject: [PATCH] feat(restart): audit lifecycle events and durable incidents (#665) Add restart_audit with mcp.restart.* event schema, redacted emission via gitea_audit, correlation ids, and incident materialization for failed drain and break-glass. Wire gitea_request_mcp_restart to always audit impact previews and fail closed on privileged apply when the audit sink is enabled but write fails. Closes #665 --- docs/mcp-restart-audit.md | 93 +++++++ docs/mcp-restart-coordinator.md | 1 + gitea_mcp_server.py | 113 +++++++++ restart_audit.py | 426 ++++++++++++++++++++++++++++++++ tests/test_restart_audit.py | 342 +++++++++++++++++++++++++ 5 files changed, 975 insertions(+) create mode 100644 docs/mcp-restart-audit.md create mode 100644 restart_audit.py create mode 100644 tests/test_restart_audit.py diff --git a/docs/mcp-restart-audit.md b/docs/mcp-restart-audit.md new file mode 100644 index 0000000..183f03e --- /dev/null +++ b/docs/mcp-restart-audit.md @@ -0,0 +1,93 @@ +# MCP restart audit events and incidents (#665) + +Restarts and recovery attempts leave a forensic trail. Failed drains and +break-glass paths also raise durable Gitea incident issues so unsafe restarts +cannot be silently repeated. + +Parent umbrella: **#655**. Related: impact coordinator **#658**, drain proof +**#661**, restart classes **#663**, break-glass **#664**, post-restart reconcile +**#662**, vision **#652**, roadmap **#653**, console recovery **#642**. + +## Components + +| Piece | Where | Responsibility | +|-------|-------|----------------| +| Event schema + emission | `restart_audit.py` | `mcp.restart.*` vocabulary, redacted payload builder, append-only sink via `gitea_audit` | +| Fail-closed privileged gate | `restart_audit.require_audit_or_deny` | When `GITEA_AUDIT_LOG` is set and the write fails, privileged apply is denied | +| Incident descriptors | `restart_audit.build_incident_descriptor` | Durable follow-up issues (failed drain, break-glass, reconcile unresolved, unguarded) | +| Materializer | `restart_audit.materialize_incident` | Injected `create_issue_fn` (network kept out of pure tests) | +| Wiring | `gitea_request_mcp_restart` | Correlation id, impact-preview audit, apply-gate / break-glass audit + incident creation | + +## Event vocabulary + +| Event type | When | +|------------|------| +| `mcp.restart.impact_preview` | Every `gitea_request_mcp_restart` evaluation | +| `mcp.restart.drain_enter` | Drain window starts (schema reserved; emit from drain path) | +| `mcp.restart.drain_exit` | Drain window ends | +| `mcp.restart.drain_proof` | Drain-proof verification result | +| `mcp.restart.apply_gate` | Apply hard gate (`dry_run=False`) | +| `mcp.restart.break_glass` | Authorized break-glass bypass | +| `mcp.restart.post_restart_reconcile` | Post-restart reconcile outcome | +| `mcp.restart.narrower_recovery` | Narrower recovery attempt recorded | +| `mcp.restart.unguarded_detected` | Unguarded restart path detected | + +All free text is redacted before sink write or issue body assembly. Emission +never raises; callers decide fail-closed policy. + +## Correlation + +Each restart lifecycle mints a short `correlation_id` (`rst-` + 16 hex) shared +across impact preview → apply gate → incident descriptors so operators can join +the trail. + +## Privileged deny-on-audit-fail + +Rollout policy (issue #665): + +1. Configure `GITEA_AUDIT_LOG` so writes land. +2. Only then enforce deny when a privileged restart path cannot audit. + +When audit is **not** configured, privileged apply still proceeds (no false +denials during rollout). When audit **is** configured and the write fails, +`apply_authorized` is cleared. + +## Incidents + +| Kind | Trigger | +|------|---------| +| `restart_failed_drain` | Apply denied by drain hard gate / failed proof | +| `restart_break_glass` | Any authorized break-glass apply | +| `restart_reconcile_unresolved` | Post-restart reconcile left work unresolved | +| `restart_unguarded_detected` | Unguarded restart attempt detected | + +Break-glass **always** creates an incident descriptor (and a Gitea issue when +the create path is available). Failed drain does the same. Incident bodies +include correlation id, session, class, scope, proof id, and redacted reasons. + +Default labels: `mcp-health`, `safety`, `observability`, `status:ready`, +`type:bug`, `workflow-hardening`. + +## Tool payload surface + +`gitea_request_mcp_restart` returns: + +* `correlation_id` — lifecycle join key +* `restart_audit.impact_preview_written` — sink success for the preview event +* `restart_audit.apply_gate_written` — sink success for apply/break-glass (apply only) +* `restart_audit.incident_result` — materialization outcome when an incident was required +* `incident` — durable descriptor (when gate requires follow-up) + +## Security + +* No secrets in audit payloads or issue bodies. +* This module never restarts a process. +* Drain proof verification remains #661; audit only records the decision. +* Incident creation failures are recorded in `incident_result.reasons` and never + crash the restart evaluation path (audit write failure still fails closed for + privileged apply when the sink is enabled). + +## Tests + +See `tests/test_restart_audit.py`: schema, redaction, emission, deny policy, +incident materialization mocks, break-glass / failed-drain selection. diff --git a/docs/mcp-restart-coordinator.md b/docs/mcp-restart-coordinator.md index 628895f..335f78b 100644 --- a/docs/mcp-restart-coordinator.md +++ b/docs/mcp-restart-coordinator.md @@ -33,6 +33,7 @@ recovery behavior for all nine classes. | `ControlPlaneDB.list_sessions` | `control_plane_db.py` | Read-only session inventory (the process-level unit a restart kills). | | `gitea_request_mcp_restart` | `gitea_mcp_server.py` | MCP tool: gathers inventory from the #613 DB, calls the coordinator, returns the report, and on `dry_run=False` runs the #661 drain-proof hard gate. Never restarts a process. | | `drain_proof.gate_apply_restart` | `drain_proof.py` | The #661 hard gate: verifies a drain proof against the current impact fingerprint, or records an authorized break-glass bypass. | +| `restart_audit` | `restart_audit.py` | #665 forensic trail: `mcp.restart.*` events via `gitea_audit`, correlation ids, durable incidents for failed drain / break-glass. See [`mcp-restart-audit.md`](./mcp-restart-audit.md). | ## Dimensions evaluated diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index aab8249..1e2498c 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -2069,6 +2069,7 @@ import lease_policy # noqa: E402 import workflow_dashboard # noqa: E402 # #605 live queue/lease dashboard import restart_coordinator # noqa: E402 # #658 MCP restart coordinator/impact import drain_proof # noqa: E402 # #661 pre-restart drain proof and hard gate +import restart_audit # noqa: E402 # #665 restart audit events + incidents import incident_bridge # noqa: E402 import sentry_observability # noqa: E402 (#606 optional Sentry observability) import sentry_incident_bridge # noqa: E402 (#607 Sentry→Gitea incident bridge) @@ -22750,6 +22751,38 @@ def gitea_request_mcp_restart( # and a durable incident is raised. Break-glass is the only bypass and its # authorization is read from the environment, never self-asserted. payload["apply_supported"] = False + # #665: correlation id threads impact preview → apply gate → incidents. + correlation_id = restart_audit.new_correlation_id() + payload["correlation_id"] = correlation_id + auth_user = None + try: + # remote-only: host override is for operator diagnostics, not required here + auth_user = (gitea_whoami(remote=remote) or {}).get("username") + except Exception: # noqa: BLE001 — identity is best-effort for audit + auth_user = None + preview_audit = restart_audit.record_restart_lifecycle( + event_type=restart_audit.EVENT_IMPACT_PREVIEW, + outcome=str(report.verdict or "unknown"), + correlation_id=correlation_id, + remote=remote, + org=o, + repo=r, + requesting_session_id=sid, + restart_class=restart_class, + profile_name=profile_name, + authenticated_username=auth_user, + reasons=list(report.reasons or []), + details={ + "dry_run": True, + "allow_restart": bool(report.allow_restart), + "inventory_complete": inventory_complete, + }, + privileged=False, + ) + payload["restart_audit"] = { + "correlation_id": correlation_id, + "impact_preview_written": preview_audit["audit_written"], + } if not dry_run: proof_obj: dict | None = None proof_parse_error: str | None = None @@ -22809,6 +22842,86 @@ def gitea_request_mcp_restart( ) if not gate.allow and gate.incident is not None: payload["incident"] = gate.incident + + # #665: audit apply-gate + materialize durable incidents for failed + # drain and break-glass. Privileged apply denies if audit is enabled + # and the sink write fails. + incident_desc = restart_audit.incident_from_apply_gate( + gate_payload={ + **gate_payload, + "incident": gate.incident, + "allow": gate.allow, + }, + break_glass=break_glass, + correlation_id=correlation_id, + requesting_session_id=sid, + restart_class=restart_class, + remote=remote, + org=o, + repo=r, + ) + if incident_desc is not None: + payload["incident"] = incident_desc + + def _create_restart_incident_issue( + *, title, body, labels, org=None, repo=None, **_kw + ): + return gitea_create_issue( + title=title, + body=body, + labels=labels, + remote=remote, + host=h, + org=org or o, + repo=repo or r, + ) + + apply_audit = restart_audit.record_restart_lifecycle( + event_type=( + restart_audit.EVENT_BREAK_GLASS + if break_glass + else restart_audit.EVENT_APPLY_GATE + ), + outcome=( + "break_glass" + if break_glass + else ("allow" if payload["apply_authorized"] else "deny") + ), + correlation_id=correlation_id, + remote=remote, + org=o, + repo=r, + requesting_session_id=sid, + restart_class=restart_class, + profile_name=profile_name, + authenticated_username=auth_user, + reasons=list(gate_payload.get("reasons") or []), + details={ + "apply_authorized": payload["apply_authorized"], + "drain_gate_allow": gate_payload.get("drain_gate_allow"), + "restart_class_authorized": restart_class_authorized, + "break_glass": break_glass, + "proof_id": gate_payload.get("proof_id"), + }, + privileged=True, + create_incident=incident_desc, + create_issue_fn=_create_restart_incident_issue + if incident_desc is not None + else None, + dry_run_incident=False, + ) + payload["restart_audit"] = { + "correlation_id": correlation_id, + "impact_preview_written": preview_audit["audit_written"], + "apply_gate_written": apply_audit["audit_written"], + "incident_result": apply_audit.get("incident_result"), + } + if apply_audit["deny_reasons"]: + payload["apply_authorized"] = False + payload["reasons"] = list(payload.get("reasons") or []) + list( + apply_audit["deny_reasons"] + ) + payload["success"] = True return payload diff --git a/restart_audit.py b/restart_audit.py new file mode 100644 index 0000000..9ed5abb --- /dev/null +++ b/restart_audit.py @@ -0,0 +1,426 @@ +"""MCP restart lifecycle audit events and incident materialization (#665). + +Restarts and recovery attempts must leave a forensic trail: impact previews, +drain enter/exit, drain-proof results, apply gate verdicts, break-glass, and +post-restart reconcile outcomes. Failed drains and break-glass must also raise +durable Gitea incident issues so they cannot be silently repeated. + +This module is the pure + sink layer for that trail: + +* **Schema** — ``mcp.restart.*`` event names and a redacted payload builder. +* **Emission** — append-only via :mod:`gitea_audit` (off when ``GITEA_AUDIT_LOG`` + is unset; privileged apply can still *require* a successful write). +* **Incidents** — descriptors for failed drain / break-glass / unguarded restart, + plus an optional materializer that creates a Gitea issue through an injected + ``create_issue_fn`` (keeps this module free of network I/O in tests). + +Design rules: + +* **No secrets.** All free text is redacted before write or issue body assembly. +* **Never raises from emission.** ``emit_restart_event`` returns False on sink + failure so callers can decide fail-closed policy for privileged restarts. +* **Does not restart.** Audit never executes a process restart. +* **Drain proof stays #661.** This module records what the gate decided; it + does not re-verify proofs. +""" + +from __future__ import annotations + +import uuid +from datetime import datetime, timezone +from typing import Any, Callable, Mapping, Sequence + +import gitea_audit + +# ── Event vocabulary (stable identifiers for operators + tests) ─────────────── + +EVENT_IMPACT_PREVIEW = "mcp.restart.impact_preview" +EVENT_DRAIN_ENTER = "mcp.restart.drain_enter" +EVENT_DRAIN_EXIT = "mcp.restart.drain_exit" +EVENT_DRAIN_PROOF = "mcp.restart.drain_proof" +EVENT_APPLY_GATE = "mcp.restart.apply_gate" +EVENT_BREAK_GLASS = "mcp.restart.break_glass" +EVENT_POST_RESTART_RECONCILE = "mcp.restart.post_restart_reconcile" +EVENT_NARROWER_RECOVERY = "mcp.restart.narrower_recovery" +EVENT_UNGUARDED_DETECTED = "mcp.restart.unguarded_detected" + +RESTART_EVENT_TYPES: frozenset[str] = frozenset( + { + EVENT_IMPACT_PREVIEW, + EVENT_DRAIN_ENTER, + EVENT_DRAIN_EXIT, + EVENT_DRAIN_PROOF, + EVENT_APPLY_GATE, + EVENT_BREAK_GLASS, + EVENT_POST_RESTART_RECONCILE, + EVENT_NARROWER_RECOVERY, + EVENT_UNGUARDED_DETECTED, + } +) + +# Incident kinds (durable Gitea issues). +INCIDENT_FAILED_DRAIN = "restart_failed_drain" +INCIDENT_BREAK_GLASS = "restart_break_glass" +INCIDENT_RECONCILE_UNRESOLVED = "restart_reconcile_unresolved" +INCIDENT_UNGUARDED = "restart_unguarded_detected" + +DEFAULT_INCIDENT_LABELS: tuple[str, ...] = ( + "mcp-health", + "safety", + "observability", + "status:ready", + "type:bug", + "workflow-hardening", +) + +CreateIssueFn = Callable[..., dict[str, Any]] + + +def _utc_now_iso() -> str: + return datetime.now(timezone.utc).isoformat() + + +def new_correlation_id() -> str: + """Mint a short correlation id shared across a restart lifecycle.""" + return f"rst-{uuid.uuid4().hex[:16]}" + + +def build_restart_event( + *, + event_type: str, + outcome: str, + correlation_id: str | None = None, + remote: str | None = None, + org: str | None = None, + repo: str | None = None, + requesting_session_id: str | None = None, + restart_class: str | None = None, + profile_name: str | None = None, + authenticated_username: str | None = None, + reasons: Sequence[str] | None = None, + details: Mapping[str, Any] | None = None, + now: str | None = None, +) -> dict[str, Any]: + """Build a redacted ``mcp.restart.*`` audit event. + + Raises ``ValueError`` on unknown event types so a typo cannot silently land + under a free-form action name. + """ + name = str(event_type or "").strip() + if name not in RESTART_EVENT_TYPES: + raise ValueError( + f"unknown restart audit event_type {name!r}; expected one of " + f"{sorted(RESTART_EVENT_TYPES)}" + ) + redacted_reasons = [ + gitea_audit.redact(str(r)) for r in (reasons or []) if str(r).strip() + ] + redacted_details = gitea_audit.redact(dict(details or {})) + if not isinstance(redacted_details, dict): + redacted_details = {"value": redacted_details} + + event = gitea_audit.build_event( + action=name, + result=str(outcome or "unknown"), + remote=remote, + repository=f"{org}/{repo}" if org and repo else None, + profile_name=profile_name, + authenticated_username=authenticated_username, + reason="; ".join(redacted_reasons) if redacted_reasons else None, + request_metadata={ + "event_family": "mcp.restart", + "correlation_id": correlation_id or new_correlation_id(), + "restart_class": restart_class, + "requesting_session_id": requesting_session_id, + "org": org, + "repo": repo, + "details": redacted_details, + "reasons": redacted_reasons, + }, + now=now or _utc_now_iso(), + operation=name, + ) + event["action_type"] = "restart_lifecycle" + event["event_type"] = name + event["correlation_id"] = (event.get("request_metadata") or {}).get( + "correlation_id" + ) + return event + + +def emit_restart_event(event: Mapping[str, Any], *, path: str | None = None) -> bool: + """Append *event* to the audit sink. Never raises. Returns write success.""" + try: + return bool(gitea_audit.write_event(dict(event), path=path)) + except Exception: + return False + + +def require_audit_or_deny( + *, + privileged: bool, + written: bool, + audit_enabled: bool | None = None, +) -> list[str]: + """Return deny reasons when a privileged restart path fails to audit. + + When audit is not configured (``GITEA_AUDIT_LOG`` unset), privileged apply + still proceeds under the rollout policy "enable audit before enforcing + deny-on-audit-fail" — but *if* audit is enabled and the write fails, + privileged apply is denied (fail closed). + """ + enabled = ( + gitea_audit.audit_enabled() if audit_enabled is None else bool(audit_enabled) + ) + if not privileged: + return [] + if not enabled: + return [] + if written: + return [] + return [ + "privileged restart path requires a successful audit write; " + "audit sink failed (fail closed, #665)" + ] + + +# ── Incident descriptors ────────────────────────────────────────────────────── + + +def build_incident_descriptor( + *, + kind: str, + reasons: Sequence[str], + correlation_id: str | None = None, + requesting_session_id: str | None = None, + restart_class: str | None = None, + remote: str | None = None, + org: str | None = None, + repo: str | None = None, + proof_id: str | None = None, + at: str | None = None, +) -> dict[str, Any]: + """Build a durable incident descriptor (no network).""" + titles = { + INCIDENT_FAILED_DRAIN: "Restart denied: drain proof failed the hard gate", + INCIDENT_BREAK_GLASS: "Break-glass MCP restart authorized", + INCIDENT_RECONCILE_UNRESOLVED: "Post-restart reconcile left unresolved work", + INCIDENT_UNGUARDED: "Unguarded MCP restart attempt detected", + } + title = titles.get(kind, f"MCP restart incident ({kind})") + redacted_reasons = [ + gitea_audit.redact(str(r)) for r in reasons if str(r).strip() + ] + return { + "kind": kind, + "title": title, + "labels": list(DEFAULT_INCIDENT_LABELS), + "reasons": redacted_reasons, + "correlation_id": correlation_id, + "requesting_session_id": requesting_session_id, + "restart_class": restart_class, + "remote": remote, + "org": org, + "repo": repo, + "proof_id": proof_id, + "at": at or _utc_now_iso(), + "source": "restart_audit#665", + } + + +def incident_body(descriptor: Mapping[str, Any]) -> str: + """Render a redacted markdown body for a Gitea incident issue.""" + reasons = descriptor.get("reasons") or [] + reason_lines = "\n".join(f"- {gitea_audit.redact(str(r))}" for r in reasons) or ( + "- (no reasons recorded)" + ) + return "\n".join( + [ + "", + f"## MCP restart incident (`{descriptor.get('kind')}`)", + "", + f"**Correlation:** `{descriptor.get('correlation_id') or 'none'}`", + f"**Session:** `{descriptor.get('requesting_session_id') or 'none'}`", + f"**Class:** `{descriptor.get('restart_class') or 'none'}`", + f"**Scope:** `{descriptor.get('remote')}/{descriptor.get('org')}/" + f"{descriptor.get('repo')}`", + f"**At:** `{descriptor.get('at')}`", + f"**Proof id:** `{descriptor.get('proof_id') or 'none'}`", + "", + "### Reasons", + reason_lines, + "", + "### Operator next steps", + "- Treat this as durable follow-up work under the restart-governance umbrella (#655).", + "- Do not invent a second restart path; use sanctioned coordinator tools only.", + "- Raw secrets must never appear in this issue (already redacted).", + "", + f"_Source: {descriptor.get('source')}_", + ] + ) + + +def materialize_incident( + descriptor: Mapping[str, Any], + *, + create_issue_fn: CreateIssueFn | None, + dry_run: bool = False, +) -> dict[str, Any]: + """Create a Gitea issue from *descriptor* when *create_issue_fn* is provided. + + Returns a result dict with ``created`` / ``issue_number`` / ``dry_run`` / + ``reasons``. Never raises. + """ + base: dict[str, Any] = { + "created": False, + "dry_run": bool(dry_run), + "issue_number": None, + "kind": descriptor.get("kind"), + "reasons": [], + "descriptor": dict(descriptor), + } + if dry_run: + base["reasons"] = ["dry-run only; no Gitea issue created"] + return base + if create_issue_fn is None: + base["reasons"] = [ + "create_issue_fn not provided; incident descriptor retained only" + ] + return base + try: + result = create_issue_fn( + title=str(descriptor.get("title") or "MCP restart incident"), + body=incident_body(descriptor), + labels=list(descriptor.get("labels") or DEFAULT_INCIDENT_LABELS), + org=descriptor.get("org"), + repo=descriptor.get("repo"), + ) + number = None + if isinstance(result, dict): + number = result.get("number") or result.get("issue_number") + if number is not None: + base["created"] = True + base["issue_number"] = int(number) + base["reasons"] = [f"created incident issue #{int(number)}"] + else: + base["reasons"] = ["create_issue_fn returned no issue number"] + except Exception as exc: # noqa: BLE001 — never break restart path here + base["reasons"] = [ + f"incident issue creation failed: {gitea_audit.redact(str(exc))}" + ] + return base + + +def record_restart_lifecycle( + *, + event_type: str, + outcome: str, + correlation_id: str, + remote: str | None = None, + org: str | None = None, + repo: str | None = None, + requesting_session_id: str | None = None, + restart_class: str | None = None, + profile_name: str | None = None, + authenticated_username: str | None = None, + reasons: Sequence[str] | None = None, + details: Mapping[str, Any] | None = None, + privileged: bool = False, + create_incident: Mapping[str, Any] | None = None, + create_issue_fn: CreateIssueFn | None = None, + dry_run_incident: bool = False, + audit_path: str | None = None, +) -> dict[str, Any]: + """Emit one restart audit event and optionally materialize an incident. + + Returns ``{event, audit_written, deny_reasons, incident_result}``. + """ + event = build_restart_event( + event_type=event_type, + outcome=outcome, + correlation_id=correlation_id, + remote=remote, + org=org, + repo=repo, + requesting_session_id=requesting_session_id, + restart_class=restart_class, + profile_name=profile_name, + authenticated_username=authenticated_username, + reasons=reasons, + details=details, + ) + written = emit_restart_event(event, path=audit_path) + deny = require_audit_or_deny(privileged=privileged, written=written) + incident_result = None + if create_incident is not None: + incident_result = materialize_incident( + create_incident, + create_issue_fn=create_issue_fn, + dry_run=dry_run_incident, + ) + return { + "event": event, + "audit_written": written, + "deny_reasons": deny, + "incident_result": incident_result, + "correlation_id": correlation_id, + } + + +def incident_from_apply_gate( + *, + gate_payload: Mapping[str, Any], + break_glass: bool, + correlation_id: str, + requesting_session_id: str | None, + restart_class: str | None, + remote: str | None, + org: str | None, + repo: str | None, +) -> dict[str, Any] | None: + """Choose an incident descriptor from an apply-gate payload, if required.""" + reasons = list(gate_payload.get("reasons") or []) + proof_id = gate_payload.get("proof_id") + if break_glass: + return build_incident_descriptor( + kind=INCIDENT_BREAK_GLASS, + reasons=reasons + or ["break-glass restart path used; durable incident required (#665)"], + correlation_id=correlation_id, + requesting_session_id=requesting_session_id, + restart_class=restart_class, + remote=remote, + org=org, + repo=repo, + proof_id=proof_id if isinstance(proof_id, str) else None, + ) + # Failed drain / deny path. + incident = gate_payload.get("incident") + if isinstance(incident, Mapping) and incident: + # Normalize gate-provided descriptor into our schema. + return build_incident_descriptor( + kind=INCIDENT_FAILED_DRAIN, + reasons=list(incident.get("reasons") or reasons), + correlation_id=correlation_id, + requesting_session_id=requesting_session_id + or incident.get("requesting_session_id"), + restart_class=restart_class, + remote=remote, + org=org, + repo=repo, + proof_id=incident.get("proof_id") or proof_id, + at=incident.get("at"), + ) + if not gate_payload.get("allow") and not gate_payload.get("drain_gate_allow", True): + return build_incident_descriptor( + kind=INCIDENT_FAILED_DRAIN, + reasons=reasons or ["restart apply denied"], + correlation_id=correlation_id, + requesting_session_id=requesting_session_id, + restart_class=restart_class, + remote=remote, + org=org, + repo=repo, + proof_id=proof_id if isinstance(proof_id, str) else None, + ) + return None diff --git a/tests/test_restart_audit.py b/tests/test_restart_audit.py new file mode 100644 index 0000000..f6a19f7 --- /dev/null +++ b/tests/test_restart_audit.py @@ -0,0 +1,342 @@ +"""Tests for MCP restart lifecycle audit events and incidents (#665).""" + +from __future__ import annotations + +import json +import os +import tempfile +import unittest +from unittest.mock import patch + +import gitea_audit +import restart_audit as ra + + +class TestEventSchema(unittest.TestCase): + def test_all_lifecycle_event_types_are_named(self): + expected = { + "mcp.restart.impact_preview", + "mcp.restart.drain_enter", + "mcp.restart.drain_exit", + "mcp.restart.drain_proof", + "mcp.restart.apply_gate", + "mcp.restart.break_glass", + "mcp.restart.post_restart_reconcile", + "mcp.restart.narrower_recovery", + "mcp.restart.unguarded_detected", + } + self.assertEqual(set(ra.RESTART_EVENT_TYPES), expected) + + def test_build_restart_event_core_fields(self): + event = ra.build_restart_event( + event_type=ra.EVENT_IMPACT_PREVIEW, + outcome="safe", + correlation_id="rst-abc123", + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + requesting_session_id="sess-1", + restart_class="full_mcp_restart", + profile_name="prgs-author", + authenticated_username="bot", + reasons=["inventory complete"], + details={"allow_restart": True}, + now="2026-07-25T12:00:00+00:00", + ) + self.assertEqual(event["event_type"], ra.EVENT_IMPACT_PREVIEW) + self.assertEqual(event["action"], ra.EVENT_IMPACT_PREVIEW) + self.assertEqual(event["action_type"], "restart_lifecycle") + self.assertEqual(event["result"], "safe") + self.assertEqual(event["correlation_id"], "rst-abc123") + self.assertEqual(event["profile_name"], "prgs-author") + self.assertEqual(event["authenticated_username"], "bot") + meta = event["request_metadata"] + self.assertEqual(meta["event_family"], "mcp.restart") + self.assertEqual(meta["correlation_id"], "rst-abc123") + self.assertEqual(meta["restart_class"], "full_mcp_restart") + self.assertEqual(meta["details"]["allow_restart"], True) + + def test_unknown_event_type_raises(self): + with self.assertRaises(ValueError) as ctx: + ra.build_restart_event( + event_type="mcp.restart.not_a_real_event", + outcome="x", + correlation_id="rst-1", + ) + self.assertIn("unknown restart audit event_type", str(ctx.exception)) + + def test_reasons_and_details_are_redacted(self): + event = ra.build_restart_event( + event_type=ra.EVENT_APPLY_GATE, + outcome="deny", + correlation_id="rst-sec", + reasons=["token secret-xyz rejected", "ok"], + details={"token": "leak-token", "status": "denied"}, + ) + self.assertNotIn("secret-xyz", event.get("reason") or "") + meta = event["request_metadata"] + self.assertEqual(meta["details"]["token"], gitea_audit.REDACTED) + self.assertEqual(meta["details"]["status"], "denied") + for reason in meta["reasons"]: + self.assertNotIn("secret-xyz", reason) + + def test_new_correlation_id_shape(self): + cid = ra.new_correlation_id() + self.assertTrue(cid.startswith("rst-")) + self.assertEqual(len(cid), len("rst-") + 16) + + +class TestEmitAndRequire(unittest.TestCase): + def test_emit_appends_json_line(self): + with tempfile.TemporaryDirectory() as d: + path = os.path.join(d, "audit.log") + event = ra.build_restart_event( + event_type=ra.EVENT_DRAIN_PROOF, + outcome="pass", + correlation_id="rst-write", + ) + self.assertTrue(ra.emit_restart_event(event, path=path)) + with open(path, encoding="utf-8") as fh: + lines = fh.read().splitlines() + self.assertEqual(len(lines), 1) + loaded = json.loads(lines[0]) + self.assertEqual(loaded["event_type"], ra.EVENT_DRAIN_PROOF) + self.assertEqual(loaded["correlation_id"], "rst-write") + + def test_emit_never_raises(self): + self.assertFalse( + ra.emit_restart_event({"action": "x"}, path="/no/such/dir/audit.log") + ) + + def test_require_audit_denies_privileged_when_write_fails_and_enabled(self): + deny = ra.require_audit_or_deny( + privileged=True, written=False, audit_enabled=True + ) + self.assertEqual(len(deny), 1) + self.assertIn("fail closed", deny[0]) + + def test_require_audit_allows_when_audit_disabled(self): + # Rollout policy: enable audit before enforcing deny-on-audit-fail. + deny = ra.require_audit_or_deny( + privileged=True, written=False, audit_enabled=False + ) + self.assertEqual(deny, []) + + def test_require_audit_noop_for_non_privileged(self): + deny = ra.require_audit_or_deny( + privileged=False, written=False, audit_enabled=True + ) + self.assertEqual(deny, []) + + def test_require_audit_allows_when_written(self): + deny = ra.require_audit_or_deny( + privileged=True, written=True, audit_enabled=True + ) + self.assertEqual(deny, []) + + +class TestIncidents(unittest.TestCase): + def test_break_glass_descriptor(self): + desc = ra.build_incident_descriptor( + kind=ra.INCIDENT_BREAK_GLASS, + reasons=["break-glass authorized"], + correlation_id="rst-bg", + requesting_session_id="s1", + restart_class="full_mcp_restart", + remote="prgs", + org="O", + repo="R", + ) + self.assertEqual(desc["kind"], ra.INCIDENT_BREAK_GLASS) + self.assertIn("Break-glass", desc["title"]) + self.assertIn("mcp-health", desc["labels"]) + self.assertEqual(desc["source"], "restart_audit#665") + + def test_incident_body_redacts_and_includes_correlation(self): + desc = ra.build_incident_descriptor( + kind=ra.INCIDENT_FAILED_DRAIN, + reasons=["token secret-xyz failed proof"], + correlation_id="rst-body", + remote="prgs", + org="O", + repo="R", + proof_id="proof-1", + ) + body = ra.incident_body(desc) + self.assertIn("rst-body", body) + self.assertIn("proof-1", body) + self.assertIn("mcp-restart-incident:v1", body) + self.assertNotIn("secret-xyz", body) + + def test_materialize_dry_run(self): + desc = ra.build_incident_descriptor( + kind=ra.INCIDENT_FAILED_DRAIN, + reasons=["denied"], + correlation_id="rst-dr", + ) + result = ra.materialize_incident(desc, create_issue_fn=lambda **k: {}, dry_run=True) + self.assertFalse(result["created"]) + self.assertTrue(result["dry_run"]) + self.assertIn("dry-run", result["reasons"][0]) + + def test_materialize_without_create_fn(self): + desc = ra.build_incident_descriptor( + kind=ra.INCIDENT_FAILED_DRAIN, + reasons=["denied"], + correlation_id="rst-nfn", + ) + result = ra.materialize_incident(desc, create_issue_fn=None) + self.assertFalse(result["created"]) + self.assertIn("create_issue_fn not provided", result["reasons"][0]) + + def test_materialize_creates_issue(self): + created = {} + + def _create(*, title, body, labels, org=None, repo=None, **_kw): + created["title"] = title + created["body"] = body + created["labels"] = labels + created["org"] = org + created["repo"] = repo + return {"number": 999} + + desc = ra.build_incident_descriptor( + kind=ra.INCIDENT_BREAK_GLASS, + reasons=["break-glass"], + correlation_id="rst-create", + org="O", + repo="R", + ) + result = ra.materialize_incident(desc, create_issue_fn=_create) + self.assertTrue(result["created"]) + self.assertEqual(result["issue_number"], 999) + self.assertIn("Break-glass", created["title"]) + self.assertIn("rst-create", created["body"]) + self.assertEqual(created["org"], "O") + + def test_materialize_never_raises_on_create_failure(self): + def _boom(**_kw): + raise RuntimeError("token secret-xyz network") + + desc = ra.build_incident_descriptor( + kind=ra.INCIDENT_FAILED_DRAIN, + reasons=["x"], + correlation_id="rst-boom", + ) + result = ra.materialize_incident(desc, create_issue_fn=_boom) + self.assertFalse(result["created"]) + self.assertIn("failed", result["reasons"][0]) + self.assertNotIn("secret-xyz", result["reasons"][0]) + + +class TestIncidentFromApplyGate(unittest.TestCase): + def test_break_glass_always_incident(self): + desc = ra.incident_from_apply_gate( + gate_payload={"allow": True, "reasons": [], "proof_id": None}, + break_glass=True, + correlation_id="rst-bg2", + requesting_session_id="s", + restart_class="full_mcp_restart", + remote="prgs", + org="O", + repo="R", + ) + self.assertIsNotNone(desc) + self.assertEqual(desc["kind"], ra.INCIDENT_BREAK_GLASS) + + def test_failed_drain_from_gate_incident(self): + desc = ra.incident_from_apply_gate( + gate_payload={ + "allow": False, + "drain_gate_allow": False, + "reasons": ["proof expired"], + "incident": { + "reasons": ["proof expired"], + "proof_id": "p1", + }, + }, + break_glass=False, + correlation_id="rst-fd", + requesting_session_id="s", + restart_class="full_mcp_restart", + remote="prgs", + org="O", + repo="R", + ) + self.assertIsNotNone(desc) + self.assertEqual(desc["kind"], ra.INCIDENT_FAILED_DRAIN) + self.assertEqual(desc["proof_id"], "p1") + + def test_allow_without_break_glass_no_incident(self): + desc = ra.incident_from_apply_gate( + gate_payload={ + "allow": True, + "drain_gate_allow": True, + "reasons": [], + }, + break_glass=False, + correlation_id="rst-ok", + requesting_session_id="s", + restart_class="full_mcp_restart", + remote="prgs", + org="O", + repo="R", + ) + self.assertIsNone(desc) + + +class TestRecordLifecycle(unittest.TestCase): + def test_record_emits_and_materializes(self): + created = [] + + def _create(**kwargs): + created.append(kwargs) + return {"number": 42} + + with tempfile.TemporaryDirectory() as d: + path = os.path.join(d, "audit.log") + with patch.dict(os.environ, {"GITEA_AUDIT_LOG": path}, clear=False): + incident = ra.build_incident_descriptor( + kind=ra.INCIDENT_BREAK_GLASS, + reasons=["bg"], + correlation_id="rst-lc", + org="O", + repo="R", + ) + out = ra.record_restart_lifecycle( + event_type=ra.EVENT_BREAK_GLASS, + outcome="break_glass", + correlation_id="rst-lc", + remote="prgs", + org="O", + repo="R", + privileged=True, + create_incident=incident, + create_issue_fn=_create, + audit_path=path, + ) + self.assertTrue(out["audit_written"]) + self.assertEqual(out["deny_reasons"], []) + self.assertTrue(out["incident_result"]["created"]) + self.assertEqual(out["incident_result"]["issue_number"], 42) + self.assertEqual(len(created), 1) + + def test_privileged_deny_when_audit_write_fails(self): + with patch.dict( + os.environ, {"GITEA_AUDIT_LOG": "/no/such/dir/a.log"}, clear=False + ): + with patch("restart_audit.emit_restart_event", return_value=False): + with patch("gitea_audit.audit_enabled", return_value=True): + out = ra.record_restart_lifecycle( + event_type=ra.EVENT_APPLY_GATE, + outcome="deny", + correlation_id="rst-deny", + privileged=True, + audit_path="/no/such/dir/a.log", + ) + self.assertFalse(out["audit_written"]) + self.assertEqual(len(out["deny_reasons"]), 1) + + +if __name__ == "__main__": + unittest.main()