Merge branch 'master' into feat/issue-641-runtime-session-view
This commit is contained in:
@@ -0,0 +1,381 @@
|
||||
"""``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()
|
||||
@@ -105,3 +105,120 @@ def test_cross_links_do_not_embed_secrets():
|
||||
text = _read(path)
|
||||
for marker in ("ghp_", "BEGIN PRIVATE KEY", "Authorization: Bearer"):
|
||||
assert marker not in text, f"{path} contains {marker!r}"
|
||||
|
||||
|
||||
# --- Coordinator doc stays in lock-step with the tool (#886 review blocker B2) --
|
||||
#
|
||||
# PR #882 moved the #661 drain-proof hard gate *into* gitea_request_mcp_restart,
|
||||
# but the coordinator document still described the proof as "a separate child"
|
||||
# and omitted both new parameters. Nothing referenced that document, so nothing
|
||||
# caught the drift. These tests bind the prose to the real signature.
|
||||
|
||||
COORDINATOR_DOC = REPO_ROOT / "docs" / "mcp-restart-coordinator.md"
|
||||
|
||||
# Affirmative claims that were accurate before #661 landed and are now false.
|
||||
# Matched against whitespace-normalized text so re-wrapping cannot hide them.
|
||||
# Deliberately not the bare phrase "a separate child": the corrected prose uses
|
||||
# it in a negation ("no longer a separate child operation"), and a guard that
|
||||
# forbids naming the old behaviour would block explaining that it changed.
|
||||
STALE_PRE_661_PHRASES = (
|
||||
"gated by a drain proof (a separate child)",
|
||||
"is a later child gated by a drain proof",
|
||||
"mutative apply path is explicitly out of scope",
|
||||
"apply is gated by a drain proof (a separate child)",
|
||||
)
|
||||
|
||||
|
||||
def _documented_signature_block() -> str:
|
||||
"""The fenced signature block for the tool, as published in the doc."""
|
||||
text = _read(COORDINATOR_DOC)
|
||||
marker = "gitea_request_mcp_restart("
|
||||
start = text.index(marker)
|
||||
end = text.index("```", start)
|
||||
return text[start:end]
|
||||
|
||||
|
||||
def test_documented_signature_matches_the_real_tool_signature():
|
||||
import inspect
|
||||
|
||||
import gitea_mcp_server
|
||||
|
||||
block = _documented_signature_block()
|
||||
real = inspect.signature(gitea_mcp_server.gitea_request_mcp_restart)
|
||||
for name in real.parameters:
|
||||
assert name in block, (
|
||||
f"docs/mcp-restart-coordinator.md documents no {name!r} parameter; "
|
||||
"the published signature has drifted from the tool"
|
||||
)
|
||||
|
||||
|
||||
def test_drain_proof_and_break_glass_parameters_are_documented():
|
||||
block = _documented_signature_block()
|
||||
for name in ("drain_proof_json", "request_break_glass"):
|
||||
assert name in block, f"signature block missing {name}"
|
||||
|
||||
|
||||
def test_restart_class_and_target_scoping_parameters_survive():
|
||||
block = _documented_signature_block()
|
||||
for name in ("restart_class", "target_session_id", "target_role",
|
||||
"target_connector"):
|
||||
assert name in block, f"signature block lost #663 parameter {name}"
|
||||
|
||||
|
||||
def test_gate_is_documented_as_executing_inside_this_tool():
|
||||
lower = _read(COORDINATOR_DOC).lower()
|
||||
assert "inside this tool" in lower, (
|
||||
"the coordinator doc must state that the drain-proof gate executes in "
|
||||
"gitea_request_mcp_restart, not in a later child"
|
||||
)
|
||||
assert "no longer a separate child operation" in lower
|
||||
|
||||
|
||||
def test_stale_pre_661_wording_cannot_return():
|
||||
normalized = " ".join(_read(COORDINATOR_DOC).split()).lower()
|
||||
for phrase in STALE_PRE_661_PHRASES:
|
||||
assert phrase not in normalized, (
|
||||
f"stale pre-#661 wording returned to the coordinator doc: {phrase!r}"
|
||||
)
|
||||
|
||||
|
||||
def test_dry_run_versus_apply_behavior_is_documented():
|
||||
lower = _read(COORDINATOR_DOC).lower()
|
||||
assert "dry_run=true" in lower and "dry_run=false" in lower
|
||||
assert "apply_supported" in lower and "restart_performed" in lower
|
||||
assert "never restarts anything" in lower
|
||||
|
||||
|
||||
def test_authorization_ordering_and_conjunction_are_documented():
|
||||
text = _read(COORDINATOR_DOC)
|
||||
lower = text.lower()
|
||||
assert "authorization ordering" in lower
|
||||
assert "allow_restart" in text
|
||||
assert "apply_authorized" in text
|
||||
# The conjunction itself, and the attribution fields behind it.
|
||||
assert "gate.allow and allow_restart" in text
|
||||
for field in ("drain_gate_allow", "restart_class_authorized"):
|
||||
assert field in text, f"doc omits apply_gate.{field}"
|
||||
|
||||
|
||||
def test_break_glass_scope_is_documented_as_drain_proof_only():
|
||||
text = _read(COORDINATOR_DOC)
|
||||
lower = text.lower()
|
||||
assert "break-glass" in lower
|
||||
assert "drain proof only" in lower, (
|
||||
"doc must state break-glass never bypasses the restart-class matrix"
|
||||
)
|
||||
assert "GITEA_BREAKGLASS_RESTART_AUTHORIZATION" in text
|
||||
|
||||
|
||||
def test_fail_closed_on_apply_is_documented():
|
||||
lower = _read(COORDINATOR_DOC).lower()
|
||||
assert "fail closed" in lower
|
||||
for condition in ("expired", "unclean", "tampered", "stale"):
|
||||
assert condition in lower, f"fail-closed list omits {condition!r}"
|
||||
|
||||
|
||||
def test_coordinator_doc_embeds_no_secrets():
|
||||
text = _read(COORDINATOR_DOC)
|
||||
for marker in ("ghp_", "BEGIN PRIVATE KEY", "Authorization: Bearer"):
|
||||
assert marker not in text, f"{COORDINATOR_DOC} contains {marker!r}"
|
||||
|
||||
@@ -0,0 +1,232 @@
|
||||
"""Permission, drain, routing, and audit matrix for restart classes (#663)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from datetime import datetime, timezone
|
||||
|
||||
import restart_coordinator as rc
|
||||
|
||||
NOW = datetime(2026, 7, 24, 20, 0, tzinfo=timezone.utc)
|
||||
|
||||
|
||||
def _inventory() -> dict:
|
||||
return {
|
||||
"inventory_complete": True,
|
||||
"sessions": [
|
||||
{
|
||||
"session_id": "requester",
|
||||
"role": "author",
|
||||
"profile": "prgs-author",
|
||||
"pid": os.getpid(),
|
||||
"status": "active",
|
||||
"last_heartbeat_at": NOW.isoformat(),
|
||||
},
|
||||
{
|
||||
"session_id": "reviewer",
|
||||
"role": "reviewer",
|
||||
"profile": "prgs-reviewer",
|
||||
"pid": os.getpid(),
|
||||
"status": "active",
|
||||
"last_heartbeat_at": NOW.isoformat(),
|
||||
},
|
||||
],
|
||||
"leases": [
|
||||
{
|
||||
"lease_id": "review-lease",
|
||||
"session_id": "reviewer",
|
||||
"role": "reviewer",
|
||||
"phase": "reviewing",
|
||||
"work_kind": "pr",
|
||||
"work_number": 900,
|
||||
"worktree_path": "/tmp/review-900",
|
||||
"freshness": {"freshness": "active"},
|
||||
}
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def _evaluate(
|
||||
restart_class: rc.RestartClass,
|
||||
*,
|
||||
role: str = "controller",
|
||||
permissions: tuple[str, ...] | None = None,
|
||||
approved: bool = True,
|
||||
operator: bool = True,
|
||||
**targets,
|
||||
):
|
||||
return rc.evaluate_restart_impact(
|
||||
_inventory(),
|
||||
now=NOW,
|
||||
requesting_session_id="requester",
|
||||
restart_class=restart_class,
|
||||
requester_role=role,
|
||||
requester_permissions=(
|
||||
permissions if permissions is not None
|
||||
else rc.permissions_for_role(role)
|
||||
),
|
||||
controller_approved=approved,
|
||||
operator_authorized=operator,
|
||||
**targets,
|
||||
)
|
||||
|
||||
|
||||
def test_policy_table_covers_exactly_all_nine_classes():
|
||||
assert set(rc.RESTART_CLASS_POLICIES) == set(rc.RestartClass)
|
||||
assert len(rc.RESTART_CLASS_POLICIES) == 9
|
||||
for restart_class, policy in rc.RESTART_CLASS_POLICIES.items():
|
||||
assert policy.restart_class is restart_class
|
||||
assert policy.required_permission
|
||||
assert policy.expected_blast_radius in {
|
||||
rc.BLAST_NONE, rc.BLAST_LOW, rc.BLAST_MEDIUM, rc.BLAST_HIGH
|
||||
}
|
||||
assert policy.drain_requirement
|
||||
assert policy.approval_requirement
|
||||
assert policy.audit_requirement
|
||||
assert policy.recovery_behavior
|
||||
|
||||
|
||||
def test_permission_matrix_allows_each_class_with_exact_permission():
|
||||
targets = {
|
||||
rc.RestartClass.WORKER_RESTART: {"target_session_id": "reviewer"},
|
||||
rc.RestartClass.ROLE_RUNTIME_RESTART: {"target_role": "reviewer"},
|
||||
rc.RestartClass.CONNECTOR_RESTART: {"target_connector": "github"},
|
||||
}
|
||||
for restart_class, policy in rc.RESTART_CLASS_POLICIES.items():
|
||||
report = _evaluate(
|
||||
restart_class,
|
||||
permissions=(policy.required_permission,),
|
||||
**targets.get(restart_class, {}),
|
||||
)
|
||||
assert report.permission_authorized, restart_class
|
||||
assert report.role_authorized, restart_class
|
||||
assert report.approval_satisfied, restart_class
|
||||
assert report.audit_record["restart_class"] == restart_class.value
|
||||
assert (
|
||||
report.audit_record["required_permission"]
|
||||
== policy.required_permission
|
||||
)
|
||||
|
||||
|
||||
def test_missing_or_nearby_permission_denies():
|
||||
report = _evaluate(
|
||||
rc.RestartClass.ROLE_RUNTIME_RESTART,
|
||||
permissions=("mcp.restart.worker.request",),
|
||||
target_role="reviewer",
|
||||
)
|
||||
assert report.verdict == rc.VERDICT_UNSAFE
|
||||
assert not report.allow_restart
|
||||
assert not report.permission_authorized
|
||||
assert any("missing required permission" in r for r in report.reasons)
|
||||
|
||||
|
||||
def test_unknown_restart_class_denies_fail_closed():
|
||||
report = rc.evaluate_restart_impact(
|
||||
_inventory(),
|
||||
now=NOW,
|
||||
restart_class="surprise_reboot",
|
||||
requester_role="admin",
|
||||
requester_permissions=("mcp.restart.host.request",),
|
||||
controller_approved=True,
|
||||
operator_authorized=True,
|
||||
)
|
||||
assert report.verdict == rc.VERDICT_UNSAFE
|
||||
assert not report.allow_restart
|
||||
assert report.restart_policy == {}
|
||||
assert any("unknown restart class" in r for r in report.reasons)
|
||||
|
||||
|
||||
def test_worker_roles_cannot_request_full_or_host_restart():
|
||||
for role in rc.WORKER_ROLES:
|
||||
granted = rc.permissions_for_role(role)
|
||||
assert "mcp.restart.full.request" not in granted
|
||||
assert "mcp.restart.host.request" not in granted
|
||||
report = _evaluate(
|
||||
rc.RestartClass.FULL_MCP_RESTART,
|
||||
role=role,
|
||||
permissions=granted,
|
||||
)
|
||||
assert not report.role_authorized
|
||||
assert not report.allow_restart
|
||||
|
||||
|
||||
def test_controller_approval_is_independent_of_permission():
|
||||
report = _evaluate(
|
||||
rc.RestartClass.WORKER_RESTART,
|
||||
approved=False,
|
||||
target_session_id="reviewer",
|
||||
)
|
||||
assert report.permission_authorized
|
||||
assert not report.approval_satisfied
|
||||
assert not report.allow_restart
|
||||
|
||||
|
||||
def test_narrow_classes_do_not_inherit_full_drain_or_peer_lease_block():
|
||||
for restart_class in (
|
||||
rc.RestartClass.CLIENT_RECONNECT,
|
||||
rc.RestartClass.SESSION_RECONNECT,
|
||||
rc.RestartClass.CONFIGURATION_RELOAD,
|
||||
):
|
||||
report = _evaluate(restart_class)
|
||||
assert not report.restart_policy["full_drain_required"]
|
||||
assert report.counts["leases_disruptive"] == 0
|
||||
assert report.counts["sessions_live_other"] == 0
|
||||
assert report.counts["critical_sections"] == 0
|
||||
assert report.counts["mutations"] == 0
|
||||
assert report.allow_restart, (restart_class, report.reasons)
|
||||
|
||||
|
||||
def test_client_reconnect_does_not_wait_for_unrelated_terminal_lock():
|
||||
inventory = _inventory()
|
||||
inventory["terminal_lock"] = {"terminal_pr": 901}
|
||||
report = rc.evaluate_restart_impact(
|
||||
inventory,
|
||||
now=NOW,
|
||||
requesting_session_id="requester",
|
||||
restart_class=rc.RestartClass.CLIENT_RECONNECT,
|
||||
requester_role="author",
|
||||
requester_permissions=rc.permissions_for_role("author"),
|
||||
)
|
||||
assert report.allow_restart
|
||||
assert report.terminal_lock is None
|
||||
|
||||
|
||||
def test_scoped_restart_only_counts_named_target():
|
||||
report = _evaluate(
|
||||
rc.RestartClass.ROLE_RUNTIME_RESTART,
|
||||
target_role="author",
|
||||
)
|
||||
assert report.counts["leases_disruptive"] == 0
|
||||
assert report.affected_prs == []
|
||||
assert report.allow_restart
|
||||
|
||||
reviewer = _evaluate(
|
||||
rc.RestartClass.ROLE_RUNTIME_RESTART,
|
||||
target_role="reviewer",
|
||||
)
|
||||
assert reviewer.counts["leases_disruptive"] == 1
|
||||
assert reviewer.affected_prs == [900]
|
||||
assert not reviewer.allow_restart
|
||||
|
||||
|
||||
def test_missing_scoped_target_denies_instead_of_widening():
|
||||
for restart_class in (
|
||||
rc.RestartClass.WORKER_RESTART,
|
||||
rc.RestartClass.ROLE_RUNTIME_RESTART,
|
||||
rc.RestartClass.CONNECTOR_RESTART,
|
||||
):
|
||||
report = _evaluate(restart_class)
|
||||
assert not report.allow_restart
|
||||
assert any("target required" in r for r in report.reasons)
|
||||
|
||||
|
||||
def test_only_full_and_host_classes_require_full_drain():
|
||||
requiring_full = {
|
||||
restart_class
|
||||
for restart_class, policy in rc.RESTART_CLASS_POLICIES.items()
|
||||
if policy.full_drain_required
|
||||
}
|
||||
assert requiring_full == {
|
||||
rc.RestartClass.FULL_MCP_RESTART,
|
||||
rc.RestartClass.HOST_RESTART,
|
||||
}
|
||||
@@ -444,6 +444,12 @@ class TestAuditEmission(unittest.TestCase):
|
||||
)
|
||||
self.assertEqual(record["target"]["namespace"], NAMESPACE)
|
||||
self.assertEqual(record["target"]["mode"], "restart")
|
||||
self.assertEqual(
|
||||
record["target"]["restart_class"], "role_runtime_restart"
|
||||
)
|
||||
self.assertEqual(
|
||||
record["metadata"]["restart_class"], "role_runtime_restart"
|
||||
)
|
||||
self.assertEqual(record["result"], console_audit.RESULT_ALLOWED)
|
||||
self.assertEqual(record["actor"]["subject"], "[email protected]")
|
||||
self.assertFalse(record["metadata"]["process_kill_executed"])
|
||||
|
||||
Reference in New Issue
Block a user