549 lines
29 KiB
Python
549 lines
29 KiB
Python
"""Tests for emergency break-glass MCP restart workflow (#664)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import unittest
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import gitea_audit
|
|
import gitea_mcp_server
|
|
|
|
|
|
class TestBreakGlassRestart(unittest.TestCase):
|
|
"""Test suite for gitea_break_glass_restart tool and guardrails (#664)."""
|
|
|
|
def setUp(self) -> None:
|
|
self.env_patcher = patch.dict(os.environ, {}, clear=False)
|
|
self.env_patcher.start()
|
|
|
|
def tearDown(self) -> None:
|
|
self.env_patcher.stop()
|
|
|
|
def test_role_authorization_matrix_with_real_resolver(self) -> None:
|
|
"""AC1: Explicit allowlist enforcement using production _profile_role_kind resolver.
|
|
|
|
Privileged controller role passes role check.
|
|
Synthetic roles (operator, admin, sysadmin), ordinary LLM roles (author, reviewer, merger, reconciler), and unknown/malformed roles fail closed.
|
|
"""
|
|
allowed_profiles = [
|
|
{"role": "controller", "allowed_operations": ["gitea.read", "gitea.issue.create", "runtime.break_glass_restart"]},
|
|
{"profile_name": "prgs-controller", "allowed_operations": ["gitea.read", "gitea.issue.create", "runtime.break_glass_restart"]},
|
|
]
|
|
|
|
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 {prof} should pass role check",
|
|
)
|
|
|
|
denied_profiles = [
|
|
{"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"]},
|
|
{"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 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")
|
|
|
|
def test_reason_validation(self) -> None:
|
|
"""AC2: Reason is required and must be at least 10 characters long."""
|
|
controller_prof = {"role": "controller", "allowed_operations": ["gitea.read", "gitea.issue.create"]}
|
|
|
|
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."""
|
|
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(
|
|
reason="Emergency restart needed due to stuck daemon processes",
|
|
confirmation="wrong_confirmation_phrase",
|
|
impact_ack=True,
|
|
remote="prgs",
|
|
)
|
|
self.assertFalse(res["success"])
|
|
self.assertEqual(res["blocker_kind"], "confirmation_mismatch")
|
|
|
|
def test_impact_ack_validation(self) -> None:
|
|
"""AC2: impact_ack=True is mandatory."""
|
|
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(
|
|
reason="Emergency restart needed due to stuck daemon processes",
|
|
confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION",
|
|
impact_ack=False,
|
|
remote="prgs",
|
|
)
|
|
self.assertFalse(res["success"])
|
|
self.assertEqual(res["blocker_kind"], "impact_ack_required")
|
|
|
|
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_supported": True, "apply_authorized": True, "restart_performed": True}
|
|
mock_recon = {"success": 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_mcp_server, "gitea_reconcile_after_restart", return_value=mock_recon
|
|
), patch.object(
|
|
gitea_audit, "audit_enabled", return_value=True
|
|
), 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, "audit_enabled", return_value=True
|
|
), 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",
|
|
)
|
|
self.assertTrue(res["success"])
|
|
self.assertTrue(res["dry_run"])
|
|
self.assertFalse(res["break_glass_executed"])
|
|
self.assertTrue(res["would_execute"])
|
|
self.assertIsNone(res["incident_issue"])
|
|
self.assertIsNone(res["saved_audit"])
|
|
mock_audit_write.assert_not_called()
|
|
mock_api_request.assert_not_called()
|
|
|
|
def test_successful_real_execution(self) -> None:
|
|
"""Requirement 4: Real execution calls canonical restart path and reports execution truthfully."""
|
|
controller_prof = {"profile_name": "prgs-controller", "allowed_operations": ["gitea.read", "gitea.issue.create", "runtime.break_glass_restart"]}
|
|
mock_api_request = MagicMock(return_value={"number": 555, "title": "[INCIDENT]"})
|
|
mock_restart_exec = {"success": True, "apply_supported": True, "apply_authorized": True, "restart_performed": True, "affected_sessions": []}
|
|
mock_recon = {"success": True, "reconciled": 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, "gitea_request_mcp_restart", return_value=mock_restart_exec
|
|
), patch.object(
|
|
gitea_mcp_server, "gitea_reconcile_after_restart", return_value=mock_recon
|
|
), patch.object(
|
|
gitea_audit, "audit_enabled", return_value=True
|
|
), 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="Privileged emergency break-glass restart execution",
|
|
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.assertFalse(res["dry_run"])
|
|
self.assertTrue(res["performed"])
|
|
self.assertTrue(res["break_glass_executed"])
|
|
self.assertEqual(res["incident_issue"]["number"], 555)
|
|
|
|
def test_unsupported_apply_returns_truthful_unsupported_result(self) -> None:
|
|
"""B6/B11: apply_supported=False returns apply_unsupported blocker and break_glass_executed=False."""
|
|
controller_prof = {"profile_name": "prgs-controller", "allowed_operations": ["gitea.read", "gitea.issue.create", "runtime.break_glass_restart"]}
|
|
mock_api_request = MagicMock(return_value={"number": 555, "title": "[INCIDENT]"})
|
|
mock_restart_unsupported = {"success": True, "apply_supported": False, "apply_authorized": True, "restart_performed": False}
|
|
|
|
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_unsupported]
|
|
), patch.object(
|
|
gitea_audit, "audit_enabled", return_value=True
|
|
), 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="Privileged restart request with unsupported coordinator apply",
|
|
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["performed"])
|
|
self.assertFalse(res["break_glass_executed"])
|
|
self.assertEqual(res["blocker_kind"], "apply_unsupported")
|
|
|
|
def test_reconciliation_failure_after_execution(self) -> None:
|
|
"""B10: Post-restart reconciliation failure reports break_glass_executed=True but success=False."""
|
|
controller_prof = {"profile_name": "prgs-controller", "allowed_operations": ["gitea.read", "gitea.issue.create", "runtime.break_glass_restart"]}
|
|
mock_api_request = MagicMock(return_value={"number": 555, "title": "[INCIDENT]"})
|
|
mock_restart_exec = {"success": True, "apply_supported": True, "apply_authorized": True, "restart_performed": True}
|
|
mock_recon = {"success": False, "error": "lease cleanup failed"}
|
|
|
|
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", return_value=mock_restart_exec
|
|
), patch.object(
|
|
gitea_mcp_server, "gitea_reconcile_after_restart", return_value=mock_recon
|
|
), patch.object(
|
|
gitea_audit, "audit_enabled", return_value=True
|
|
), 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="Privileged restart request with failing reconciliation",
|
|
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.assertTrue(res["performed"])
|
|
self.assertTrue(res["break_glass_executed"])
|
|
self.assertEqual(res["blocker_kind"], "reconciliation_failed")
|
|
|
|
def test_bare_secret_redaction(self) -> None:
|
|
"""B8: Bare token shapes like ghp_... and sk-live-... are redacted on all surfaces."""
|
|
controller_prof = {"profile_name": "prgs-controller", "allowed_operations": ["gitea.read", "gitea.issue.create", "runtime.break_glass_restart"]}
|
|
raw_reason = "Emergency restart reason containing bare tokens ghp_1234567890abcdef12345678 and sk-live-abcdef1234567890abcdef"
|
|
mock_api_request = MagicMock(return_value={"number": 101, "title": "[INCIDENT]"})
|
|
mock_restart_exec = {"success": True, "apply_supported": True, "apply_authorized": True, "restart_performed": True}
|
|
mock_recon = {"success": 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_mcp_server, "gitea_reconcile_after_restart", return_value=mock_recon
|
|
), patch.object(
|
|
gitea_audit, "audit_enabled", return_value=True
|
|
), 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_1234567890abcdef12345678", res["reason"])
|
|
self.assertNotIn("sk-live-abcdef1234567890abcdef", res["reason"])
|
|
self.assertIn("[REDACTED]", res["reason"])
|
|
|
|
def test_audit_disabled_fails_closed(self) -> None:
|
|
"""B9: Disabling audit recording blocks execution fail-closed."""
|
|
controller_prof = {"profile_name": "prgs-controller", "allowed_operations": ["gitea.read", "gitea.issue.create", "runtime.break_glass_restart"]}
|
|
|
|
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_audit, "audit_enabled", return_value=False
|
|
):
|
|
res = gitea_mcp_server.gitea_break_glass_restart(
|
|
reason="Emergency restart with disabled audit logging",
|
|
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")
|
|
|
|
def test_capability_map_registration(self) -> None:
|
|
"""B12: Ensure task_capability_map maps gitea_break_glass_restart to runtime.break_glass_restart."""
|
|
from task_capability_map import TASK_CAPABILITY_MAP
|
|
entry = TASK_CAPABILITY_MAP.get("gitea_break_glass_restart")
|
|
self.assertIsNotNone(entry)
|
|
self.assertEqual(entry["permission"], "runtime.break_glass_restart")
|
|
self.assertEqual(entry["role"], "controller")
|
|
|
|
def test_delegated_execution_failure(self) -> None:
|
|
"""Requirement 4: Delegated restart execution failure produces distinct terminal state."""
|
|
controller_prof = {"profile_name": "prgs-controller", "allowed_operations": ["gitea.read", "gitea.issue.create", "runtime.break_glass_restart"]}
|
|
mock_api_request = MagicMock(return_value={"number": 555, "title": "[INCIDENT]"})
|
|
mock_impact_eval = {"affected_sessions": []}
|
|
mock_restart_denied = {"success": False, "apply_supported": True, "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, "_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=[mock_impact_eval, mock_restart_denied]
|
|
), patch.object(
|
|
gitea_audit, "audit_enabled", return_value=True
|
|
), 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="Privileged break-glass restart with denied delegation",
|
|
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["performed"])
|
|
self.assertFalse(res["break_glass_executed"])
|
|
self.assertEqual(res["blocker_kind"], "restart_delegation_failed")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|