Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 77 Pull Requests 9 Actions Packages Projects Releases Wiki Activity

fix(workflow): fail-closed key version, strict incident evidence, archive-gated clear (#709)

Browse Source 
Address formal review 438 REQUEST_CHANGES on PR #710 (F6/F7/F8).

F6 — HMAC key-version validation fails closed:
- verify_authorization_artifact validates key_version BEFORE any MAC work, so an
  attacker-chosen version can never select the signing key.
- Require exactly one nonempty, well-formed key-version field; missing, empty,
  unknown, malformed, duplicated (including identical-valued and nested
  aliases), and mismatched versions all fail. No versionless legacy fallback.
- Artifact version must equal the configured active version; production now
  requires GITEA_IRRECOVERABLE_AUTH_HMAC_KEY_VERSION explicitly (an implicit
  default made rotation ambiguous). Version stays inside the signed material.

F7 — strictly canonical incident evidence:
- Replace substring/first-match parsing with an exact schema: marker on line 1,
  every field once, fixed order, no duplicate/unknown/empty/conflicting fields
  in or outside the signed block. The parsed body is re-rendered and compared
  for exact equality before acceptance.
- content_digest now binds the full recovery scope: repository identity, PR,
  decision-lock identity, destroyed subject, recovery action, recorded and
  expected head, incident issue, evidence author, minting actor, key version,
  nonce and issued_at.
- Actor identity is the immutable user id with login consistency; conflicting
  ids/logins and display-name-only identities fail closed. Edited comments are
  rejected. The independent-author rule is preserved and enforced by stable id.
- build_canonical_incident_body is the single source of the accepted format and
  refuses to emit ambiguous evidence.

F8 — archival is a prerequisite for clearing terminal evidence:
- _clear_decision_lock_for_profile no longer swallows archive failures. It
  requires a successful write plus a durable read-back matching the PR/head,
  and otherwise returns a structured, retry-safe failure that retains the lock
  and records actionable recovery evidence.
- Fixes a latent bug the read-back exposed: the archive payload inherited the
  source lock's session_profile_lock, so save_state keyed the archive under the
  reviewer profile instead of the archive identity and it never read back.

Adversarial regressions added for every listed case: key-version missing/empty/
unknown/malformed/duplicate/rotation/wrong-key-after-restart, reordered fields,
duplicate identical and conflicting fields, conflicting actor ids/names,
digest-preserving substitution, decision-lock and recovery-action substitution,
cross-PR/repo/org/remote/head replay, archive exception/timeout/false/empty/
partial-readback with proof the lock survives, exactly-one permitted clear, and
retry after archive failure. No existing assertion was weakened.

Validation: focused 150 passed in three module orders; full tests/ 2809 passed,
6 skipped, 1 warning (pre-existing StarletteDeprecationWarning in
tests/test_webui_audit.py:8), 161 subtests passed.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
sysadmin
2026-07-15 13:55:06 -04:00
co-authored by Claude Opus 4.8
parent 2b359e0c26
commit 573e721437
3 changed files with 1659 additions and 211 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_issue_709_decision_lock_cross_profile.py
+788 -3
View File
@@ -7,6 +7,7 @@ live PR numbers in production code.
from __future__ import annotations
import json
import os
import subprocess
import sys
@@ -69,6 +70,19 @@ DEDICATED_RECOVERY_OPS = RECONCILER_OPS + [
]
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(
*,
@@ -78,6 +92,17 @@ def _canonical_incident_body(
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(
@@ -85,8 +110,19 @@ def _canonical_incident_body(
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,
)
@@ -94,26 +130,36 @@ def _canonical_incident_body(
def _incident_comment_payload(
*,
comment_id=11489,
author="controller-ops",
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": _canonical_incident_body(
"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": {"login": author},
"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}",
}
@@ -1446,5 +1492,744 @@ class TestClearProfileHelperF3(unittest.TestCase):
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()