sysadminandGrok 4.5
461e1dac78
Add recovery_playbook.py with the narrow-to-broad recovery ladder, symptom routing, attempt-log helpers, and escalation metrics. Wire the attempt-log gate into restart_coordinator so rolling/full/host restarts require prior insufficient narrower attempts (or break-glass). Document the ladder and update gitea_request_mcp_restart for prior_recovery_attempts_json. Co-Authored-By: Grok 4.5 <[email protected]>
396 lines
17 KiB
Python
396 lines
17 KiB
Python
"""``apply_authorized`` requires BOTH authorizations (#886 review blocker B1).
|
|
|
|
The #663 restart-class matrix and the #661 drain-proof hard gate are independent
|
|
authorizations that first coexisted when PR #882 landed on master and PR #886
|
|
merged it into the restart-class branch. The union preserved both, but the apply
|
|
decision consulted only the drain gate::
|
|
|
|
payload["apply_authorized"] = gate.allow # pre-fix
|
|
|
|
so a clean drain proof — or an authorized break-glass, which needs no proof at
|
|
all — reported ``apply_authorized: True`` for a restart class the least-privilege
|
|
matrix had just denied, in the same payload that carried
|
|
``allow_restart: False`` and "role 'author' may not request full_mcp_restart".
|
|
|
|
These tests pin the conjunction and the properties that must survive it. They
|
|
exercise the real MCP tool, which previously had no test coverage at all — that
|
|
absence is why the defect shipped.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
import drain_proof
|
|
import gitea_mcp_server as srv
|
|
|
|
CONTROLLER_APPROVAL_ENV = "GITEA_CONTROLLER_RESTART_APPROVAL_AUTHORIZATION"
|
|
BREAK_GLASS_ENV = "GITEA_BREAKGLASS_RESTART_AUTHORIZATION"
|
|
|
|
# A quiet control plane: nothing live, so the blast radius never masks the
|
|
# authorization outcome under test.
|
|
QUIET_SESSIONS: list[dict] = []
|
|
QUIET_LEASES: list[dict] = []
|
|
|
|
# #669: broad restarts need a prior narrow-attempt log (unless break-glass).
|
|
PRIOR_NARROW_ATTEMPTS_JSON = json.dumps(
|
|
[
|
|
{
|
|
"action": "client_reconnect",
|
|
"outcome": "insufficient",
|
|
"reason": "still flapping after reconnect",
|
|
}
|
|
]
|
|
)
|
|
|
|
|
|
class _FakeDB:
|
|
"""Minimal control-plane DB stand-in for the restart inventory."""
|
|
|
|
def __init__(self, sessions=QUIET_SESSIONS, terminal=None):
|
|
self._sessions = list(sessions)
|
|
self._terminal = terminal
|
|
|
|
def list_sessions(self, statuses=None, limit=None):
|
|
return list(self._sessions)
|
|
|
|
def get_active_terminal_lock(self, remote=None, org=None, repo=None):
|
|
return self._terminal
|
|
|
|
|
|
def _profile(role: str) -> dict:
|
|
return {"profile_name": f"prgs-{role}", "role_kind": role, "role": role}
|
|
|
|
|
|
class _RestartToolHarness(unittest.TestCase):
|
|
"""Drives the real ``gitea_request_mcp_restart`` with a stubbed inventory."""
|
|
|
|
def _call(self, *, role: str, env: dict | None = None, **kwargs) -> dict:
|
|
environ = {k: v for k, v in os.environ.items()
|
|
if k not in (CONTROLLER_APPROVAL_ENV, BREAK_GLASS_ENV)}
|
|
environ.update(env or {})
|
|
with patch.object(srv, "_profile_operation_gate", return_value=None), \
|
|
patch.object(srv, "_resolve",
|
|
return_value=("gitea.prgs.cc",
|
|
"Scaled-Tech-Consulting",
|
|
"Gitea-Tools")), \
|
|
patch.object(srv, "get_profile", return_value=_profile(role)), \
|
|
patch.object(srv, "_control_plane_db_or_error",
|
|
return_value=(_FakeDB(), [])), \
|
|
patch.object(srv.lease_lifecycle, "list_active_leases",
|
|
return_value={"leases": list(QUIET_LEASES)}), \
|
|
patch.dict(os.environ, environ, clear=True):
|
|
return srv.gitea_request_mcp_restart(
|
|
remote="prgs",
|
|
org="Scaled-Tech-Consulting",
|
|
repo="Gitea-Tools",
|
|
session_id="probe-session",
|
|
**kwargs,
|
|
)
|
|
|
|
def _clean_proof_for(self, preview: dict) -> str:
|
|
"""Mint a genuinely clean, signature-valid proof bound to *preview*.
|
|
|
|
Built from the tool's own dry-run report, so the fingerprint matches and
|
|
the proof is rejected for authorization reasons only — never because it
|
|
was stale or forged.
|
|
"""
|
|
proof = drain_proof.build_drain_proof(
|
|
impact_report=preview,
|
|
drain_state={
|
|
"assignments_stopped": True,
|
|
"checkpoints_complete": True,
|
|
"handoffs_verified": True,
|
|
"leases_handled": True,
|
|
"acks": {},
|
|
},
|
|
requesting_session_id="probe-session",
|
|
)
|
|
self.assertTrue(proof.clean, "harness must mint a clean proof")
|
|
return json.dumps(proof.as_dict())
|
|
|
|
|
|
class TestConjunction(_RestartToolHarness):
|
|
"""AC1/AC2 — the two authorizations are ANDed, in both directions."""
|
|
|
|
def test_gate_allow_with_class_denied_yields_apply_authorized_false(self):
|
|
# An author may not request full_mcp_restart (CONTROL_ROLES only).
|
|
preview = self._call(role="author", restart_class="full_mcp_restart")
|
|
self.assertFalse(preview["allow_restart"])
|
|
|
|
result = self._call(
|
|
role="author",
|
|
restart_class="full_mcp_restart",
|
|
dry_run=False,
|
|
drain_proof_json=self._clean_proof_for(preview),
|
|
)
|
|
|
|
self.assertTrue(result["apply_gate"]["drain_gate_allow"],
|
|
"drain gate itself should have allowed this proof")
|
|
self.assertFalse(result["apply_gate"]["restart_class_authorized"])
|
|
self.assertFalse(result["apply_authorized"],
|
|
"a clean proof must not authorize a denied class")
|
|
self.assertFalse(result["allow_restart"])
|
|
|
|
def test_gate_allow_with_class_allowed_can_yield_apply_authorized_true(self):
|
|
preview = self._call(
|
|
role="operator",
|
|
restart_class="full_mcp_restart",
|
|
prior_recovery_attempts_json=PRIOR_NARROW_ATTEMPTS_JSON,
|
|
env={CONTROLLER_APPROVAL_ENV: "operator-approved"},
|
|
)
|
|
self.assertTrue(preview["allow_restart"],
|
|
"operator + controller approval must authorize the class")
|
|
|
|
result = self._call(
|
|
role="operator",
|
|
restart_class="full_mcp_restart",
|
|
prior_recovery_attempts_json=PRIOR_NARROW_ATTEMPTS_JSON,
|
|
dry_run=False,
|
|
drain_proof_json=self._clean_proof_for(preview),
|
|
env={CONTROLLER_APPROVAL_ENV: "operator-approved"},
|
|
)
|
|
|
|
self.assertTrue(result["apply_gate"]["drain_gate_allow"])
|
|
self.assertTrue(result["apply_gate"]["restart_class_authorized"])
|
|
self.assertTrue(result["apply_authorized"],
|
|
"both authorizations pass; apply must be authorized")
|
|
|
|
def test_denial_is_attributable_to_the_authorization_that_caused_it(self):
|
|
preview = self._call(role="author", restart_class="full_mcp_restart")
|
|
result = self._call(
|
|
role="author",
|
|
restart_class="full_mcp_restart",
|
|
dry_run=False,
|
|
drain_proof_json=self._clean_proof_for(preview),
|
|
)
|
|
blob = " ".join(result["apply_gate"]["reasons"]).lower()
|
|
self.assertIn("restart class authorization denied", blob)
|
|
self.assertIn("full_mcp_restart", blob)
|
|
|
|
|
|
class TestProofCannotOverrideAuthorization(_RestartToolHarness):
|
|
"""AC3 — a clean proof never overrides a class or requester-role denial."""
|
|
|
|
def test_clean_proof_cannot_override_role_denial(self):
|
|
for role in ("author", "reviewer", "merger", "reconciler"):
|
|
with self.subTest(role=role):
|
|
preview = self._call(role=role, restart_class="full_mcp_restart")
|
|
result = self._call(
|
|
role=role,
|
|
restart_class="full_mcp_restart",
|
|
dry_run=False,
|
|
drain_proof_json=self._clean_proof_for(preview),
|
|
)
|
|
self.assertFalse(result["apply_authorized"])
|
|
|
|
def test_clean_proof_cannot_override_missing_controller_approval(self):
|
|
# Correct role, but the class demands controller approval and the
|
|
# environment carries none.
|
|
preview = self._call(role="operator", restart_class="full_mcp_restart")
|
|
self.assertFalse(preview["allow_restart"])
|
|
result = self._call(
|
|
role="operator",
|
|
restart_class="full_mcp_restart",
|
|
dry_run=False,
|
|
drain_proof_json=self._clean_proof_for(preview),
|
|
)
|
|
self.assertFalse(result["apply_authorized"])
|
|
|
|
def test_clean_proof_cannot_override_unknown_class(self):
|
|
preview = self._call(role="operator", restart_class="not_a_real_class",
|
|
env={CONTROLLER_APPROVAL_ENV: "yes"})
|
|
self.assertFalse(preview["allow_restart"])
|
|
result = self._call(
|
|
role="operator",
|
|
restart_class="not_a_real_class",
|
|
dry_run=False,
|
|
drain_proof_json=self._clean_proof_for(preview),
|
|
env={CONTROLLER_APPROVAL_ENV: "yes"},
|
|
)
|
|
self.assertFalse(result["apply_authorized"])
|
|
|
|
def test_clean_proof_cannot_override_missing_scope_target(self):
|
|
# worker_restart without target_session_id fails closed on scoping.
|
|
preview = self._call(role="operator", restart_class="worker_restart",
|
|
env={CONTROLLER_APPROVAL_ENV: "yes"})
|
|
self.assertFalse(preview["allow_restart"])
|
|
result = self._call(
|
|
role="operator",
|
|
restart_class="worker_restart",
|
|
dry_run=False,
|
|
drain_proof_json=self._clean_proof_for(preview),
|
|
env={CONTROLLER_APPROVAL_ENV: "yes"},
|
|
)
|
|
self.assertFalse(result["apply_authorized"])
|
|
|
|
|
|
class TestBreakGlassDoesNotCollapseTheMatrix(_RestartToolHarness):
|
|
"""AC4 — break-glass bypasses the drain proof only, never the class matrix."""
|
|
|
|
def test_break_glass_does_not_authorize_a_denied_class(self):
|
|
result = self._call(
|
|
role="author",
|
|
restart_class="host_restart",
|
|
dry_run=False,
|
|
request_break_glass=True,
|
|
env={BREAK_GLASS_ENV: "operator-issued"},
|
|
)
|
|
self.assertTrue(result["break_glass_authorized"])
|
|
self.assertTrue(result["apply_gate"]["drain_gate_allow"],
|
|
"break-glass does satisfy the drain gate")
|
|
self.assertFalse(result["apply_gate"]["restart_class_authorized"])
|
|
self.assertFalse(result["apply_authorized"],
|
|
"break-glass must not collapse the class matrix")
|
|
|
|
def test_break_glass_across_every_worker_role_and_restricted_class(self):
|
|
for role in ("author", "reviewer", "merger", "reconciler"):
|
|
for klass in ("rolling_mcp_restart", "full_mcp_restart",
|
|
"host_restart"):
|
|
with self.subTest(role=role, restart_class=klass):
|
|
result = self._call(
|
|
role=role,
|
|
restart_class=klass,
|
|
dry_run=False,
|
|
request_break_glass=True,
|
|
env={BREAK_GLASS_ENV: "operator-issued"},
|
|
)
|
|
self.assertFalse(result["apply_authorized"])
|
|
|
|
def test_break_glass_still_works_when_the_class_is_authorized(self):
|
|
# Break-glass keeps its purpose: skipping the drain proof for a caller
|
|
# the matrix does allow.
|
|
result = self._call(
|
|
role="operator",
|
|
restart_class="full_mcp_restart",
|
|
dry_run=False,
|
|
request_break_glass=True,
|
|
env={BREAK_GLASS_ENV: "operator-issued",
|
|
CONTROLLER_APPROVAL_ENV: "operator-approved"},
|
|
)
|
|
self.assertTrue(result["apply_authorized"])
|
|
self.assertEqual(result["apply_gate"]["verdict"], "break_glass")
|
|
|
|
def test_break_glass_is_not_self_assertable(self):
|
|
# Requested but no environment authorization -> no bypass, and the
|
|
# unproven apply is denied.
|
|
result = self._call(
|
|
role="operator",
|
|
restart_class="full_mcp_restart",
|
|
dry_run=False,
|
|
request_break_glass=True,
|
|
env={CONTROLLER_APPROVAL_ENV: "operator-approved"},
|
|
)
|
|
self.assertTrue(result["break_glass_requested"])
|
|
self.assertFalse(result["break_glass_authorized"])
|
|
self.assertFalse(result["apply_authorized"])
|
|
self.assertIn("incident", result)
|
|
|
|
|
|
class TestRestrictedClassesStayDenied(_RestartToolHarness):
|
|
"""AC5 — restricted classes remain denied to unauthorized requesters."""
|
|
|
|
def test_restricted_classes_denied_for_worker_roles(self):
|
|
for role in ("author", "reviewer", "merger", "reconciler"):
|
|
for klass in ("rolling_mcp_restart", "full_mcp_restart",
|
|
"host_restart"):
|
|
with self.subTest(role=role, restart_class=klass):
|
|
preview = self._call(
|
|
role=role,
|
|
restart_class=klass,
|
|
env={CONTROLLER_APPROVAL_ENV: "yes"},
|
|
)
|
|
self.assertFalse(preview["allow_restart"])
|
|
self.assertFalse(preview["permission_authorized"])
|
|
self.assertFalse(preview["role_authorized"])
|
|
|
|
def test_host_restart_needs_controller_and_infrastructure_operator(self):
|
|
# controller approval alone is not enough for host_restart.
|
|
preview = self._call(role="controller", restart_class="host_restart",
|
|
env={CONTROLLER_APPROVAL_ENV: "yes"})
|
|
self.assertFalse(preview["approval_satisfied"])
|
|
self.assertFalse(preview["allow_restart"])
|
|
|
|
|
|
class TestExistingPathsStillWork(_RestartToolHarness):
|
|
"""AC6 — valid scoped and unscoped restart paths are unaffected."""
|
|
|
|
def test_dry_run_never_reports_apply_authorization(self):
|
|
result = self._call(role="operator", restart_class="full_mcp_restart",
|
|
env={CONTROLLER_APPROVAL_ENV: "yes"})
|
|
self.assertNotIn("apply_authorized", result)
|
|
self.assertNotIn("apply_gate", result)
|
|
self.assertFalse(result["apply_supported"])
|
|
self.assertFalse(result["restart_performed"])
|
|
|
|
def test_self_service_unscoped_classes_authorize_for_every_role(self):
|
|
for role in ("author", "reviewer", "merger", "reconciler",
|
|
"controller", "operator", "admin"):
|
|
for klass in ("client_reconnect", "session_reconnect"):
|
|
with self.subTest(role=role, restart_class=klass):
|
|
preview = self._call(role=role, restart_class=klass)
|
|
self.assertTrue(preview["allow_restart"])
|
|
|
|
def test_scoped_class_with_target_authorizes_and_applies(self):
|
|
env = {CONTROLLER_APPROVAL_ENV: "operator-approved"}
|
|
preview = self._call(role="operator", restart_class="worker_restart",
|
|
target_session_id="worker-1", env=env)
|
|
self.assertTrue(preview["allow_restart"])
|
|
|
|
result = self._call(
|
|
role="operator",
|
|
restart_class="worker_restart",
|
|
target_session_id="worker-1",
|
|
dry_run=False,
|
|
drain_proof_json=self._clean_proof_for(preview),
|
|
env=env,
|
|
)
|
|
self.assertTrue(result["apply_authorized"])
|
|
|
|
def test_apply_still_denies_without_any_proof(self):
|
|
# The #661 hard gate is untouched by the conjunction.
|
|
result = self._call(
|
|
role="operator",
|
|
restart_class="full_mcp_restart",
|
|
prior_recovery_attempts_json=PRIOR_NARROW_ATTEMPTS_JSON,
|
|
dry_run=False,
|
|
env={CONTROLLER_APPROVAL_ENV: "operator-approved"},
|
|
)
|
|
self.assertFalse(result["apply_gate"]["drain_gate_allow"])
|
|
self.assertTrue(result["apply_gate"]["restart_class_authorized"])
|
|
self.assertFalse(result["apply_authorized"])
|
|
self.assertEqual(result["incident"]["kind"], "restart_drain_gate_denied")
|
|
|
|
def test_apply_denies_on_malformed_proof(self):
|
|
result = self._call(
|
|
role="operator",
|
|
restart_class="full_mcp_restart",
|
|
dry_run=False,
|
|
drain_proof_json="{not valid json",
|
|
env={CONTROLLER_APPROVAL_ENV: "operator-approved"},
|
|
)
|
|
self.assertFalse(result["apply_authorized"])
|
|
self.assertTrue(any("invalid drain_proof_json" in reason
|
|
for reason in result["apply_gate"]["reasons"]))
|
|
|
|
def test_tool_never_restarts_on_any_path(self):
|
|
for kwargs in (
|
|
{"restart_class": "client_reconnect"},
|
|
{"restart_class": "full_mcp_restart", "dry_run": False},
|
|
{"restart_class": "host_restart", "dry_run": False,
|
|
"request_break_glass": True},
|
|
):
|
|
with self.subTest(**kwargs):
|
|
result = self._call(role="operator", env={
|
|
CONTROLLER_APPROVAL_ENV: "yes", BREAK_GLASS_ENV: "yes"},
|
|
**kwargs)
|
|
self.assertFalse(result["restart_performed"])
|
|
self.assertFalse(result["apply_supported"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|