sysadminandClaude Opus 5
220361ad94
Addresses the two blockers raised in the PR #886 review (comment 16559) for
issue #663.
B1 — apply_authorized ignored restart-class authorization.
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 this branch merged it. The union preserved both, but the apply
decision consulted only the drain gate:
payload["apply_authorized"] = gate.allow
so a clean drain proof — or an authorized break-glass, which needs no proof
at all — reported apply_authorized: True for a class the least-privilege
matrix had just denied, in the same payload carrying allow_restart: False
and "role 'author' may not request full_mcp_restart". One environment
variable therefore collapsed the whole nine-class matrix for the apply
decision, including host_restart.
The apply decision is now the conjunction of both authorizations, and
apply_gate carries drain_gate_allow and restart_class_authorized so a denial
is attributable to the authorization that produced it. Break-glass keeps its
purpose — bypassing the drain proof — and never bypasses the class matrix.
No existing fail-closed behaviour is weakened: allow_restart, drain-proof
verification, fingerprint binding, and requester authorization are untouched.
B2 — docs/mcp-restart-coordinator.md described pre-#661 behaviour.
The document still called the drain proof "a separate child" and omitted
drain_proof_json and request_break_glass from the published signature, so a
safety document asserted there was no gate where a gate now exists. It now
documents both parameters, states that the gate executes inside this tool,
and records dry-run versus apply behaviour, authorization ordering, the
break-glass scope, and fail-closed conditions as implemented.
Regression coverage.
tests/test_issue_886_apply_authorization_conjunction.py exercises the MCP
tool itself, which previously had no test at all — that absence is why the
defect shipped. It pins both conjunction directions, proves a clean proof
cannot override a role, approval, unknown-class, or missing-target denial,
proves break-glass does not collapse the matrix for any worker role or
restricted class, and proves the existing scoped and unscoped paths and the
#661 denials still hold. Against the pre-fix tree 24 of these fail; against
this commit all 19 pass with 45 subtests.
tests/test_mcp_restart_governance_docs.py now binds the published signature
to inspect.signature() of the real tool and forbids the stale pre-#661
phrasing, so the drift that produced B2 cannot return unnoticed.
Verification: targeted restart/drain/governance/webui suites 194 passed,
113 subtests. Full suite 23 failed, 5230 passed, 6 skipped, 912 subtests —
the failure set is identical to the reviewed baseline at 9bc021e
(23 failed, 5201 passed), with +29 passing from the added tests and no new
or changed failure. Zero conflict markers; py_compile passes; the #882
union remains intact in both directions.
Refs #663, PR #886
Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_01V6xFqovhbArPv61j9KCGkL
382 lines
16 KiB
Python
382 lines
16 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] = []
|
|
|
|
|
|
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",
|
|
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",
|
|
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",
|
|
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()
|