fix(mcp): remediate break-glass restart authorization and audit findings (#664)
This commit is contained in:
@@ -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:[email protected]/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:[email protected]", 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:[email protected]", 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__":
|
||||
|
||||
Reference in New Issue
Block a user