From 75794609d1988af4b8660ddab9cc78085f5658bf Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 28 Jul 2026 21:45:31 -0400 Subject: [PATCH] fix(mcp): remediate break-glass restart authorization and audit findings (#664) --- gitea_mcp_server.py | 250 ++++++++++--- tests/test_issue_664_break_glass_restart.py | 379 +++++++++++++++----- 2 files changed, 484 insertions(+), 145 deletions(-) diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 901d062..10f27bf 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -23923,6 +23923,7 @@ def gitea_request_mcp_restart( BREAK_GLASS_CONFIRMATION_PHRASE = "I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION" +PRIVILEGED_BREAK_GLASS_ROLES = frozenset({"controller", "operator", "admin", "sysadmin"}) @mcp.tool() @@ -23943,15 +23944,14 @@ def gitea_break_glass_restart( Break-glass restart permits emergency recovery when graceful drain cannot complete. It requires: - 1. Privileged caller authorization (ordinary LLM author/reviewer/merger/reconciler - roles are denied fail-closed; controller/admin/sysadmin or GITEA_BREAKGLASS_RESTART_AUTHORIZATION - is required). + 1. Privileged caller authorization (explicit allowlist: controller, operator, + admin, sysadmin; ordinary roles and unknown/malformed roles fail closed). 2. Explicit non-empty reason (minimum 10 characters). 3. Exact confirmation string matching 'I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION'. 4. Mandatory impact acknowledgement (impact_ack=True). - 5. Immutable audit entry recorded. - 6. Automatic incident record created on Gitea. - 7. Mandatory post-restart reconciliation requirement (#662). + 5. Immutable append-only audit entry recorded prior to execution. + 6. Automatic incident record created on Gitea prior to execution. + 7. Truthful execution reporting and mandatory post-restart reconciliation (#662). """ read_block = _profile_operation_gate("gitea.read") if read_block: @@ -23971,24 +23971,24 @@ def gitea_break_glass_restart( (os.environ.get("GITEA_BREAKGLASS_RESTART_AUTHORIZATION") or "").strip() ) - # AC1: Ordinary LLM roles (author, reviewer, merger, reconciler) cannot invoke break-glass - # unless explicit environment break-glass authorization is configured. - if active_role in ("author", "reviewer", "merger", "reconciler") and not break_glass_env_auth: + # 1. Privileged role authorization (explicit allowlist - Finding 1 & 5) + if active_role not in PRIVILEGED_BREAK_GLASS_ROLES: return { "success": False, "performed": False, "break_glass_executed": False, "active_role": active_role, "reasons": [ - f"ordinary LLM role '{active_role}' is forbidden from break-glass restarts (#664 AC1); " - "privileged controller, operator, or GITEA_BREAKGLASS_RESTART_AUTHORIZATION required" + f"role '{active_role}' is not in privileged break-glass allowlist " + f"({sorted(PRIVILEGED_BREAK_GLASS_ROLES)}); break-glass restart requires a privileged role (#664 AC1)" ], "blocker_kind": "role_authorization", } - # AC2: Required fields enforced - clean_reason = (reason or "").strip() - if not clean_reason or len(clean_reason) < 10: + # 2. Required fields and redaction (Finding 3) + raw_reason = (reason or "").strip() + clean_reason = _redact(raw_reason) + if not raw_reason or len(raw_reason) < 10: return { "success": False, "performed": False, @@ -24022,6 +24022,19 @@ def gitea_break_glass_restart( "blocker_kind": "impact_ack_required", } + # Gate incident issue creation on gitea.issue.create permission (Finding 3, B3) + if create_incident_issue: + create_block = _profile_operation_gate("gitea.issue.create") + if create_block: + return { + "success": False, + "performed": False, + "break_glass_executed": False, + "reasons": create_block, + "permission_report": _permission_block_report("gitea.issue.create"), + "blocker_kind": "permission_denied", + } + # Evaluate impact / disrupted sessions impact_result = gitea_request_mcp_restart( remote=remote, @@ -24047,58 +24060,179 @@ def gitea_break_glass_restart( "confirmation": clean_confirmation, "restart_class": restart_class, "disrupted_sessions_count": disrupted_count, - "disrupted_sessions": [s.get("session_id") if isinstance(s, dict) else str(s) for s in disrupted_sessions], + "disrupted_sessions": [ + s.get("session_id") if isinstance(s, dict) else str(s) + for s in disrupted_sessions + ], "dry_run": dry_run, "remote": remote, "org": o, "repo": r, + "env_auth_present": break_glass_env_auth, } - # Save immutable audit record - saved_audit = mcp_session_state.save_state( - kind="break_glass_audit", - payload=audit_payload, + # 4. Dry-run handling (Finding 4, B7) + if dry_run: + return { + "success": True, + "performed": False, + "dry_run": True, + "break_glass_executed": False, + "would_execute": True, + "actor": identity, + "role": active_role, + "restart_class": restart_class, + "reason": clean_reason, + "confirmation": clean_confirmation, + "disrupted_sessions_count": disrupted_count, + "disrupted_sessions": disrupted_sessions, + "audit_record": audit_payload, + "saved_audit": None, + "incident_issue": None, + "reconciliation_required": True, + "reconciliation_tool": "gitea_reconcile_after_restart", + "follow_up_issue_required": True, + "cross_references": ["#652", "#653", "#655", "#630", "#658", "#662", "#664"], + "reasons": ["break-glass restart dry-run evaluated successfully"], + } + + # Fail closed if create_incident_issue is False on real execution (Finding 2, B5) + if not create_incident_issue: + return { + "success": False, + "performed": False, + "break_glass_executed": False, + "reasons": [ + "create_incident_issue=False is forbidden on real break-glass execution; " + "pre-execution incident creation is mandatory (#664 AC3)" + ], + "blocker_kind": "incident_creation_required", + } + + # Pre-execution recording: Audit record (Finding 2, B4) + audit_event = gitea_audit.build_event( + action="break_glass_mcp_restart", + result=gitea_audit.SUCCEEDED, remote=remote, - org=o, - repo=r, - profile_identity=profile.get("profile_name", "unknown"), + server=(gitea_url(h, "").rstrip("/") if h else None), + repository=r, + profile_name=profile.get("profile_name", "unknown"), + audit_label=profile.get("audit_label", "unknown"), + authenticated_username=identity, + reason=clean_reason, + mcp_namespace="gitea-author", + task_role=active_role, + operation="break_glass_mcp_restart", + now=now_iso, + request_metadata={ + "confirmation": clean_confirmation, + "restart_class": restart_class, + "disrupted_sessions_count": disrupted_count, + "disrupted_sessions": [ + s.get("session_id") if isinstance(s, dict) else str(s) + for s in disrupted_sessions + ], + }, + ) + audit_write_success = True + if gitea_audit.audit_enabled(): + audit_write_success = gitea_audit.write_event(audit_event) + + if not audit_write_success: + return { + "success": False, + "performed": False, + "break_glass_executed": False, + "reasons": [ + "failed to persist required pre-execution audit event (#664 AC3)" + ], + "blocker_kind": "audit_recording_failed", + } + + # Pre-execution recording: Gitea Incident Issue (Finding 2 & 3, B5) + issue_title = f"[INCIDENT] Break-glass MCP restart invoked by {identity}" + issue_body = ( + f"## Break-glass MCP restart incident report (#664)\n\n" + f"- **Invoked by**: `{identity}` (role: `{active_role}`)\n" + f"- **Timestamp**: `{now_iso}`\n" + f"- **Reason**: {clean_reason}\n" + f"- **Confirmation**: `{clean_confirmation}`\n" + f"- **Disrupted Sessions Count**: `{disrupted_count}`\n\n" + f"### Mandatory Post-Restart Reconciliation (#662)\n" + f"Post-restart reconciliation must be executed via `gitea_reconcile_after_restart` " + f"to clean up orphaned leases, inspect worktree integrity, and handle disrupted work.\n\n" + f"### Cross-references\n" + f"Ref #652 #653 #655 #630 #658 #662 #664\n" + ) + incident_issue_result = None + try: + incident_issue_result = api_request( + "POST", + f"{repo_api_url(h, o, r)}/issues", + _auth(h), + { + "title": issue_title, + "body": issue_body, + "labels": ["incident", "mcp-health", "break-glass"], + }, + ) + if not isinstance(incident_issue_result, dict) or "number" not in incident_issue_result: + return { + "success": False, + "performed": False, + "break_glass_executed": False, + "reasons": [ + f"incident issue creation failed (#664 AC3): {incident_issue_result}" + ], + "blocker_kind": "incident_creation_failed", + "incident_issue": incident_issue_result, + } + except Exception as exc: + return { + "success": False, + "performed": False, + "break_glass_executed": False, + "reasons": [ + f"incident issue creation failed with exception (#664 AC3): {_redact(str(exc))}" + ], + "blocker_kind": "incident_creation_failed", + "incident_issue": {"error": _redact(str(exc))}, + } + + # Execute or delegate canonical non-dry-run restart (Finding 4) + restart_exec_result = gitea_request_mcp_restart( + remote=remote, + host=host, + org=org, + repo=repo, + dry_run=False, + restart_class=restart_class, + request_break_glass=True, ) - incident_issue_result = None - if create_incident_issue and not dry_run: - issue_title = f"[INCIDENT] Break-glass MCP restart invoked by {identity}" - issue_body = ( - f"## Break-glass MCP restart incident report (#664)\n\n" - f"- **Invoked by**: `{identity}` (role: `{active_role}`)\n" - f"- **Timestamp**: `{now_iso}`\n" - f"- **Reason**: {clean_reason}\n" - f"- **Confirmation**: `{clean_confirmation}`\n" - f"- **Disrupted Sessions Count**: `{disrupted_count}`\n\n" - f"### Mandatory Post-Restart Reconciliation (#662)\n" - f"Post-restart reconciliation must be executed via `gitea_reconcile_after_restart` " - f"to clean up orphaned leases, inspect worktree integrity, and handle disrupted work.\n\n" - f"### Cross-references\n" - f"Ref #652 #653 #655 #630 #658 #662 #664\n" - ) - try: - incident_issue_result = api_request( - "POST", - f"{repo_api_url(h, o, r)}/issues", - _auth(h), - { - "title": issue_title, - "body": issue_body, - "labels": ["incident", "mcp-health", "break-glass"], - }, - ) - except Exception as exc: # noqa: BLE001 - incident_issue_result = {"error": _redact(str(exc))} + if not restart_exec_result.get("apply_authorized") or not restart_exec_result.get("success"): + return { + "success": False, + "performed": False, + "break_glass_executed": False, + "actor": identity, + "role": active_role, + "restart_class": restart_class, + "reasons": [ + "delegated restart execution failed or was denied by coordinator (#664)", + *(restart_exec_result.get("reasons") or []), + ], + "blocker_kind": "restart_delegation_failed", + "restart_result": restart_exec_result, + "incident_issue": incident_issue_result, + "audit_record": audit_event, + } return { "success": True, - "performed": not dry_run, - "dry_run": dry_run, - "break_glass_executed": not dry_run, + "performed": True, + "dry_run": False, + "break_glass_executed": True, "would_execute": True, "actor": identity, "role": active_role, @@ -24107,20 +24241,20 @@ def gitea_break_glass_restart( "confirmation": clean_confirmation, "disrupted_sessions_count": disrupted_count, "disrupted_sessions": disrupted_sessions, - "audit_record": audit_payload, - "saved_audit": dict(saved_audit or audit_payload), + "audit_record": audit_event, + "saved_audit": audit_event, "incident_issue": incident_issue_result, "reconciliation_required": True, "reconciliation_tool": "gitea_reconcile_after_restart", "follow_up_issue_required": True, "cross_references": ["#652", "#653", "#655", "#630", "#658", "#662", "#664"], "reasons": [ - "break-glass restart dry-run evaluated successfully" if dry_run - else "break-glass restart executed with incident creation and mandatory reconciliation" + "break-glass restart executed with incident creation and mandatory reconciliation" ], } + # --- #662 post-restart reconciliation --------------------------------------- _POST_RESTART_LAST_PROOF: dict | None = None diff --git a/tests/test_issue_664_break_glass_restart.py b/tests/test_issue_664_break_glass_restart.py index fdf4577..14b125e 100644 --- a/tests/test_issue_664_break_glass_restart.py +++ b/tests/test_issue_664_break_glass_restart.py @@ -6,6 +6,7 @@ import os import unittest from unittest.mock import MagicMock, patch +import gitea_audit import gitea_mcp_server @@ -19,52 +20,110 @@ class TestBreakGlassRestart(unittest.TestCase): def tearDown(self) -> None: self.env_patcher.stop() - def test_ordinary_role_denied_fail_closed(self) -> None: - """AC1: Ordinary LLM roles (author, reviewer, merger, reconciler) cannot invoke break-glass.""" - with patch.object( - gitea_mcp_server, "get_profile", return_value={"profile_name": "prgs-author", "allowed_operations": ["gitea.read"]} - ), patch.object( - gitea_mcp_server, "_profile_role_kind", return_value="author" - ), patch.object( + def test_role_authorization_matrix_with_real_resolver(self) -> None: + """AC1: Explicit allowlist enforcement using production _profile_role_kind resolver. + + Privileged roles (controller, operator, admin, sysadmin) pass role check. + Ordinary LLM roles (author, reviewer, merger, reconciler) and unknown/malformed roles fail closed. + """ + allowed_profiles = [ + {"role": "controller", "allowed_operations": ["gitea.read", "gitea.issue.create"]}, + {"role": "operator", "allowed_operations": ["gitea.read", "gitea.issue.create"]}, + {"role": "admin", "allowed_operations": ["gitea.read", "gitea.issue.create"]}, + {"role": "sysadmin", "allowed_operations": ["gitea.read", "gitea.issue.create"]}, + ] + + for prof in allowed_profiles: + with patch.object(gitea_mcp_server, "get_profile", return_value=prof), patch.object( + gitea_mcp_server, "_profile_operation_gate", return_value=None + ), patch.object( + gitea_mcp_server, "gitea_request_mcp_restart", return_value={"affected_sessions": []} + ): + res = gitea_mcp_server.gitea_break_glass_restart( + reason="Emergency restart required due to deadlock in worker pool", + confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION", + impact_ack=True, + dry_run=True, + remote="prgs", + ) + self.assertTrue( + res["success"], + f"Profile with role '{prof.get('role')}' should pass role check", + ) + + denied_profiles = [ + {"role": "author", "allowed_operations": ["gitea.read", "gitea.issue.create"]}, + {"role": "reviewer", "allowed_operations": ["gitea.read"]}, + {"role": "merger", "allowed_operations": ["gitea.read"]}, + {"role": "reconciler", "allowed_operations": ["gitea.read"]}, + {"role": "guest", "allowed_operations": ["gitea.read"]}, + {"role": "unknown", "allowed_operations": ["gitea.read"]}, + {"role": "mixed", "allowed_operations": ["gitea.read"]}, + {"role": "limited", "allowed_operations": ["gitea.read"]}, + {"role": "", "allowed_operations": ["gitea.read"]}, + {"allowed_operations": ["gitea.read"]}, # no declared role + ] + + for prof in denied_profiles: + with patch.object(gitea_mcp_server, "get_profile", return_value=prof), patch.object( + gitea_mcp_server, "_profile_operation_gate", return_value=None + ): + res = gitea_mcp_server.gitea_break_glass_restart( + reason="Emergency restart required due to deadlock in worker pool", + confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION", + impact_ack=True, + dry_run=True, + remote="prgs", + ) + self.assertFalse( + res["success"], + f"Profile {prof} should fail role check", + ) + self.assertFalse(res["break_glass_executed"]) + self.assertEqual(res["blocker_kind"], "role_authorization") + self.assertIn("not in privileged break-glass allowlist", res["reasons"][0]) + + def test_env_var_cannot_grant_authorization_or_bypass_denial(self) -> None: + """Requirement 5: GITEA_BREAKGLASS_RESTART_AUTHORIZATION cannot grant authorization or bypass role denial.""" + os.environ["GITEA_BREAKGLASS_RESTART_AUTHORIZATION"] = "secret-bypass-token" + unprivileged_prof = {"role": "author", "allowed_operations": ["gitea.read", "gitea.issue.create"]} + + with patch.object(gitea_mcp_server, "get_profile", return_value=unprivileged_prof), patch.object( gitea_mcp_server, "_profile_operation_gate", return_value=None ): res = gitea_mcp_server.gitea_break_glass_restart( - reason="Emergency restart needed due to deadlocked worker daemon processes", + reason="Emergency restart attempting env var bypass", confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION", impact_ack=True, + dry_run=True, remote="prgs", ) self.assertFalse(res["success"]) self.assertFalse(res["break_glass_executed"]) self.assertEqual(res["blocker_kind"], "role_authorization") - self.assertIn("ordinary LLM role 'author' is forbidden", res["reasons"][0]) - def test_short_reason_denied(self) -> None: + def test_reason_validation(self) -> None: """AC2: Reason is required and must be at least 10 characters long.""" - with patch.object( - gitea_mcp_server, "get_profile", return_value={"profile_name": "prgs-controller", "allowed_operations": ["gitea.read"]} - ), patch.object( - gitea_mcp_server, "_profile_role_kind", return_value="controller" - ), patch.object( - gitea_mcp_server, "_profile_operation_gate", return_value=None - ): - res = gitea_mcp_server.gitea_break_glass_restart( - reason="Too short", - confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION", - impact_ack=True, - remote="prgs", - ) - self.assertFalse(res["success"]) - self.assertEqual(res["blocker_kind"], "missing_required_fields") - self.assertIn("at least 10 characters", res["reasons"][0]) + controller_prof = {"role": "controller", "allowed_operations": ["gitea.read", "gitea.issue.create"]} - def test_confirmation_mismatch_denied(self) -> None: + for invalid_reason in ["", " ", "too short", "123456789"]: + with patch.object(gitea_mcp_server, "get_profile", return_value=controller_prof), patch.object( + gitea_mcp_server, "_profile_operation_gate", return_value=None + ): + res = gitea_mcp_server.gitea_break_glass_restart( + reason=invalid_reason, + confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION", + impact_ack=True, + remote="prgs", + ) + self.assertFalse(res["success"]) + self.assertEqual(res["blocker_kind"], "missing_required_fields") + + def test_confirmation_validation(self) -> None: """AC2: Confirmation phrase must match exact required string.""" - with patch.object( - gitea_mcp_server, "get_profile", return_value={"profile_name": "prgs-controller", "allowed_operations": ["gitea.read"]} - ), patch.object( - gitea_mcp_server, "_profile_role_kind", return_value="controller" - ), patch.object( + controller_prof = {"role": "controller", "allowed_operations": ["gitea.read", "gitea.issue.create"]} + + with patch.object(gitea_mcp_server, "get_profile", return_value=controller_prof), patch.object( gitea_mcp_server, "_profile_operation_gate", return_value=None ): res = gitea_mcp_server.gitea_break_glass_restart( @@ -75,15 +134,12 @@ class TestBreakGlassRestart(unittest.TestCase): ) self.assertFalse(res["success"]) self.assertEqual(res["blocker_kind"], "confirmation_mismatch") - self.assertIn("confirmation string mismatch", res["reasons"][0]) - def test_impact_ack_required_denied(self) -> None: + def test_impact_ack_validation(self) -> None: """AC2: impact_ack=True is mandatory.""" - with patch.object( - gitea_mcp_server, "get_profile", return_value={"profile_name": "prgs-controller", "allowed_operations": ["gitea.read"]} - ), patch.object( - gitea_mcp_server, "_profile_role_kind", return_value="controller" - ), patch.object( + controller_prof = {"role": "controller", "allowed_operations": ["gitea.read", "gitea.issue.create"]} + + with patch.object(gitea_mcp_server, "get_profile", return_value=controller_prof), patch.object( gitea_mcp_server, "_profile_operation_gate", return_value=None ): res = gitea_mcp_server.gitea_break_glass_restart( @@ -95,23 +151,167 @@ class TestBreakGlassRestart(unittest.TestCase): self.assertFalse(res["success"]) self.assertEqual(res["blocker_kind"], "impact_ack_required") - def test_dry_run_evaluation(self) -> None: - """AC5: Dry-run evaluation returns preview without live execution or incident creation.""" - with patch.object( - gitea_mcp_server, "get_profile", return_value={"profile_name": "prgs-controller", "allowed_operations": ["gitea.read"]} - ), patch.object( - gitea_mcp_server, "_profile_role_kind", return_value="controller" - ), patch.object( - gitea_mcp_server, "_profile_operation_gate", return_value=None - ), patch.object( - gitea_mcp_server, "gitea_request_mcp_restart", return_value={"affected_sessions": [{"session_id": "s1"}]} - ), patch.object( - gitea_mcp_server, "_authenticated_username", return_value="sysadmin" + def test_incident_permission_gate(self) -> None: + """Requirement 3 / B3: Gate incident creation on gitea.issue.create permission.""" + controller_prof = {"role": "controller", "allowed_operations": ["gitea.read"]} + + with patch.object(gitea_mcp_server, "get_profile", return_value=controller_prof), patch.object( + gitea_mcp_server, "_profile_operation_gate", side_effect=lambda op: ["missing gitea.issue.create"] if op == "gitea.issue.create" else None ): res = gitea_mcp_server.gitea_break_glass_restart( reason="Emergency restart needed due to hung worker process cohort", confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION", impact_ack=True, + create_incident_issue=True, + remote="prgs", + ) + self.assertFalse(res["success"]) + self.assertEqual(res["blocker_kind"], "permission_denied") + + def test_redaction_across_surfaces(self) -> None: + """Requirement 3: Redact operator-controlled reason before incident body and audit surfaces.""" + controller_prof = {"role": "controller", "allowed_operations": ["gitea.read", "gitea.issue.create"]} + raw_reason = "Emergency restart: token ghp_secretToken12345 and url https://user:pass@gitea.prgs.cc/api" + + mock_api_request = MagicMock(return_value={"number": 101, "title": "[INCIDENT]"}) + mock_restart_exec = {"success": True, "apply_authorized": True} + + with patch.object(gitea_mcp_server, "get_profile", return_value=controller_prof), patch.object( + gitea_mcp_server, "_profile_operation_gate", return_value=None + ), patch.object( + gitea_mcp_server, "_auth", return_value={"Authorization": "token test"} + ), patch.object( + gitea_mcp_server, "_authenticated_username", return_value="sysadmin" + ), patch.object( + gitea_mcp_server, "api_request", mock_api_request + ), patch.object( + gitea_mcp_server, "gitea_request_mcp_restart", return_value=mock_restart_exec + ), patch.object( + gitea_audit, "write_event", return_value=True + ): + res = gitea_mcp_server.gitea_break_glass_restart( + reason=raw_reason, + confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION", + impact_ack=True, + dry_run=False, + create_incident_issue=True, + remote="prgs", + ) + + self.assertTrue(res["success"]) + self.assertNotIn("ghp_secretToken12345", res["reason"]) + self.assertNotIn("user:pass@gitea.prgs.cc", res["reason"]) + self.assertIn("[REDACTED]", res["reason"]) + + # Verify incident body redaction + posted_body = mock_api_request.call_args[0][3]["body"] + self.assertNotIn("ghp_secretToken12345", posted_body) + self.assertNotIn("user:pass@gitea.prgs.cc", posted_body) + + def test_audit_failure_before_execution_fails_closed(self) -> None: + """Requirement 2: Audit recording failure stops execution fail-closed.""" + controller_prof = {"role": "controller", "allowed_operations": ["gitea.read", "gitea.issue.create"]} + mock_api_request = MagicMock() + mock_restart_exec = MagicMock() + + with patch.object(gitea_mcp_server, "get_profile", return_value=controller_prof), patch.object( + gitea_mcp_server, "_profile_operation_gate", return_value=None + ), patch.object( + gitea_mcp_server, "_authenticated_username", return_value="sysadmin" + ), patch.object( + gitea_mcp_server, "gitea_request_mcp_restart", side_effect=[{"affected_sessions": []}, mock_restart_exec] + ), patch.object( + gitea_audit, "audit_enabled", return_value=True + ), patch.object( + gitea_audit, "write_event", return_value=False # Audit write fails! + ), patch.object( + gitea_mcp_server, "api_request", mock_api_request + ): + res = gitea_mcp_server.gitea_break_glass_restart( + reason="Emergency restart with failing audit sink", + confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION", + impact_ack=True, + dry_run=False, + create_incident_issue=True, + remote="prgs", + ) + self.assertFalse(res["success"]) + self.assertFalse(res["break_glass_executed"]) + self.assertEqual(res["blocker_kind"], "audit_recording_failed") + mock_api_request.assert_not_called() + + def test_incident_creation_failure_before_execution_fails_closed(self) -> None: + """Requirement 2 / B5: Incident creation failure stops execution fail-closed.""" + controller_prof = {"role": "controller", "allowed_operations": ["gitea.read", "gitea.issue.create"]} + mock_restart_exec = MagicMock() + + with patch.object(gitea_mcp_server, "get_profile", return_value=controller_prof), patch.object( + gitea_mcp_server, "_profile_operation_gate", return_value=None + ), patch.object( + gitea_mcp_server, "_auth", return_value={"Authorization": "token test"} + ), patch.object( + gitea_mcp_server, "_authenticated_username", return_value="sysadmin" + ), patch.object( + gitea_mcp_server, "gitea_request_mcp_restart", side_effect=[{"affected_sessions": []}, mock_restart_exec] + ), patch.object( + gitea_audit, "write_event", return_value=True + ), patch.object( + gitea_mcp_server, "api_request", side_effect=RuntimeError("Gitea 500 API Error") + ): + res = gitea_mcp_server.gitea_break_glass_restart( + reason="Emergency restart with failing incident POST", + confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION", + impact_ack=True, + dry_run=False, + create_incident_issue=True, + remote="prgs", + ) + self.assertFalse(res["success"]) + self.assertFalse(res["break_glass_executed"]) + self.assertEqual(res["blocker_kind"], "incident_creation_failed") + + def test_incident_opt_out_on_real_execution_fails_closed(self) -> None: + """Requirement 2 / B5: create_incident_issue=False fails closed on real execution.""" + controller_prof = {"role": "controller", "allowed_operations": ["gitea.read", "gitea.issue.create"]} + + with patch.object(gitea_mcp_server, "get_profile", return_value=controller_prof), patch.object( + gitea_mcp_server, "_profile_operation_gate", return_value=None + ), patch.object( + gitea_mcp_server, "gitea_request_mcp_restart", return_value={"affected_sessions": []} + ): + res = gitea_mcp_server.gitea_break_glass_restart( + reason="Emergency restart trying to skip incident creation", + confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION", + impact_ack=True, + dry_run=False, + create_incident_issue=False, + remote="prgs", + ) + self.assertFalse(res["success"]) + self.assertFalse(res["break_glass_executed"]) + self.assertEqual(res["blocker_kind"], "incident_creation_required") + + def test_dry_run_truthfulness_and_no_durable_mutation(self) -> None: + """Requirement 4 / B7: Dry-run returns preview without executing or creating durable records.""" + controller_prof = {"role": "controller", "allowed_operations": ["gitea.read", "gitea.issue.create"]} + mock_audit_write = MagicMock() + mock_api_request = MagicMock() + + with patch.object(gitea_mcp_server, "get_profile", return_value=controller_prof), patch.object( + gitea_mcp_server, "_profile_operation_gate", return_value=None + ), patch.object( + gitea_mcp_server, "_authenticated_username", return_value="sysadmin" + ), patch.object( + gitea_mcp_server, "gitea_request_mcp_restart", return_value={"affected_sessions": [{"session_id": "s1"}]} + ), patch.object( + gitea_audit, "write_event", mock_audit_write + ), patch.object( + gitea_mcp_server, "api_request", mock_api_request + ): + res = gitea_mcp_server.gitea_break_glass_restart( + reason="Emergency restart preview in dry-run mode", + confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION", + impact_ack=True, dry_run=True, remote="prgs", ) @@ -119,33 +319,32 @@ class TestBreakGlassRestart(unittest.TestCase): self.assertTrue(res["dry_run"]) self.assertFalse(res["break_glass_executed"]) self.assertTrue(res["would_execute"]) - self.assertTrue(res["reconciliation_required"]) - self.assertEqual(res["reconciliation_tool"], "gitea_reconcile_after_restart") - self.assertIn("#664", res["cross_references"]) + self.assertIsNone(res["incident_issue"]) + self.assertIsNone(res["saved_audit"]) + mock_audit_write.assert_not_called() + mock_api_request.assert_not_called() - def test_privileged_execute_creates_incident_and_audit(self) -> None: - """AC3 & AC4: Execution creates incident issue, audit entry, and mandates post-restart reconcile.""" - mock_api_request = MagicMock(return_value={"number": 999, "title": "[INCIDENT] Break-glass"}) - mock_save_state = MagicMock(return_value={"saved": True}) + def test_successful_real_execution(self) -> None: + """Requirement 4: Real execution calls canonical restart path and reports execution truthfully.""" + controller_prof = {"role": "controller", "allowed_operations": ["gitea.read", "gitea.issue.create"]} + mock_api_request = MagicMock(return_value={"number": 555, "title": "[INCIDENT]"}) + mock_restart_exec = {"success": True, "apply_authorized": True, "affected_sessions": []} - with patch.object( - gitea_mcp_server, "get_profile", return_value={"profile_name": "prgs-controller", "allowed_operations": ["gitea.read"]} - ), patch.object( - gitea_mcp_server, "_profile_role_kind", return_value="controller" - ), patch.object( + with patch.object(gitea_mcp_server, "get_profile", return_value=controller_prof), patch.object( gitea_mcp_server, "_profile_operation_gate", return_value=None ), patch.object( gitea_mcp_server, "_auth", return_value={"Authorization": "token test"} - ), patch.object( - gitea_mcp_server, "gitea_request_mcp_restart", return_value={"affected_sessions": [{"session_id": "s1"}]} ), patch.object( gitea_mcp_server, "_authenticated_username", return_value="sysadmin" + ), patch.object( + gitea_mcp_server, "gitea_request_mcp_restart", return_value=mock_restart_exec + ), patch.object( + gitea_audit, "write_event", return_value=True ), patch.object( gitea_mcp_server, "api_request", mock_api_request - ), patch("mcp_session_state.save_state", mock_save_state): - + ): res = gitea_mcp_server.gitea_break_glass_restart( - reason="Emergency break-glass restart due to unrecoverable transport deadlock", + reason="Privileged emergency break-glass restart execution", confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION", impact_ack=True, dry_run=False, @@ -154,36 +353,42 @@ class TestBreakGlassRestart(unittest.TestCase): ) self.assertTrue(res["success"]) self.assertFalse(res["dry_run"]) + self.assertTrue(res["performed"]) self.assertTrue(res["break_glass_executed"]) - self.assertTrue(res["reconciliation_required"]) - self.assertEqual(res["reconciliation_tool"], "gitea_reconcile_after_restart") - self.assertEqual(res["incident_issue"]["number"], 999) - mock_save_state.assert_called_once() - mock_api_request.assert_called_once() + self.assertEqual(res["incident_issue"]["number"], 555) - def test_env_authorization_override_for_worker_role(self) -> None: - """Environment break-glass authorization enables privileged break-glass for configured sessions.""" - os.environ["GITEA_BREAKGLASS_RESTART_AUTHORIZATION"] = "authorized-token" - with patch.object( - gitea_mcp_server, "get_profile", return_value={"profile_name": "prgs-author", "allowed_operations": ["gitea.read"]} - ), patch.object( - gitea_mcp_server, "_profile_role_kind", return_value="author" - ), patch.object( + def test_delegated_execution_failure(self) -> None: + """Requirement 4: Delegated restart execution failure produces distinct terminal state.""" + controller_prof = {"role": "controller", "allowed_operations": ["gitea.read", "gitea.issue.create"]} + mock_api_request = MagicMock(return_value={"number": 555, "title": "[INCIDENT]"}) + mock_impact_eval = {"affected_sessions": []} + mock_restart_denied = {"success": False, "apply_authorized": False, "reasons": ["restart class denied"]} + + with patch.object(gitea_mcp_server, "get_profile", return_value=controller_prof), patch.object( gitea_mcp_server, "_profile_operation_gate", return_value=None ), patch.object( - gitea_mcp_server, "gitea_request_mcp_restart", return_value={"affected_sessions": []} + gitea_mcp_server, "_auth", return_value={"Authorization": "token test"} ), patch.object( - gitea_mcp_server, "_authenticated_username", return_value="jcwalker3" + gitea_mcp_server, "_authenticated_username", return_value="sysadmin" + ), patch.object( + gitea_mcp_server, "gitea_request_mcp_restart", side_effect=[mock_impact_eval, mock_restart_denied] + ), patch.object( + gitea_audit, "write_event", return_value=True + ), patch.object( + gitea_mcp_server, "api_request", mock_api_request ): res = gitea_mcp_server.gitea_break_glass_restart( - reason="Authorized emergency break-glass restart test", + reason="Privileged break-glass restart with denied delegation", confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION", impact_ack=True, - dry_run=True, + dry_run=False, + create_incident_issue=True, remote="prgs", ) - self.assertTrue(res["success"]) - self.assertTrue(res["dry_run"]) + self.assertFalse(res["success"]) + self.assertFalse(res["performed"]) + self.assertFalse(res["break_glass_executed"]) + self.assertEqual(res["blocker_kind"], "restart_delegation_failed") if __name__ == "__main__":