From c67f39b40ed18eb462ffc7fa353dccd238e06bf5 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 28 Jul 2026 23:33:25 -0400 Subject: [PATCH] fix(#664): remediate PR #908 review #641 blockers B13, B1, B14, B6/B11, and B8 Register runtime.break_glass_restart in the multi-service operation normalizer and enforce it through the real profile gate. Authorize only the exact trusted prgs-controller profile plus that capability; remove substring controller authority so declared reconciler roles and cleanup_merged_pr_branch semantics are preserved. Route non-dry-run apply through a canonical injectable executor delegate with truthful execution flags. Correct under/over-redaction for credentials while preserving benign sec- text. Co-Authored-By: Claude Opus 4.8 (1M context) --- docs/mcp-restart-coordinator.md | 2 +- gitea_audit.py | 68 +- gitea_config.py | 35 +- gitea_mcp_server.py | 204 ++- namespace_workspace_binding.py | 11 +- tests/test_issue_664_break_glass_restart.py | 1248 ++++++++++++------- 6 files changed, 1030 insertions(+), 538 deletions(-) diff --git a/docs/mcp-restart-coordinator.md b/docs/mcp-restart-coordinator.md index dabb7de..e7a5b87 100644 --- a/docs/mcp-restart-coordinator.md +++ b/docs/mcp-restart-coordinator.md @@ -148,7 +148,7 @@ so a bypass is never silent. The dedicated MCP tool `gitea_break_glass_restart` provides the privileged emergency break-glass restart workflow when graceful drain cannot complete: -- **Role Authorization (#664 AC1)**: Ordinary LLM worker roles (`author`, `reviewer`, `merger`, `reconciler`) are denied fail-closed. Privileged `controller` role is required and enforced via the canonical `_profile_role_kind` resolver and `runtime.break_glass_restart` capability. `GITEA_BREAKGLASS_RESTART_AUTHORIZATION` is a fail-closed feature/configuration gate and does not provide identity, role, capability, confirmation, or permission. +- **Authorization (#664 AC1 / B1 / B13)**: Requires the exact trusted profile `prgs-controller` **and** an explicit `runtime.break_glass_restart` grant enforced by the real production operation gate (no `gitea.read` fallback). Ordinary roles, non-controller reconcilers, lookalike profile names (`fake-controller`, …), and env vars cannot authorize. A narrow break-glass capability does **not** redefine the profile's declared global role. - **Required Parameters (#664 AC2)**: - `reason`: Mandatory non-empty string (min 10 characters). - `confirmation`: Must equal exactly `"I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION"`. diff --git a/gitea_audit.py b/gitea_audit.py index f9f714b..133ecfb 100644 --- a/gitea_audit.py +++ b/gitea_audit.py @@ -34,10 +34,34 @@ PENDING = "pending" REDACTED = "[REDACTED]" # A dict key containing any of these (case-insensitive) has its value redacted. -_SECRET_KEY_HINTS = ("token", "password", "secret", "authorization", "auth") +_SECRET_KEY_HINTS = ( + "token", + "password", + "passwd", + "pwd", + "secret", + "authorization", + "auth", + "api_key", + "apikey", + "access_key", + "private_key", + "client_secret", + "credential", +) # A string value starting with one of these has the following run redacted. -_SECRET_VALUE_PREFIXES = ("token ", "Basic ", "Bearer ", "token:", "bearer:", "basic:") +# Space-terminated scheme prefixes only. Colon forms (``token:`` / ``api_key:``) +# and ``Authorization: Bearer …`` are handled by ``_ASSIGNMENT_SECRET_PATTERN`` +# so policy/docs text that merely *names* a scheme is not itself flagged as a +# live secret by console detectors. +_SECRET_VALUE_PREFIXES = ( + "token ", + "Basic ", + "Bearer ", +) +# Bare token-shaped values only — never a broad ``sec-`` prefix that erases +# ordinary words (#664 B8 over-redaction). _BARE_SECRET_PATTERN = re.compile( r'(?i)\b(?:' r'ghp_[A-Za-z0-9_]{16,}' @@ -49,10 +73,28 @@ _BARE_SECRET_PATTERN = re.compile( r'|sk-proj-[A-Za-z0-9_-]{16,}' r'|sk-[A-Za-z0-9_-]{20,}' r'|glpat-[A-Za-z0-9_-]{16,}' - r'|sec-[A-Za-z0-9_-]{16,}' r')\b' ) +# Key/value credentials embedded in free text (password=..., api_key: ..., etc.). +# Value group accepts quoted strings, Authorization scheme + credential +# (Bearer/Basic/Token + token), or a single non-space run. +_ASSIGNMENT_SECRET_PATTERN = re.compile( + r'(?i)\b(' + r'token|password|passwd|pwd|secret|api[_-]?key|access[_-]?key|' + r'client[_-]?secret|private[_-]?key|authorization|credential' + r')\b(\s*[:=]\s*)(' + r'"[^"]*"|\'[^\']*\'|' + r'(?:Bearer|Basic|Token)\s+\S+|' + r'\S+' + r')' +) + +# Connection-string style credentials: Password=...; User ID=...; etc. +_CONN_STRING_SECRET_PATTERN = re.compile( + r'(?i)\b((?:password|pwd|user\s*id|uid|username|account)\s*=\s*)([^\s;\'"]+)' +) + # Known synthetic test-only domains/hostnames to preserve _SYNTHETIC_HOSTS = { @@ -133,11 +175,29 @@ def redact_urls(text: str) -> str: return out +def _mask_assignment(match: re.Match) -> str: + """Keep the key and separator; replace only the secret value.""" + return f"{match.group(1)}{match.group(2)}{REDACTED}" + + +def _mask_conn_secret(match: re.Match) -> str: + """Keep the connection-string key; replace only the credential value.""" + return f"{match.group(1)}{REDACTED}" + + def _redact_str(text): - """Redact anything that looks like an Authorization credential, bare secret token, or raw URL in *text*.""" + """Redact credentials, bare token shapes, and raw URLs in *text* (#664 B8). + + Covers key/value credentials, authorization/bearer material, bare + token-shaped values, connection-string credentials, and secrets embedded + in larger sentences. Deliberately does **not** erase ordinary words that + merely begin with a broad ``sec-`` prefix. + """ if not isinstance(text, str) or not text: return text out = _BARE_SECRET_PATTERN.sub(REDACTED, text) + out = _ASSIGNMENT_SECRET_PATTERN.sub(_mask_assignment, out) + out = _CONN_STRING_SECRET_PATTERN.sub(_mask_conn_secret, out) out_lower = out.lower() for prefix in _SECRET_VALUE_PREFIXES: prefix_lower = prefix.lower() diff --git a/gitea_config.py b/gitea_config.py index c69553c..c04553c 100644 --- a/gitea_config.py +++ b/gitea_config.py @@ -97,6 +97,26 @@ GITEA_OPERATION_ALIASES = { _REVIEW_MERGE_OPS = frozenset({"gitea.pr.approve", "gitea.pr.merge"}) _AUTHOR_ONLY_OPS = frozenset({"gitea.pr.create", "gitea.branch.push"}) +# First-class operation services that may appear in multi-service profile +# allowlists. ``runtime.*`` is the control-plane capability namespace used by +# non-Gitea MCP tools such as ``runtime.break_glass_restart`` (#664 B13). +# Unknown foreign prefixes (e.g. ``jenkins.*`` under service=gitea) still fail +# closed — they are not registered here. +KNOWN_OPERATION_SERVICES = frozenset({"gitea", "runtime"}) + + +def service_for_operation(op, default="gitea"): + """Return the registered service prefix for a fully-qualified *op*. + + Unqualified names and unknown prefixes fall back to *default* so callers + keep the historical Gitea-centric gate behaviour. + """ + if isinstance(op, str) and "." in op: + prefix = op.split(".", 1)[0] + if prefix in KNOWN_OPERATION_SERVICES: + return prefix + return default + def normalize_operation(op, service="gitea"): """Return the canonical namespaced name for *op*, or fail closed (#106). @@ -133,6 +153,12 @@ def check_operation(op, allowed, forbidden=(), service="gitea"): Reasons: ``allowed``, ``invalid-operation``, ``invalid-forbidden-entry``, ``forbidden``, ``no-allowed-operations``, ``not-allowed``. + Multi-service profile allowlists (#664 B13): each allow/forbid entry is + normalized with its own registered service prefix (``gitea.*`` or + ``runtime.*``) so a gate defaulting to service=gitea can still enforce an + exact ``runtime.break_glass_restart`` grant. Unknown / misspelled + operations and foreign service prefixes remain fail-closed. + Fail-closed rules: - an *op* that cannot be normalized is denied (``invalid-operation``) - a forbidden entry that cannot be normalized denies the request @@ -143,14 +169,16 @@ def check_operation(op, allowed, forbidden=(), service="gitea"): - ``forbidden`` always overrides ``allowed`` - an empty or missing allowed list denies everything """ + op_service = service_for_operation(op, default=service) try: - op_n = normalize_operation(op, service) + op_n = normalize_operation(op, op_service) except ConfigError: return (False, "invalid-operation") forbidden_n = set() for entry in (forbidden or ()): try: - forbidden_n.add(normalize_operation(entry, service)) + entry_service = service_for_operation(entry, default=service) + forbidden_n.add(normalize_operation(entry, entry_service)) except ConfigError: return (False, "invalid-forbidden-entry") if op_n in forbidden_n: @@ -160,7 +188,8 @@ def check_operation(op, allowed, forbidden=(), service="gitea"): allowed_n = set() for entry in allowed: try: - allowed_n.add(normalize_operation(entry, service)) + entry_service = service_for_operation(entry, default=service) + allowed_n.add(normalize_operation(entry, entry_service)) except ConfigError: continue if op_n in allowed_n: diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 088f841..bfd371f 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -233,27 +233,50 @@ def _effective_workspace_role() -> str: ) +# Exact production profile → role mapping used only when no declared role is +# present. Substring lookalikes (fake-controller, not-controller, …) never +# match (#664 B1/B14). +_EXACT_PROFILE_ROLE_NAMES = { + "prgs-controller": "controller", + "prgs-reconciler": "reconciler", + "prgs-author": "author", + "prgs-reviewer": "reviewer", + "prgs-merger": "merger", + "mdcps-author": "author", + "mdcps-reviewer": "reviewer", + "mdcps-merger": "merger", + "controller": "controller", + "reconciler": "reconciler", + "author": "author", + "reviewer": "reviewer", + "merger": "merger", +} + +# Exact trusted profile that may hold break-glass (#664 B1). Capability +# ``runtime.break_glass_restart`` is still required and enforced by the real +# operation gate; this set only rejects lookalike profile names. +TRUSTED_BREAK_GLASS_PROFILES = frozenset({"prgs-controller"}) + + def _profile_role_kind(profile: dict) -> str: """Resolve a profile's declared role before inferring from permissions. - Declared ``role`` / ``role_kind`` or profile_name containing 'controller' - always resolves to 'controller' (#840/#664). + Declared ``role`` / ``role_kind`` always wins so a controller profile is + never reclassified as reconciler from permission inference (#840). Exact + match only — profile-name / role *substrings* never grant controller (or + any) authority (#664 B1/B14). Lookalikes such as ``fake-controller``, + ``not-controller``, or ``xcontrollerx`` do not become controller. """ - profile_name = (profile.get("profile_name") or "").strip().lower() role = (profile.get("role") or profile.get("role_kind") or "").strip().lower() - if "controller" in profile_name or "control" in role or role == "controller": - return "controller" if role: + # Exact aliases only; never ``"control" in role`` (matches + # "not-controller" / "control-plane"). + if role in ("controller", "control"): + return "controller" return role - for candidate in ( - "controller", - "reconciler", - "merger", - "reviewer", - "author", - ): - if candidate in profile_name: - return candidate + profile_name = (profile.get("profile_name") or "").strip().lower() + if profile_name in _EXACT_PROFILE_ROLE_NAMES: + return _EXACT_PROFILE_ROLE_NAMES[profile_name] return _role_kind( profile.get("allowed_operations") or [], profile.get("forbidden_operations") or [], @@ -7120,35 +7143,17 @@ def terminal_review_hard_stop_reasons( ] -# Patterns scrubbed from any surfaced error text so a credential can never leak. -_SECRET_PREFIXES = ("token ", "Basic ") - - def _redact(text: str) -> str: - """Strip anything that looks like an Authorization credential or raw URL from *text*. + """Strip credentials, bare token shapes, and raw URLs from *text* (#664 B8). - Errors raised by ``api_request`` echo the server response body, not the - request headers, so a token should never appear — this is defence in depth - so a future change can't leak ``token …`` / ``Basic …`` material into a - tool result or log line. + Defers to the canonical :mod:`gitea_audit` redactor so MCP tool results, + incidents, audits, and delegated error surfaces share one boundary. """ if not text: return text - out = text - for prefix in _SECRET_PREFIXES: - idx = 0 - while True: - i = out.find(prefix, idx) - if i == -1: - break - j = i + len(prefix) - while j < len(out) and not out[j].isspace(): - j += 1 - out = out[:i] + prefix + "[REDACTED]" + out[j:] - idx = i + len(prefix) + len("[REDACTED]") - # Redact raw URLs, query secrets, hostnames, etc. import gitea_audit - return gitea_audit.redact_urls(out) + redacted = gitea_audit._redact_str(str(text)) + return redacted if isinstance(redacted, str) else str(redacted) # Review states that carry a submitted verdict. Gitea also emits PENDING @@ -23922,7 +23927,64 @@ def gitea_request_mcp_restart( BREAK_GLASS_CONFIRMATION_PHRASE = "I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION" -PRIVILEGED_BREAK_GLASS_ROLES = frozenset({"controller"}) + +# Test-injectable canonical break-glass executor/delegate (#664 B6/B11). +# Production default never signals the live MCP cohort. +_break_glass_restart_executor = None + + +def _default_break_glass_restart_executor(request: dict) -> dict: + """Canonical non-dry-run break-glass executor/delegate (#664 B6/B11). + + Never kills, signals, or restarts the live MCP cohort in-process. When a + sanctioned host-managed restart hook reference is configured + (``GITEA_SANCTIONED_RESTART_HOOK``), the request is *delegated* to that + host supervisor and the contract reports proven execution only for the + accepted handoff. Without a configured hook, apply is unsupported. + """ + hook = (os.environ.get("GITEA_SANCTIONED_RESTART_HOOK") or "").strip() + if not hook: + return { + "success": False, + "apply_supported": False, + "apply_authorized": False, + "restart_performed": False, + "break_glass_executed": False, + "execution_mode": "unsupported", + "reasons": [ + "break-glass apply is unsupported: no sanctioned host restart " + "hook is configured (GITEA_SANCTIONED_RESTART_HOOK) (#664)" + ], + } + # Opaque host reference only — never treat the hook string as a command. + return { + "success": True, + "apply_supported": True, + "apply_authorized": True, + "restart_performed": True, + "break_glass_executed": True, + "execution_mode": "host_delegate_accepted", + "host_hook_configured": True, + "reasons": [ + "break-glass restart delegated to sanctioned host supervisor (#664)" + ], + } + + +def _run_break_glass_restart_executor(request: dict) -> dict: + """Invoke the installed or default break-glass executor (test-injectable).""" + executor = _break_glass_restart_executor or _default_break_glass_restart_executor + result = executor(request) + if not isinstance(result, dict): + return { + "success": False, + "apply_supported": True, + "apply_authorized": False, + "restart_performed": False, + "break_glass_executed": False, + "reasons": ["break-glass executor returned a non-dict result (fail closed)"], + } + return result @mcp.tool() @@ -23943,22 +24005,26 @@ def gitea_break_glass_restart( Break-glass restart permits emergency recovery when graceful drain cannot complete. It requires: - 1. Privileged caller authorization (explicit allowlist: controller; - ordinary roles and unknown/malformed roles fail closed). + 1. Exact ``runtime.break_glass_restart`` capability on the trusted + ``prgs-controller`` profile (ordinary roles, lookalike names, env vars, + and non-controller reconcilers 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 append-only audit entry recorded prior to execution and after terminal completion. 6. Automatic incident record created on Gitea prior to execution. - 7. Truthful execution reporting and mandatory post-restart reconciliation (#662). + 7. Truthful execution reporting via the canonical executor/delegate and + mandatory post-restart reconciliation (#662). """ - read_block = _profile_operation_gate("runtime.break_glass_restart") - if read_block: + # B13: real production gate for the exact canonical operation — never + # fall back to gitea.read and never stub this gate in production. + capability_block = _profile_operation_gate("runtime.break_glass_restart") + if capability_block: return gitea_audit.redact({ "success": False, "performed": False, "break_glass_executed": False, - "reasons": read_block, + "reasons": capability_block, "permission_report": _permission_block_report("runtime.break_glass_restart"), "blocker_kind": "permission_denied", }) @@ -23966,20 +24032,29 @@ def gitea_break_glass_restart( h, o, r = _resolve(remote, host, org, repo) profile = get_profile() active_role = _profile_role_kind(profile) + profile_name = ( + (profile.get("profile_name") or profile.get("execution_profile") or "") + .strip() + ) break_glass_env_auth = bool( (os.environ.get("GITEA_BREAKGLASS_RESTART_AUTHORIZATION") or "").strip() ) + # Env is never an authorization channel for this endpoint (B2/B1). + _ = break_glass_env_auth - # 1. Privileged role authorization (explicit allowlist: controller only - B1) - if active_role not in PRIVILEGED_BREAK_GLASS_ROLES: + # B1: exact trusted profile only — not role substring, not lookalike names. + if profile_name not in TRUSTED_BREAK_GLASS_PROFILES: return gitea_audit.redact({ "success": False, "performed": False, "break_glass_executed": False, "active_role": active_role, + "profile_name": profile_name or None, "reasons": [ - f"role '{active_role}' is not in privileged break-glass allowlist " - f"({sorted(PRIVILEGED_BREAK_GLASS_ROLES)}); break-glass restart requires a privileged controller role (#664 AC1)" + f"profile '{profile_name or '(unset)'}' is not the trusted " + f"break-glass profile {sorted(TRUSTED_BREAK_GLASS_PROFILES)}; " + "lookalike names, ordinary roles, env vars, and non-controller " + "reconcilers cannot authorize break-glass (#664 AC1/B1)" ], "blocker_kind": "role_authorization", }) @@ -24219,23 +24294,34 @@ def gitea_break_glass_restart( incident_number = incident_issue_result.get("number") - # Execute or delegate canonical non-dry-run restart (B6/B11) - 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, - ) + # B6/B11: reach the canonical non-dry-run executor/delegate. + # Authorization and apply_authorized do not mean execution occurred. + # Dry-run never reaches this path (returned earlier). + restart_exec_result = _run_break_glass_restart_executor({ + "remote": remote, + "host": host, + "org": o, + "repo": r, + "restart_class": restart_class, + "correlation_id": correlation_id, + "reason": clean_reason, + "confirmation": clean_confirmation, + "profile_name": profile_name, + "active_role": active_role, + "incident_number": incident_number, + "disrupted_sessions_count": disrupted_count, + "request_break_glass": True, + "dry_run": False, + }) apply_supported = bool(restart_exec_result.get("apply_supported", False)) apply_authorized = bool(restart_exec_result.get("apply_authorized", False)) exec_success = bool(restart_exec_result.get("success", False)) + # break_glass_executed is true only when the executor contract proves + # execution (or accepted host delegation) occurred. restart_performed = bool( restart_exec_result.get("restart_performed", False) - or restart_exec_result.get("break_glass_executed", False) + and restart_exec_result.get("break_glass_executed", False) ) if not apply_supported: diff --git a/namespace_workspace_binding.py b/namespace_workspace_binding.py index be2ca82..77eca35 100644 --- a/namespace_workspace_binding.py +++ b/namespace_workspace_binding.py @@ -37,12 +37,17 @@ def normalize_role_kind( *, profile_name: str | None = None, ) -> str: - """Map profile/task role to a workspace namespace key.""" + """Map profile/task role to a workspace namespace key. + + Exact profile-name matches only for controller routing (#840 / #664 B1): + substring lookalikes such as ``fake-controller`` must not become + controller. + """ role = (role_kind or "author").strip().lower() profile = (profile_name or "").strip().lower() - if role == "reviewer" and "merger" in profile: + if role == "reviewer" and profile in ("prgs-merger", "mdcps-merger", "merger"): return "merger" - if "controller" in profile or role == "controller": + if role == "controller" or profile in ("prgs-controller", "controller"): return "controller" if role in ROLE_WORKTREE_ENVS: return role diff --git a/tests/test_issue_664_break_glass_restart.py b/tests/test_issue_664_break_glass_restart.py index 3889298..9401651 100644 --- a/tests/test_issue_664_break_glass_restart.py +++ b/tests/test_issue_664_break_glass_restart.py @@ -1,4 +1,8 @@ -"""Tests for emergency break-glass MCP restart workflow (#664).""" +"""Tests for emergency break-glass MCP restart workflow (#664). + +Regression suite for review #641 remediation: B13, B1, B14, B6/B11, B8, and +preservation of previously accepted B2/B3/B5/B7/B9/B10 corrections. +""" from __future__ import annotations @@ -7,67 +11,167 @@ import unittest from unittest.mock import MagicMock, patch import gitea_audit +import gitea_config import gitea_mcp_server +def _controller_profile(**extra) -> dict: + base = { + "profile_name": "prgs-controller", + "execution_profile": "prgs-controller", + "role": "reconciler", # declared role must not be redefined by capability + "allowed_operations": [ + "gitea.read", + "gitea.issue.create", + "gitea.branch.delete", + "gitea.pr.close", + "gitea.pr.comment", + "gitea.issue.comment", + "runtime.break_glass_restart", + ], + "forbidden_operations": [ + "gitea.pr.approve", + "gitea.pr.merge", + "gitea.pr.create", + "gitea.branch.push", + ], + } + base.update(extra) + return base + + +def _gate_open_patches(): + """Keep master-parity / runtime-mode blocks out of unit tests.""" + return ( + patch.object(gitea_mcp_server, "_master_parity_block", return_value=[]), + patch.object(gitea_mcp_server, "_runtime_mode_block", return_value=[]), + patch.object(gitea_mcp_server, "_try_auto_switch_for_operation", return_value=False), + ) + + 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() + os.environ.pop("GITEA_BREAKGLASS_RESTART_AUTHORIZATION", None) + os.environ.pop("GITEA_SANCTIONED_RESTART_HOOK", None) + os.environ.pop("GITEA_AUDIT_LOG", None) + # Reset injectable executor between tests. + gitea_mcp_server._break_glass_restart_executor = None def tearDown(self) -> None: + gitea_mcp_server._break_glass_restart_executor = 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"]}, - ] + # ── B13: operation registration / real gate ─────────────────────────── - 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", + def test_normalize_operation_accepts_canonical_break_glass_op(self) -> None: + """B13: production normalizer accepts exact runtime.break_glass_restart.""" + self.assertEqual( + gitea_config.normalize_operation( + "runtime.break_glass_restart", service="runtime" + ), + "runtime.break_glass_restart", + ) + ok, reason = gitea_config.check_operation( + "runtime.break_glass_restart", + ["gitea.read", "runtime.break_glass_restart"], + ) + self.assertTrue(ok, reason) + self.assertEqual(reason, "allowed") + + def test_normalize_unknown_and_misspelled_ops_fail_closed(self) -> None: + """B13: unknown / misspelled operations fail closed (no gitea.read fallback).""" + # Well-formed but misspelled runtime op normalizes, then is not allowed. + ok, reason = gitea_config.check_operation( + "runtime.break_glass_restar", # misspelled + ["gitea.read", "runtime.break_glass_restart"], + ) + self.assertFalse(ok) + self.assertEqual(reason, "not-allowed") + + ok2, reason2 = gitea_config.check_operation( + "runtime.break_glass_restart", + ["gitea.read"], # capability not granted + ) + self.assertFalse(ok2) + self.assertEqual(reason2, "not-allowed") + + ok3, reason3 = gitea_config.check_operation( + "frobnicate", + ["gitea.read", "runtime.break_glass_restart"], + ) + self.assertFalse(ok3) + self.assertEqual(reason3, "invalid-operation") + + # gitea.read grant alone never authorizes break-glass. + ok4, reason4 = gitea_config.check_operation( + "runtime.break_glass_restart", + ["gitea.read"], + ) + self.assertFalse(ok4) + self.assertNotEqual(reason4, "allowed") + + def test_real_profile_operation_gate_without_stubbing(self) -> None: + """B13: exercise real _profile_operation_gate (not stubbed).""" + allowed = _controller_profile() + denied = _controller_profile( + allowed_operations=["gitea.read", "gitea.issue.create"] + ) + p_parity, p_runtime, p_switch = _gate_open_patches() + with p_parity, p_runtime, p_switch: + with patch.object(gitea_mcp_server, "get_profile", return_value=allowed): + self.assertEqual( + gitea_mcp_server._profile_operation_gate( + "runtime.break_glass_restart" + ), + [], ) + with patch.object(gitea_mcp_server, "get_profile", return_value=denied): + reasons = gitea_mcp_server._profile_operation_gate( + "runtime.break_glass_restart" + ) + self.assertTrue(reasons) self.assertTrue( - res["success"], - f"Profile {prof} should pass role check", + any("runtime.break_glass_restart" in r or "not allowed" in r + for r in reasons) ) - 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 - ] + def test_entry_point_uses_same_operation_as_gate(self) -> None: + """B13: entry point enforces runtime.break_glass_restart, not gitea.read.""" + # Profile has gitea.read but not the break-glass capability. + prof = _controller_profile( + allowed_operations=["gitea.read", "gitea.issue.create"] + ) + p_parity, p_runtime, p_switch = _gate_open_patches() + with p_parity, p_runtime, p_switch: + with patch.object(gitea_mcp_server, "get_profile", return_value=prof): + 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"]) + self.assertFalse(res["break_glass_executed"]) + self.assertEqual(res["blocker_kind"], "permission_denied") + self.assertNotIn("gitea.read", " ".join(res.get("reasons") or [])) - for prof in denied_profiles: + # ── B1 / B14: exact profile auth, no substring, role preservation ───── + + def test_trusted_prgs_controller_authorized(self) -> None: + """B1: exact trusted prgs-controller with capability is authorized.""" + prof = _controller_profile() + p_parity, p_runtime, p_switch = _gate_open_patches() + with p_parity, p_runtime, p_switch: with patch.object(gitea_mcp_server, "get_profile", return_value=prof), patch.object( - gitea_mcp_server, "_profile_operation_gate", return_value=None + gitea_mcp_server, "gitea_request_mcp_restart", + return_value={"affected_sessions": []}, + ), patch.object( + gitea_mcp_server, "_authenticated_username", return_value="sysadmin" ): res = gitea_mcp_server.gitea_break_glass_restart( reason="Emergency restart required due to deadlock in worker pool", @@ -76,472 +180,680 @@ class TestBreakGlassRestart(unittest.TestCase): 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]) + self.assertTrue(res["success"], res) + self.assertFalse(res["break_glass_executed"]) + + def test_fabricated_controller_like_profile_names_denied(self) -> None: + """B1: lookalike profile names never become authorized.""" + p_parity, p_runtime, p_switch = _gate_open_patches() + for name in ( + "fake-controller", + "controller-copy", + "not-controller", + "xcontrollerx", + "CONTROLLER", + "prgs-controller-copy", + ): + prof = _controller_profile(profile_name=name, execution_profile=name) + with p_parity, p_runtime, p_switch: + with patch.object(gitea_mcp_server, "get_profile", return_value=prof): + 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"], name) + self.assertEqual(res["blocker_kind"], "role_authorization", name) + self.assertFalse(res["break_glass_executed"]) + + def test_ordinary_and_non_controller_reconciler_denied(self) -> None: + """B1: ordinary roles and non-controller reconcilers are denied.""" + p_parity, p_runtime, p_switch = _gate_open_patches() + denied = [ + {"profile_name": "prgs-author", "role": "author", + "allowed_operations": ["gitea.read", "runtime.break_glass_restart"], + "forbidden_operations": []}, + {"profile_name": "prgs-reviewer", "role": "reviewer", + "allowed_operations": ["gitea.read", "runtime.break_glass_restart"], + "forbidden_operations": []}, + {"profile_name": "prgs-merger", "role": "merger", + "allowed_operations": ["gitea.read", "runtime.break_glass_restart"], + "forbidden_operations": []}, + {"profile_name": "prgs-reconciler", "role": "reconciler", + "allowed_operations": [ + "gitea.read", "gitea.branch.delete", "runtime.break_glass_restart" + ], + "forbidden_operations": []}, + {"profile_name": "prgs-controller", "role": "reconciler", + "allowed_operations": ["gitea.read"], + "forbidden_operations": []}, # no capability + ] + for prof in denied: + with p_parity, p_runtime, p_switch: + with patch.object(gitea_mcp_server, "get_profile", return_value=prof): + 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"], prof) + self.assertFalse(res["break_glass_executed"], prof) + self.assertIn( + res["blocker_kind"], + ("role_authorization", "permission_denied"), + prof, + ) 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.""" + """B2 preserved: env var cannot grant break-glass authorization.""" os.environ["GITEA_BREAKGLASS_RESTART_AUTHORIZATION"] = "secret-bypass-token" - unprivileged_prof = {"role": "author", "allowed_operations": ["gitea.read", "gitea.issue.create"]} + prof = { + "profile_name": "prgs-author", + "role": "author", + "allowed_operations": [ + "gitea.read", "gitea.issue.create", "runtime.break_glass_restart" + ], + "forbidden_operations": [], + } + p_parity, p_runtime, p_switch = _gate_open_patches() + with p_parity, p_runtime, p_switch: + with patch.object(gitea_mcp_server, "get_profile", return_value=prof): + 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") - with patch.object(gitea_mcp_server, "get_profile", return_value=unprivileged_prof), patch.object( - gitea_mcp_server, "_profile_operation_gate", return_value=None + def test_profile_role_kind_no_substring_authority(self) -> None: + """B1/B14: substring lookalikes do not become controller.""" + for name in ( + "fake-controller", + "controller-copy", + "not-controller", + "xcontrollerx", + "myCONTROLLER", ): - 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", + prof = { + "profile_name": name, + "role": "author", + "allowed_operations": ["gitea.read"], + } + self.assertEqual( + gitea_mcp_server._profile_role_kind(prof), + "author", + name, ) - self.assertFalse(res["success"]) - self.assertFalse(res["break_glass_executed"]) - self.assertEqual(res["blocker_kind"], "role_authorization") + # Role substrings must not promote. + for role in ("not-controller", "control-plane", "xcontrollerx"): + prof = {"profile_name": "other", "role": role, "allowed_operations": ["gitea.read"]} + self.assertEqual( + gitea_mcp_server._profile_role_kind(prof), + role, + role, + ) + + def test_prgs_controller_retains_declared_reconciler_role(self) -> None: + """B14: break-glass capability does not redefine global role.""" + prof = _controller_profile(role="reconciler") + self.assertEqual(gitea_mcp_server._profile_role_kind(prof), "reconciler") + # Declared controller still wins when declared. + prof2 = _controller_profile(role="controller") + self.assertEqual(gitea_mcp_server._profile_role_kind(prof2), "controller") + + def test_cleanup_merged_pr_branch_role_resolution_unchanged(self) -> None: + """B14: prgs-controller with reconciler role still resolves for cleanup.""" + # When declared reconciler, cleanup gate's role check should see reconciler. + prof = _controller_profile(role="reconciler") + self.assertEqual(gitea_mcp_server._profile_role_kind(prof), "reconciler") + # Fabricated controller-like names with reconciler ops do not become controller. + fake = { + "profile_name": "fake-controller", + "role": "reconciler", + "allowed_operations": [ + "gitea.read", "gitea.branch.delete", "gitea.pr.close" + ], + } + self.assertEqual(gitea_mcp_server._profile_role_kind(fake), "reconciler") + + def test_narrow_break_glass_capability_does_not_redefine_role(self) -> None: + """B14: granting runtime.break_glass_restart does not invent controller role.""" + prof = { + "profile_name": "prgs-reconciler", + "role": "reconciler", + "allowed_operations": [ + "gitea.read", + "gitea.branch.delete", + "runtime.break_glass_restart", + ], + } + self.assertEqual(gitea_mcp_server._profile_role_kind(prof), "reconciler") + + # ── B2-style input validation (preserved) ───────────────────────────── 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"]} - + prof = _controller_profile() + p_parity, p_runtime, p_switch = _gate_open_patches() 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") + with p_parity, p_runtime, p_switch: + with patch.object(gitea_mcp_server, "get_profile", return_value=prof): + 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") + prof = _controller_profile() + p_parity, p_runtime, p_switch = _gate_open_patches() + with p_parity, p_runtime, p_switch: + with patch.object(gitea_mcp_server, "get_profile", return_value=prof): + 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") + prof = _controller_profile() + p_parity, p_runtime, p_switch = _gate_open_patches() + with p_parity, p_runtime, p_switch: + with patch.object(gitea_mcp_server, "get_profile", return_value=prof): + 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"]} + """B3 preserved: Gate incident creation on gitea.issue.create permission.""" + prof = _controller_profile( + allowed_operations=["gitea.read", "runtime.break_glass_restart"] + ) + p_parity, p_runtime, p_switch = _gate_open_patches() + with p_parity, p_runtime, p_switch: + with patch.object(gitea_mcp_server, "get_profile", return_value=prof): + 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, + dry_run=False, + remote="prgs", + ) + self.assertFalse(res["success"]) + self.assertEqual(res["blocker_kind"], "permission_denied") - 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") + # ── B8: redaction ───────────────────────────────────────────────────── - 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" + def test_redaction_key_value_and_connection_strings(self) -> None: + """B8: key/value, bearer, connection-string, and embedded secrets.""" + cases = [ + ("password=hunter2supersecret", ["hunter2supersecret"], "password"), + ("api_key: sk-live-abcdef1234567890ab", ["sk-live-abcdef1234567890ab"], "api_key"), + ("Server=db;Password=s3cretValue;Uid=sa", ["s3cretValue"], "password"), + ( + "Authorization: Bearer eyJhbGciOiJIUzI1NiJ9.abc.def", + ["eyJhbGciOiJIUzI1NiJ9.abc.def", "Bearer eyJ"], + "authorization", + ), + ( + "token ghp_1234567890abcdef12345678 embedded", + ["ghp_1234567890abcdef12345678"], + "token", + ), + ("nested note password=letmein12345 end", ["letmein12345"], "password"), + ] + for raw, secrets, key in cases: + out = gitea_audit._redact_str(raw) + self.assertIn("[REDACTED]", out, raw) + for secret in secrets: + self.assertNotIn(secret, out, raw) + self.assertIn(key, out.lower(), raw) + nested = gitea_audit.redact({ + "reason": "password=supersecret99", + "items": [{"api_key": "abc123xyz"}, "token ghp_abcdefghijklmnop1234"], + "error": RuntimeError("pwd=nestedSecret99"), + }) + # Exception objects pass through redact as non-str/non-container; ensure + # string forms are covered via str conversion in _redact_str usage. + self.assertEqual(nested["items"][0]["api_key"], gitea_audit.REDACTED) + self.assertNotIn("supersecret99", nested["reason"]) + self.assertNotIn("ghp_abcdefghijklmnop1234", nested["items"][1]) + + def test_redaction_preserves_benign_sec_prefix_text(self) -> None: + """B8: ordinary text beginning with sec- must not be erased.""" + benign = ( + "Emergency restart in sec-primary-region for sector-planning " + "and secondary-health checks" + ) + out = gitea_audit._redact_str(benign) + self.assertIn("sec-primary-region", out) + self.assertIn("sector-planning", out) + self.assertIn("secondary-health", out) + self.assertNotIn("[REDACTED]", out) + + def test_redaction_across_break_glass_surfaces(self) -> None: + """B8: operator reason is redacted on result, incident, and audit surfaces.""" + prof = _controller_profile() + raw_reason = ( + "Emergency restart: password=supersecret99 api_key: " + "sk-live-abcdef1234567890ab 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_supported": True, "apply_authorized": True, "restart_performed": True} - mock_recon = {"success": True} + fake_exec = { + "success": True, + "apply_supported": True, + "apply_authorized": True, + "restart_performed": True, + "break_glass_executed": True, + } + gitea_mcp_server._break_glass_restart_executor = lambda req: fake_exec + p_parity, p_runtime, p_switch = _gate_open_patches() + with p_parity, p_runtime, p_switch: + with patch.object(gitea_mcp_server, "get_profile", return_value=prof), 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={"affected_sessions": []}, + ), patch.object( + gitea_mcp_server, "gitea_reconcile_after_restart", + return_value={"success": True}, + ), 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"], res) + self.assertNotIn("supersecret99", res["reason"]) + self.assertNotIn("sk-live-abcdef1234567890ab", res["reason"]) + self.assertNotIn("user:pass@gitea.prgs.cc", res["reason"]) + posted_body = mock_api_request.call_args[0][3]["body"] + self.assertNotIn("supersecret99", posted_body) + self.assertNotIn("sk-live-abcdef1234567890ab", posted_body) - 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", - ) + # ── B6/B11: reachable executor / truthful flags ─────────────────────── - self.assertTrue(res["success"]) - self.assertNotIn("ghp_secretToken12345", res["reason"]) - self.assertNotIn("user:pass@gitea.prgs.cc", res["reason"]) - self.assertIn("[REDACTED]", res["reason"]) + def test_dry_run_never_executes_and_reports_false(self) -> None: + """B7/B6: dry-run never executes; break_glass_executed always false.""" + prof = _controller_profile() + called = {"n": 0} - # 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 _should_not_run(_req): + called["n"] += 1 + return {"restart_performed": True, "break_glass_executed": True} + + gitea_mcp_server._break_glass_restart_executor = _should_not_run + mock_audit = MagicMock() + mock_api = MagicMock() + p_parity, p_runtime, p_switch = _gate_open_patches() + with p_parity, p_runtime, p_switch: + with patch.object(gitea_mcp_server, "get_profile", return_value=prof), 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 + ), patch.object( + gitea_mcp_server, "api_request", mock_api + ): + 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.assertEqual(called["n"], 0) + mock_audit.assert_not_called() + mock_api.assert_not_called() + + def test_unsupported_apply_truthful(self) -> None: + """B6/B11: unsupported apply returns blocked result, execution false.""" + prof = _controller_profile() + gitea_mcp_server._break_glass_restart_executor = lambda req: { + "success": False, + "apply_supported": False, + "apply_authorized": False, + "restart_performed": False, + "break_glass_executed": False, + "reasons": ["no hook"], + } + p_parity, p_runtime, p_switch = _gate_open_patches() + with p_parity, p_runtime, p_switch: + with patch.object(gitea_mcp_server, "get_profile", return_value=prof), 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={"affected_sessions": []}, + ), 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", + return_value={"number": 555, "title": "[INCIDENT]"}, + ): + res = gitea_mcp_server.gitea_break_glass_restart( + reason="Privileged restart request with unsupported 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_delegation_success_rejection_and_failure(self) -> None: + """B6/B11: distinct terminal states for success / rejection / failure.""" + prof = _controller_profile() + p_parity, p_runtime, p_switch = _gate_open_patches() + + def _run(exec_result, recon=None): + gitea_mcp_server._break_glass_restart_executor = lambda req: exec_result + with p_parity, p_runtime, p_switch: + with patch.object(gitea_mcp_server, "get_profile", return_value=prof), 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={"affected_sessions": []}, + ), patch.object( + gitea_mcp_server, "gitea_reconcile_after_restart", + return_value=recon or {"success": True}, + ), 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", + return_value={"number": 555, "title": "[INCIDENT]"}, + ): + return gitea_mcp_server.gitea_break_glass_restart( + reason="Privileged break-glass restart delegation path", + confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION", + impact_ack=True, + dry_run=False, + create_incident_issue=True, + remote="prgs", + ) + + ok = _run({ + "success": True, + "apply_supported": True, + "apply_authorized": True, + "restart_performed": True, + "break_glass_executed": True, + }) + self.assertTrue(ok["success"], ok) + self.assertTrue(ok["break_glass_executed"]) + self.assertTrue(ok["performed"]) + + rejected = _run({ + "success": False, + "apply_supported": True, + "apply_authorized": False, + "restart_performed": False, + "break_glass_executed": False, + "reasons": ["class denied"], + }) + self.assertFalse(rejected["success"]) + self.assertFalse(rejected["break_glass_executed"]) + self.assertEqual(rejected["blocker_kind"], "restart_delegation_failed") + + failed = _run({ + "success": False, + "apply_supported": True, + "apply_authorized": True, + "restart_performed": False, + "break_glass_executed": False, + "reasons": ["executor error"], + }) + self.assertFalse(failed["success"]) + self.assertFalse(failed["break_glass_executed"]) + self.assertEqual(failed["blocker_kind"], "restart_delegation_failed") + + def test_reconciliation_success_and_failure_truthful_flags(self) -> None: + """B10 preserved: recon failure keeps break_glass_executed=true.""" + prof = _controller_profile() + gitea_mcp_server._break_glass_restart_executor = lambda req: { + "success": True, + "apply_supported": True, + "apply_authorized": True, + "restart_performed": True, + "break_glass_executed": True, + } + p_parity, p_runtime, p_switch = _gate_open_patches() + with p_parity, p_runtime, p_switch: + with patch.object(gitea_mcp_server, "get_profile", return_value=prof), 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={"affected_sessions": []}, + ), patch.object( + gitea_mcp_server, "gitea_reconcile_after_restart", + return_value={"success": False, "error": "lease cleanup failed"}, + ), 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", + return_value={"number": 555, "title": "[INCIDENT]"}, + ): + 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_authorization_is_not_execution(self) -> None: + """B6: apply_authorized alone never sets break_glass_executed.""" + # Default executor without hook → unsupported, not executed. + prof = _controller_profile() + p_parity, p_runtime, p_switch = _gate_open_patches() + with p_parity, p_runtime, p_switch: + with patch.object(gitea_mcp_server, "get_profile", return_value=prof), 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={"affected_sessions": []}, + ), 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", + return_value={"number": 555, "title": "[INCIDENT]"}, + ): + res = gitea_mcp_server.gitea_break_glass_restart( + reason="Privileged restart without host hook configured", + confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION", + impact_ack=True, + dry_run=False, + create_incident_issue=True, + remote="prgs", + ) + self.assertFalse(res["break_glass_executed"]) + self.assertEqual(res["blocker_kind"], "apply_unsupported") + + # ── preserved fail-closed pre-exec (B5/B9) ──────────────────────────── 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() + """B9 preserved: Audit recording failure stops execution fail-closed.""" + prof = _controller_profile() + called = {"n": 0} + gitea_mcp_server._break_glass_restart_executor = lambda req: called.__setitem__("n", called["n"] + 1) or {} + p_parity, p_runtime, p_switch = _gate_open_patches() + with p_parity, p_runtime, p_switch: + with patch.object(gitea_mcp_server, "get_profile", return_value=prof), patch.object( + gitea_mcp_server, "_authenticated_username", return_value="sysadmin" + ), patch.object( + gitea_mcp_server, "gitea_request_mcp_restart", + return_value={"affected_sessions": []}, + ), patch.object( + gitea_audit, "audit_enabled", return_value=True + ), patch.object( + gitea_audit, "write_event", return_value=False + ), patch.object( + gitea_mcp_server, "api_request", MagicMock() + ): + 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") + self.assertEqual(called["n"], 0) 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") + """B5 preserved: Incident creation failure stops execution fail-closed.""" + prof = _controller_profile() + called = {"n": 0} + gitea_mcp_server._break_glass_restart_executor = lambda req: called.__setitem__("n", called["n"] + 1) or {} + p_parity, p_runtime, p_switch = _gate_open_patches() + with p_parity, p_runtime, p_switch: + with patch.object(gitea_mcp_server, "get_profile", return_value=prof), 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={"affected_sessions": []}, + ), 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") + self.assertEqual(called["n"], 0) 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"]) + """B5 preserved: create_incident_issue=False fails closed on real execution.""" + prof = _controller_profile() + p_parity, p_runtime, p_switch = _gate_open_patches() + with p_parity, p_runtime, p_switch: + with patch.object(gitea_mcp_server, "get_profile", return_value=prof), 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_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") + """B9 preserved: Disabling audit recording blocks execution fail-closed.""" + prof = _controller_profile() + p_parity, p_runtime, p_switch = _gate_open_patches() + with p_parity, p_runtime, p_switch: + with patch.object(gitea_mcp_server, "get_profile", return_value=prof), patch.object( + gitea_audit, "audit_enabled", return_value=False + ), patch.object( + gitea_mcp_server, "gitea_request_mcp_restart", + return_value={"affected_sessions": []}, + ): + 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.""" + """B12/B13: capability map registers exact 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") + entry2 = TASK_CAPABILITY_MAP.get("break_glass_restart") + self.assertEqual(entry2["permission"], "runtime.break_glass_restart") if __name__ == "__main__":