"""#709: cross-profile decision-lock cleanup, overwrite protection, recovery. Covers AC1–AC8 plus review-434 F1/F2/F3 and review-435 F3-residual/F4/F5 remediations without fabricating historical PR provenance or special-casing live PR numbers in production code. """ from __future__ import annotations import json import os import subprocess import sys import tempfile import unittest from unittest.mock import patch import irrecoverable_provenance as irp import mcp_session_state as ss import stale_review_decision_lock as srdl def _lock( mutations=None, *, profile="prgs-reviewer", remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", head=None, ): muts = [] for m in mutations or []: row = dict(m) if head and "head_sha" not in row: row["head_sha"] = head muts.append(row) return { "task": "review_pr", "remote": remote, "org": org, "repo": repo, "session_pid": os.getpid(), "session_profile": profile, "session_profile_lock": profile, "profile_identity": profile, "final_review_decision_ready": False, "ready_pr_number": None, "ready_action": None, "ready_expected_head_sha": None, "live_mutations": muts, "correction_authorized": False, "correction_reason": None, "kind": ss.KIND_DECISION_LOCK, } APPROVE = {"pr_number": 100, "action": "approve", "review_id": 9} APPROVE_OTHER = {"pr_number": 200, "action": "approve", "review_id": 10} HEAD_A = "a" * 40 HEAD_B = "b" * 40 RECONCILER_OPS = [ "gitea.read", "gitea.pr.close", "gitea.pr.comment", "gitea.issue.comment", ] DEDICATED_RECOVERY_OPS = RECONCILER_OPS + [ irp.CAPABILITY_IRRECOVERABLE_RECOVERY, ] DURABLE_TEST_HMAC_KEY = "0" * 64 # 32-byte hex durable key for F4 tests # #709 F7 (review 438): canonical incident evidence binds the full recovery # scope, so the fixtures carry stable actor ids, the decision-lock identity, # the recovery action, both heads, the key version, and a replay nonce. DECISION_LOCK_ID = "review_decision_lock-prgs-reviewer" DESTROYED_SUBJECT = "prgs-reviewer terminal approval ledger" RECOVERY_ACTION = irp.RECOVERY_ACTION_IRRECOVERABLE_PROVENANCE INCIDENT_NONCE = "11111111-2222-3333-4444-555555555555" INCIDENT_ISSUED_AT = "2026-07-13T00:00:00+00:00" AUTHOR_LOGIN = "controller-ops" AUTHOR_ID = 4242 MINT_ACTOR_LOGIN = "sysadmin" MINT_ACTOR_ID = 7 def _canonical_incident_body( *, pr_number=42, head=HEAD_A, remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", incident_issue=700, decision_lock_id=DECISION_LOCK_ID, destroyed_subject=DESTROYED_SUBJECT, recovery_action=RECOVERY_ACTION, recorded_head_sha=HEAD_A, evidence_author_id=AUTHOR_ID, evidence_author_login=AUTHOR_LOGIN, mint_actor_id=MINT_ACTOR_ID, mint_actor_login=MINT_ACTOR_LOGIN, key_version=None, nonce=INCIDENT_NONCE, issued_at=INCIDENT_ISSUED_AT, narrative="forensic diagnosis", ): return irp.build_canonical_incident_body( remote=remote, org=org, repo=repo, pr_number=pr_number, decision_lock_id=decision_lock_id, destroyed_subject=destroyed_subject, recovery_action=recovery_action, recorded_head_sha=recorded_head_sha, expected_head_sha=head, incident_issue=incident_issue, evidence_author_id=evidence_author_id, evidence_author_login=evidence_author_login, mint_actor_id=mint_actor_id, mint_actor_login=mint_actor_login, key_version=key_version or irp.auth_key_version(), nonce=nonce, issued_at=issued_at, narrative=narrative, ) def _incident_comment_payload( *, comment_id=11489, author=AUTHOR_LOGIN, author_id=None, pr_number=42, head=HEAD_A, remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", incident_issue=700, body=None, **body_kwargs, ): uid = AUTHOR_ID if author_id is None else author_id return { "id": comment_id, "body": body if body is not None else _canonical_incident_body( pr_number=pr_number, head=head, remote=remote, org=org, repo=repo, incident_issue=incident_issue, evidence_author_id=uid, evidence_author_login=author, **body_kwargs, ), "user": {"id": uid, "login": author}, "created_at": "2026-07-13T00:00:00Z", "updated_at": "2026-07-13T00:00:00Z", "html_url": f"https://gitea.example/{org}/{repo}/issues/{incident_issue}#issuecomment-{comment_id}", } def _mint_auth( *, pr_number=42, head=HEAD_A, remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", incident_issue=700, incident_comment_id=11489, issuer="sysadmin", profile="prgs-reconciler", destroyed_subject=None, ): return irp.build_authorization_artifact( remote=remote, org=org, repo=repo, pr_number=pr_number, expected_head_sha=head, incident_issue=incident_issue, incident_comment_id=incident_comment_id, destroyed_subject=destroyed_subject, issuer_username=issuer, issuer_profile=profile, native_provenance={ "native_mcp_transport": True, "production_native_mcp_transport": False, "pytest": True, "token_fingerprint": "testfp", "entrypoint": "pytest", "pid": os.getpid(), }, ) class TestAC2InitOverwrite(unittest.TestCase): def test_empty_lock_allows_reinit(self): a = srdl.assess_init_overwrite(_lock([]), force=True) self.assertTrue(a["overwrite_allowed"]) def test_terminal_lock_blocks_force_reinit(self): a = srdl.assess_init_overwrite(_lock([APPROVE]), force=True) self.assertFalse(a["overwrite_allowed"]) self.assertTrue(a["has_unresolved_terminal"]) self.assertEqual(a["last_terminal_pr"], 100) def test_none_lock_allows_init(self): a = srdl.assess_init_overwrite(None) self.assertTrue(a["overwrite_allowed"]) class TestAC1TargetApproval(unittest.TestCase): def test_targets_matching_approve(self): self.assertTrue( srdl.lock_targets_merged_pr_approval( _lock([APPROVE], head=HEAD_A), pr_number=100, expected_head_sha=HEAD_A, ) ) def test_rejects_other_pr(self): self.assertFalse( srdl.lock_targets_merged_pr_approval( _lock([APPROVE_OTHER]), pr_number=100 ) ) def test_rejects_head_mismatch(self): self.assertFalse( srdl.lock_targets_merged_pr_approval( _lock([APPROVE], head=HEAD_A), pr_number=100, expected_head_sha=HEAD_B, ) ) def test_f3_residual_rejects_legacy_no_head_when_expected_head_given(self): """Primary approve-match requires recorded-head (#709 F3 residual / 435).""" # APPROVE without head fields — legacy ledger. legacy = _lock([APPROVE]) # no head= self.assertIsNone(srdl.mutation_head_sha(APPROVE, legacy)) self.assertFalse( srdl.lock_targets_merged_pr_approval( legacy, pr_number=100, expected_head_sha=HEAD_A, ) ) # Without expected_head_sha, PR-number-only match still works for # non-destructive callers that do not pass a head pin. self.assertTrue( srdl.lock_targets_merged_pr_approval(legacy, pr_number=100) ) class TestF1AuthorizationNotSelfAssertable(unittest.TestCase): def test_operator_authorized_true_cannot_authorize_via_build(self): rec = srdl.build_irrecoverable_provenance_record( pr_number=42, head_sha=HEAD_A, remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", actor_username="sysadmin", profile_name="prgs-reconciler", reason="evidence destroyed", incident_ref="anything", operator_authorized=True, ) self.assertFalse(rec["applied"]) self.assertFalse(rec["historical_cleanup_proven"]) self.assertFalse(rec["merger_may_accept"]) def test_confirmation_string_not_authorization(self): # Confirmation is only intent text; capability assess ignores it. conf = irp.expected_confirmation(99) self.assertEqual(conf, "IRRECOVERABLE DECISION PROVENANCE PR 99") # Without auth artifact, merger cannot accept. rec = srdl.build_irrecoverable_provenance_record( pr_number=99, head_sha=HEAD_A, remote="prgs", org="o", repo="r", actor_username="x", profile_name="y", reason="r", operator_authorized=False, ) self.assertFalse(rec["merger_may_accept"]) def test_gitea_read_alone_insufficient(self): a = irp.assess_capability_for_irrecoverable_recovery( allowed_operations=["gitea.read"], forbidden_operations=[], role_kind="author", profile_name="prgs-author", ) self.assertFalse(a["allowed"]) def test_expected_head_missing_fails(self): g = irp.assess_live_head_binding( expected_head_sha=None, live_head_sha=HEAD_A, ) self.assertFalse(g["valid"]) def test_expected_head_differs_fails(self): g = irp.assess_live_head_binding( expected_head_sha=HEAD_A, live_head_sha=HEAD_B, ) self.assertFalse(g["valid"]) def test_missing_incident_fails(self): g = irp.assess_incident_evidence( incident_issue=None, incident_comment_id=None, comment_payload=None, ) self.assertFalse(g["valid"]) def test_nonexistent_incident_fails(self): g = irp.assess_incident_evidence( incident_issue=1, incident_comment_id=2, comment_payload=None, comment_lookup_error="404", ) self.assertFalse(g["valid"]) def test_valid_auth_succeeds_exact_scope(self): auth = _mint_auth() v = irp.verify_authorization_artifact( auth, remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", pr_number=42, expected_head_sha=HEAD_A, incident_issue=700, incident_comment_id=11489, ) self.assertTrue(v["valid"], v) rec = irp.build_irrecoverable_provenance_record( pr_number=42, head_sha=HEAD_A, remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", actor_username="sysadmin", profile_name="prgs-reconciler", reason="evidence destroyed", incident_issue=700, incident_comment_id=11489, authorization=auth, ) self.assertTrue(rec["merger_may_accept"]) self.assertFalse(rec["applied"]) body = irp.format_irrecoverable_audit_comment(rec) self.assertIn("applied: `False`", body) def test_wrong_repo_auth_fails(self): auth = _mint_auth(repo="Other-Repo") v = irp.verify_authorization_artifact( auth, remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", pr_number=42, expected_head_sha=HEAD_A, incident_issue=700, incident_comment_id=11489, ) self.assertFalse(v["valid"]) def test_wrong_pr_auth_fails(self): auth = _mint_auth(pr_number=1) v = irp.verify_authorization_artifact( auth, remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", pr_number=42, expected_head_sha=HEAD_A, ) self.assertFalse(v["valid"]) def test_wrong_head_auth_fails(self): auth = _mint_auth(head=HEAD_B) v = irp.verify_authorization_artifact( auth, remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", pr_number=42, expected_head_sha=HEAD_A, ) self.assertFalse(v["valid"]) def test_altered_signature_fails(self): auth = _mint_auth() auth["server_signature"] = "0" * 64 v = irp.verify_authorization_artifact( auth, remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", pr_number=42, expected_head_sha=HEAD_A, incident_issue=700, incident_comment_id=11489, ) self.assertFalse(v["valid"]) def test_fresh_non_pytest_process_cannot_mint_accepted_record(self): """Ordinary Python process: merger_may_accept stays False without server auth.""" script = ( "import stale_review_decision_lock as s\n" "r=s.build_irrecoverable_provenance_record(\n" " pr_number=1, head_sha=None, remote='prgs', org=None, repo=None,\n" " actor_username='x', profile_name='y', reason='r',\n" " incident_ref=None, operator_authorized=True)\n" "print(r.get('merger_may_accept'), r.get('head_sha'))\n" ) env = {k: v for k, v in os.environ.items() if not k.startswith("PYTEST")} env.pop("PYTEST_CURRENT_TEST", None) proc = subprocess.run( [sys.executable, "-c", script], cwd=os.path.dirname(os.path.dirname(os.path.abspath(__file__))), capture_output=True, text=True, env=env, timeout=30, ) self.assertEqual(proc.returncode, 0, proc.stderr) out = (proc.stdout or "").strip() self.assertTrue(out.startswith("False"), msg=out) def test_unauthorized_profile_capability(self): a = irp.assess_capability_for_irrecoverable_recovery( allowed_operations=["gitea.read", "gitea.pr.comment"], forbidden_operations=["gitea.pr.close"], role_kind="author", profile_name="prgs-author", ) self.assertFalse(a["allowed"]) def test_f5_reconciler_equivalence_rejected(self): """Reconciler profile without dedicated capability cannot mint (#709 F5).""" a = irp.assess_capability_for_irrecoverable_recovery( allowed_operations=RECONCILER_OPS, forbidden_operations=[ "gitea.pr.approve", "gitea.pr.merge", "gitea.pr.review", ], role_kind="reconciler", profile_name="prgs-reconciler", ) self.assertFalse(a["allowed"], a) self.assertTrue( any("dedicated" in r for r in a["reasons"]), msg=a["reasons"], ) def test_f5_dedicated_capability_allowed(self): a = irp.assess_capability_for_irrecoverable_recovery( allowed_operations=DEDICATED_RECOVERY_OPS, forbidden_operations=[ "gitea.pr.approve", "gitea.pr.merge", "gitea.pr.review", ], role_kind="reconciler", profile_name="prgs-reconciler", ) self.assertTrue(a["allowed"], a) self.assertEqual(a["via"], "dedicated_capability") class TestF5AuthoritativeIncidentEvidence(unittest.TestCase): def test_any_nonempty_body_rejected(self): g = irp.assess_incident_evidence( incident_issue=700, incident_comment_id=11489, comment_payload={ "id": 11489, "body": "random forensic note without canonical fields", "user": {"login": "controller-ops"}, }, expected_remote="prgs", expected_org="Scaled-Tech-Consulting", expected_repo="Gitea-Tools", expected_pr_number=42, expected_head_sha=HEAD_A, ) self.assertFalse(g["valid"], g) def test_missing_author_rejected(self): body = _canonical_incident_body() g = irp.assess_incident_evidence( incident_issue=700, incident_comment_id=11489, comment_payload={"id": 11489, "body": body, "user": {}}, expected_remote="prgs", expected_org="Scaled-Tech-Consulting", expected_repo="Gitea-Tools", expected_pr_number=42, expected_head_sha=HEAD_A, ) self.assertFalse(g["valid"], g) self.assertTrue(any("author" in r for r in g["reasons"]), g["reasons"]) def test_self_authored_rejected(self): g = irp.assess_incident_evidence( incident_issue=700, incident_comment_id=11489, comment_payload=_incident_comment_payload(author="sysadmin"), expected_remote="prgs", expected_org="Scaled-Tech-Consulting", expected_repo="Gitea-Tools", expected_pr_number=42, expected_head_sha=HEAD_A, mint_actor_username="sysadmin", reject_self_authored=True, ) self.assertFalse(g["valid"], g) self.assertTrue( any("self-authored" in r for r in g["reasons"]), g["reasons"] ) def test_canonical_body_with_independent_author_accepted(self): g = irp.assess_incident_evidence( incident_issue=700, incident_comment_id=11489, comment_payload=_incident_comment_payload(author="controller-ops"), expected_remote="prgs", expected_org="Scaled-Tech-Consulting", expected_repo="Gitea-Tools", expected_pr_number=42, expected_head_sha=HEAD_A, mint_actor_username="sysadmin", reject_self_authored=True, ) self.assertTrue(g["valid"], g) def test_tampered_content_digest_rejected(self): payload = _incident_comment_payload() payload["body"] = payload["body"].replace( "content_digest: ", "content_digest: " + "f" * 64 + "x" ) # Force bad digest line lines = [] for line in payload["body"].splitlines(): if line.startswith("content_digest:"): lines.append("content_digest: " + "0" * 64) else: lines.append(line) payload["body"] = "\n".join(lines) g = irp.assess_incident_evidence( incident_issue=700, incident_comment_id=11489, comment_payload=payload, expected_remote="prgs", expected_org="Scaled-Tech-Consulting", expected_repo="Gitea-Tools", expected_pr_number=42, expected_head_sha=HEAD_A, mint_actor_username="sysadmin", ) self.assertFalse(g["valid"], g) self.assertTrue( any("content_digest" in r for r in g["reasons"]), g["reasons"] ) class TestF4DurableHmacKey(unittest.TestCase): def tearDown(self): os.environ.pop(irp.ENV_AUTH_HMAC_KEY, None) os.environ.pop(irp.ENV_AUTH_HMAC_KEY_VERSION, None) irp.reset_process_auth_secret_for_tests() def test_key_version_bound_into_artifact(self): irp.reset_process_auth_secret_for_tests() auth = _mint_auth() self.assertIn("key_version", auth) self.assertTrue(auth["key_version"]) self.assertNotIn("server_secret", auth) self.assertNotIn("hmac_key", auth) def test_production_fails_closed_without_durable_key(self): """Non-pytest process without env key must not generate ephemeral secret.""" script = ( "import os, sys\n" "os.environ.pop('PYTEST_CURRENT_TEST', None)\n" "os.environ.pop('GITEA_IRRECOVERABLE_AUTH_HMAC_KEY', None)\n" # Force non-pytest path by patching guard after import. "import mcp_daemon_guard as g\n" "g.is_pytest_runtime = lambda: False\n" "import irrecoverable_provenance as irp\n" "irp.reset_process_auth_secret_for_tests()\n" "try:\n" " irp._process_secret()\n" " print('UNEXPECTED_OK')\n" "except irp.AuthSecretError as e:\n" " print('FAIL_CLOSED', 'ephemeral' in str(e).lower() or 'required' in str(e).lower())\n" ) env = {k: v for k, v in os.environ.items() if not k.startswith("PYTEST")} env.pop("PYTEST_CURRENT_TEST", None) env.pop(irp.ENV_AUTH_HMAC_KEY, None) proc = subprocess.run( [sys.executable, "-c", script], cwd=os.path.dirname(os.path.dirname(os.path.abspath(__file__))), capture_output=True, text=True, env=env, timeout=30, ) self.assertEqual(proc.returncode, 0, proc.stderr) out = (proc.stdout or "").strip() self.assertTrue(out.startswith("FAIL_CLOSED"), msg=out) def test_cross_process_verify_with_durable_key(self): """Mint in process A, verify in process B with same durable key (#709 F4).""" import json as _json mint_script = ( "import json, os, irrecoverable_provenance as irp\n" "irp.reset_process_auth_secret_for_tests()\n" "auth = irp.build_authorization_artifact(\n" " remote='prgs', org='o', repo='r', pr_number=10,\n" f" expected_head_sha={HEAD_A!r}, incident_issue=1, incident_comment_id=2,\n" " destroyed_subject=None, issuer_username='sysadmin',\n" " issuer_profile='prgs-reconciler',\n" " native_provenance={'native_mcp_transport': True, 'pytest': True,\n" " 'token_fingerprint': 'fp', 'entrypoint': 'pytest', 'pid': 1})\n" "print(json.dumps(auth))\n" ) env = dict(os.environ) env[irp.ENV_AUTH_HMAC_KEY] = DURABLE_TEST_HMAC_KEY env[irp.ENV_AUTH_HMAC_KEY_VERSION] = "test-v1" mint = subprocess.run( [sys.executable, "-c", mint_script], cwd=os.path.dirname(os.path.dirname(os.path.abspath(__file__))), capture_output=True, text=True, env=env, timeout=30, ) self.assertEqual(mint.returncode, 0, mint.stderr) auth = _json.loads(mint.stdout.strip()) self.assertEqual(auth.get("key_version"), "test-v1") verify_script = ( "import json, sys, irrecoverable_provenance as irp\n" "irp.reset_process_auth_secret_for_tests()\n" "auth = json.loads(sys.stdin.read())\n" "v = irp.verify_authorization_artifact(\n" " auth, remote='prgs', org='o', repo='r', pr_number=10,\n" f" expected_head_sha={HEAD_A!r}, incident_issue=1, incident_comment_id=2)\n" "print(v.get('valid'), v.get('reasons'))\n" ) verify = subprocess.run( [sys.executable, "-c", verify_script], cwd=os.path.dirname(os.path.dirname(os.path.abspath(__file__))), input=_json.dumps(auth), capture_output=True, text=True, env=env, timeout=30, ) self.assertEqual(verify.returncode, 0, verify.stderr) self.assertTrue( (verify.stdout or "").strip().startswith("True"), msg=verify.stdout + verify.stderr, ) # Different durable key must fail verification (cross-process mismatch). env_bad = dict(env) env_bad[irp.ENV_AUTH_HMAC_KEY] = "1" * 64 verify_bad = subprocess.run( [sys.executable, "-c", verify_script], cwd=os.path.dirname(os.path.dirname(os.path.abspath(__file__))), input=_json.dumps(auth), capture_output=True, text=True, env=env_bad, timeout=30, ) self.assertEqual(verify_bad.returncode, 0, verify_bad.stderr) self.assertTrue( (verify_bad.stdout or "").strip().startswith("False"), msg=verify_bad.stdout, ) class TestF2MergerConsumer(unittest.TestCase): def test_merger_rejects_without_valid_auth(self): rec = srdl.build_irrecoverable_provenance_record( pr_number=10, head_sha=HEAD_A, remote="prgs", org="o", repo="r", actor_username="a", profile_name="p", reason="x", operator_authorized=True, ) a = irp.assess_merger_consumption( rec, None, remote="prgs", org="o", repo="r", pr_number=10, live_head_sha=HEAD_A, approval_at_current_head=True, has_blocking_change_requests=False, ) self.assertFalse(a["allowed"]) def test_merger_rejects_wrong_head(self): auth = _mint_auth(pr_number=10, head=HEAD_A, org="o", repo="r") rec = irp.build_irrecoverable_provenance_record( pr_number=10, head_sha=HEAD_A, remote="prgs", org="o", repo="r", actor_username="a", profile_name="p", reason="x", incident_issue=700, incident_comment_id=1, authorization=auth, ) a = irp.assess_merger_consumption( rec, auth, remote="prgs", org="o", repo="r", pr_number=10, live_head_sha=HEAD_B, approval_at_current_head=True, has_blocking_change_requests=False, ) self.assertFalse(a["allowed"]) def test_merger_rejects_replayed_auth(self): auth = _mint_auth(pr_number=10, head=HEAD_A, org="o", repo="r", incident_comment_id=1) consumed = irp.mark_consumed(auth, consumer_username="m", consumer_profile="merger") rec = irp.build_irrecoverable_provenance_record( pr_number=10, head_sha=HEAD_A, remote="prgs", org="o", repo="r", actor_username="a", profile_name="p", reason="x", incident_issue=700, incident_comment_id=1, authorization=auth, # original unconsumed for record build ) a = irp.assess_merger_consumption( rec, consumed, remote="prgs", org="o", repo="r", pr_number=10, live_head_sha=HEAD_A, approval_at_current_head=True, has_blocking_change_requests=False, ) self.assertFalse(a["allowed"]) def test_recovery_cannot_bypass_missing_approval(self): auth = _mint_auth(pr_number=10, head=HEAD_A, org="o", repo="r", incident_comment_id=1) rec = irp.build_irrecoverable_provenance_record( pr_number=10, head_sha=HEAD_A, remote="prgs", org="o", repo="r", actor_username="a", profile_name="p", reason="x", incident_issue=700, incident_comment_id=1, authorization=auth, ) a = irp.assess_merger_consumption( rec, auth, remote="prgs", org="o", repo="r", pr_number=10, live_head_sha=HEAD_A, approval_at_current_head=False, has_blocking_change_requests=False, ) self.assertFalse(a["allowed"]) self.assertTrue(any("approval" in r for r in a["reasons"])) def test_recovery_cannot_bypass_blocking_crs(self): auth = _mint_auth(pr_number=10, head=HEAD_A, org="o", repo="r", incident_comment_id=1) rec = irp.build_irrecoverable_provenance_record( pr_number=10, head_sha=HEAD_A, remote="prgs", org="o", repo="r", actor_username="a", profile_name="p", reason="x", incident_issue=700, incident_comment_id=1, authorization=auth, ) a = irp.assess_merger_consumption( rec, auth, remote="prgs", org="o", repo="r", pr_number=10, live_head_sha=HEAD_A, approval_at_current_head=True, has_blocking_change_requests=True, ) self.assertFalse(a["allowed"]) def test_valid_recovery_resolves_only_prior_provenance(self): auth = _mint_auth(pr_number=10, head=HEAD_A, org="o", repo="r", incident_comment_id=1) rec = irp.build_irrecoverable_provenance_record( pr_number=10, head_sha=HEAD_A, remote="prgs", org="o", repo="r", actor_username="a", profile_name="p", reason="x", incident_issue=700, incident_comment_id=1, authorization=auth, ) a = irp.assess_merger_consumption( rec, auth, remote="prgs", org="o", repo="r", pr_number=10, live_head_sha=HEAD_A, approval_at_current_head=True, has_blocking_change_requests=False, mergeable=True, lease_ok=True, runtime_ok=True, workspace_ok=True, anti_stomp_ok=True, ) self.assertTrue(a["allowed"], a) self.assertTrue(a["resolves_prior_provenance_blocker"]) self.assertFalse(a["historical_cleanup_proven"]) def test_duplicate_consume_idempotent(self): auth = _mint_auth(pr_number=10, head=HEAD_A, org="o", repo="r", incident_comment_id=1) rec = irp.build_irrecoverable_provenance_record( pr_number=10, head_sha=HEAD_A, remote="prgs", org="o", repo="r", actor_username="a", profile_name="p", reason="x", incident_issue=700, incident_comment_id=1, authorization=auth, ) consumed_auth = irp.mark_consumed(auth, consumer_username="m", consumer_profile="mer") consumed_rec = irp.mark_consumed(rec, consumer_username="m", consumer_profile="mer") a = irp.assess_merger_consumption( consumed_rec, consumed_auth, remote="prgs", org="o", repo="r", pr_number=10, live_head_sha=HEAD_A, approval_at_current_head=True, has_blocking_change_requests=False, ) self.assertTrue(a["allowed"], a) self.assertTrue(a["recovery_record_consumed"]) class TestF3ExactScopeEnforcement(unittest.TestCase): def setUp(self): self._tmp = tempfile.TemporaryDirectory() self.state_dir = self._tmp.name os.chmod(self.state_dir, 0o700) def tearDown(self): self._tmp.cleanup() def test_same_pr_other_remote_not_loaded(self): ss.save_state( kind=ss.KIND_DECISION_LOCK, payload=_lock([APPROVE], profile="prgs-reviewer", remote="other"), remote="other", org="Scaled-Tech-Consulting", repo="Gitea-Tools", profile_identity="prgs-reviewer", state_dir=self.state_dir, ) loaded = ss.load_state_for_profile( kind=ss.KIND_DECISION_LOCK, profile_identity="prgs-reviewer", remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", state_dir=self.state_dir, skip_identity_match=True, ) self.assertIsNone(loaded) def test_same_pr_other_org_not_loaded(self): ss.save_state( kind=ss.KIND_DECISION_LOCK, payload=_lock( [APPROVE], profile="prgs-reviewer", org="Other-Org", ), remote="prgs", org="Other-Org", repo="Gitea-Tools", profile_identity="prgs-reviewer", state_dir=self.state_dir, ) loaded = ss.load_state_for_profile( kind=ss.KIND_DECISION_LOCK, profile_identity="prgs-reviewer", remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", state_dir=self.state_dir, skip_identity_match=True, ) self.assertIsNone(loaded) def test_same_pr_other_repo_not_loaded(self): ss.save_state( kind=ss.KIND_DECISION_LOCK, payload=_lock( [APPROVE], profile="prgs-reviewer", repo="Other-Repo", ), remote="prgs", org="Scaled-Tech-Consulting", repo="Other-Repo", profile_identity="prgs-reviewer", state_dir=self.state_dir, ) loaded = ss.load_state_for_profile( kind=ss.KIND_DECISION_LOCK, profile_identity="prgs-reviewer", remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", state_dir=self.state_dir, skip_identity_match=True, ) self.assertIsNone(loaded) def test_path_traversal_profile_fails(self): g = irp.assess_profile_path_identity("../evil") self.assertFalse(g["valid"]) loaded = ss.load_state_for_profile( kind=ss.KIND_DECISION_LOCK, profile_identity="../evil", remote="prgs", org="o", repo="r", state_dir=self.state_dir, skip_identity_match=True, ) self.assertIsNone(loaded) def test_malformed_legacy_missing_repo_fails_closed(self): # Record without repo identity when caller requires repo. payload = _lock([APPROVE], profile="prgs-reviewer") del payload["repo"] ss.save_state( kind=ss.KIND_DECISION_LOCK, payload=payload, remote="prgs", org="Scaled-Tech-Consulting", repo=None, profile_identity="prgs-reviewer", state_dir=self.state_dir, ) loaded = ss.load_state_for_profile( kind=ss.KIND_DECISION_LOCK, profile_identity="prgs-reviewer", remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", state_dir=self.state_dir, skip_identity_match=True, ) self.assertIsNone(loaded) def test_list_and_load_foreign_profile_lock(self): ss.save_state( kind=ss.KIND_DECISION_LOCK, payload=_lock([APPROVE], profile="prgs-reviewer"), remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", profile_identity="prgs-reviewer", state_dir=self.state_dir, ) ss.save_state( kind=ss.KIND_DECISION_LOCK, payload=_lock([], profile="prgs-merger"), remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", profile_identity="prgs-merger", state_dir=self.state_dir, ) foreign = ss.load_state_for_profile( kind=ss.KIND_DECISION_LOCK, profile_identity="prgs-reviewer", remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", state_dir=self.state_dir, skip_identity_match=True, ) self.assertIsNotNone(foreign) self.assertTrue( srdl.lock_targets_merged_pr_approval(foreign, pr_number=100) ) class TestAC3PostMergeRecoveryRecord(unittest.TestCase): def test_recovery_record_is_not_applied_cleanup(self): rec = srdl.build_post_merge_recovery_record( pr_number=10, head_sha=HEAD_A, merge_commit_sha="m" * 40, target_profile_identity="prgs-reviewer", failed_step="audit_comment_publish", error="timeout", remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", actor_username="sysadmin", profile_name="prgs-merger", ) self.assertEqual(rec["status"], "recovery_required") self.assertFalse(rec["applied"]) self.assertTrue(rec["recovery_critical"]) class TestSessionStateTTL(unittest.TestCase): def test_recovery_critical_kinds_ttl_exempt(self): auth = _mint_auth(pr_number=1) rec = irp.build_irrecoverable_provenance_record( pr_number=1, head_sha=HEAD_A, remote="prgs", org="o", repo="r", actor_username="a", profile_name="p", reason="gone", incident_issue=700, incident_comment_id=1, authorization=auth, ) rec["kind"] = ss.KIND_IRRECOVERABLE_DECISION_PROVENANCE rec["recorded_at"] = "2000-01-01T00:00:00Z" rec["updated_at"] = rec["recorded_at"] rec["profile_identity"] = "prgs-reconciler" rec["session_profile_lock"] = "prgs-reconciler" reasons = ss.identity_match_reasons( rec, profile_identity="prgs-reconciler" ) self.assertFalse(any("expired" in r for r in reasons), msg=reasons) class TestInitReviewDecisionLockIntegration(unittest.TestCase): def setUp(self): self._tmp = tempfile.TemporaryDirectory() self.env = patch.dict( os.environ, { "GITEA_MCP_SESSION_STATE_DIR": self._tmp.name, "GITEA_SESSION_PROFILE_LOCK": "prgs-reviewer", "GITEA_PROFILE_NAME": "prgs-reviewer", }, clear=False, ) self.env.start() import mcp_server self.mcp = mcp_server self.mcp._REVIEW_DECISION_LOCK = None def tearDown(self): self.mcp._REVIEW_DECISION_LOCK = None self.env.stop() self._tmp.cleanup() def test_init_does_not_wipe_terminal_ledger(self): self.mcp._save_review_decision_lock( _lock([APPROVE], profile="prgs-reviewer") ) self.mcp.init_review_decision_lock("prgs", "review_pr", force=True) loaded = self.mcp._load_review_decision_lock() self.assertIsNotNone(loaded) last = srdl.last_terminal_mutation(loaded) self.assertIsNotNone(last) self.assertEqual(last.get("pr_number"), 100) def test_init_creates_empty_when_no_terminal(self): self.mcp._save_review_decision_lock(None) self.mcp.init_review_decision_lock("prgs", "review_pr", force=True) loaded = self.mcp._load_review_decision_lock() self.assertIsNotNone(loaded) self.assertEqual(loaded.get("live_mutations"), []) class TestIrrecoverableToolF1(unittest.TestCase): def setUp(self): self._tmp = tempfile.TemporaryDirectory() self.env = patch.dict( os.environ, { "GITEA_MCP_SESSION_STATE_DIR": self._tmp.name, "GITEA_SESSION_PROFILE_LOCK": "prgs-reconciler", "GITEA_PROFILE_NAME": "prgs-reconciler", "GITEA_ALLOWED_OPERATIONS": ",".join(DEDICATED_RECOVERY_OPS), }, clear=False, ) self.env.start() import mcp_server self.mcp = mcp_server def tearDown(self): self.env.stop() self._tmp.cleanup() def _profile(self): return { "profile_name": "prgs-reconciler", "role": "reconciler", "allowed_operations": DEDICATED_RECOVERY_OPS, "forbidden_operations": [ "gitea.pr.approve", "gitea.pr.merge", "gitea.pr.review", ], } def test_operator_authorized_true_rejected(self): with patch.object(self.mcp, "get_profile", return_value=self._profile()): r = self.mcp.gitea_record_irrecoverable_decision_lock_provenance( pr_number=50, reason="lost", confirmation=irp.expected_confirmation(50), operator_authorized=True, expected_head_sha=HEAD_A, incident_issue=700, incident_comment_id=11489, remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", post_audit_comment=False, ) self.assertFalse(r["success"]) self.assertTrue(any("operator_authorized" in x for x in r["reasons"])) def test_missing_expected_head_fails(self): with patch.object(self.mcp, "get_profile", return_value=self._profile()), patch.object( self.mcp, "_authenticated_username", return_value="sysadmin" ), patch.object( self.mcp, "_irrecoverable_capability_gate", return_value=None ), patch.object( irp, "assess_transport_for_auth_mint", return_value={"allowed": True, "reasons": []} ), patch.object( self.mcp, "_resolve", return_value=("h", "Scaled-Tech-Consulting", "Gitea-Tools") ): r = self.mcp.gitea_record_irrecoverable_decision_lock_provenance( pr_number=50, reason="lost", confirmation=irp.expected_confirmation(50), expected_head_sha=None, incident_issue=700, incident_comment_id=11489, remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", post_audit_comment=False, ) self.assertFalse(r["success"]) def test_wrong_confirmation_fails(self): with patch.object(self.mcp, "get_profile", return_value=self._profile()), patch.object( self.mcp, "_irrecoverable_capability_gate", return_value=None ), patch.object( irp, "assess_transport_for_auth_mint", return_value={"allowed": True, "reasons": []} ), patch.object( self.mcp, "_resolve", return_value=("h", "o", "r") ): r = self.mcp.gitea_record_irrecoverable_decision_lock_provenance( pr_number=50, reason="x", confirmation=irp.expected_confirmation(51), expected_head_sha=HEAD_A, incident_issue=1, incident_comment_id=2, remote="prgs", post_audit_comment=False, ) self.assertFalse(r["success"]) def test_records_with_server_auth(self): auth = _mint_auth( pr_number=50, head=HEAD_A, org="Scaled-Tech-Consulting", repo="Gitea-Tools", incident_issue=700, incident_comment_id=11489, ) auth_profile = irp.auth_state_profile_identity( remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", pr_number=50, expected_head_sha=HEAD_A, ) ss.save_state( kind=ss.KIND_IRRECOVERABLE_PROVENANCE_AUTH, payload=auth, remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", profile_identity=auth_profile, state_dir=self._tmp.name, ) def _api(method, url, auth=None, data=None, **kwargs): if "/pulls/" in str(url): return {"head": {"sha": HEAD_A}, "state": "open"} if "/issues/comments/" in str(url): return _incident_comment_payload( comment_id=11489, author="controller-ops", pr_number=50, head=HEAD_A, remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", incident_issue=700, ) raise AssertionError(f"unexpected API {method} {url}") with patch.object(self.mcp, "get_profile", return_value=self._profile()), patch.object( self.mcp, "_authenticated_username", return_value="sysadmin" ), patch.object( self.mcp, "_irrecoverable_capability_gate", return_value=None ), patch.object( irp, "assess_transport_for_auth_mint", return_value={"allowed": True, "reasons": []} ), patch.object( self.mcp, "_resolve", return_value=("h", "Scaled-Tech-Consulting", "Gitea-Tools") ), patch.object( self.mcp, "_auth", return_value={"Authorization": "token test"} ), patch.object( self.mcp, "api_request", side_effect=_api ), patch.object( self.mcp, "repo_api_url", return_value="https://example.test/api/v1/repos/o/r" ): r = self.mcp.gitea_record_irrecoverable_decision_lock_provenance( pr_number=50, reason="terminal evidence overwritten", confirmation=irp.expected_confirmation(50), expected_head_sha=HEAD_A, incident_issue=700, incident_comment_id=11489, remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", post_audit_comment=False, ) self.assertTrue(r["success"], r) self.assertFalse(r["applied"]) self.assertFalse(r["historical_cleanup_proven"]) self.assertTrue(r["merger_may_accept"]) self.assertEqual(r["record"]["status"], "provenance_irrecoverable") # Idempotent with patch.object(self.mcp, "get_profile", return_value=self._profile()), patch.object( self.mcp, "_authenticated_username", return_value="sysadmin" ), patch.object( self.mcp, "_irrecoverable_capability_gate", return_value=None ), patch.object( irp, "assess_transport_for_auth_mint", return_value={"allowed": True, "reasons": []} ), patch.object( self.mcp, "_resolve", return_value=("h", "Scaled-Tech-Consulting", "Gitea-Tools") ), patch.object( self.mcp, "_auth", return_value={"Authorization": "token test"} ), patch.object( self.mcp, "api_request", side_effect=_api ), patch.object( self.mcp, "repo_api_url", return_value="https://example.test/api/v1/repos/o/r" ): r2 = self.mcp.gitea_record_irrecoverable_decision_lock_provenance( pr_number=50, reason="terminal evidence overwritten", confirmation=irp.expected_confirmation(50), expected_head_sha=HEAD_A, incident_issue=700, incident_comment_id=11489, remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", post_audit_comment=False, ) self.assertTrue(r2["success"], r2) self.assertFalse(r2["performed"]) class TestClearProfileHelperF3(unittest.TestCase): def setUp(self): self._tmp = tempfile.TemporaryDirectory() self.env = patch.dict( os.environ, { "GITEA_MCP_SESSION_STATE_DIR": self._tmp.name, "GITEA_SESSION_PROFILE_LOCK": "prgs-merger", }, clear=False, ) self.env.start() import mcp_server self.mcp = mcp_server self.mcp._REVIEW_DECISION_LOCK = None def tearDown(self): self.mcp._REVIEW_DECISION_LOCK = None self.env.stop() self._tmp.cleanup() def test_clear_only_matching_reviewer_approve(self): ss.save_state( kind=ss.KIND_DECISION_LOCK, payload=_lock([APPROVE], profile="prgs-reviewer", head=HEAD_A), remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", profile_identity="prgs-reviewer", state_dir=self._tmp.name, ) ss.save_state( kind=ss.KIND_DECISION_LOCK, payload=_lock([], profile="prgs-merger"), remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", profile_identity="prgs-merger", state_dir=self._tmp.name, ) out = self.mcp._clear_decision_lock_for_profile( profile_identity="prgs-reviewer", pr_number=100, expected_head_sha=HEAD_A, remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", ) self.assertTrue(out["cleared"], out) skip = self.mcp._clear_decision_lock_for_profile( profile_identity="prgs-merger", pr_number=100, expected_head_sha=HEAD_A, remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", ) self.assertFalse(skip["cleared"]) def test_pr_number_only_fallback_impossible(self): ss.save_state( kind=ss.KIND_DECISION_LOCK, payload=_lock([APPROVE], profile="prgs-reviewer", head=HEAD_A), remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", profile_identity="prgs-reviewer", state_dir=self._tmp.name, ) # Missing expected_head_sha out = self.mcp._clear_decision_lock_for_profile( profile_identity="prgs-reviewer", pr_number=100, expected_head_sha=None, remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", ) self.assertFalse(out["cleared"]) self.assertIn("PR-number-only", out["reason"]) def test_wrong_head_not_cleared(self): ss.save_state( kind=ss.KIND_DECISION_LOCK, payload=_lock([APPROVE], profile="prgs-reviewer", head=HEAD_A), remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", profile_identity="prgs-reviewer", state_dir=self._tmp.name, ) out = self.mcp._clear_decision_lock_for_profile( profile_identity="prgs-reviewer", pr_number=100, expected_head_sha=HEAD_B, remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", ) self.assertFalse(out["cleared"]) def test_cross_repo_same_pr_number_not_cleared(self): ss.save_state( kind=ss.KIND_DECISION_LOCK, payload=_lock( [APPROVE], profile="prgs-reviewer", head=HEAD_A, repo="Other-Repo", ), remote="prgs", org="Scaled-Tech-Consulting", repo="Other-Repo", profile_identity="prgs-reviewer", state_dir=self._tmp.name, ) out = self.mcp._clear_decision_lock_for_profile( profile_identity="prgs-reviewer", pr_number=100, expected_head_sha=HEAD_A, remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", ) self.assertFalse(out["cleared"]) # Original lock still present under Other-Repo scope still = ss.load_state_for_profile( kind=ss.KIND_DECISION_LOCK, profile_identity="prgs-reviewer", remote="prgs", org="Scaled-Tech-Consulting", repo="Other-Repo", state_dir=self._tmp.name, skip_identity_match=True, ) self.assertIsNotNone(still) def _reorder_canonical_lines(body, first, second): """Swap two canonical field lines, leaving the digest untouched.""" lines = body.split("\n") i = next(i for i, l in enumerate(lines) if l.startswith(f"{first}: ")) j = next(i for i, l in enumerate(lines) if l.startswith(f"{second}: ")) lines[i], lines[j] = lines[j], lines[i] return "\n".join(lines) def _insert_after_canonical_line(body, after_field, extra_line): lines = body.split("\n") i = next(i for i, l in enumerate(lines) if l.startswith(f"{after_field}: ")) lines.insert(i + 1, extra_line) return "\n".join(lines) class TestF6KeyVersionFailsClosed(unittest.TestCase): """#709 F6 (review 438): key-version validation must fail closed.""" def _verify(self, auth, **kwargs): params = { "remote": "prgs", "org": "Scaled-Tech-Consulting", "repo": "Gitea-Tools", "pr_number": 42, "expected_head_sha": HEAD_A, } params.update(kwargs) return irp.verify_authorization_artifact(auth, **params) def test_valid_artifact_verifies(self): self.assertTrue(self._verify(_mint_auth())["valid"]) def test_missing_key_version_rejected(self): auth = _mint_auth() del auth["key_version"] v = self._verify(auth) self.assertFalse(v["valid"], v) self.assertTrue( any("missing key_version" in r for r in v["reasons"]), v["reasons"] ) def test_empty_key_version_rejected(self): auth = _mint_auth() auth["key_version"] = "" v = self._verify(auth) self.assertFalse(v["valid"], v) self.assertTrue(any("empty" in r for r in v["reasons"]), v["reasons"]) def test_unknown_key_version_rejected(self): auth = _mint_auth() auth["key_version"] = "totally-unknown-version" v = self._verify(auth) self.assertFalse(v["valid"], v) self.assertTrue( any("unknown or does not match" in r for r in v["reasons"]), v["reasons"] ) def test_malformed_key_version_rejected(self): auth = _mint_auth() auth["key_version"] = "bad version!" v = self._verify(auth) self.assertFalse(v["valid"], v) self.assertTrue(any("malformed" in r for r in v["reasons"]), v["reasons"]) def test_non_string_key_version_rejected(self): auth = _mint_auth() auth["key_version"] = ["v1", "v2"] v = self._verify(auth) self.assertFalse(v["valid"], v) def test_duplicate_key_version_fields_rejected(self): auth = _mint_auth() auth["keyVersion"] = auth["key_version"] # identical value, still duplicate v = self._verify(auth) self.assertFalse(v["valid"], v) self.assertTrue( any("duplicate key-version" in r for r in v["reasons"]), v["reasons"] ) def test_duplicate_conflicting_key_version_fields_rejected(self): auth = _mint_auth() auth["auth_key_version"] = "v9" v = self._verify(auth) self.assertFalse(v["valid"], v) self.assertTrue( any("duplicate key-version" in r for r in v["reasons"]), v["reasons"] ) def test_nested_key_version_counts_as_duplicate(self): auth = _mint_auth() auth["native_provenance"] = dict(auth["native_provenance"]) auth["native_provenance"]["key_version"] = auth["key_version"] v = self._verify(auth) self.assertFalse(v["valid"], v) def test_rotation_invalidates_prior_version_artifact(self): """Rotating the configured version must reject artifacts minted under the old one.""" auth = _mint_auth() self.assertTrue(self._verify(auth)["valid"]) irp.reset_process_auth_secret_for_tests() try: with patch.dict( os.environ, { irp.ENV_AUTH_HMAC_KEY: DURABLE_TEST_HMAC_KEY, irp.ENV_AUTH_HMAC_KEY_VERSION: "v2", }, clear=False, ): v = self._verify(auth) self.assertFalse(v["valid"], v) self.assertTrue( any("does not match the configured" in r for r in v["reasons"]), v["reasons"], ) finally: irp.reset_process_auth_secret_for_tests() def test_wrong_version_fails_even_when_key_unchanged(self): """Version mismatch alone fails: the durable key staying the same is not enough.""" irp.reset_process_auth_secret_for_tests() try: with patch.dict( os.environ, { irp.ENV_AUTH_HMAC_KEY: DURABLE_TEST_HMAC_KEY, irp.ENV_AUTH_HMAC_KEY_VERSION: "v1", }, clear=False, ): auth = _mint_auth() self.assertEqual(auth["key_version"], "v1") self.assertTrue(self._verify(auth)["valid"]) irp.reset_process_auth_secret_for_tests() with patch.dict( os.environ, { irp.ENV_AUTH_HMAC_KEY: DURABLE_TEST_HMAC_KEY, # same key irp.ENV_AUTH_HMAC_KEY_VERSION: "v2", # rotated version }, clear=False, ): self.assertFalse(self._verify(auth)["valid"]) finally: irp.reset_process_auth_secret_for_tests() def test_wrong_key_fails_after_restart(self): """A different durable key must reject the artifact even at the same version.""" irp.reset_process_auth_secret_for_tests() try: with patch.dict( os.environ, { irp.ENV_AUTH_HMAC_KEY: DURABLE_TEST_HMAC_KEY, irp.ENV_AUTH_HMAC_KEY_VERSION: "v1", }, clear=False, ): auth = _mint_auth() irp.reset_process_auth_secret_for_tests() # simulate restart with patch.dict( os.environ, { irp.ENV_AUTH_HMAC_KEY: "1" * 64, # different durable key irp.ENV_AUTH_HMAC_KEY_VERSION: "v1", # same version }, clear=False, ): v = self._verify(auth) self.assertFalse(v["valid"], v) self.assertTrue( any("server_signature invalid" in r for r in v["reasons"]), v["reasons"], ) finally: irp.reset_process_auth_secret_for_tests() def test_same_durable_key_verifies_after_restart(self): irp.reset_process_auth_secret_for_tests() try: env = { irp.ENV_AUTH_HMAC_KEY: DURABLE_TEST_HMAC_KEY, irp.ENV_AUTH_HMAC_KEY_VERSION: "v1", } with patch.dict(os.environ, env, clear=False): auth = _mint_auth() irp.reset_process_auth_secret_for_tests() # simulate restart with patch.dict(os.environ, env, clear=False): self.assertTrue(self._verify(auth)["valid"]) finally: irp.reset_process_auth_secret_for_tests() def test_key_version_is_inside_authenticated_data(self): """Editing key_version must break the MAC, not just the version check.""" irp.reset_process_auth_secret_for_tests() try: with patch.dict( os.environ, { irp.ENV_AUTH_HMAC_KEY: DURABLE_TEST_HMAC_KEY, irp.ENV_AUTH_HMAC_KEY_VERSION: "v1", }, clear=False, ): auth = _mint_auth() signature_v1 = auth["server_signature"] irp.reset_process_auth_secret_for_tests() with patch.dict( os.environ, { irp.ENV_AUTH_HMAC_KEY: DURABLE_TEST_HMAC_KEY, irp.ENV_AUTH_HMAC_KEY_VERSION: "v2", }, clear=False, ): rotated = _mint_auth() # Same key + same scope, different version => different MAC. self.assertNotEqual(signature_v1, rotated["server_signature"]) finally: irp.reset_process_auth_secret_for_tests() def test_production_requires_configured_key_version(self): irp.reset_process_auth_secret_for_tests() try: with patch.object( irp.mcp_daemon_guard, "is_pytest_runtime", return_value=False ), patch.dict( os.environ, {irp.ENV_AUTH_HMAC_KEY: DURABLE_TEST_HMAC_KEY}, clear=False ): os.environ.pop(irp.ENV_AUTH_HMAC_KEY_VERSION, None) with self.assertRaises(irp.AuthSecretError) as ctx: irp.auth_key_version() self.assertIn(irp.ENV_AUTH_HMAC_KEY_VERSION, str(ctx.exception)) finally: irp.reset_process_auth_secret_for_tests() def test_production_requires_durable_key(self): irp.reset_process_auth_secret_for_tests() try: with patch.object( irp.mcp_daemon_guard, "is_pytest_runtime", return_value=False ), patch.dict(os.environ, {}, clear=False): os.environ.pop(irp.ENV_AUTH_HMAC_KEY, None) with self.assertRaises(irp.AuthSecretError): irp.auth_key_version() finally: irp.reset_process_auth_secret_for_tests() def test_errors_never_leak_key_material(self): irp.reset_process_auth_secret_for_tests() try: secret = "s3cr3t" + "9" * 58 with patch.dict( os.environ, { irp.ENV_AUTH_HMAC_KEY: secret, irp.ENV_AUTH_HMAC_KEY_VERSION: "v1", }, clear=False, ): auth = _mint_auth() self.assertNotIn("key", {k.lower(): 1 for k in ()}) # no-op guard blob = json.dumps(auth) self.assertNotIn(secret, blob) auth["key_version"] = "nope" v = self._verify(auth) self.assertNotIn(secret, json.dumps(v["reasons"])) finally: irp.reset_process_auth_secret_for_tests() class TestF7StrictCanonicalIncidentEvidence(unittest.TestCase): """#709 F7 (review 438): only the exact canonical representation is accepted.""" def _assess(self, payload, **kwargs): params = { "incident_issue": 700, "incident_comment_id": 11489, "comment_payload": payload, "expected_remote": "prgs", "expected_org": "Scaled-Tech-Consulting", "expected_repo": "Gitea-Tools", "expected_pr_number": 42, "expected_head_sha": HEAD_A, "expected_decision_lock_id": DECISION_LOCK_ID, "expected_recovery_action": RECOVERY_ACTION, "mint_actor_id": MINT_ACTOR_ID, "mint_actor_username": MINT_ACTOR_LOGIN, } params.update(kwargs) return irp.assess_incident_evidence(**params) def test_canonical_evidence_accepted(self): g = self._assess(_incident_comment_payload()) self.assertTrue(g["valid"], g) def test_reordered_fields_rejected(self): body = _reorder_canonical_lines(_canonical_incident_body(), "org", "repo") g = self._assess(_incident_comment_payload(body=body)) self.assertFalse(g["valid"], g) self.assertTrue( any("canonical order" in r for r in g["reasons"]), g["reasons"] ) def test_duplicate_identical_field_rejected(self): body = _canonical_incident_body() body = _insert_after_canonical_line(body, "repo", "repo: Gitea-Tools") g = self._assess(_incident_comment_payload(body=body)) self.assertFalse(g["valid"], g) def test_duplicate_conflicting_field_rejected(self): body = _canonical_incident_body() body = _insert_after_canonical_line(body, "repo", "repo: Other-Repo") g = self._assess(_incident_comment_payload(body=body)) self.assertFalse(g["valid"], g) def test_conflicting_field_in_narrative_rejected(self): body = _canonical_incident_body(narrative="context") + "\nrepo: Other-Repo" g = self._assess(_incident_comment_payload(body=body)) self.assertFalse(g["valid"], g) self.assertTrue( any("outside the signed block" in r for r in g["reasons"]), g["reasons"] ) def test_second_marker_rejected(self): body = _canonical_incident_body(narrative="context") body = f"{body}\n\n{irp.INCIDENT_MARKER}" g = self._assess(_incident_comment_payload(body=body)) self.assertFalse(g["valid"], g) def test_marker_not_first_rejected(self): body = "preamble\n" + _canonical_incident_body() g = self._assess(_incident_comment_payload(body=body)) self.assertFalse(g["valid"], g) self.assertTrue( any("marker position" in r for r in g["reasons"]), g["reasons"] ) def test_unknown_extra_field_rejected(self): body = _insert_after_canonical_line( _canonical_incident_body(), "repo", "sneaky: value" ) g = self._assess(_incident_comment_payload(body=body)) self.assertFalse(g["valid"], g) def test_missing_field_rejected(self): body = "\n".join( l for l in _canonical_incident_body().split("\n") if not l.startswith("nonce: ") ) g = self._assess(_incident_comment_payload(body=body)) self.assertFalse(g["valid"], g) def test_empty_field_rejected(self): body = _canonical_incident_body().replace( f"decision_lock_id: {DECISION_LOCK_ID}", "decision_lock_id: " ) g = self._assess(_incident_comment_payload(body=body)) self.assertFalse(g["valid"], g) def test_conflicting_actor_logins_rejected(self): payload = _incident_comment_payload() payload["user"] = {"id": AUTHOR_ID, "login": AUTHOR_LOGIN, "username": "someone-else"} g = self._assess(payload) self.assertFalse(g["valid"], g) self.assertTrue( any("conflicting logins" in r for r in g["reasons"]), g["reasons"] ) def test_conflicting_actor_ids_rejected(self): payload = _incident_comment_payload() payload["user"] = {"id": AUTHOR_ID, "user_id": 999, "login": AUTHOR_LOGIN} g = self._assess(payload) self.assertFalse(g["valid"], g) self.assertTrue( any("conflicting user ids" in r for r in g["reasons"]), g["reasons"] ) def test_display_name_only_actor_rejected(self): payload = _incident_comment_payload() payload["user"] = {"login": AUTHOR_LOGIN} # no stable id g = self._assess(payload) self.assertFalse(g["valid"], g) self.assertTrue( any("stable user id" in r for r in g["reasons"]), g["reasons"] ) def test_author_id_substitution_rejected(self): """Body claims one author id, live comment is authored by another.""" payload = _incident_comment_payload() payload["user"] = {"id": 5150, "login": AUTHOR_LOGIN} g = self._assess(payload) self.assertFalse(g["valid"], g) self.assertTrue( any("evidence_author_id" in r for r in g["reasons"]), g["reasons"] ) def test_edited_comment_rejected(self): payload = _incident_comment_payload() payload["updated_at"] = "2026-07-14T00:00:00Z" g = self._assess(payload) self.assertFalse(g["valid"], g) self.assertTrue(any("edited" in r for r in g["reasons"]), g["reasons"]) def test_self_authored_by_stable_id_rejected(self): payload = _incident_comment_payload(author=MINT_ACTOR_LOGIN, author_id=MINT_ACTOR_ID) g = self._assess(payload) self.assertFalse(g["valid"], g) self.assertTrue( any("self-authored" in r for r in g["reasons"]), g["reasons"] ) def test_mint_actor_substitution_rejected(self): g = self._assess(_incident_comment_payload(), mint_actor_id=999, mint_actor_username="someone") self.assertFalse(g["valid"], g) self.assertTrue( any("mint_actor" in r for r in g["reasons"]), g["reasons"] ) def test_decision_lock_substitution_rejected(self): payload = _incident_comment_payload(decision_lock_id="review_decision_lock-prgs-merger") g = self._assess(payload) self.assertFalse(g["valid"], g) self.assertTrue( any("decision_lock_id" in r for r in g["reasons"]), g["reasons"] ) def test_recovery_action_substitution_rejected(self): body = _canonical_incident_body().replace( f"recovery_action: {RECOVERY_ACTION}", "recovery_action: clear_decision_lock" ) g = self._assess(_incident_comment_payload(body=body)) self.assertFalse(g["valid"], g) def test_unsupported_recovery_action_cannot_be_built(self): with self.assertRaises(ValueError): _canonical_incident_body(recovery_action="merge_pr") def test_cross_pr_replay_rejected(self): payload = _incident_comment_payload(pr_number=99) g = self._assess(payload) self.assertFalse(g["valid"], g) self.assertTrue( any("pr_number" in r for r in g["reasons"]), g["reasons"] ) def test_cross_repository_replay_rejected(self): payload = _incident_comment_payload(repo="Other-Repo") g = self._assess(payload) self.assertFalse(g["valid"], g) def test_cross_org_replay_rejected(self): payload = _incident_comment_payload(org="Other-Org") g = self._assess(payload) self.assertFalse(g["valid"], g) def test_cross_remote_replay_rejected(self): payload = _incident_comment_payload(remote="dadeschools") g = self._assess(payload) self.assertFalse(g["valid"], g) def test_cross_head_replay_rejected(self): payload = _incident_comment_payload(head=HEAD_B) g = self._assess(payload) self.assertFalse(g["valid"], g) def test_recorded_head_substitution_rejected(self): payload = _incident_comment_payload(recorded_head_sha=HEAD_B) g = self._assess(payload, expected_recorded_head_sha=HEAD_A) self.assertFalse(g["valid"], g) self.assertTrue( any("recorded_head_sha" in r for r in g["reasons"]), g["reasons"] ) def test_key_version_substitution_rejected(self): payload = _incident_comment_payload(key_version="v9") g = self._assess(payload, expected_key_version=irp.auth_key_version()) self.assertFalse(g["valid"], g) def test_digest_preserving_substitution_rejected(self): """Swap a field *and* its digest from another scope: still refused. The attacker mints a fully valid canonical body for a different PR (so the digest is internally consistent) and presents it for this scope. """ foreign = _canonical_incident_body(pr_number=99) parsed = irp.parse_canonical_incident_body(foreign) self.assertTrue(parsed["valid"], parsed) # internally consistent g = self._assess(_incident_comment_payload(body=foreign)) self.assertFalse(g["valid"], g) def test_field_swap_without_digest_update_rejected(self): body = _canonical_incident_body().replace( "repo: Gitea-Tools", "repo: Other-Repo" ) g = self._assess( _incident_comment_payload(body=body), expected_repo="Other-Repo" ) self.assertFalse(g["valid"], g) self.assertTrue( any("content_digest" in r for r in g["reasons"]), g["reasons"] ) def test_nonce_binds_digest(self): body = _canonical_incident_body().replace( f"nonce: {INCIDENT_NONCE}", "nonce: 00000000-0000-0000-0000-000000000000" ) g = self._assess(_incident_comment_payload(body=body)) self.assertFalse(g["valid"], g) self.assertTrue( any("content_digest" in r for r in g["reasons"]), g["reasons"] ) def test_builder_refuses_ambiguous_narrative(self): with self.assertRaises(ValueError): _canonical_incident_body(narrative=f"{irp.INCIDENT_MARKER}\nrepo: evil") def test_builder_refuses_multiline_field(self): with self.assertRaises(ValueError): _canonical_incident_body(destroyed_subject="line1\nrepo: evil") def test_builder_refuses_empty_field(self): with self.assertRaises(ValueError): _canonical_incident_body(decision_lock_id="") def test_builder_output_is_the_accepted_format(self): body = _canonical_incident_body() parsed = irp.parse_canonical_incident_body(body) self.assertTrue(parsed["valid"], parsed) self.assertEqual( parsed["canonical_block"], irp.render_canonical_incident_block(parsed["fields"]), ) class TestF8ArchivePrerequisiteForClear(unittest.TestCase): """#709 F8 (review 438): never clear terminal evidence without a durable archive.""" def setUp(self): self._tmp = tempfile.TemporaryDirectory() self.env = patch.dict( os.environ, { "GITEA_MCP_SESSION_STATE_DIR": self._tmp.name, "GITEA_SESSION_PROFILE_LOCK": "prgs-merger", }, clear=False, ) self.env.start() import mcp_server self.mcp = mcp_server self.mcp._REVIEW_DECISION_LOCK = None ss.save_state( kind=ss.KIND_DECISION_LOCK, payload=_lock([APPROVE], profile="prgs-reviewer", head=HEAD_A), remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", profile_identity="prgs-reviewer", state_dir=self._tmp.name, ) def tearDown(self): self.mcp._REVIEW_DECISION_LOCK = None self.env.stop() self._tmp.cleanup() def _clear(self): return self.mcp._clear_decision_lock_for_profile( profile_identity="prgs-reviewer", pr_number=100, expected_head_sha=HEAD_A, remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", ) def _lock_still_present(self): return ss.load_state_for_profile( kind=ss.KIND_DECISION_LOCK, profile_identity="prgs-reviewer", remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", skip_identity_match=True, ) def _fail_archive_only(self, behavior): """Patch save_state so archive writes fail but other writes pass through.""" real = ss.save_state def _fake(**kwargs): if kwargs.get("kind") == ss.KIND_DECISION_LOCK_ARCHIVE: return behavior() return real(**kwargs) return patch.object(self.mcp.mcp_session_state, "save_state", side_effect=_fake) def test_archive_exception_retains_lock(self): def _boom(): raise OSError("disk failure") with self._fail_archive_only(_boom): out = self._clear() self.assertFalse(out["cleared"], out) self.assertTrue(out["terminal_lock_retained"], out) self.assertTrue(out["recovery_required"], out) self.assertEqual(out["archive_failed_step"], "archive_save_state") self.assertIsNotNone(self._lock_still_present(), "terminal lock destroyed") def test_archive_false_response_retains_lock(self): with self._fail_archive_only(lambda: False): out = self._clear() self.assertFalse(out["cleared"], out) self.assertIsNotNone(self._lock_still_present(), "terminal lock destroyed") def test_archive_empty_response_retains_lock(self): with self._fail_archive_only(lambda: {}): out = self._clear() self.assertFalse(out["cleared"], out) self.assertIsNotNone(self._lock_still_present(), "terminal lock destroyed") def test_archive_timeout_retains_lock(self): def _timeout(): raise TimeoutError("session state write timed out") with self._fail_archive_only(_timeout): out = self._clear() self.assertFalse(out["cleared"], out) self.assertIsNotNone(self._lock_still_present(), "terminal lock destroyed") def test_archive_unreadable_retains_lock(self): """Write claims success but read-back finds nothing: still refuse to clear.""" real = ss.load_state_for_profile def _fake(**kwargs): if kwargs.get("kind") == ss.KIND_DECISION_LOCK_ARCHIVE: return None return real(**kwargs) with patch.object( self.mcp.mcp_session_state, "load_state_for_profile", side_effect=_fake ): out = self._clear() self.assertFalse(out["cleared"], out) self.assertEqual(out["archive_failed_step"], "archive_readback") self.assertIsNotNone(self._lock_still_present(), "terminal lock destroyed") def test_partial_archive_readback_retains_lock(self): """Read-back returns a record for a different PR/head: refuse to clear.""" real = ss.load_state_for_profile def _fake(**kwargs): if kwargs.get("kind") == ss.KIND_DECISION_LOCK_ARCHIVE: return {"archived_for_pr": 999, "archived_for_head": HEAD_B} return real(**kwargs) with patch.object( self.mcp.mcp_session_state, "load_state_for_profile", side_effect=_fake ): out = self._clear() self.assertFalse(out["cleared"], out) self.assertEqual(out["archive_failed_step"], "archive_readback") self.assertIsNotNone(self._lock_still_present(), "terminal lock destroyed") def test_archive_failure_records_actionable_recovery_evidence(self): with self._fail_archive_only(lambda: False): out = self._clear() self.assertFalse(out["cleared"], out) self.assertTrue(out["retry_safe"], out) self.assertIn("archival failed", out["reason"]) self.assertIsNotNone(out.get("prior_summary"), out) # The recovery row is keyed by the active (merging) session profile. recovery = ss.load_state_for_profile( kind=ss.KIND_POST_MERGE_DECISION_RECOVERY, profile_identity="prgs-merger", remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", skip_identity_match=True, ) self.assertIsNotNone(recovery, "no durable recovery evidence recorded") self.assertEqual(recovery["failed_step"], "archive_save_state") self.assertEqual(recovery["target_profile_identity"], "prgs-reviewer") def test_successful_archive_clears_exactly_once(self): out = self._clear() self.assertTrue(out["cleared"], out) self.assertTrue(out["archive_ok"], out) self.assertIsNone(self._lock_still_present(), "lock should be cleared") archived = ss.load_state_for_profile( kind=ss.KIND_DECISION_LOCK_ARCHIVE, profile_identity="prgs-reviewer-archive-pr100", remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", skip_identity_match=True, ) self.assertIsNotNone(archived, "archive not durable") self.assertEqual(archived["archived_for_pr"], 100) # A second clear is a no-op, not a duplicate clear. again = self._clear() self.assertFalse(again["cleared"], again) def test_retry_after_archive_failure_succeeds(self): with self._fail_archive_only(lambda: False): first = self._clear() self.assertFalse(first["cleared"], first) self.assertIsNotNone(self._lock_still_present()) retry = self._clear() self.assertTrue(retry["cleared"], retry) self.assertIsNone(self._lock_still_present()) def test_no_alternate_path_clears_after_archive_failure(self): """The post-merge reconciler must not clear the lock when archival failed.""" with self._fail_archive_only(lambda: False), patch.object( self.mcp, "api_request", return_value={} ), patch.object( self.mcp, "repo_api_url", return_value="https://example.test/api/v1/repos/o/r" ): report = self.mcp._reconcile_decision_locks_after_merge( pr_number=100, head_sha=HEAD_A, merge_commit_sha="c" * 40, remote="prgs", host="h", org="Scaled-Tech-Consulting", repo="Gitea-Tools", auth={"Authorization": "token test"}, ) self.assertFalse(report.get("cleared_any"), report) self.assertIsNotNone(self._lock_still_present(), "terminal lock destroyed") if __name__ == "__main__": unittest.main()