yuqei/ai-contract-platform
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

fix(release): close RC06 trust replay and transaction gaps

Browse Source
This commit is contained in:
yuqei
2026-09-27 14:47:15 +08:00
parent 19a500cbf8
commit c90cf87fb3
4 changed files with 1590 additions and 129 deletions
Download Patch File Download Diff File Expand all files Collapse all files
yuqei-infra/scripts/build-rc06-publication-batch.py
+145 -4
View File
@@ -35,14 +35,19 @@ APPROVAL_SCHEMA_VERSION = "yuqei.rc06-publication-batch-approval.v4"
APPROVAL_PAYLOAD_SCHEMA_VERSION = "yuqei.rc06-publication-batch-approval-payload.v3"
APPROVAL_REVIEW_SUMMARY_SCHEMA_VERSION = "yuqei.rc06-single-reviewer-summary.v1"
APPROVAL_KEYRING_SCHEMA_VERSION = "yuqei.rc06-publication-approval-keyring.v3"
APPLY_SCHEMA_VERSION = "yuqei.rc06-publication-system-item-apply-receipt.v4"
APPLY_PAYLOAD_SCHEMA_VERSION = "yuqei.rc06-publication-system-item-apply-payload.v2"
ONLINE_TRANSACTION_RECEIPT_SCHEMA_VERSION = "yuqei.rc06-online-transaction-receipt.v1"
ONLINE_TRANSACTION_PAYLOAD_SCHEMA_VERSION = "yuqei.rc06-online-transaction-payload.v1"
APPLY_SCHEMA_VERSION = "yuqei.rc06-publication-system-item-apply-receipt.v5"
APPLY_PAYLOAD_SCHEMA_VERSION = "yuqei.rc06-publication-system-item-apply-payload.v3"
ONLINE_TRANSACTION_RECEIPT_SCHEMA_VERSION = "yuqei.rc06-online-transaction-receipt.v2"
ONLINE_TRANSACTION_PAYLOAD_SCHEMA_VERSION = "yuqei.rc06-online-transaction-payload.v2"
APPLY_CHALLENGE_SCHEMA_VERSION = "yuqei.rc06-apply-challenge.v1"
APPLY_CHALLENGE_PAYLOAD_SCHEMA_VERSION = "yuqei.rc06-apply-challenge-payload.v1"
APPROVAL_SIGNATURE_DOMAIN = b"YUQEI_RC06_PUBLICATION_BATCH_APPROVAL_V1\0"
APPROVAL_KEYRING_SIGNATURE_DOMAIN = b"YUQEI_RC06_APPROVAL_KEYRING_V1\0"
APPLY_SIGNATURE_DOMAIN = b"YUQEI_RC06_SYSTEM_APPLY_RECEIPT_V1\0"
ONLINE_TRANSACTION_SIGNATURE_DOMAIN = b"YUQEI_RC06_ONLINE_TRANSACTION_RECEIPT_V1\0"
APPLY_CHALLENGE_SIGNATURE_DOMAIN = b"YUQEI_RC06_APPLY_CHALLENGE_V1\0"
APPLY_OPERATION = "apply_rc06_publication_batch"
MAX_APPLY_CHALLENGE_TTL = timedelta(minutes=10)
TRUSTED_APPROVAL_ROOT_KEY_ID = "yuqei-rc06-approval-root-v1"
# Provision the long-lived root/apply pins and exact candidate keyring epoch
# before source freeze. A root-signed keyring then authorizes the exact frozen
@@ -2874,10 +2879,113 @@ def _expected_apply_item_results(manifest: Mapping[str, Any]) -> list[dict[str,
return results
def validate_apply_challenge(
manifest: Mapping[str, Any],
manifest_raw_sha256: str,
approval_raw_sha256: str,
apply_service_public_key: Ed25519PublicKey,
challenge: Any,
) -> dict[str, Any]:
if not HEX64_RE.fullmatch(TRUSTED_APPLY_SERVICE_PUBLIC_KEY_SHA256):
raise PublicationBatchError("apply service public key pin is not provisioned")
apply_public_raw = _public_key_raw(apply_service_public_key)
if _sha256_bytes(apply_public_raw) != TRUSTED_APPLY_SERVICE_PUBLIC_KEY_SHA256:
raise PublicationBatchError("apply challenge key does not match the frozen pin")
envelope = _require_keys(
challenge,
"apply challenge",
("schema_version", "challenge_id", "payload", "signature"),
)
if envelope["schema_version"] != APPLY_CHALLENGE_SCHEMA_VERSION:
raise PublicationBatchError("apply challenge schema is unsupported")
payload = _require_keys(
envelope["payload"],
"apply challenge payload",
(
"schema_version",
"nonce_base64",
"candidate_commit",
"candidate_tag",
"batch_manifest_raw_sha256",
"batch_manifest_payload_sha256",
"approval_receipt_sha256",
"operation",
"environment_id",
"actor",
"issued_at",
"expires_at",
),
)
if (
payload["schema_version"] != APPLY_CHALLENGE_PAYLOAD_SCHEMA_VERSION
or payload["candidate_commit"] != manifest["candidate"]["commit"]
or payload["candidate_tag"] != manifest["candidate"]["tag"]
or payload["batch_manifest_payload_sha256"]
!= manifest["manifest_payload_sha256"]
or payload["approval_receipt_sha256"]
!= _require_sha(approval_raw_sha256, "approval raw SHA-256")
or payload["operation"] != APPLY_OPERATION
):
raise PublicationBatchError("apply challenge candidate or operation binding drift")
if payload["batch_manifest_raw_sha256"] != _require_sha(
manifest_raw_sha256, "apply challenge manifest raw SHA-256"
):
raise PublicationBatchError("apply challenge manifest raw binding drift")
environment_id = _require_string(
payload["environment_id"], "apply challenge environment_id", max_length=256
)
if environment_id != payload["environment_id"]:
raise PublicationBatchError("apply challenge environment_id must be canonical")
actor = _require_keys(payload["actor"], "apply challenge actor", ("type", "component"))
if actor != {
"type": "system",
"component": TRUSTED_ONLINE_APPLY_SERVICE_COMPONENT,
}:
raise PublicationBatchError("apply challenge actor is not the trusted online service")
issued_at = _parse_time(payload["issued_at"], "apply challenge issued_at")
expires_at = _parse_time(payload["expires_at"], "apply challenge expires_at")
if (
expires_at <= issued_at
or expires_at - issued_at > MAX_APPLY_CHALLENGE_TTL
):
raise PublicationBatchError("apply challenge TTL is invalid")
try:
nonce = base64.b64decode(
_require_string(
payload["nonce_base64"], "apply challenge nonce", max_length=128
),
validate=True,
)
signature = base64.b64decode(
_require_string(
envelope["signature"], "apply challenge signature", max_length=1024
),
validate=True,
)
if len(nonce) != 32:
raise ValueError("challenge nonce must contain 32 bytes")
apply_service_public_key.verify(
signature,
APPLY_CHALLENGE_SIGNATURE_DOMAIN + _canonical(payload),
)
except (TypeError, ValueError, InvalidSignature) as exc:
raise PublicationBatchError("apply challenge signature verification failed") from exc
expected_id = (
"rc06-apply-challenge-"
+ _sha256_bytes(_canonical(payload) + signature)[:24]
)
if envelope["challenge_id"] != expected_id:
raise PublicationBatchError("apply challenge_id does not match its signed material")
_assert_no_secret_material(envelope, "apply challenge")
return envelope
def validate_online_transaction_receipt(
manifest: Mapping[str, Any],
manifest_raw_sha256: str,
approval_raw_sha256: str,
approval: Mapping[str, Any],
apply_service_public_key: Ed25519PublicKey,
online_service_public_key: Ed25519PublicKey,
receipt: Any,
) -> dict[str, Any]:
@@ -2928,6 +3036,7 @@ def validate_online_transaction_receipt(
"transaction_id",
"database_commit_id",
"committed_at",
"authority_challenge",
"generated_by",
"item_results",
"write_count",
@@ -2982,6 +3091,36 @@ def validate_online_transaction_receipt(
raise PublicationBatchError(
"online transaction receipt was generated outside the approval window"
)
challenge = validate_apply_challenge(
manifest,
manifest_raw_sha256,
approval_raw_sha256,
apply_service_public_key,
payload["authority_challenge"],
)
challenge_payload = challenge["payload"]
challenge_issued_at = _parse_time(
challenge_payload["issued_at"], "apply challenge issued_at"
)
challenge_expires_at = _parse_time(
challenge_payload["expires_at"], "apply challenge expires_at"
)
approval_approved_at = _parse_time(
approval_payload["approved_at"], "approval approved_at"
)
approval_expires_at = _parse_time(
approval_payload["expires_at"], "approval expires_at"
)
if (
challenge_payload["environment_id"] != environment_id
or challenge_issued_at < approval_approved_at
or challenge_expires_at > approval_expires_at
or committed_at < challenge_issued_at
or committed_at >= challenge_expires_at
):
raise PublicationBatchError(
"online transaction receipt does not satisfy its apply challenge"
)
actor = _require_keys(
payload["generated_by"], "online transaction generator", ("type", "component")
)
@@ -3084,8 +3223,10 @@ def validate_apply_receipt(
raise PublicationBatchError("approval/apply batch identity mismatch")
online_receipt = validate_online_transaction_receipt(
manifest,
manifest_raw_sha256,
approval_raw_sha256,
approval,
apply_service_public_key,
online_service_public_key,
row["online_transaction_receipt"],
)
yuqei-infra/scripts/rc06-release-trust-bootstrap.py
+1003 -100
View File
File diff suppressed because it is too large Load Diff
yuqei-infra/tests/test_rc06_publication_batch.py
+29
View File
@@ -988,6 +988,34 @@ def test_one_reviewer_batch_signature_and_system_item_receipts_remove_human_repe
for item in manifest["prompt_module_batch"]["items"]
]
item_results.sort(key=lambda item: (item["kind"], item["item_id"]))
challenge_payload = {
"schema_version": batch.APPLY_CHALLENGE_PAYLOAD_SCHEMA_VERSION,
"nonce_base64": base64.b64encode(b"c" * 32).decode("ascii"),
"candidate_commit": manifest["candidate"]["commit"],
"candidate_tag": manifest["candidate"]["tag"],
"batch_manifest_raw_sha256": manifest_raw_sha,
"batch_manifest_payload_sha256": manifest["manifest_payload_sha256"],
"approval_receipt_sha256": approval_raw_sha,
"operation": batch.APPLY_OPERATION,
"environment_id": "isolated-uat-fixture",
"actor": {"type": "system", "component": "rc06-publication-service"},
"issued_at": "2026-09-21T09:55:00+08:00",
"expires_at": "2026-09-21T10:05:00+08:00",
}
challenge_signature = apply_service_private_key.sign(
batch.APPLY_CHALLENGE_SIGNATURE_DOMAIN + batch._canonical(challenge_payload)
)
authority_challenge = {
"schema_version": batch.APPLY_CHALLENGE_SCHEMA_VERSION,
"challenge_id": (
"rc06-apply-challenge-"
+ batch._sha256_bytes(
batch._canonical(challenge_payload) + challenge_signature
)[:24]
),
"payload": challenge_payload,
"signature": base64.b64encode(challenge_signature).decode("ascii"),
}
online_payload = {
"schema_version": batch.ONLINE_TRANSACTION_PAYLOAD_SCHEMA_VERSION,
"key_id": batch.TRUSTED_ONLINE_APPLY_SERVICE_KEY_ID,
@@ -997,6 +1025,7 @@ def test_one_reviewer_batch_signature_and_system_item_receipts_remove_human_repe
"transaction_id": "fixture-transaction-001",
"database_commit_id": "fixture-database-commit-001",
"committed_at": "2026-09-21T10:00:00+08:00",
"authority_challenge": authority_challenge,
"generated_by": {"type": "system", "component": "rc06-publication-service"},
"item_results": item_results,
"write_count": 74,
yuqei-infra/tests/test_rc06_release_trust_bootstrap.py
+413 -25
View File
@@ -6,6 +6,8 @@ import importlib.util
import json
import os
import re
import stat
import subprocess
import sys
from datetime import UTC, datetime, timedelta
from pathlib import Path
@@ -21,6 +23,20 @@ EXISTING_BATCH_TEST = REPO_ROOT / "yuqei-infra" / "tests" / "test_rc06_publicati
AUTHORITY_LAUNCHER = REPO_ROOT / "yuqei-infra" / "scripts" / "Launch-ReleaseAuthority.ps1"
def _acl_snapshot(path: Path) -> bytes | int:
if os.name != "nt":
return stat.S_IMODE(path.stat().st_mode)
completed = subprocess.run(
["icacls", os.fspath(path)],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
check=False,
)
if completed.returncode != 0:
pytest.fail(f"cannot snapshot fixture ACL: {completed.stderr!r}")
return completed.stdout
def _load_script():
spec = importlib.util.spec_from_file_location("rc06_release_trust_bootstrap", SCRIPT)
assert spec is not None and spec.loader is not None
@@ -216,6 +232,24 @@ def _formal_apply_materials(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) ->
approval_file = tmp_path / "rc06-batch-approval.json"
approval_raw = authority._canonical(approval)
approval_file.write_bytes(approval_raw)
approval_sha = authority.batch._sha256_bytes(approval_raw)
challenge_result = authority.issue_apply_challenge(
state,
manifest_file=manifest_file,
manifest_sha256=manifest_sha,
approval_file=approval_file,
approval_sha256=approval_sha,
approval_keyring_file=keyring_file,
approval_keyring_sha256=authority.batch._sha256_bytes(keyring_raw),
approval_root_public_key_file=(
keyring_directory / "rc06-approval-root-public-key.raw"
),
environment_id="isolated-uat-fixture",
unprotect=_unprotect,
now=datetime(2026, 9, 27, 3, 55, tzinfo=UTC),
)
challenge_file = Path(challenge_result["challenge_file"])
challenge = json.loads(challenge_file.read_bytes())
item_results = [
{
"kind": "asset",
@@ -245,6 +279,7 @@ def _formal_apply_materials(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) ->
"transaction_id": "fixture-transaction-001",
"database_commit_id": "fixture-database-commit-001",
"committed_at": "2026-09-27T04:00:00Z",
"authority_challenge": challenge,
"generated_by": {"type": "system", "component": "rc06-publication-service"},
"item_results": item_results,
"write_count": 74,
@@ -282,7 +317,9 @@ def _formal_apply_materials(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) ->
"root_file": keyring_directory / "rc06-approval-root-public-key.raw",
"approval": approval,
"approval_file": approval_file,
"approval_sha": authority.batch._sha256_bytes(approval_raw),
"approval_sha": approval_sha,
"challenge": challenge,
"challenge_file": challenge_file,
"online_private": online_private,
"online_public_file": online_public_file,
"online_receipt": online_receipt,
@@ -344,6 +381,28 @@ def test_authority_rejects_partial_or_tampered_bundle(tmp_path: Path) -> None:
)
def test_authority_manifest_requires_exact_role_array_objects(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
state, _result, _keys = _initialize(tmp_path)
manifest_path = state / authority.KEYS_DIRECTORY_NAME / authority.AUTHORITY_MANIFEST_FILENAME
original = json.loads(manifest_path.read_bytes())
monkeypatch.setattr(authority.signer, "_verify_complete_bundle", lambda *args, **kwargs: None)
extra = copy.deepcopy(original)
extra["roles"].append(copy.deepcopy(extra["roles"][0]))
manifest_path.write_bytes(authority._canonical(extra))
with pytest.raises(authority.AuthorityError, match="role inventory drift"):
authority.load_authority(state, unprotect=_unprotect, unseal_roles=())
missing_field = copy.deepcopy(original)
del missing_field["roles"][0]["algorithm"]
manifest_path.write_bytes(authority._canonical(missing_field))
with pytest.raises(authority.AuthorityError, match="role inventory drift"):
authority.load_authority(state, unprotect=_unprotect, unseal_roles=())
def test_recovery_keyrings_share_only_the_recovery_root_and_verify_both_schemas(
tmp_path: Path,
) -> None:
@@ -521,6 +580,71 @@ def test_pin_planner_refuses_unknown_existing_pin_without_overwriting(tmp_path:
)
@pytest.mark.parametrize(
("injection", "message"),
(
(
'\nTRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256 = ""\n',
"one unique module-level literal",
),
(
'\nglobals()["TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256"] = "' + "f" * 64 + '"\n',
"dynamic namespace mutation",
),
(
'\nexec("TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256 = \\\"' + "f" * 64 + '\\\"")\n',
"untrusted dynamic exec",
),
(
"\nfrom attacker_pin_side_effect import *\n",
"star import",
),
),
)
def test_pin_planner_rejects_duplicate_dynamic_and_import_side_effect_assignments(
tmp_path: Path,
injection: str,
message: str,
) -> None:
state, repo, _plan = _pin_plan_fixture(tmp_path, include_online=False)
rc06 = repo / "yuqei-infra/scripts/build-rc06-publication-batch.py"
rc06.write_text(rc06.read_text(encoding="utf-8") + injection, encoding="utf-8")
with pytest.raises(authority.AuthorityError, match=message):
authority.plan_pin_updates(
repo,
state,
state / "keyrings" / "recovery-epoch-1",
epoch=1,
unprotect=_unprotect,
)
def test_pin_planner_isolated_runtime_detects_import_side_effect_override(
tmp_path: Path,
) -> None:
state, repo, _plan = _pin_plan_fixture(tmp_path, include_online=False)
rc06 = repo / "yuqei-infra/scripts/build-rc06-publication-batch.py"
side_effect = rc06.parent / "attacker_pin_side_effect.py"
side_effect.write_text(
"import sys\n"
"module = sys.modules['_yuqei_pin_runtime_validation']\n"
"module.TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256 = '" + "f" * 64 + "'\n",
encoding="utf-8",
)
rc06.write_text(
rc06.read_text(encoding="utf-8") + "\nimport attacker_pin_side_effect\n",
encoding="utf-8",
)
with pytest.raises(authority.AuthorityError, match="runtime values drift"):
authority.plan_pin_updates(
repo,
state,
state / "keyrings" / "recovery-epoch-1",
epoch=1,
unprotect=_unprotect,
)
def test_pin_apply_rolls_back_completed_files_when_later_replace_fails(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
@@ -547,17 +671,17 @@ def test_pin_apply_rolls_back_completed_files_when_later_replace_fails(
online_service_registration_file=online_registration_file,
online_environment_ids=("isolated-uat-fixture",),
)
real_replace = authority._write_atomic_replacement
real_replace = authority._write_bound_source_replacement
calls = 0
def fail_second(path: Path, raw: bytes) -> None:
def fail_second(path: Path, repo: Path, expected: bytes, raw: bytes) -> None:
nonlocal calls
calls += 1
if calls == 2:
raise OSError("fixture replacement failure")
real_replace(path, raw)
real_replace(path, repo, expected, raw)
monkeypatch.setattr(authority, "_write_atomic_replacement", fail_second)
monkeypatch.setattr(authority, "_write_bound_source_replacement", fail_second)
with pytest.raises(authority.AuthorityError, match="transaction failed"):
authority.apply_pin_plan(plan)
assert all(change.path.read_bytes() == change.original for change in plan.changes)
@@ -668,7 +792,7 @@ def test_apply_payload_signing_is_bound_verified_atomic_and_idempotent(
online_transaction_receipts_sha256=materials["online_receipts_sha"],
output_directory=output,
unprotect=_unprotect,
now=datetime(2026, 9, 27, 4, 30, tzinfo=UTC),
now=datetime(2026, 9, 27, 4, 1, tzinfo=UTC),
)
assert result["status"] == "apply_receipts_signed_and_verified_not_published"
assert result["receipt_count"] == 1
@@ -701,11 +825,32 @@ def test_apply_payload_signing_is_bound_verified_atomic_and_idempotent(
online_transaction_receipts_sha256=materials["online_receipts_sha"],
output_directory=output,
unprotect=_unprotect,
now=datetime(2026, 9, 27, 4, 30, tzinfo=UTC),
now=datetime(2026, 9, 27, 4, 1, tzinfo=UTC),
)
assert repeated["status"] == "apply_receipts_already_signed_and_verified_not_published"
assert repeated["receipts"] == result["receipts"]
with pytest.raises(
authority.AuthorityError,
match="already consumed or reserved for another output",
):
authority.sign_apply_payloads(
materials["state"],
manifest_file=materials["manifest_file"],
manifest_sha256=materials["manifest_sha"],
approval_file=materials["approval_file"],
approval_sha256=materials["approval_sha"],
approval_keyring_file=materials["keyring_file"],
approval_keyring_sha256=materials["keyring_sha"],
approval_root_public_key_file=materials["root_file"],
online_service_public_key_file=materials["online_public_file"],
online_transaction_receipts_file=materials["online_receipts_file"],
online_transaction_receipts_sha256=materials["online_receipts_sha"],
output_directory=tmp_path / "replayed-to-another-output",
unprotect=_unprotect,
now=datetime(2026, 9, 27, 4, 2, tzinfo=UTC),
)
def test_apply_payload_signing_rejects_tamper_cross_approval_and_partial_bundle(
tmp_path: Path,
@@ -734,10 +879,51 @@ def test_apply_payload_signing_rejects_tamper_cross_approval_and_partial_bundle(
online_transaction_receipts_sha256=authority.batch._sha256_bytes(tampered_raw),
output_directory=tampered_output,
unprotect=_unprotect,
now=datetime(2026, 9, 27, 4, 30, tzinfo=UTC),
now=datetime(2026, 9, 27, 4, 1, tzinfo=UTC),
)
assert not tampered_output.exists()
challenge_tampered = copy.deepcopy(materials["online_receipt"])
challenge_tampered["payload"]["authority_challenge"]["payload"][
"operation"
] = "attacker-operation"
challenge_online_signature = materials["online_private"].sign(
authority.batch.ONLINE_TRANSACTION_SIGNATURE_DOMAIN
+ authority._canonical(challenge_tampered["payload"])
)
challenge_tampered["signature"] = base64.b64encode(
challenge_online_signature
).decode("ascii")
challenge_tampered["receipt_id"] = (
"rc06-online-transaction-"
+ authority.batch._sha256_bytes(
authority._canonical(challenge_tampered["payload"])
+ challenge_online_signature
)[:24]
)
challenge_tampered_file = tmp_path / "tampered-challenge-receipt.json"
challenge_tampered_raw = authority._canonical(challenge_tampered)
challenge_tampered_file.write_bytes(challenge_tampered_raw)
with pytest.raises(authority.AuthorityError, match="challenge.*binding drift"):
authority.sign_apply_payloads(
materials["state"],
manifest_file=materials["manifest_file"],
manifest_sha256=materials["manifest_sha"],
approval_file=materials["approval_file"],
approval_sha256=materials["approval_sha"],
approval_keyring_file=materials["keyring_file"],
approval_keyring_sha256=materials["keyring_sha"],
approval_root_public_key_file=materials["root_file"],
online_service_public_key_file=materials["online_public_file"],
online_transaction_receipts_file=challenge_tampered_file,
online_transaction_receipts_sha256=authority.batch._sha256_bytes(
challenge_tampered_raw
),
output_directory=tmp_path / "tampered-challenge-output",
unprotect=_unprotect,
now=datetime(2026, 9, 27, 4, 1, tzinfo=UTC),
)
second_payload = dict(materials["approval"]["payload"])
second_payload["approved_at"] = "2026-09-27T03:15:00Z"
second_signature = materials["reviewer_private"].sign(
@@ -772,7 +958,7 @@ def test_apply_payload_signing_rejects_tamper_cross_approval_and_partial_bundle(
online_transaction_receipts_sha256=materials["online_receipts_sha"],
output_directory=tmp_path / "cross-approval-output",
unprotect=_unprotect,
now=datetime(2026, 9, 27, 4, 30, tzinfo=UTC),
now=datetime(2026, 9, 27, 4, 1, tzinfo=UTC),
)
partial = copy.deepcopy(materials["online_receipt"])
@@ -797,11 +983,81 @@ def test_apply_payload_signing_rejects_tamper_cross_approval_and_partial_bundle(
online_transaction_receipts_sha256=authority.batch._sha256_bytes(payload_list_raw),
output_directory=partial_output,
unprotect=_unprotect,
now=datetime(2026, 9, 27, 4, 30, tzinfo=UTC),
now=datetime(2026, 9, 27, 4, 1, tzinfo=UTC),
)
assert not partial_output.exists()
def test_apply_challenge_consumption_recovers_after_finalize_interrupt(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
materials = _formal_apply_materials(tmp_path, monkeypatch)
output = tmp_path / "interrupted-consumption-output"
real_finalize = authority._finalize_apply_consumption
interrupted = False
def interrupt_finalize(*args, **kwargs):
nonlocal interrupted
if not interrupted:
interrupted = True
raise KeyboardInterrupt()
return real_finalize(*args, **kwargs)
monkeypatch.setattr(authority, "_finalize_apply_consumption", interrupt_finalize)
with pytest.raises(KeyboardInterrupt):
authority.sign_apply_payloads(
materials["state"],
manifest_file=materials["manifest_file"],
manifest_sha256=materials["manifest_sha"],
approval_file=materials["approval_file"],
approval_sha256=materials["approval_sha"],
approval_keyring_file=materials["keyring_file"],
approval_keyring_sha256=materials["keyring_sha"],
approval_root_public_key_file=materials["root_file"],
online_service_public_key_file=materials["online_public_file"],
online_transaction_receipts_file=materials["online_receipts_file"],
online_transaction_receipts_sha256=materials["online_receipts_sha"],
output_directory=output,
unprotect=_unprotect,
now=datetime(2026, 9, 27, 4, 1, tzinfo=UTC),
)
assert output.is_dir()
consumption_path = (
materials["state"]
/ "apply-challenge-consumption"
/ f"{materials['challenge']['challenge_id']}.json"
)
assert json.loads(consumption_path.read_bytes())["status"] == "reserved"
monkeypatch.setattr(authority, "_finalize_apply_consumption", real_finalize)
calls: list[str] = []
def observed_unprotect(raw: bytes, role: str) -> bytes:
calls.append(role)
return _unprotect(raw, role)
recovered = authority.sign_apply_payloads(
materials["state"],
manifest_file=materials["manifest_file"],
manifest_sha256=materials["manifest_sha"],
approval_file=materials["approval_file"],
approval_sha256=materials["approval_sha"],
approval_keyring_file=materials["keyring_file"],
approval_keyring_sha256=materials["keyring_sha"],
approval_root_public_key_file=materials["root_file"],
online_service_public_key_file=materials["online_public_file"],
online_transaction_receipts_file=materials["online_receipts_file"],
online_transaction_receipts_sha256=materials["online_receipts_sha"],
output_directory=output,
unprotect=observed_unprotect,
now=datetime(2026, 9, 27, 4, 2, tzinfo=UTC),
)
assert recovered["status"] == "apply_receipts_already_signed_and_verified_not_published"
assert json.loads(consumption_path.read_bytes())["status"] == "consumed"
assert calls == []
def test_authority_binds_exact_reviewer_registration_and_derived_key_id(
tmp_path: Path,
) -> None:
@@ -876,7 +1132,7 @@ def test_public_status_and_apply_validation_use_least_private_key_unsealing(
online_transaction_receipts_sha256=materials["online_receipts_sha"],
output_directory=output,
unprotect=observed_unprotect,
now=datetime(2026, 9, 27, 4, 30, tzinfo=UTC),
now=datetime(2026, 9, 27, 4, 1, tzinfo=UTC),
)
assert calls == [authority.ROLE_APPLY_SERVICE]
@@ -933,7 +1189,7 @@ def test_apply_signing_rejects_unprovisioned_custom_environment_and_expired_wind
authority.sign_apply_payloads(
materials["state"],
output_directory=tmp_path / "unprovisioned-output",
now=datetime(2026, 9, 27, 4, 30, tzinfo=UTC),
now=datetime(2026, 9, 27, 4, 1, tzinfo=UTC),
**common,
)
assert calls == []
@@ -963,7 +1219,7 @@ def test_apply_signing_rejects_unprovisioned_custom_environment_and_expired_wind
authority.sign_apply_payloads(
materials["state"],
output_directory=tmp_path / "custom-output",
now=datetime(2026, 9, 27, 4, 30, tzinfo=UTC),
now=datetime(2026, 9, 27, 4, 1, tzinfo=UTC),
**{
**common,
"online_transaction_receipts_file": custom_file,
@@ -990,11 +1246,14 @@ def test_apply_signing_rejects_unprovisioned_custom_environment_and_expired_wind
future_file = tmp_path / "future-online-receipt.json"
future_raw = authority._canonical(future)
future_file.write_bytes(future_raw)
with pytest.raises(authority.AuthorityError, match="after the current signing time"):
with pytest.raises(
authority.AuthorityError,
match="does not satisfy its apply challenge|after the current signing time",
):
authority.sign_apply_payloads(
materials["state"],
output_directory=tmp_path / "future-output",
now=datetime(2026, 9, 27, 4, 30, tzinfo=UTC),
now=datetime(2026, 9, 27, 4, 1, tzinfo=UTC),
**{
**common,
"online_transaction_receipts_file": future_file,
@@ -1005,6 +1264,15 @@ def test_apply_signing_rejects_unprovisioned_custom_environment_and_expired_wind
)
assert calls == []
with pytest.raises(authority.AuthorityError, match="short-lived apply challenge"):
authority.sign_apply_payloads(
materials["state"],
output_directory=tmp_path / "expired-challenge-output",
now=datetime(2026, 9, 27, 4, 6, tzinfo=UTC),
**common,
)
assert calls == []
with pytest.raises(authority.AuthorityError, match="outside the approval window"):
authority.sign_apply_payloads(
materials["state"],
@@ -1050,21 +1318,21 @@ def test_pin_transaction_handles_keyboard_interrupt_and_concurrent_writer(
monkeypatch: pytest.MonkeyPatch,
) -> None:
state, repo, plan = _pin_plan_fixture(tmp_path)
real_replace = authority._write_atomic_replacement
real_replace = authority._write_bound_source_replacement
interrupted = False
def interrupt_first_source(path: Path, raw: bytes) -> None:
def interrupt_first_source(path: Path, repo: Path, expected: bytes, raw: bytes) -> None:
nonlocal interrupted
if not interrupted and any(path == change.path for change in plan.changes):
interrupted = True
raise KeyboardInterrupt()
real_replace(path, raw)
real_replace(path, repo, expected, raw)
monkeypatch.setattr(authority, "_write_atomic_replacement", interrupt_first_source)
monkeypatch.setattr(authority, "_write_bound_source_replacement", interrupt_first_source)
with pytest.raises(KeyboardInterrupt):
authority.apply_pin_plan(plan)
assert all(change.path.read_bytes() == change.original for change in plan.changes)
monkeypatch.setattr(authority, "_write_atomic_replacement", real_replace)
monkeypatch.setattr(authority, "_write_bound_source_replacement", real_replace)
with authority._repository_pin_lock(repo, state):
with pytest.raises(authority.AuthorityError, match="repository lock"):
@@ -1101,6 +1369,64 @@ def test_pin_apply_rechecks_ancestor_reparse_state_immediately_before_write(
assert all(change.path.read_bytes() == change.original for change in plan.changes)
def test_bound_source_publish_rejects_parent_path_swap_without_touching_outside_file(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
state, _repo, plan = _pin_plan_fixture(tmp_path)
target = next(change.path for change in plan.changes if change.original != change.proposed)
outside = tmp_path / "outside-protected.py"
outside.write_bytes(b"outside-content-must-not-change\n")
os.chmod(outside, 0o444)
outside_before = outside.read_bytes()
outside_mode = stat.S_IMODE(outside.stat().st_mode)
outside_attributes = getattr(outside.stat(), "st_file_attributes", None)
outside_acl = _acl_snapshot(outside)
real_assert_bound = authority._assert_descriptor_still_bound
swapped = False
swap_blocked_by_open_handle = False
def swap_parent_after_open(descriptor, path, repo, label, *, original_resolved):
nonlocal swapped, swap_blocked_by_open_handle
if path == target and not swapped:
swapped = True
original_parent = path.parent.with_name(path.parent.name + "-opened-original")
try:
path.parent.rename(original_parent)
path.parent.mkdir()
os.link(outside, path)
except PermissionError as exc:
swap_blocked_by_open_handle = True
raise authority.AuthorityError(
"source path or parent identity changed; operating system blocked the swap"
) from exc
return real_assert_bound(
descriptor,
path,
repo,
label,
original_resolved=original_resolved,
)
monkeypatch.setattr(
authority,
"_assert_descriptor_still_bound",
swap_parent_after_open,
)
with pytest.raises(authority.AuthorityError, match="path or parent identity changed"):
authority.apply_pin_plan(plan)
assert swapped is True
assert swap_blocked_by_open_handle is True or not target.parent.samefile(
target.parent.with_name(target.parent.name + "-opened-original")
)
assert outside.read_bytes() == outside_before
assert stat.S_IMODE(outside.stat().st_mode) == outside_mode
assert getattr(outside.stat(), "st_file_attributes", None) == outside_attributes
assert _acl_snapshot(outside) == outside_acl
control = authority._pin_control_directory(plan.repo, state)
assert not (control / "transaction.json").exists()
def test_pin_transaction_persistent_journal_blocks_and_recovers_after_crash(
tmp_path: Path,
) -> None:
@@ -1110,7 +1436,12 @@ def test_pin_transaction_persistent_journal_blocks_and_recovers_after_crash(
journal = authority._journal_payload(plan)
journal_path = control / "transaction.json"
authority._create_pin_journal(journal_path, journal)
authority._write_atomic_replacement(first.path, first.proposed)
authority._write_bound_source_replacement(
first.path,
repo,
first.original,
first.proposed,
)
journal["status"] = "applying"
journal["completed_paths"].append(first.path.relative_to(repo).as_posix())
authority._write_pin_journal(journal_path, journal)
@@ -1126,6 +1457,53 @@ def test_pin_transaction_persistent_journal_blocks_and_recovers_after_crash(
)
def test_pin_initial_journal_publish_interrupt_is_inside_recovery_boundary(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
state, repo, plan = _pin_plan_fixture(tmp_path)
real_create = authority._create_pin_journal
interrupted = False
def publish_then_interrupt(path: Path, payload: dict) -> None:
nonlocal interrupted
real_create(path, payload)
if not interrupted:
interrupted = True
raise KeyboardInterrupt()
monkeypatch.setattr(authority, "_create_pin_journal", publish_then_interrupt)
with pytest.raises(KeyboardInterrupt):
authority.apply_pin_plan(plan)
assert all(change.path.read_bytes() == change.original for change in plan.changes)
control = authority._pin_control_directory(repo, state)
assert not (control / "transaction.json").exists()
assert list((control / "history").glob("*.rolled_back.json"))
def test_partial_pin_journal_is_quarantined_and_blocks_future_apply(
tmp_path: Path,
) -> None:
state, repo, plan = _pin_plan_fixture(tmp_path)
with authority._repository_pin_lock(repo, state) as control:
journal_path = control / "transaction.json"
authority._publish_private_no_replace(
journal_path,
b'{"partial":',
"partial test journal",
)
before = {change.path: change.path.read_bytes() for change in plan.changes}
result = authority.recover_pin_transaction(repo, state)
assert result["status"] == (
"source_pin_transaction_quarantined_manual_recovery_required"
)
assert Path(result["quarantine_marker"]).is_file()
assert Path(result["quarantine_file"]).read_bytes() == b'{"partial":'
assert all(path.read_bytes() == raw for path, raw in before.items())
with pytest.raises(authority.AuthorityError, match="quarantine requires manual recovery"):
authority.apply_pin_plan(plan)
def test_pin_commit_point_survives_archive_failure_without_rollback(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
@@ -1168,7 +1546,12 @@ def test_pin_recovery_validates_entire_journal_before_restoring_any_file(
with authority._repository_pin_lock(repo, state) as control:
journal = authority._journal_payload(plan)
for change in changed[:2]:
authority._write_atomic_replacement(change.path, change.proposed)
authority._write_bound_source_replacement(
change.path,
repo,
change.original,
change.proposed,
)
journal["completed_paths"].append(change.path.relative_to(repo).as_posix())
journal["status"] = "applying"
journal["changes"][-1]["original_base64"] = base64.b64encode(
@@ -1176,8 +1559,10 @@ def test_pin_recovery_validates_entire_journal_before_restoring_any_file(
).decode("ascii")
authority._create_pin_journal(control / "transaction.json", journal)
with pytest.raises(authority.AuthorityError, match="journal hash drift"):
authority.recover_pin_transaction(repo, state)
result = authority.recover_pin_transaction(repo, state)
assert result["status"] == (
"source_pin_transaction_quarantined_manual_recovery_required"
)
assert all(change.path.read_bytes() == change.proposed for change in changed[:2])
@@ -1284,7 +1669,10 @@ def test_pin_planner_rejects_hardlinked_source_and_recovery_refuses_unknown_byte
os.link(first.path, alias)
except OSError:
pytest.skip("hard links are unavailable on this platform")
with pytest.raises(authority.AuthorityError, match="single-link regular"):
with pytest.raises(
authority.AuthorityError,
match="single-link regular|path or parent identity changed",
):
authority._read_source_file(first.path, repo, first.path.name)
alias.unlink()