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

fix(rc06): close publication authority recovery gaps

Browse Source
This commit is contained in:
yuqei
2026-10-06 21:51:51 +08:00
parent f7d5946c42
commit a965b13645
5 changed files with 1076 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
yuqei-infra/scripts/apply-rc06-publication-batch.py
+268 -37
View File
@@ -304,7 +304,20 @@ class ApplyLedger:
elif metadata[0] != LEDGER_SCHEMA_VERSION:
raise ApplyPublicationError("apply ledger schema is unsupported")
def reserve(self, key: str, request_sha256: str, now: datetime) -> dict[str, Any] | None:
def reserve(
self,
key: str,
request_sha256: str,
now: datetime,
*,
initial_dispatch_authorized: bool,
) -> tuple[str, dict[str, Any] | None]:
"""Reserve a new request or identify an exact crash-recovery replay.
An expired short-lived challenge may recover an already-dispatched exact
request, but it must never create a new dispatch reservation.
"""
self.connection.execute("BEGIN IMMEDIATE")
try:
row = self.connection.execute(
@@ -320,9 +333,23 @@ class ApplyLedger:
}:
receipt = json.loads(row[2])
self.connection.execute("COMMIT")
return receipt
return "replay", receipt
if row is not None and row[1] in {"dispatching", "verifying"}:
timestamp = now.astimezone(UTC).isoformat(timespec="seconds")
self._event(
key,
f"recovering_{row[1]}",
request_sha256,
timestamp,
)
self.connection.execute("COMMIT")
return "recover", None
if row is not None:
raise ApplyPublicationError("an apply with this idempotency key needs reconciliation")
if not initial_dispatch_authorized:
raise ApplyPublicationError(
"apply challenge is not currently valid for an initial dispatch"
)
timestamp = now.astimezone(UTC).isoformat(timespec="seconds")
self.connection.execute(
"INSERT INTO rc06_apply_operations "
@@ -332,7 +359,7 @@ class ApplyLedger:
)
self._event(key, "dispatching", request_sha256, timestamp)
self.connection.execute("COMMIT")
return None
return "dispatch", None
except Exception:
self.connection.execute("ROLLBACK")
raise
@@ -553,6 +580,45 @@ def _remote_status_state(
return status["state"]
def _validate_compensation_hashes(
item: Mapping[str, Any],
label: str,
*,
unchanged_preexisting_sha256: str | None = None,
) -> tuple[int, int, int]:
if item["status"] != "verified":
raise ApplyPublicationError(f"{label} status drift")
if item["action"] == "restored_previous":
batch._require_sha(
item["pre_apply_content_sha256"],
f"{label} restored content SHA-256",
)
if item["post_compensation_content_sha256"] != item["pre_apply_content_sha256"]:
raise ApplyPublicationError(f"{label} restored content hash drift")
return 1, 0, 0
if item["action"] == "removed_new":
if (
item["pre_apply_content_sha256"] is not None
or item["post_compensation_content_sha256"] is not None
):
raise ApplyPublicationError(f"{label} removal evidence drift")
return 0, 1, 0
if item["action"] == "unchanged_preexisting":
if unchanged_preexisting_sha256 is None:
raise ApplyPublicationError(f"{label} cannot be unchanged preexisting")
expected_sha256 = batch._require_sha(
unchanged_preexisting_sha256,
f"{label} expected preexisting content SHA-256",
)
if (
item["pre_apply_content_sha256"] != expected_sha256
or item["post_compensation_content_sha256"] != expected_sha256
):
raise ApplyPublicationError(f"{label} unchanged content hash drift")
return 0, 0, 1
raise ApplyPublicationError(f"{label} action is unsupported")
def _compensate(
client: JsonHttpClient,
*,
@@ -595,8 +661,10 @@ def _compensate(
"participant_states_before",
"participant_states_after",
"compensation_item_results",
"runtime_snapshot_compensation_results",
"restored_item_count",
"removed_item_count",
"unchanged_item_count",
"rollback_evidence_sha256",
),
)
@@ -611,6 +679,9 @@ def _compensate(
or not isinstance(row["removed_item_count"], int)
or isinstance(row["removed_item_count"], bool)
or row["removed_item_count"] < 0
or not isinstance(row["unchanged_item_count"], int)
or isinstance(row["unchanged_item_count"], bool)
or row["unchanged_item_count"] < 0
):
raise ApplyPublicationError("online rollback evidence is invalid")
expected_participants = ["ai-platform", "contract-platform"]
@@ -618,24 +689,33 @@ def _compensate(
after = row["participant_states_after"]
if not isinstance(before, list) or not isinstance(after, list):
raise ApplyPublicationError("online rollback participant evidence is invalid")
for label, states, allowed in (
("before", before, {"not_started", "prepared", "committed", "failed"}),
("after", after, {"unchanged", "compensated_and_verified"}),
):
validated_states: dict[str, dict[str, dict[str, Any]]] = {}
for label, states in (("before", before), ("after", after)):
if [item.get("participant") for item in states if isinstance(item, dict)] != expected_participants:
raise ApplyPublicationError(f"online rollback {label} participant identities drift")
validated_states[label] = {}
for index, value in enumerate(states):
state_row = _require_exact(
value,
f"online rollback {label} participant {index}",
("participant", "state", "write_count", "durable_state_sha256"),
(
"participant",
"state",
"write_count",
"runtime_snapshot_count",
"durable_state_sha256",
),
)
if (
state_row["state"] not in allowed
not isinstance(state_row["state"], str)
or not isinstance(state_row["write_count"], int)
or isinstance(state_row["write_count"], bool)
or state_row["write_count"] < 0
or state_row["write_count"] > 74
or not isinstance(state_row["runtime_snapshot_count"], int)
or isinstance(state_row["runtime_snapshot_count"], bool)
or state_row["runtime_snapshot_count"] < 0
or state_row["runtime_snapshot_count"] > 14
):
raise ApplyPublicationError(
f"online rollback {label} participant state drift"
@@ -654,16 +734,60 @@ def _compensate(
raise ApplyPublicationError(
f"online rollback {label} participant durable state hash drift"
)
validated_states[label][state_row["participant"]] = state_row
ai_before = validated_states["before"]["ai-platform"]
contract_before = validated_states["before"]["contract-platform"]
ai_was_committed = ai_before["state"] == "committed"
if ai_was_committed:
if (
ai_before["write_count"] != 14
or ai_before["runtime_snapshot_count"] != 14
):
raise ApplyPublicationError(
"online rollback AI committed state must contain exact 14+14 counts"
)
elif (
ai_before["state"] not in {"not_started", "prepared"}
or ai_before["write_count"] != 0
or ai_before["runtime_snapshot_count"] != 0
):
raise ApplyPublicationError("online rollback AI before state is not compensatable")
if (
contract_before["state"] not in {"not_started", "prepared"}
or contract_before["write_count"] != 0
or contract_before["runtime_snapshot_count"] != 0
):
raise ApplyPublicationError(
"online rollback Contract before state crossed the irreversible boundary"
)
expected_after = {
"ai-platform": "compensated_and_verified" if ai_was_committed else "unchanged",
"contract-platform": "unchanged",
}
for participant, expected_state in expected_after.items():
state_row = validated_states["after"][participant]
if (
state_row["state"] != expected_state
or state_row["write_count"] != 0
or state_row["runtime_snapshot_count"] != 0
):
raise ApplyPublicationError(
f"online rollback {participant} after state is not exact"
)
item_results = row["compensation_item_results"]
if not isinstance(item_results, list):
raise ApplyPublicationError("online rollback item results are invalid")
identities: set[tuple[str, str]] = set()
expected_identities = {
expected_identities = sorted(
(item["kind"], item["item_id"])
for item in batch._expected_apply_item_results(manifest)
}
if item["kind"] == "prompt_module"
)
supplied_identities: list[tuple[str, str]] = []
restored_count = 0
removed_count = 0
unchanged_count = 0
for index, value in enumerate(item_results):
item = _require_exact(
value,
@@ -678,33 +802,74 @@ def _compensate(
),
)
identity = (item["kind"], item["item_id"])
if (
identity in identities
or identity not in expected_identities
or item["status"] != "verified"
):
raise ApplyPublicationError("online rollback item identity/status drift")
identities.add(identity)
if item["action"] == "restored_previous":
restored_count += 1
batch._require_sha(
item["pre_apply_content_sha256"],
"online rollback restored content SHA-256",
supplied_identities.append(identity)
restored, removed, unchanged = _validate_compensation_hashes(
item, f"online rollback item {index}"
)
restored_count += restored
removed_count += removed
unchanged_count += unchanged
required_publication_identities = expected_identities if ai_was_committed else []
if supplied_identities != required_publication_identities:
raise ApplyPublicationError(
"online rollback Prompt compensation identities/order drift"
)
runtime_results = row["runtime_snapshot_compensation_results"]
if not isinstance(runtime_results, list):
raise ApplyPublicationError(
"online rollback runtime snapshot results are invalid"
)
runtime_expectations = {
(item["kind"], item["runtime_key"], item["version"]): item
for item in expected_runtime_snapshot_results(manifest)
}
expected_runtime_identities = sorted(runtime_expectations)
supplied_runtime_identities: list[tuple[str, str, str]] = []
for index, value in enumerate(runtime_results):
item = _require_exact(
value,
f"online rollback runtime snapshot item {index}",
(
"kind",
"runtime_key",
"version",
"pre_apply_content_sha256",
"post_compensation_content_sha256",
"action",
"status",
),
)
identity = (item["kind"], item["runtime_key"], item["version"])
supplied_runtime_identities.append(identity)
expected_runtime = runtime_expectations.get(identity)
unchanged_sha256 = (
expected_runtime["expected_content_sha256"]
if expected_runtime is not None
and expected_runtime["kind"] == "existing_domain_prompt"
else None
)
restored, removed, unchanged = _validate_compensation_hashes(
item,
f"online rollback runtime snapshot item {index}",
unchanged_preexisting_sha256=unchanged_sha256,
)
if unchanged_sha256 is not None and item["action"] != "unchanged_preexisting":
raise ApplyPublicationError(
"online rollback preexisting runtime dependency was falsely compensated"
)
if item["post_compensation_content_sha256"] != item["pre_apply_content_sha256"]:
raise ApplyPublicationError("online rollback restored content hash drift")
elif item["action"] == "removed_new":
removed_count += 1
if (
item["pre_apply_content_sha256"] is not None
or item["post_compensation_content_sha256"] is not None
):
raise ApplyPublicationError("online rollback removal evidence drift")
else:
raise ApplyPublicationError("online rollback action is unsupported")
restored_count += restored
removed_count += removed
unchanged_count += unchanged
required_runtime_identities = expected_runtime_identities if ai_was_committed else []
if supplied_runtime_identities != required_runtime_identities:
raise ApplyPublicationError(
"online rollback runtime snapshot identities/order drift"
)
if (
restored_count != row["restored_item_count"]
or removed_count != row["removed_item_count"]
or unchanged_count != row["unchanged_item_count"]
):
raise ApplyPublicationError("online rollback item counts drift")
rollback_sha = batch._require_sha(
@@ -826,6 +991,9 @@ def execute_online_publication_batch(
) -> dict[str, Any]:
"""Execute the online saga and persist evidence pending offline countersign."""
if now.tzinfo is None or now.utcoffset() is None:
raise ApplyPublicationError("online apply time must include a timezone")
checked_at = now.astimezone(UTC)
batch.validate_manifest(manifest)
if manifest["runtime_prompt_dependencies"]["items"] != batch._load_executable_prompt_dependencies(
REPO_ROOT
@@ -844,6 +1012,14 @@ def execute_online_publication_batch(
approval_keyring,
approval_root_public_key,
)
approval_approved_at = batch._parse_time(
approval["payload"]["approved_at"], "approval approved_at"
)
approval_expires_at = batch._parse_time(
approval["payload"]["expires_at"], "approval expires_at"
)
if checked_at < approval_approved_at or checked_at >= approval_expires_at:
raise ApplyPublicationError("approval is not currently valid for online apply")
legal_review_summary, legal_review_binding_sha256 = validate_scene_legal_review(
manifest,
approval,
@@ -869,7 +1045,7 @@ def execute_online_publication_batch(
tenant_id,
)
transaction_id = _transaction_id(key)
batch.validate_apply_challenge(
validated_challenge = batch.validate_apply_challenge(
manifest,
manifest_raw_sha256,
approval_raw_sha256,
@@ -880,6 +1056,13 @@ def execute_online_publication_batch(
environment_id=environment_id,
tenant_id=tenant_id,
)
challenge_issued_at = batch._parse_time(
validated_challenge["payload"]["issued_at"], "apply challenge issued_at"
)
challenge_expires_at = batch._parse_time(
validated_challenge["payload"]["expires_at"], "apply challenge expires_at"
)
challenge_is_current = challenge_issued_at <= checked_at < challenge_expires_at
request_payload = {
"schema_version": APPLY_REQUEST_SCHEMA_VERSION,
"idempotency_key": key,
@@ -906,7 +1089,12 @@ def execute_online_publication_batch(
"expected_runtime_snapshot_dependency_count": 14,
}
request_sha256 = batch._sha256_json(request_payload)
replay = ledger.reserve(key, request_sha256, now)
reservation, replay = ledger.reserve(
key,
request_sha256,
now,
initial_dispatch_authorized=challenge_is_current,
)
if replay is not None:
if replay.get("schema_version") == FINAL_EVIDENCE_SCHEMA_VERSION:
replay = replay.get("online_execution")
@@ -932,6 +1120,49 @@ def execute_online_publication_batch(
legal_review_binding_sha256,
tenant_id,
)
if reservation == "recover" and not challenge_is_current:
try:
reconciliation = client.request(
"GET",
f"/internal/rc06/publication-transactions/{transaction_id}/status",
payload=None,
headers=headers,
)
remote_state = _remote_status_state(
reconciliation,
key=key,
transaction_id=transaction_id,
)
except Exception as exc:
ledger.state(
key,
"commit_status_unknown",
{"reconciliation_error": str(exc)},
now,
error="expired challenge recovery could not establish remote commit status",
)
raise ApplyPublicationError(
"expired challenge cannot be retransmitted until an existing terminal commit is proven"
) from exc
if remote_state == "rolled_back":
ledger.state(key, "rolled_back", reconciliation, now)
raise ApplyPublicationError("online apply was already rolled back")
if remote_state not in {
"contract_committed",
"evidence_pending",
"committed_and_verified",
"online_committed_pending_countersign",
}:
ledger.state(
key,
"expired_challenge_recovery_no_go",
reconciliation,
now,
error="remote transaction was not terminally committed before challenge expiry",
)
raise ApplyPublicationError(
"expired challenge cannot authorize a new or non-terminal retransmission"
)
dispatch_started = False
response: Any = None
try:
yuqei-infra/scripts/build-rc06-publication-batch.py
+151 -17
View File
@@ -40,10 +40,14 @@ APPROVAL_REVIEW_SUMMARY_SCHEMA_VERSION = "yuqei.rc06-single-reviewer-summary.v2"
APPROVAL_KEYRING_SCHEMA_VERSION = "yuqei.rc06-publication-approval-keyring.v4"
APPLY_SCHEMA_VERSION = "yuqei.rc06-publication-system-item-apply-receipt.v6"
APPLY_PAYLOAD_SCHEMA_VERSION = "yuqei.rc06-publication-system-item-apply-payload.v4"
ONLINE_TRANSACTION_RECEIPT_SCHEMA_VERSION = "yuqei.rc06-online-transaction-receipt.v4"
ONLINE_TRANSACTION_RECEIPT_SCHEMA_VERSION = "yuqei.rc06-online-transaction-receipt.v5"
ONLINE_TRANSACTION_PAYLOAD_SCHEMA_VERSION = "yuqei.rc06-online-transaction-payload.v4"
RUNTIME_RECEIPT_SCHEMA_VERSION = "yuqei.rc06-runtime-snapshot-receipt.v2"
RUNTIME_RECEIPT_SCHEMA_VERSION = "yuqei.rc06-runtime-snapshot-receipt.v3"
RUNTIME_PAYLOAD_SCHEMA_VERSION = "yuqei.rc06-runtime-snapshot-payload.v2"
KMS_REQUEST_ID_ATTESTATION_SCHEMA_VERSION = "yuqei.rc06-kms-request-id-attestation.v1"
KMS_REQUEST_ID_ATTESTATION_PAYLOAD_SCHEMA_VERSION = (
"yuqei.rc06-kms-request-id-attestation-payload.v1"
)
APPLY_CHALLENGE_SCHEMA_VERSION = "yuqei.rc06-apply-challenge.v2"
APPLY_CHALLENGE_PAYLOAD_SCHEMA_VERSION = "yuqei.rc06-apply-challenge-payload.v2"
APPROVAL_SIGNATURE_DOMAIN = b"YUQEI_RC06_PUBLICATION_BATCH_APPROVAL_V1\0"
@@ -51,6 +55,12 @@ APPROVAL_KEYRING_SIGNATURE_DOMAIN = b"YUQEI_RC06_APPROVAL_KEYRING_V1\0"
APPLY_SIGNATURE_DOMAIN = b"YUQEI_RC06_SYSTEM_APPLY_RECEIPT_V2\0"
ONLINE_TRANSACTION_SIGNATURE_DOMAIN = b"YUQEI_RC06_ONLINE_TRANSACTION_RECEIPT_V2\0"
RUNTIME_SIGNATURE_DOMAIN = b"YUQEI_RC06_RUNTIME_SNAPSHOT_RECEIPT_V2\0"
ONLINE_REQUEST_ID_ATTESTATION_DOMAIN = (
b"YUQEI_RC06_ONLINE_TRANSACTION_REQUEST_ID_ATTESTATION_V1\0"
)
RUNTIME_REQUEST_ID_ATTESTATION_DOMAIN = (
b"YUQEI_RC06_RUNTIME_SNAPSHOT_REQUEST_ID_ATTESTATION_V1\0"
)
APPLY_CHALLENGE_SIGNATURE_DOMAIN = b"YUQEI_RC06_APPLY_CHALLENGE_V2\0"
APPLY_OPERATION = "apply_rc06_publication_batch"
TRANSACTION_MODE = "durable_saga"
@@ -3144,27 +3154,28 @@ def _read_pinned_spki_public_key(path: Path, *, expected_sha256: str, label: str
return public_key
def _verify_online_kms_signature(
def _verify_frozen_kms_signature(
public_key: Any,
signature_block: Any,
*,
domain: bytes,
payload: Mapping[str, Any],
label: str,
) -> dict[str, Any]:
include_request_id: bool,
) -> tuple[dict[str, Any], bytes]:
"""Verify one KMS digest signature against exact frozen signer metadata."""
fields = (
"provider",
"key_reference",
"key_version",
"algorithm",
"value_base64",
) + (("request_id",) if include_request_id else ())
signature = _require_keys(
signature_block,
f"{label} signature",
(
"provider",
"key_reference",
"key_version",
"algorithm",
"value_base64",
"request_id",
),
fields,
)
required_trust = (
TRUSTED_ONLINE_APPLY_SERVICE_PROVIDER,
@@ -3188,7 +3199,10 @@ def _verify_online_kms_signature(
or signature["algorithm"] != TRUSTED_ONLINE_APPLY_SERVICE_ALGORITHM
):
raise PublicationBatchError(f"{label} KMS signer identity drift")
_require_string(signature["request_id"], f"{label} KMS request_id", max_length=512)
if include_request_id:
_require_string(
signature["request_id"], f"{label} KMS request_id", max_length=512
)
if (
_sha256_bytes(_public_key_spki(public_key))
!= TRUSTED_ONLINE_APPLY_SERVICE_PUBLIC_KEY_SPKI_SHA256
@@ -3234,9 +3248,83 @@ def _verify_online_kms_signature(
raise PublicationBatchError(f"{label} KMS algorithm/key type is unsupported")
except (TypeError, ValueError, InvalidSignature) as exc:
raise PublicationBatchError(f"{label} KMS signature verification failed") from exc
return signature, signature_bytes
def _verify_online_kms_signature(
public_key: Any,
signature_block: Any,
*,
domain: bytes,
payload: Mapping[str, Any],
label: str,
) -> dict[str, Any]:
signature, _ = _verify_frozen_kms_signature(
public_key,
signature_block,
domain=domain,
payload=payload,
label=label,
include_request_id=True,
)
return signature
def _verify_kms_request_id_attestation(
public_key: Any,
attestation_value: Any,
*,
receipt_payload: Mapping[str, Any],
primary_signature: Mapping[str, Any],
attestation_domain: bytes,
label: str,
) -> dict[str, Any]:
attestation = _require_keys(
attestation_value,
f"{label} request_id attestation",
("schema_version", "payload", "signature"),
)
if attestation["schema_version"] != KMS_REQUEST_ID_ATTESTATION_SCHEMA_VERSION:
raise PublicationBatchError(f"{label} request_id attestation schema is unsupported")
try:
primary_signature_bytes = base64.b64decode(
_require_string(
primary_signature["value_base64"],
f"{label} primary signature value",
max_length=16384,
),
validate=True,
)
except (TypeError, ValueError) as exc:
raise PublicationBatchError(f"{label} primary signature encoding is invalid") from exc
expected_payload = {
"schema_version": KMS_REQUEST_ID_ATTESTATION_PAYLOAD_SCHEMA_VERSION,
"receipt_payload_sha256": _sha256_json(receipt_payload),
"primary_signature_sha256": _sha256_bytes(primary_signature_bytes),
"primary_provider": primary_signature["provider"],
"primary_key_reference": primary_signature["key_reference"],
"primary_key_version": primary_signature["key_version"],
"primary_algorithm": primary_signature["algorithm"],
"primary_request_id": primary_signature["request_id"],
}
payload = _require_keys(
attestation["payload"],
f"{label} request_id attestation payload",
tuple(expected_payload),
)
if payload != expected_payload:
raise PublicationBatchError(f"{label} request_id attestation binding drift")
_verify_frozen_kms_signature(
public_key,
attestation["signature"],
domain=attestation_domain,
payload=payload,
label=f"{label} request_id attestation",
include_request_id=False,
)
return attestation
def _read_pinned_raw_public_key(path: Path, *, expected_sha256: str, label: str) -> Ed25519PublicKey:
if not HEX64_RE.fullmatch(expected_sha256):
raise PublicationBatchError(f"{label} pin is not provisioned in the frozen source")
@@ -4059,7 +4147,13 @@ def validate_online_transaction_receipt(
envelope = _require_keys(
receipt,
"online transaction receipt",
("schema_version", "receipt_id", "payload", "signature"),
(
"schema_version",
"receipt_id",
"payload",
"signature",
"request_id_attestation",
),
)
if envelope["schema_version"] != ONLINE_TRANSACTION_RECEIPT_SCHEMA_VERSION:
raise PublicationBatchError("online transaction receipt schema is unsupported")
@@ -4325,9 +4419,21 @@ def validate_online_transaction_receipt(
payload=payload,
label="online transaction receipt",
)
request_id_attestation = _verify_kms_request_id_attestation(
online_service_public_key,
envelope["request_id_attestation"],
receipt_payload=payload,
primary_signature=signature,
attestation_domain=ONLINE_REQUEST_ID_ATTESTATION_DOMAIN,
label="online transaction receipt",
)
expected_id = (
"rc06-online-transaction-"
+ _sha256_bytes(_canonical(payload) + _canonical(signature))[:24]
+ _sha256_bytes(
_canonical(payload)
+ _canonical(signature)
+ _canonical(request_id_attestation)
)[:24]
)
if envelope["receipt_id"] != expected_id:
raise PublicationBatchError(
@@ -4370,7 +4476,13 @@ def validate_runtime_snapshot_receipt(
envelope = _require_keys(
receipt,
"runtime snapshot receipt",
("schema_version", "receipt_id", "payload", "signature"),
(
"schema_version",
"receipt_id",
"payload",
"signature",
"request_id_attestation",
),
)
if envelope["schema_version"] != RUNTIME_RECEIPT_SCHEMA_VERSION:
raise PublicationBatchError("runtime snapshot receipt schema is unsupported")
@@ -4458,9 +4570,21 @@ def validate_runtime_snapshot_receipt(
payload=payload,
label="runtime snapshot receipt",
)
request_id_attestation = _verify_kms_request_id_attestation(
online_service_public_key,
envelope["request_id_attestation"],
receipt_payload=payload,
primary_signature=signature,
attestation_domain=RUNTIME_REQUEST_ID_ATTESTATION_DOMAIN,
label="runtime snapshot receipt",
)
expected_id = (
"rc06-runtime-snapshot-"
+ _sha256_bytes(_canonical(payload) + _canonical(signature))[:24]
+ _sha256_bytes(
_canonical(payload)
+ _canonical(signature)
+ _canonical(request_id_attestation)
)[:24]
)
if envelope["receipt_id"] != expected_id:
raise PublicationBatchError(
@@ -4570,6 +4694,12 @@ def validate_apply_receipt(
row["countersigned_at"], "apply countersigned_at"
)
approval_payload = approval["payload"]
approval_expires_at = _parse_time(
approval_payload["expires_at"], "approval expires_at"
)
legal_expires_at = _parse_time(
legal_review_expires_at, "legal review expires_at"
)
actor = _require_keys(row["generated_by"], "apply generator", ("type", "component"))
if actor != {"type": "system", "component": "rc06-apply-authority"}:
raise PublicationBatchError("per-item apply evidence must be authority generated")
@@ -4633,6 +4763,10 @@ def validate_apply_receipt(
)
if countersigned_at < authorized_commit_at:
raise PublicationBatchError("apply countersign predates runtime read-back evidence")
if countersigned_at >= approval_expires_at:
raise PublicationBatchError("apply countersign is outside the approval window")
if countersigned_at >= legal_expires_at:
raise PublicationBatchError("apply countersign is outside the legal review window")
if not HEX64_RE.fullmatch(TRUSTED_APPLY_SERVICE_PUBLIC_KEY_SHA256):
raise PublicationBatchError("apply service public key pin is not provisioned")
service_public_raw = _public_key_raw(apply_service_public_key)
yuqei-infra/tests/test_rc06_publication_apply.py
+435 -4
View File
@@ -93,14 +93,54 @@ def _kms_envelope(
"value_base64": base64.b64encode(signature_value).decode("ascii"),
"request_id": request_id,
}
attestation_payload = {
"schema_version": apply.batch.KMS_REQUEST_ID_ATTESTATION_PAYLOAD_SCHEMA_VERSION,
"receipt_payload_sha256": apply.batch._sha256_json(payload),
"primary_signature_sha256": apply.batch._sha256_bytes(signature_value),
"primary_provider": signature["provider"],
"primary_key_reference": signature["key_reference"],
"primary_key_version": signature["key_version"],
"primary_algorithm": signature["algorithm"],
"primary_request_id": request_id,
}
attestation_domain = (
apply.batch.ONLINE_REQUEST_ID_ATTESTATION_DOMAIN
if domain == apply.batch.ONLINE_TRANSACTION_SIGNATURE_DOMAIN
else apply.batch.RUNTIME_REQUEST_ID_ATTESTATION_DOMAIN
)
attestation_digest = hashlib.sha256(
attestation_domain + apply.batch._canonical(attestation_payload)
).digest()
attestation_signature_value = private_key.sign(
attestation_digest,
padding.PSS(
mgf=padding.MGF1(hashes.SHA256()),
salt_length=hashes.SHA256().digest_size,
),
utils.Prehashed(hashes.SHA256()),
)
request_id_attestation = {
"schema_version": apply.batch.KMS_REQUEST_ID_ATTESTATION_SCHEMA_VERSION,
"payload": attestation_payload,
"signature": {
"provider": signature["provider"],
"key_reference": signature["key_reference"],
"key_version": signature["key_version"],
"algorithm": signature["algorithm"],
"value_base64": base64.b64encode(attestation_signature_value).decode("ascii"),
},
}
return {
"schema_version": schema,
"receipt_id": id_prefix
+ apply.batch._sha256_bytes(
apply.batch._canonical(payload) + apply.batch._canonical(signature)
apply.batch._canonical(payload)
+ apply.batch._canonical(signature)
+ apply.batch._canonical(request_id_attestation)
)[:24],
"payload": payload,
"signature": signature,
"request_id_attestation": request_id_attestation,
}
@@ -619,11 +659,17 @@ class FakePublicationService:
return 200, self.response
@staticmethod
def _durable_state(participant: str, state: str, write_count: int) -> dict:
def _durable_state(
participant: str,
state: str,
write_count: int,
runtime_snapshot_count: int = 0,
) -> dict:
projection = {
"participant": participant,
"state": state,
"write_count": write_count,
"runtime_snapshot_count": runtime_snapshot_count,
}
return {
**projection,
@@ -637,8 +683,8 @@ class FakePublicationService:
self._durable_state("contract-platform", "prepared", 0),
]
after = [
self._durable_state("ai-platform", "compensated_and_verified", 0),
self._durable_state("contract-platform", "compensated_and_verified", 0),
self._durable_state("ai-platform", "unchanged", 0),
self._durable_state("contract-platform", "unchanged", 0),
]
projection = {
"schema_version": apply.ROLLBACK_RESPONSE_SCHEMA_VERSION,
@@ -648,8 +694,10 @@ class FakePublicationService:
"participant_states_before": before,
"participant_states_after": after,
"compensation_item_results": [],
"runtime_snapshot_compensation_results": [],
"restored_item_count": 0,
"removed_item_count": 0,
"unchanged_item_count": 0,
}
self.state = "rolled_back"
return 200, {
@@ -913,6 +961,136 @@ def _service(material: dict, **kwargs) -> FakePublicationService:
)
def _rollback_state(
participant: str,
state: str,
write_count: int,
runtime_snapshot_count: int,
) -> dict:
projection = {
"participant": participant,
"state": state,
"write_count": write_count,
"runtime_snapshot_count": runtime_snapshot_count,
}
return {
**projection,
"durable_state_sha256": apply.batch._sha256_json(projection),
}
def _rehash_rollback(response: dict) -> None:
for states in (
response["participant_states_before"],
response["participant_states_after"],
):
for state in states:
state["durable_state_sha256"] = apply.batch._sha256_json(
{
key: value
for key, value in state.items()
if key != "durable_state_sha256"
}
)
actions = [
item["action"]
for key in (
"compensation_item_results",
"runtime_snapshot_compensation_results",
)
for item in response[key]
]
response["restored_item_count"] = actions.count("restored_previous")
response["removed_item_count"] = actions.count("removed_new")
response["unchanged_item_count"] = actions.count("unchanged_preexisting")
response["rollback_evidence_sha256"] = apply.batch._sha256_json(
{
key: value
for key, value in response.items()
if key != "rollback_evidence_sha256"
}
)
def _ai_committed_rollback_response(material: dict) -> dict:
prompt_results = [
{
"kind": item["kind"],
"item_id": item["item_id"],
"pre_apply_content_sha256": None,
"post_compensation_content_sha256": None,
"action": "removed_new",
"status": "verified",
}
for item in apply.batch._expected_apply_item_results(material["manifest"])
if item["kind"] == "prompt_module"
]
prompt_results.sort(key=lambda item: (item["kind"], item["item_id"]))
runtime_results = []
for item in apply.expected_runtime_snapshot_results(material["manifest"]):
preexisting = item["kind"] == "existing_domain_prompt"
runtime_results.append(
{
"kind": item["kind"],
"runtime_key": item["runtime_key"],
"version": item["version"],
"pre_apply_content_sha256": (
item["expected_content_sha256"] if preexisting else None
),
"post_compensation_content_sha256": (
item["expected_content_sha256"] if preexisting else None
),
"action": "unchanged_preexisting" if preexisting else "removed_new",
"status": "verified",
}
)
runtime_results.sort(
key=lambda item: (item["kind"], item["runtime_key"], item["version"])
)
response = {
"schema_version": apply.ROLLBACK_RESPONSE_SCHEMA_VERSION,
"idempotency_key": material["idempotency_key"],
"transaction_id": material["transaction_id"],
"state": "rolled_back",
"participant_states_before": [
_rollback_state("ai-platform", "committed", 14, 14),
_rollback_state("contract-platform", "prepared", 0, 0),
],
"participant_states_after": [
_rollback_state("ai-platform", "compensated_and_verified", 0, 0),
_rollback_state("contract-platform", "unchanged", 0, 0),
],
"compensation_item_results": prompt_results,
"runtime_snapshot_compensation_results": runtime_results,
"restored_item_count": 0,
"removed_item_count": 0,
"unchanged_item_count": 0,
"rollback_evidence_sha256": "0" * 64,
}
_rehash_rollback(response)
return response
def _validate_rollback_response(material: dict, response: dict) -> dict:
class StaticRollbackClient:
def request(self, method, path, *, payload, headers):
assert method == "POST" and path.endswith("/rollback")
return response
return apply._compensate(
StaticRollbackClient(),
manifest=material["manifest"],
key=material["idempotency_key"],
transaction_id=material["transaction_id"],
headers={},
manifest_raw_sha256=material["manifest_raw"],
approval_raw_sha256=material["approval_raw"],
legal_review_binding_sha256=material["legal_binding"],
environment_id=material["environment_id"],
tenant_id=material["tenant_id"],
)
def test_online_execute_then_offline_countersign_is_durable_and_idempotent(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
@@ -933,6 +1111,9 @@ def test_online_execute_then_offline_countersign_is_durable_and_idempotent(
# The short challenge has expired, but the signed commit and runtime
# facts were captured inside it. Delayed offline finalization remains valid.
countersigned_at = material["now"] + timedelta(minutes=10)
assert countersigned_at > datetime.fromisoformat(
material["authority_challenge"]["payload"]["expires_at"]
)
receipt = _countersign(material, execution, countersigned_at)
final = _finalize(
material, client, ledger, execution, receipt, countersigned_at
@@ -954,6 +1135,172 @@ def test_online_execute_then_offline_countersign_is_durable_and_idempotent(
assert service.rollback_calls == 0
@pytest.mark.parametrize(
("expiry_field", "expected_error"),
(
("legal_expires_at", "outside the legal review window"),
("approval_expires_at", "outside the approval window"),
),
)
def test_validly_resigned_countersign_cannot_outlive_approval_or_legal_authority(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
expiry_field: str,
expected_error: str,
) -> None:
material = _material(monkeypatch, datetime(2026, 10, 6, 8, 0, tzinfo=UTC))
service = _service(material)
expiry = (
datetime.fromisoformat(material["legal_expires_at"])
if expiry_field == "legal_expires_at"
else datetime.fromisoformat(material["approval"]["payload"]["expires_at"])
)
with _serve(service) as url:
client = apply.JsonHttpClient(url, "test-token", allow_http_loopback=True)
ledger = apply.ApplyLedger(tmp_path / f"{expiry_field}.sqlite3")
try:
execution = _execute(material, client, ledger)
receipt = _countersign(material, execution, expiry)
with pytest.raises(apply.batch.PublicationBatchError, match=expected_error):
_finalize(material, client, ledger, execution, receipt, expiry)
finally:
ledger.close()
assert service.finalize_calls == 0
assert service.rollback_calls == 0
def test_expired_challenge_is_rejected_before_initial_dispatch(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
material = _material(monkeypatch, datetime(2026, 10, 6, 8, 0, tzinfo=UTC))
payload = copy.deepcopy(material["authority_challenge"]["payload"])
payload["issued_at"] = (material["now"] - timedelta(minutes=10)).isoformat(
timespec="seconds"
)
payload["expires_at"] = (material["now"] - timedelta(minutes=5)).isoformat(
timespec="seconds"
)
material["authority_challenge"] = _ed_envelope(
material["apply_private"],
schema=apply.batch.APPLY_CHALLENGE_SCHEMA_VERSION,
id_field="challenge_id",
id_prefix="rc06-apply-challenge-",
domain=apply.batch.APPLY_CHALLENGE_SIGNATURE_DOMAIN,
payload=payload,
)
service = _service(material)
with _serve(service) as url:
client = apply.JsonHttpClient(url, "test-token", allow_http_loopback=True)
ledger = apply.ApplyLedger(tmp_path / "ledger.sqlite3")
try:
with pytest.raises(
apply.ApplyPublicationError,
match="not currently valid for an initial dispatch",
):
_execute(material, client, ledger)
operation_count = ledger.connection.execute(
"SELECT COUNT(*) FROM rc06_apply_operations"
).fetchone()[0]
finally:
ledger.close()
assert operation_count == 0
assert service.apply_calls == 0
def test_expired_approval_is_rejected_independently_before_http(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
material = _material(monkeypatch, datetime(2026, 10, 6, 8, 0, tzinfo=UTC))
material["now"] = datetime.fromisoformat(
material["approval"]["payload"]["expires_at"]
)
service = _service(material)
with _serve(service) as url:
client = apply.JsonHttpClient(url, "test-token", allow_http_loopback=True)
ledger = apply.ApplyLedger(tmp_path / "ledger.sqlite3")
try:
with pytest.raises(
apply.ApplyPublicationError,
match="approval is not currently valid",
):
_execute(material, client, ledger)
finally:
ledger.close()
assert service.apply_calls == 0
@pytest.mark.parametrize("crash_stage", ("dispatching", "verifying"))
def test_restart_recovers_committed_dispatch_after_short_challenge_expires(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, crash_stage: str
) -> None:
material = _material(monkeypatch, datetime(2026, 10, 6, 8, 0, tzinfo=UTC))
service = _service(material)
class CrashAfterCommittedResponse:
def __init__(self, delegate) -> None:
self.delegate = delegate
self.crashed = False
def request(self, method, path, *, payload, headers):
response = self.delegate.request(
method, path, payload=payload, headers=headers
)
if (
not self.crashed
and method == "POST"
and path == "/internal/rc06/publication-transactions/apply"
):
self.crashed = True
raise KeyboardInterrupt("simulated process loss after remote commit")
return response
ledger_path = tmp_path / "ledger.sqlite3"
with _serve(service) as url:
client = apply.JsonHttpClient(url, "test-token", allow_http_loopback=True)
ledger = apply.ApplyLedger(ledger_path)
try:
crash_client = client
if crash_stage == "dispatching":
crash_client = CrashAfterCommittedResponse(client)
else:
original_state = ledger.state
def crash_after_verifying_state(*args, **kwargs):
original_state(*args, **kwargs)
if args[1] == "verifying":
raise KeyboardInterrupt(
"simulated process loss after verifying state"
)
monkeypatch.setattr(ledger, "state", crash_after_verifying_state)
with pytest.raises(KeyboardInterrupt, match="simulated process loss"):
_execute(material, crash_client, ledger)
state = ledger.connection.execute(
"SELECT state FROM rc06_apply_operations"
).fetchone()[0]
assert state == crash_stage
finally:
ledger.close()
material["now"] += timedelta(minutes=10)
assert material["now"] > datetime.fromisoformat(
material["authority_challenge"]["payload"]["expires_at"]
)
ledger = apply.ApplyLedger(ledger_path)
try:
execution = _execute(material, client, ledger)
state = ledger.connection.execute(
"SELECT state FROM rc06_apply_operations"
).fetchone()[0]
finally:
ledger.close()
assert execution["schema_version"] == apply.EXECUTION_EVIDENCE_SCHEMA_VERSION
assert state == "online_committed_pending_countersign"
assert service.apply_calls == 2
assert service.rollback_calls == 0
@pytest.mark.parametrize("tamper_status", ["publication", "runtime"])
def test_terminal_commit_verification_failure_never_rolls_back(
tmp_path: Path,
@@ -978,6 +1325,90 @@ def test_terminal_commit_verification_failure_never_rolls_back(
assert service.rollback_calls == 0
def test_ai_committed_compensation_requires_exact_14_plus_14_evidence(
monkeypatch: pytest.MonkeyPatch,
) -> None:
material = _material(monkeypatch, datetime(2026, 10, 6, 8, 0, tzinfo=UTC))
response = _ai_committed_rollback_response(material)
assert _validate_rollback_response(material, response) == response
assert len(response["compensation_item_results"]) == 14
assert len(response["runtime_snapshot_compensation_results"]) == 14
assert response["removed_item_count"] == 23
assert response["unchanged_item_count"] == 5
@pytest.mark.parametrize(
"mutation",
(
"zero_zero_empty",
"missing_prompt",
"duplicate_prompt",
"extra_prompt",
"missing_runtime",
"duplicate_runtime",
"wrong_runtime_count",
"contract_committed",
"preexisting_falsely_removed",
),
)
def test_ai_committed_compensation_rejects_incomplete_or_false_evidence(
monkeypatch: pytest.MonkeyPatch,
mutation: str,
) -> None:
material = _material(monkeypatch, datetime(2026, 10, 6, 8, 0, tzinfo=UTC))
response = _ai_committed_rollback_response(material)
if mutation == "zero_zero_empty":
response["participant_states_before"][0]["write_count"] = 0
response["participant_states_before"][0]["runtime_snapshot_count"] = 0
response["compensation_item_results"] = []
response["runtime_snapshot_compensation_results"] = []
elif mutation == "missing_prompt":
response["compensation_item_results"].pop()
elif mutation == "duplicate_prompt":
response["compensation_item_results"][-1] = copy.deepcopy(
response["compensation_item_results"][0]
)
elif mutation == "extra_prompt":
response["compensation_item_results"].append(
{
"kind": "asset",
"item_id": material["manifest"]["asset_batch"]["items"][0][
"template_id"
],
"pre_apply_content_sha256": None,
"post_compensation_content_sha256": None,
"action": "removed_new",
"status": "verified",
}
)
elif mutation == "missing_runtime":
response["runtime_snapshot_compensation_results"].pop()
elif mutation == "duplicate_runtime":
response["runtime_snapshot_compensation_results"][-1] = copy.deepcopy(
response["runtime_snapshot_compensation_results"][0]
)
elif mutation == "wrong_runtime_count":
response["participant_states_before"][0]["runtime_snapshot_count"] = 13
elif mutation == "contract_committed":
contract = response["participant_states_before"][1]
contract["state"] = "committed"
contract["write_count"] = 60
elif mutation == "preexisting_falsely_removed":
item = next(
value
for value in response["runtime_snapshot_compensation_results"]
if value["kind"] == "existing_domain_prompt"
)
item["action"] = "removed_new"
item["pre_apply_content_sha256"] = None
item["post_compensation_content_sha256"] = None
else: # pragma: no cover - the parameter inventory is closed above.
raise AssertionError(mutation)
_rehash_rollback(response)
with pytest.raises(apply.ApplyPublicationError, match="rollback"):
_validate_rollback_response(material, response)
def test_only_pre_terminal_state_is_compensated_with_structured_evidence(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
yuqei-infra/tests/test_rc06_publication_batch.py
+181 -1
View File
@@ -1115,6 +1115,45 @@ def test_one_reviewer_batch_signature_and_system_item_receipts_remove_human_repe
"value_base64": base64.b64encode(signature_value).decode("ascii"),
"request_id": request_id,
}
attestation_payload = {
"schema_version": batch.KMS_REQUEST_ID_ATTESTATION_PAYLOAD_SCHEMA_VERSION,
"receipt_payload_sha256": batch._sha256_json(payload),
"primary_signature_sha256": batch._sha256_bytes(signature_value),
"primary_provider": signature_block["provider"],
"primary_key_reference": signature_block["key_reference"],
"primary_key_version": signature_block["key_version"],
"primary_algorithm": signature_block["algorithm"],
"primary_request_id": request_id,
}
attestation_domain = (
batch.ONLINE_REQUEST_ID_ATTESTATION_DOMAIN
if domain == batch.ONLINE_TRANSACTION_SIGNATURE_DOMAIN
else batch.RUNTIME_REQUEST_ID_ATTESTATION_DOMAIN
)
attestation_digest = hashlib.sha256(
attestation_domain + batch._canonical(attestation_payload)
).digest()
attestation_signature_value = online_service_private_key.sign(
attestation_digest,
padding.PSS(
mgf=padding.MGF1(hashes.SHA256()),
salt_length=hashes.SHA256().digest_size,
),
utils.Prehashed(hashes.SHA256()),
)
request_id_attestation = {
"schema_version": batch.KMS_REQUEST_ID_ATTESTATION_SCHEMA_VERSION,
"payload": attestation_payload,
"signature": {
"provider": signature_block["provider"],
"key_reference": signature_block["key_reference"],
"key_version": signature_block["key_version"],
"algorithm": signature_block["algorithm"],
"value_base64": base64.b64encode(attestation_signature_value).decode(
"ascii"
),
},
}
return {
"schema_version": (
batch.ONLINE_TRANSACTION_RECEIPT_SCHEMA_VERSION
@@ -1123,10 +1162,13 @@ def test_one_reviewer_batch_signature_and_system_item_receipts_remove_human_repe
),
"receipt_id": prefix
+ batch._sha256_bytes(
batch._canonical(payload) + batch._canonical(signature_block)
batch._canonical(payload)
+ batch._canonical(signature_block)
+ batch._canonical(request_id_attestation)
)[:24],
"payload": payload,
"signature": signature_block,
"request_id_attestation": request_id_attestation,
}
image_digests = {
@@ -1392,6 +1434,144 @@ def test_one_reviewer_batch_signature_and_system_item_receipts_remove_human_repe
tenant_id=tenant_id,
)
def reidentify_online(value: dict) -> None:
value["receipt_id"] = "rc06-online-transaction-" + batch._sha256_bytes(
batch._canonical(value["payload"])
+ batch._canonical(value["signature"])
+ batch._canonical(value["request_id_attestation"])
)[:24]
request_id_drift = copy.deepcopy(online_receipt)
request_id_drift["signature"]["request_id"] = "forged-audit-request-id"
reidentify_online(request_id_drift)
with pytest.raises(batch.PublicationBatchError, match="attestation binding drift"):
batch.validate_online_transaction_receipt(
manifest,
manifest_raw_sha,
approval_raw_sha,
approval,
apply_service_public_key,
online_service_public_key,
request_id_drift,
legal_review_binding_sha256=legal_review_binding_sha256,
legal_review_expires_at=legal_review_expires_at,
environment_id=environment_id,
tenant_id=tenant_id,
)
runtime_request_id_drift = copy.deepcopy(runtime_receipt)
runtime_request_id_drift["signature"]["request_id"] = (
"forged-runtime-audit-request-id"
)
runtime_request_id_drift["receipt_id"] = (
"rc06-runtime-snapshot-"
+ batch._sha256_bytes(
batch._canonical(runtime_request_id_drift["payload"])
+ batch._canonical(runtime_request_id_drift["signature"])
+ batch._canonical(runtime_request_id_drift["request_id_attestation"])
)[:24]
)
with pytest.raises(batch.PublicationBatchError, match="attestation binding drift"):
batch.validate_runtime_snapshot_receipt(
manifest,
manifest_raw_sha,
approval_raw_sha,
approval,
apply_service_public_key,
online_service_public_key,
online_receipt,
runtime_request_id_drift,
legal_review_binding_sha256=legal_review_binding_sha256,
legal_review_expires_at=legal_review_expires_at,
environment_id=environment_id,
tenant_id=tenant_id,
)
# RSA-PSS can create another valid signature for the same digest. The
# second-layer attestation must still bind the exact primary signature bytes.
valid_primary_resign = copy.deepcopy(online_receipt)
primary_digest = hashlib.sha256(
batch.ONLINE_TRANSACTION_SIGNATURE_DOMAIN
+ batch._canonical(valid_primary_resign["payload"])
).digest()
replacement_signature = online_service_private_key.sign(
primary_digest,
padding.PSS(
mgf=padding.MGF1(hashes.SHA256()),
salt_length=hashes.SHA256().digest_size,
),
utils.Prehashed(hashes.SHA256()),
)
assert replacement_signature != base64.b64decode(
valid_primary_resign["signature"]["value_base64"]
)
valid_primary_resign["signature"]["value_base64"] = base64.b64encode(
replacement_signature
).decode("ascii")
reidentify_online(valid_primary_resign)
with pytest.raises(batch.PublicationBatchError, match="attestation binding drift"):
batch.validate_online_transaction_receipt(
manifest,
manifest_raw_sha,
approval_raw_sha,
approval,
apply_service_public_key,
online_service_public_key,
valid_primary_resign,
legal_review_binding_sha256=legal_review_binding_sha256,
legal_review_expires_at=legal_review_expires_at,
environment_id=environment_id,
tenant_id=tenant_id,
)
attestation_signature_drift = copy.deepcopy(online_receipt)
attestation_bytes = bytearray(
base64.b64decode(
attestation_signature_drift["request_id_attestation"]["signature"][
"value_base64"
]
)
)
attestation_bytes[0] ^= 1
attestation_signature_drift["request_id_attestation"]["signature"][
"value_base64"
] = base64.b64encode(bytes(attestation_bytes)).decode("ascii")
reidentify_online(attestation_signature_drift)
with pytest.raises(batch.PublicationBatchError, match="KMS signature verification failed"):
batch.validate_online_transaction_receipt(
manifest,
manifest_raw_sha,
approval_raw_sha,
approval,
apply_service_public_key,
online_service_public_key,
attestation_signature_drift,
legal_review_binding_sha256=legal_review_binding_sha256,
legal_review_expires_at=legal_review_expires_at,
environment_id=environment_id,
tenant_id=tenant_id,
)
attestation_identity_drift = copy.deepcopy(online_receipt)
attestation_identity_drift["request_id_attestation"]["signature"][
"key_version"
] = "v2"
reidentify_online(attestation_identity_drift)
with pytest.raises(batch.PublicationBatchError, match="KMS signer identity drift"):
batch.validate_online_transaction_receipt(
manifest,
manifest_raw_sha,
approval_raw_sha,
approval,
apply_service_public_key,
online_service_public_key,
attestation_identity_drift,
legal_review_binding_sha256=legal_review_binding_sha256,
legal_review_expires_at=legal_review_expires_at,
environment_id=environment_id,
tenant_id=tenant_id,
)
forged_apply = copy.deepcopy(apply_receipt)
attacker = Ed25519PrivateKey.generate()
forged_apply["signature"] = base64.b64encode(
yuqei-infra/tests/test_rc06_release_trust_bootstrap.py
+41 -1
View File
@@ -220,14 +220,54 @@ def _kms_envelope(
"value_base64": base64.b64encode(signature_value).decode("ascii"),
"request_id": request_id,
}
attestation_payload = {
"schema_version": authority.batch.KMS_REQUEST_ID_ATTESTATION_PAYLOAD_SCHEMA_VERSION,
"receipt_payload_sha256": authority.batch._sha256_json(payload),
"primary_signature_sha256": authority.batch._sha256_bytes(signature_value),
"primary_provider": signature["provider"],
"primary_key_reference": signature["key_reference"],
"primary_key_version": signature["key_version"],
"primary_algorithm": signature["algorithm"],
"primary_request_id": request_id,
}
attestation_domain = (
authority.batch.ONLINE_REQUEST_ID_ATTESTATION_DOMAIN
if domain == authority.batch.ONLINE_TRANSACTION_SIGNATURE_DOMAIN
else authority.batch.RUNTIME_REQUEST_ID_ATTESTATION_DOMAIN
)
attestation_digest = hashlib.sha256(
attestation_domain + authority._canonical(attestation_payload)
).digest()
attestation_signature_value = private_key.sign(
attestation_digest,
padding.PSS(
mgf=padding.MGF1(hashes.SHA256()),
salt_length=hashes.SHA256().digest_size,
),
utils.Prehashed(hashes.SHA256()),
)
request_id_attestation = {
"schema_version": authority.batch.KMS_REQUEST_ID_ATTESTATION_SCHEMA_VERSION,
"payload": attestation_payload,
"signature": {
"provider": signature["provider"],
"key_reference": signature["key_reference"],
"key_version": signature["key_version"],
"algorithm": signature["algorithm"],
"value_base64": base64.b64encode(attestation_signature_value).decode("ascii"),
},
}
return {
"schema_version": schema,
"receipt_id": prefix
+ authority.batch._sha256_bytes(
authority._canonical(payload) + authority._canonical(signature)
authority._canonical(payload)
+ authority._canonical(signature)
+ authority._canonical(request_id_attestation)
)[:24],
"payload": payload,
"signature": signature,
"request_id_attestation": request_id_attestation,
}