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

Harden RC06 durable publication authority protocol

Browse Source
This commit is contained in:
yuqei
2026-10-06 20:56:40 +08:00
parent ebad424350
commit 65ee37a527
6 changed files with 4113 additions and 1207 deletions
Download Patch File Download Diff File Expand all files Collapse all files
yuqei-infra/scripts/apply-rc06-publication-batch.py
+867 -281
View File
File diff suppressed because it is too large Load Diff
yuqei-infra/scripts/build-rc06-publication-batch.py
+1222 -92
View File
File diff suppressed because it is too large Load Diff
yuqei-infra/scripts/rc06-release-trust-bootstrap.py
+587 -203
View File
File diff suppressed because it is too large Load Diff
yuqei-infra/tests/test_rc06_publication_apply.py
+627 -405
View File
File diff suppressed because it is too large Load Diff
yuqei-infra/tests/test_rc06_publication_batch.py
+371 -88
View File
@@ -2,6 +2,7 @@ from __future__ import annotations
import copy
import base64
import hashlib
import importlib.util
import json
import os
@@ -12,7 +13,8 @@ from datetime import UTC, datetime, timedelta
from pathlib import Path
import pytest
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding, rsa, utils
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
@@ -90,6 +92,26 @@ def _image_identity() -> dict:
return {"images": images, "identity_set_sha256": batch._sha256_json(images)}
def _rc05_candidate_binding() -> dict:
images = _image_identity()
return {
"schema_version": batch.RC05_CANDIDATE_SCHEMA_VERSION,
"commit": "a" * 40,
"tag": "v2-fixture",
"source_manifest": {
"schema_version": "yuqei.release-source-manifest.v1",
"raw_sha256": "0" * 64,
"payload_sha256": "1" * 64,
},
"image_identities": {
"schema_version": batch.IMAGE_SCHEMA_VERSION,
"raw_sha256": "2" * 64,
"identity_set_sha256": images["identity_set_sha256"],
"images": images["images"],
},
}
def _rc05_plan() -> dict:
cases = []
@@ -120,6 +142,7 @@ def _rc05_plan() -> dict:
"auto_repair": auto_repair,
"operations": operations,
"provider_call_budget": len(operations),
"failure_class": "drafting",
}
for index in range(90):
@@ -145,9 +168,21 @@ def _rc05_plan() -> dict:
"request": request,
"request_sha256": batch._sha256_json(request),
"expectations": {},
"failure_class": operation,
})
for position, canary_id in zip(
(0, 1, 2, 3, 120, 136),
batch.EXPECTED_RC05_CANARY_CASE_IDS,
strict=True,
):
cases[position]["case_id"] = canary_id
cases[position]["source_case_id"] = canary_id
failure_classes = {
case["case_id"]: case["failure_class"] for case in cases
}
return {
"schema_version": "multiscene-evaluation-plan.v1",
"schema_version": batch.RC05_PLAN_SCHEMA_VERSION,
"prepared_at": "2026-09-27T03:00:00Z",
"model": "gpt-5.6-sol",
"reasoning_effort": "medium",
"timeout_seconds": 300,
@@ -156,8 +191,15 @@ def _rc05_plan() -> dict:
"supplemental_case_count": 24,
"operation_limits": batch.EXPECTED_RC05_OPERATION_LIMITS,
"total_provider_call_budget": 280,
"candidate": _rc05_candidate_binding(),
"release_controls": {
"schema_version": batch.RC05_RELEASE_CONTROLS_SCHEMA_VERSION,
"canary_case_ids": list(batch.EXPECTED_RC05_CANARY_CASE_IDS),
"failure_classes": failure_classes,
},
"source_pins": {"source.py": "0" * 64},
"cases": cases,
"limitations": ["fixture only"],
}
@@ -182,6 +224,7 @@ def _terminal_identity(plan_sha: str = "3" * 64) -> dict:
def _terminal_documents(plan: dict, plan_sha: str = "3" * 64):
replacement_case_id = plan["release_controls"]["canary_case_ids"][0]
attempts = []
for index in range(4):
resolved = index == 3
@@ -189,7 +232,7 @@ def _terminal_documents(plan: dict, plan_sha: str = "3" * 64):
"resolution_status": "resolved" if resolved else "requires_replacement",
"release_eligible": resolved,
"reason": "fixture resolution" if resolved else "fixture replacement",
"replacement_case_id": None if resolved else "draft-000",
"replacement_case_id": None if resolved else replacement_case_id,
"replacement_operation": None if resolved else "draft",
}
attempts.append({
@@ -243,13 +286,15 @@ def _terminal_documents(plan: dict, plan_sha: str = "3" * 64):
},
"source_pins": dict(sorted(plan["source_pins"].items())),
"real_provider": True,
"candidate_binding": plan["candidate"],
"release_controls": plan["release_controls"],
"authority": authority,
"integrity_not_authentication": True,
"historical_usage": historical,
}
campaign_identity = {
"schema_version": batch.RC05_CAMPAIGN_SCHEMA_VERSION,
"campaign_id": "rc05-real-usage-v1",
"campaign_id": "rc05-real-usage-v2",
"historical_receipt_sha256": historical["receipt_sha256"],
"historical_attempts_sha256": historical["attempts_sha256"],
"historical_provider_call_count": 4,
@@ -373,7 +418,7 @@ def _terminal_documents(plan: dict, plan_sha: str = "3" * 64):
current_calls = sum(current_operations.values())
combined_calls = current_calls + 4
ledger_snapshot = {
"schema_version": "multiscene-call-ledger-snapshot.v1",
"schema_version": batch.RC05_LEDGER_SNAPSHOT_SCHEMA_VERSION,
"ledger_identity_sha256": batch._sha256_json(ledger_identity),
"real_provider": True,
"authority": authority,
@@ -395,6 +440,13 @@ def _terminal_documents(plan: dict, plan_sha: str = "3" * 64):
"attempted": current_calls,
"terminal": current_calls,
"unknown": 0,
"release_controls": {
"schema_version": batch.RC05_CONTROL_STATE_SCHEMA_VERSION,
"phase": "full_authorized",
"canary_case_ids": list(batch.EXPECTED_RC05_CANARY_CASE_IDS),
"blocked_failure_classes": [],
"canary_transition_sha256": "9" * 64,
},
}
remediation = [
({
@@ -407,7 +459,7 @@ def _terminal_documents(plan: dict, plan_sha: str = "3" * 64):
"attempt_id": attempt["attempt_id"],
"status": "replacement_verified",
"passed": True,
"replacement_case_id": "draft-000",
"replacement_case_id": replacement_case_id,
"replacement_operation": "draft",
"provider_passed": True,
"business_passed": True,
@@ -512,19 +564,7 @@ def _manifest(inventory: dict, *, resolved: bool = False) -> dict:
image_identities_sha256="2" * 64,
image_identity=_image_identity(),
rc05_plan_sha256="3" * 64,
rc05_plan_identity={
"schema_version": "multiscene-evaluation-plan.v1",
"model": "gpt-5.6-sol",
"reasoning_effort": "medium",
"timeout_seconds": 300,
"case_count": 144,
"drafting_case_count": 120,
"supplemental_case_count": 24,
"operation_limits": batch.EXPECTED_RC05_OPERATION_LIMITS,
"total_provider_call_budget": 280,
"case_inventory_sha256": "5" * 64,
"source_pins_sha256": "4" * 64,
},
rc05_plan_identity=batch.validate_rc05_plan(_rc05_plan()),
rc05_terminal_evidence=_terminal_identity(),
inventory=inventory,
dispositions=dispositions,
@@ -701,7 +741,9 @@ def test_one_reviewer_batch_signature_and_system_item_receipts_remove_human_repe
root_private_key = Ed25519PrivateKey.generate()
private_key = Ed25519PrivateKey.generate()
apply_service_private_key = Ed25519PrivateKey.generate()
online_service_private_key = Ed25519PrivateKey.generate()
online_service_private_key = rsa.generate_private_key(
public_exponent=65537, key_size=2048
)
root_public_key = root_private_key.public_key()
apply_service_public_key = apply_service_private_key.public_key()
online_service_public_key = online_service_private_key.public_key()
@@ -713,9 +755,9 @@ def test_one_reviewer_batch_signature_and_system_item_receipts_remove_human_repe
serialization.Encoding.Raw,
serialization.PublicFormat.Raw,
)
online_service_public_raw = online_service_public_key.public_bytes(
serialization.Encoding.Raw,
serialization.PublicFormat.Raw,
online_service_public_spki = online_service_public_key.public_bytes(
serialization.Encoding.DER,
serialization.PublicFormat.SubjectPublicKeyInfo,
)
public_pem = private_key.public_key().public_bytes(
serialization.Encoding.PEM,
@@ -763,8 +805,16 @@ def test_one_reviewer_batch_signature_and_system_item_receipts_remove_human_repe
)
monkeypatch.setattr(
batch,
"TRUSTED_ONLINE_APPLY_SERVICE_PUBLIC_KEY_SHA256",
batch._sha256_bytes(online_service_public_raw),
"TRUSTED_ONLINE_APPLY_SERVICE_PUBLIC_KEY_SPKI_SHA256",
batch._sha256_bytes(online_service_public_spki),
)
monkeypatch.setattr(batch, "TRUSTED_ONLINE_APPLY_SERVICE_PROVIDER", "fixture-kms")
monkeypatch.setattr(
batch, "TRUSTED_ONLINE_APPLY_SERVICE_KEY_REFERENCE", "kms://fixture/rc06"
)
monkeypatch.setattr(batch, "TRUSTED_ONLINE_APPLY_SERVICE_KEY_VERSION", "v1")
monkeypatch.setattr(
batch, "TRUSTED_ONLINE_APPLY_SERVICE_ALGORITHM", "RSA_PSS_SHA_256"
)
monkeypatch.setattr(
batch,
@@ -999,26 +1049,20 @@ def test_one_reviewer_batch_signature_and_system_item_receipts_remove_human_repe
)
approval_raw_sha = batch._sha256_bytes(batch._canonical(approval))
item_results = [
{
"kind": "asset",
"item_id": item["template_id"],
"expected_content_sha256": item["content_sha256"],
"online_content_sha256": item["content_sha256"],
"status": "applied_and_verified",
}
for item in manifest["asset_batch"]["items"]
] + [
{
"kind": "prompt_module",
"item_id": item["key"],
"expected_content_sha256": item["content_sha256"],
"online_content_sha256": item["content_sha256"],
"status": "applied_and_verified",
}
for item in manifest["prompt_module_batch"]["items"]
]
item_results.sort(key=lambda item: (item["kind"], item["item_id"]))
item_results = batch._expected_apply_item_results(manifest)
runtime_results = batch.expected_runtime_snapshot_results(manifest)
legal_review_binding_sha256 = "d" * 64
legal_review_expires_at = "2026-09-28T08:00:00+08:00"
environment_id = "isolated-uat-fixture"
tenant_id = "tenant-fixture"
idempotency_key = batch.publication_idempotency_key(
manifest_raw_sha,
approval_raw_sha,
legal_review_binding_sha256,
environment_id,
tenant_id,
)
transaction_id = batch.publication_transaction_id(idempotency_key)
challenge_payload = {
"schema_version": batch.APPLY_CHALLENGE_PAYLOAD_SCHEMA_VERSION,
"nonce_base64": base64.b64encode(b"c" * 32).decode("ascii"),
@@ -1027,8 +1071,13 @@ def test_one_reviewer_batch_signature_and_system_item_receipts_remove_human_repe
"batch_manifest_raw_sha256": manifest_raw_sha,
"batch_manifest_payload_sha256": manifest["manifest_payload_sha256"],
"approval_receipt_sha256": approval_raw_sha,
"legal_source_review_binding_sha256": legal_review_binding_sha256,
"legal_source_review_expires_at": legal_review_expires_at,
"operation": batch.APPLY_OPERATION,
"environment_id": "isolated-uat-fixture",
"environment_id": environment_id,
"tenant_id": tenant_id,
"idempotency_key": idempotency_key,
"transaction_id": transaction_id,
"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",
@@ -1047,14 +1096,100 @@ def test_one_reviewer_batch_signature_and_system_item_receipts_remove_human_repe
"payload": challenge_payload,
"signature": base64.b64encode(challenge_signature).decode("ascii"),
}
def kms_envelope(payload: dict, *, domain: bytes, prefix: str, request_id: str) -> dict:
digest = hashlib.sha256(domain + batch._canonical(payload)).digest()
signature_value = online_service_private_key.sign(
digest,
padding.PSS(
mgf=padding.MGF1(hashes.SHA256()),
salt_length=hashes.SHA256().digest_size,
),
utils.Prehashed(hashes.SHA256()),
)
signature_block = {
"provider": "fixture-kms",
"key_reference": "kms://fixture/rc06",
"key_version": "v1",
"algorithm": "RSA_PSS_SHA_256",
"value_base64": base64.b64encode(signature_value).decode("ascii"),
"request_id": request_id,
}
return {
"schema_version": (
batch.ONLINE_TRANSACTION_RECEIPT_SCHEMA_VERSION
if prefix == "rc06-online-transaction-"
else batch.RUNTIME_RECEIPT_SCHEMA_VERSION
),
"receipt_id": prefix
+ batch._sha256_bytes(
batch._canonical(payload) + batch._canonical(signature_block)
)[:24],
"payload": payload,
"signature": signature_block,
}
image_digests = {
item["role"]: item["manifest_digest"]
for item in manifest["candidate"]["images"]
}
participant_evidence = []
participant_specs = (
(
"ai-platform",
"2026-09-21T09:58:00+08:00",
14,
14,
image_digests["ai-platform-api"],
[item for item in item_results if item["kind"] == "prompt_module"],
),
(
"contract-platform",
"2026-09-21T09:59:00+08:00",
60,
0,
image_digests["contract-api"],
[item for item in item_results if item["kind"] == "asset"],
),
)
for participant, committed_at, write_count, runtime_count, image, results in participant_specs:
projection = {
"participant": participant,
"state": "committed",
"transaction_id": transaction_id,
"candidate_commit": manifest["candidate"]["commit"],
"release_image_digest": image,
"committed_at": committed_at,
"write_count": write_count,
"runtime_snapshot_count": runtime_count,
"committed_item_results_sha256": batch._sha256_json(results),
}
participant_evidence.append(
{
**projection,
"durable_commit_evidence_sha256": batch._sha256_json(projection),
}
)
online_payload = {
"schema_version": batch.ONLINE_TRANSACTION_PAYLOAD_SCHEMA_VERSION,
"key_id": batch.TRUSTED_ONLINE_APPLY_SERVICE_KEY_ID,
"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,
"environment_id": "isolated-uat-fixture",
"transaction_id": "fixture-transaction-001",
"database_commit_id": "fixture-database-commit-001",
"legal_source_review_binding_sha256": legal_review_binding_sha256,
"environment_id": environment_id,
"tenant_id": tenant_id,
"idempotency_key": idempotency_key,
"transaction_id": transaction_id,
"transaction_mode": batch.TRANSACTION_MODE,
"cross_database_atomicity": False,
"coordinator_deployment": {
"component": batch.TRUSTED_ONLINE_APPLY_SERVICE_COMPONENT,
"candidate_commit": manifest["candidate"]["commit"],
"release_image_digest": image_digests["contract-api"],
},
"participant_evidence": participant_evidence,
"committed_at": "2026-09-21T10:00:00+08:00",
"authority_challenge": authority_challenge,
"generated_by": {"type": "system", "component": "rc06-publication-service"},
@@ -1062,29 +1197,63 @@ def test_one_reviewer_batch_signature_and_system_item_receipts_remove_human_repe
"write_count": 74,
"status": "committed_and_verified",
}
online_signature = online_service_private_key.sign(
batch.ONLINE_TRANSACTION_SIGNATURE_DOMAIN + batch._canonical(online_payload)
online_receipt = kms_envelope(
online_payload,
domain=batch.ONLINE_TRANSACTION_SIGNATURE_DOMAIN,
prefix="rc06-online-transaction-",
request_id="kms-online-fixture-001",
)
online_receipt = {
"schema_version": batch.ONLINE_TRANSACTION_RECEIPT_SCHEMA_VERSION,
"receipt_id": (
"rc06-online-transaction-"
+ batch._sha256_bytes(batch._canonical(online_payload) + online_signature)[:24]
),
"payload": online_payload,
"signature": base64.b64encode(online_signature).decode("ascii"),
}
apply_payload = {
"schema_version": batch.APPLY_PAYLOAD_SCHEMA_VERSION,
runtime_payload = {
"schema_version": batch.RUNTIME_PAYLOAD_SCHEMA_VERSION,
"batch_manifest_raw_sha256": manifest_raw_sha,
"batch_manifest_payload_sha256": manifest["manifest_payload_sha256"],
"approval_receipt_sha256": approval_raw_sha,
"environment_id": "isolated-uat-fixture",
"applied_at": "2026-09-21T10:00:00+08:00",
"legal_source_review_binding_sha256": legal_review_binding_sha256,
"environment_id": environment_id,
"tenant_id": tenant_id,
"idempotency_key": idempotency_key,
"transaction_id": transaction_id,
"online_transaction_receipt_sha256": batch._sha256_json(online_receipt),
"captured_at": "2026-09-21T10:01:00+08:00",
"generated_by": {
"type": "system",
"component": batch.TRUSTED_ONLINE_APPLY_SERVICE_COMPONENT,
},
"transaction_mode": batch.TRANSACTION_MODE,
"cross_database_atomicity": False,
"publication_item_count": 74,
"runtime_snapshot_dependencies": runtime_results,
"runtime_snapshot_dependency_count": 14,
"status": "published_and_resolved",
}
runtime_receipt = kms_envelope(
runtime_payload,
domain=batch.RUNTIME_SIGNATURE_DOMAIN,
prefix="rc06-runtime-snapshot-",
request_id="kms-runtime-fixture-001",
)
apply_payload = {
"schema_version": batch.APPLY_PAYLOAD_SCHEMA_VERSION,
"batch_manifest_raw_sha256": manifest_raw_sha,
"batch_manifest_payload_sha256": manifest["manifest_payload_sha256"],
"approval_receipt_sha256": approval_raw_sha,
"legal_source_review_binding_sha256": legal_review_binding_sha256,
"environment_id": environment_id,
"tenant_id": tenant_id,
"idempotency_key": idempotency_key,
"transaction_id": transaction_id,
"authorized_commit_at": runtime_payload["captured_at"],
# A durable online commit remains countersignable after the short challenge
# window; only the signed online/runtime facts must fall inside that window.
"countersigned_at": "2026-09-21T10:30:00+08:00",
"generated_by": {"type": "system", "component": "rc06-apply-authority"},
"transaction_mode": batch.TRANSACTION_MODE,
"cross_database_atomicity": False,
"item_results": item_results,
"write_count": 74,
"status": "applied_and_verified",
"online_transaction_receipt": online_receipt,
"runtime_snapshot_receipt": runtime_receipt,
}
apply_signature = apply_service_private_key.sign(
batch.APPLY_SIGNATURE_DOMAIN + batch._canonical(apply_payload)
@@ -1108,31 +1277,12 @@ def test_one_reviewer_batch_signature_and_system_item_receipts_remove_human_repe
online_service_public_key,
approval_raw_sha,
apply_receipt,
legal_review_binding_sha256=legal_review_binding_sha256,
legal_review_expires_at=legal_review_expires_at,
environment_id=environment_id,
tenant_id=tenant_id,
) == apply_receipt
boundary_apply = copy.deepcopy(apply_receipt)
boundary_apply["payload"]["applied_at"] = approval_payload["expires_at"]
boundary_signature = apply_service_private_key.sign(
batch.APPLY_SIGNATURE_DOMAIN + batch._canonical(boundary_apply["payload"])
)
boundary_apply["signature"] = base64.b64encode(boundary_signature).decode("ascii")
boundary_apply["receipt_id"] = (
"rc06-apply-receipt-"
+ batch._sha256_bytes(batch._canonical(boundary_apply["payload"]) + boundary_signature)[:24]
)
with pytest.raises(batch.PublicationBatchError, match="outside the approval window"):
batch.validate_apply_receipt(
manifest,
manifest_raw_sha,
approval,
keyring,
root_public_key,
apply_service_public_key,
online_service_public_key,
approval_raw_sha,
boundary_apply,
)
another_reviewer = copy.deepcopy(approval)
another_reviewer["payload"]["approver"]["approver_id"] = "another-reviewer"
with pytest.raises(batch.PublicationBatchError, match="sole release reviewer"):
@@ -1146,6 +1296,16 @@ def test_one_reviewer_batch_signature_and_system_item_receipts_remove_human_repe
tampered_apply = copy.deepcopy(apply_receipt)
tampered_apply["payload"]["item_results"][0]["online_content_sha256"] = "7" * 64
tampered_signature = apply_service_private_key.sign(
batch.APPLY_SIGNATURE_DOMAIN + batch._canonical(tampered_apply["payload"])
)
tampered_apply["signature"] = base64.b64encode(tampered_signature).decode("ascii")
tampered_apply["receipt_id"] = (
"rc06-apply-receipt-"
+ batch._sha256_bytes(
batch._canonical(tampered_apply["payload"]) + tampered_signature
)[:24]
)
with pytest.raises(batch.PublicationBatchError, match="74-item online evidence drift"):
batch.validate_apply_receipt(
manifest,
@@ -1157,6 +1317,79 @@ def test_one_reviewer_batch_signature_and_system_item_receipts_remove_human_repe
online_service_public_key,
approval_raw_sha,
tampered_apply,
legal_review_binding_sha256=legal_review_binding_sha256,
legal_review_expires_at=legal_review_expires_at,
environment_id=environment_id,
tenant_id=tenant_id,
)
# These mismatches are signed by the trusted fixture key so the tests prove
# semantic binding checks, not merely signature failure.
deployment_drift_payload = copy.deepcopy(online_payload)
deployment_drift_payload["participant_evidence"][0]["release_image_digest"] = (
"sha256:" + "f" * 64
)
drift_projection = {
key: value
for key, value in deployment_drift_payload["participant_evidence"][0].items()
if key != "durable_commit_evidence_sha256"
}
deployment_drift_payload["participant_evidence"][0][
"durable_commit_evidence_sha256"
] = batch._sha256_json(drift_projection)
deployment_drift = kms_envelope(
deployment_drift_payload,
domain=batch.ONLINE_TRANSACTION_SIGNATURE_DOMAIN,
prefix="rc06-online-transaction-",
request_id="kms-online-fixture-deployment-drift",
)
with pytest.raises(batch.PublicationBatchError, match="participant commit evidence drift"):
batch.validate_online_transaction_receipt(
manifest,
manifest_raw_sha,
approval_raw_sha,
approval,
apply_service_public_key,
online_service_public_key,
deployment_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,
)
stale_schema = copy.deepcopy(online_receipt)
stale_schema["schema_version"] = "yuqei.rc06-online-transaction-receipt.v3"
with pytest.raises(batch.PublicationBatchError, match="schema is unsupported"):
batch.validate_online_transaction_receipt(
manifest,
manifest_raw_sha,
approval_raw_sha,
approval,
apply_service_public_key,
online_service_public_key,
stale_schema,
legal_review_binding_sha256=legal_review_binding_sha256,
legal_review_expires_at=legal_review_expires_at,
environment_id=environment_id,
tenant_id=tenant_id,
)
metadata_drift = copy.deepcopy(online_receipt)
metadata_drift["signature"]["key_version"] = "v2"
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,
metadata_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)
@@ -1175,6 +1408,10 @@ def test_one_reviewer_batch_signature_and_system_item_receipts_remove_human_repe
online_service_public_key,
approval_raw_sha,
forged_apply,
legal_review_binding_sha256=legal_review_binding_sha256,
legal_review_expires_at=legal_review_expires_at,
environment_id=environment_id,
tenant_id=tenant_id,
)
assert manifest["publication_control"]["blocked_controls"] == list(batch.BLOCKED_CONTROLS)
@@ -2189,6 +2426,25 @@ def test_rc05_plan_requires_exact_budget_operations_and_unique_case_inventory()
identity = batch.validate_rc05_plan(plan)
assert identity["total_provider_call_budget"] == 280
assert identity["operation_limits"] == batch.EXPECTED_RC05_OPERATION_LIMITS
assert identity["candidate"] == _rc05_candidate_binding()
assert identity["release_controls"] == plan["release_controls"]
legacy = copy.deepcopy(plan)
legacy["schema_version"] = "multiscene-evaluation-plan.v1"
with pytest.raises(batch.PublicationBatchError, match="read-only diagnostic"):
batch.validate_rc05_plan(legacy)
candidate_drift = copy.deepcopy(plan)
candidate_drift["candidate"]["commit"] = "f" * 40
with pytest.raises(batch.PublicationBatchError, match="candidate binding differs"):
batch.validate_rc05_plan(
candidate_drift, expected_candidate=_rc05_candidate_binding()
)
control_drift = copy.deepcopy(plan)
control_drift["release_controls"]["canary_case_ids"].pop()
with pytest.raises(batch.PublicationBatchError, match="release controls"):
batch.validate_rc05_plan(control_drift)
drifted = copy.deepcopy(plan)
drifted["total_provider_call_budget"] = 281
@@ -2251,6 +2507,33 @@ def test_rc05_terminal_evidence_requires_full_real_green_chain_and_remediation()
sorted(summary["case_operation_coverage"], key=lambda item: item["case_id"])
)
legacy_ledger = copy.deepcopy(ledger)
legacy_ledger["schema_version"] = "multiscene-call-ledger.v1"
with pytest.raises(batch.PublicationBatchError, match="ledger identity"):
batch.validate_rc05_terminal_evidence(
plan, "3" * 64, summary, "6" * 64,
legacy_ledger, "7" * 64, campaign, "8" * 64,
)
legacy_campaign = copy.deepcopy(campaign)
legacy_campaign["schema_version"] = "multiscene-call-campaign.v1"
legacy_campaign["campaign_id"] = "rc05-real-usage-v1"
with pytest.raises(batch.PublicationBatchError, match="campaign identity"):
batch.validate_rc05_terminal_evidence(
plan, "3" * 64, summary, "6" * 64,
ledger, "7" * 64, legacy_campaign, "8" * 64,
)
blocked_control_state = copy.deepcopy(summary)
blocked_control_state["ledger"]["release_controls"][
"blocked_failure_classes"
] = ["provider_outage"]
with pytest.raises(batch.PublicationBatchError, match="release controls"):
batch.validate_rc05_terminal_evidence(
plan, "3" * 64, blocked_control_state, "6" * 64,
ledger, "7" * 64, campaign, "8" * 64,
)
incomplete = copy.deepcopy(summary)
incomplete["historical_remediation"][0]["passed"] = False
with pytest.raises(batch.PublicationBatchError, match="duplicate or not passed"):
yuqei-infra/tests/test_rc06_release_trust_bootstrap.py
+439 -138
View File
@@ -2,6 +2,7 @@ from __future__ import annotations
import base64
import copy
import hashlib
import importlib.util
import json
import os
@@ -13,7 +14,8 @@ from datetime import UTC, datetime, timedelta
from pathlib import Path
import pytest
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding, rsa, utils
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
@@ -108,22 +110,125 @@ def _reviewer_registration(path: Path, *, private_key: Ed25519PrivateKey | None
def _online_service_registration(path: Path) -> dict:
private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
public_pem = private_key.public_key().public_bytes(
serialization.Encoding.PEM,
serialization.PublicFormat.SubjectPublicKeyInfo,
)
public_spki = private_key.public_key().public_bytes(
serialization.Encoding.DER,
serialization.PublicFormat.SubjectPublicKeyInfo,
)
registration = {
"schema_version": authority.ONLINE_SERVICE_REGISTRATION_SCHEMA,
"provider": "aliyun",
"key_reference": "kms://fixture/rc06-online",
"key_version": "version-1",
"algorithm": "RSA_PSS_SHA_256",
"component": authority.batch.TRUSTED_ONLINE_APPLY_SERVICE_COMPONENT,
"created_at": "2026-09-27T02:30:00Z",
"public_key_pem": public_pem.decode("ascii"),
"public_key_spki_sha256": authority.batch._sha256_bytes(public_spki),
}
path.write_bytes(authority._canonical(registration))
return registration
def _legal_keyring(path: Path) -> tuple[dict, Ed25519PrivateKey]:
private_key = Ed25519PrivateKey.generate()
public_raw = private_key.public_key().public_bytes(
serialization.Encoding.Raw,
serialization.PublicFormat.Raw,
)
registration = {
"schema_version": authority.ONLINE_SERVICE_REGISTRATION_SCHEMA,
"key_id": authority.batch.TRUSTED_ONLINE_APPLY_SERVICE_KEY_ID,
"algorithm": "Ed25519",
"component": authority.batch.TRUSTED_ONLINE_APPLY_SERVICE_COMPONENT,
"created_at": "2026-09-27T02:30:00Z",
"public_key_raw_base64": base64.b64encode(public_raw).decode("ascii"),
"public_key_raw_sha256": authority.batch._sha256_bytes(public_raw),
keyring = {
"schema_version": "yuqei.scene-source-legal-review-keyring.v1",
"keys": [
{
"key_id": "fixture-legal-reviewer",
"status": "trusted",
"revoked_at": None,
"valid_from": "2026-09-26T00:00:00Z",
"valid_until": "2026-09-28T00:00:00Z",
"reviewer_name": "测试法源复核人（非真实身份）",
"reviewer_organization": "测试组织（非真实法律服务主体）",
"reviewer_qualification": "测试资格声明（不代表真实法律资质）",
"credential_reference": "TEST-FIXTURE-NOT-A-CREDENTIAL",
"public_key_raw_base64": base64.b64encode(public_raw).decode("ascii"),
}
],
}
path.write_bytes(authority._canonical(keyring))
return keyring, private_key
class _FakeCanonicalLegalReviewValidator:
@staticmethod
def verify_request(request: dict, _repo: Path) -> dict:
projection = {key: value for key, value in request.items() if key != "request_sha256"}
if request.get("request_sha256") != authority.batch._sha256_json(projection):
raise ValueError("review request embedded hash drift")
if len(request.get("occurrences", [])) != 43:
raise ValueError("review request must contain 43 occurrences")
return request
@staticmethod
def verify_signed_decision(
request: dict, decision: dict, _keyring: dict, *, checked_at: str
) -> dict:
if decision.get("request_sha256") != request["request_sha256"]:
raise ValueError("decision request binding drift")
if datetime.fromisoformat(checked_at) > datetime.fromisoformat(
decision["expires_at"]
):
raise ValueError("decision expired")
return {
"status": "verified_qualified_human_review",
"candidate_commit": request["candidate_commit"],
"request_sha256": request["request_sha256"],
"reviewer_name": decision["reviewer"]["name"],
"reviewed_occurrence_count": 43,
"all_current_effectiveness_verified": True,
"all_transaction_applicability_verified": True,
"expires_at": decision["expires_at"],
"authorizes_publication_by_itself": False,
}
def _kms_envelope(
private_key,
*,
schema: str,
prefix: str,
domain: bytes,
payload: dict,
request_id: str,
) -> dict:
digest = hashlib.sha256(domain + authority._canonical(payload)).digest()
signature_value = private_key.sign(
digest,
padding.PSS(
mgf=padding.MGF1(hashes.SHA256()),
salt_length=hashes.SHA256().digest_size,
),
utils.Prehashed(hashes.SHA256()),
)
signature = {
"provider": "fixture-kms",
"key_reference": "kms://fixture/rc06-online",
"key_version": "version-1",
"algorithm": "RSA_PSS_SHA_256",
"value_base64": base64.b64encode(signature_value).decode("ascii"),
"request_id": request_id,
}
return {
"schema_version": schema,
"receipt_id": prefix
+ authority.batch._sha256_bytes(
authority._canonical(payload) + authority._canonical(signature)
)[:24],
"payload": payload,
"signature": signature,
}
path.write_bytes(authority._canonical(registration))
return registration
def _initialize(tmp_path: Path):
@@ -181,17 +286,35 @@ def _formal_apply_materials(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) ->
"TRUSTED_APPLY_SERVICE_PUBLIC_KEY_SHA256",
keys[authority.ROLE_APPLY_SERVICE].public_sha256,
)
online_private = Ed25519PrivateKey.generate()
online_raw = online_private.public_key().public_bytes(
serialization.Encoding.Raw,
serialization.PublicFormat.Raw,
online_private = rsa.generate_private_key(public_exponent=65537, key_size=2048)
online_public_pem = online_private.public_key().public_bytes(
serialization.Encoding.PEM,
serialization.PublicFormat.SubjectPublicKeyInfo,
)
online_public_spki = online_private.public_key().public_bytes(
serialization.Encoding.DER,
serialization.PublicFormat.SubjectPublicKeyInfo,
)
online_public_file = tmp_path / "online-service-public.pem"
online_public_file.write_bytes(online_public_pem)
monkeypatch.setattr(
authority.batch, "TRUSTED_ONLINE_APPLY_SERVICE_PROVIDER", "fixture-kms"
)
online_public_file = tmp_path / "online-service-public.raw"
online_public_file.write_bytes(online_raw)
monkeypatch.setattr(
authority.batch,
"TRUSTED_ONLINE_APPLY_SERVICE_PUBLIC_KEY_SHA256",
authority.batch._sha256_bytes(online_raw),
"TRUSTED_ONLINE_APPLY_SERVICE_KEY_REFERENCE",
"kms://fixture/rc06-online",
)
monkeypatch.setattr(
authority.batch, "TRUSTED_ONLINE_APPLY_SERVICE_KEY_VERSION", "version-1"
)
monkeypatch.setattr(
authority.batch, "TRUSTED_ONLINE_APPLY_SERVICE_ALGORITHM", "RSA_PSS_SHA_256"
)
monkeypatch.setattr(
authority.batch,
"TRUSTED_ONLINE_APPLY_SERVICE_PUBLIC_KEY_SPKI_SHA256",
authority.batch._sha256_bytes(online_public_spki),
)
monkeypatch.setattr(
authority.batch,
@@ -241,6 +364,69 @@ def _formal_apply_materials(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) ->
approval_raw = authority._canonical(approval)
approval_file.write_bytes(approval_raw)
approval_sha = authority.batch._sha256_bytes(approval_raw)
legal_keyring_file = tmp_path / "scene-legal-keyring.json"
legal_keyring, _legal_private = _legal_keyring(legal_keyring_file)
legal_keyring_raw = legal_keyring_file.read_bytes()
legal_keyring_sha = authority.batch._sha256_bytes(legal_keyring_raw)
monkeypatch.setattr(
authority.batch,
"TRUSTED_SCENE_LEGAL_REVIEW_KEYRING_SHA256",
legal_keyring_sha,
)
monkeypatch.setattr(
authority.batch,
"_load_scene_legal_review_module",
lambda: _FakeCanonicalLegalReviewValidator,
)
occurrences = [
{
"occurrence_id": f"fixture-occurrence-{index:02d}",
"source_record_sha256": f"{index + 1:064x}",
}
for index in range(43)
]
legal_request = {
"schema_version": "yuqei.scene-source-legal-review-request.v1",
"status": "pending_qualified_human_review",
"candidate_commit": manifest["candidate"]["commit"],
"created_at": "2026-09-27T03:00:00Z",
"scope": {
"scene_count": 30,
"reference_occurrence_count": 43,
"distinct_source_record_count": 43,
"distinct_url_count": 43,
},
"source_files": ["fixture-only/not-production.json"],
"occurrences": occurrences,
"guardrails": {
"not_legal_approval": True,
"authorizes_publication": False,
"ai_must_not_supply_human_decisions": True,
"all_occurrences_require_qualified_human_review": True,
},
}
legal_request["request_sha256"] = authority.batch._sha256_json(legal_request)
legal_decision = {
"schema_version": "yuqei.scene-source-legal-review-decision.v1",
"request_sha256": legal_request["request_sha256"],
"candidate_commit": manifest["candidate"]["commit"],
"reviewed_at": "2026-09-27T03:30:00Z",
"expires_at": "2026-09-27T05:00:00Z",
"reviewer": {"name": "测试法源复核人（非真实身份）"},
"items": [],
"signature": {
"algorithm": "Ed25519",
"key_id": "fixture-legal-reviewer",
"value_base64": base64.b64encode(b"fixture-signature").decode("ascii"),
},
}
legal_request_file = tmp_path / "scene-legal-request.json"
legal_decision_file = tmp_path / "scene-legal-decision.json"
legal_request_file.write_bytes(authority._canonical(legal_request))
legal_decision_file.write_bytes(authority._canonical(legal_decision))
legal_request_sha = authority.batch._sha256_bytes(legal_request_file.read_bytes())
legal_decision_sha = authority.batch._sha256_bytes(legal_decision_file.read_bytes())
challenge_result = authority.issue_apply_challenge(
state,
manifest_file=manifest_file,
@@ -252,40 +438,86 @@ def _formal_apply_materials(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) ->
approval_root_public_key_file=(
keyring_directory / "rc06-approval-root-public-key.raw"
),
legal_review_request_file=legal_request_file,
legal_review_request_sha256=legal_request_sha,
legal_review_decision_file=legal_decision_file,
legal_review_decision_sha256=legal_decision_sha,
legal_review_keyring_file=legal_keyring_file,
legal_review_keyring_sha256=legal_keyring_sha,
environment_id="isolated-uat-fixture",
tenant_id="tenant-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",
"item_id": item["template_id"],
"expected_content_sha256": item["content_sha256"],
"online_content_sha256": item["content_sha256"],
"status": "applied_and_verified",
item_results = authority.batch._expected_apply_item_results(manifest)
runtime_results = authority.batch.expected_runtime_snapshot_results(manifest)
challenge_payload = challenge["payload"]
images = {
item["role"]: item["manifest_digest"]
for item in manifest["candidate"]["images"]
}
participant_evidence = []
for participant, committed_at, write_count, runtime_count, image, results in (
(
"ai-platform",
"2026-09-27T03:58:00Z",
14,
14,
images["ai-platform-api"],
[item for item in item_results if item["kind"] == "prompt_module"],
),
(
"contract-platform",
"2026-09-27T03:59:00Z",
60,
0,
images["contract-api"],
[item for item in item_results if item["kind"] == "asset"],
),
):
projection = {
"participant": participant,
"state": "committed",
"transaction_id": challenge_payload["transaction_id"],
"candidate_commit": manifest["candidate"]["commit"],
"release_image_digest": image,
"committed_at": committed_at,
"write_count": write_count,
"runtime_snapshot_count": runtime_count,
"committed_item_results_sha256": authority.batch._sha256_json(results),
}
for item in manifest["asset_batch"]["items"]
] + [
{
"kind": "prompt_module",
"item_id": item["key"],
"expected_content_sha256": item["content_sha256"],
"online_content_sha256": item["content_sha256"],
"status": "applied_and_verified",
}
for item in manifest["prompt_module_batch"]["items"]
]
item_results.sort(key=lambda item: (item["kind"], item["item_id"]))
participant_evidence.append(
{
**projection,
"durable_commit_evidence_sha256": authority.batch._sha256_json(
projection
),
}
)
online_payload = {
"schema_version": authority.batch.ONLINE_TRANSACTION_PAYLOAD_SCHEMA_VERSION,
"key_id": authority.batch.TRUSTED_ONLINE_APPLY_SERVICE_KEY_ID,
"candidate_commit": manifest["candidate"]["commit"],
"candidate_tag": manifest["candidate"]["tag"],
"batch_manifest_raw_sha256": manifest_sha,
"batch_manifest_payload_sha256": manifest["manifest_payload_sha256"],
"approval_receipt_sha256": authority.batch._sha256_bytes(approval_raw),
"approval_receipt_sha256": approval_sha,
"legal_source_review_binding_sha256": challenge_payload[
"legal_source_review_binding_sha256"
],
"environment_id": "isolated-uat-fixture",
"transaction_id": "fixture-transaction-001",
"database_commit_id": "fixture-database-commit-001",
"tenant_id": "tenant-fixture",
"idempotency_key": challenge_payload["idempotency_key"],
"transaction_id": challenge_payload["transaction_id"],
"transaction_mode": authority.batch.TRANSACTION_MODE,
"cross_database_atomicity": False,
"coordinator_deployment": {
"component": authority.batch.TRUSTED_ONLINE_APPLY_SERVICE_COMPONENT,
"candidate_commit": manifest["candidate"]["commit"],
"release_image_digest": images["contract-api"],
},
"participant_evidence": participant_evidence,
"committed_at": "2026-09-27T04:00:00Z",
"authority_challenge": challenge,
"generated_by": {"type": "system", "component": "rc06-publication-service"},
@@ -293,24 +525,55 @@ def _formal_apply_materials(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) ->
"write_count": 74,
"status": "committed_and_verified",
}
online_signature = online_private.sign(
authority.batch.ONLINE_TRANSACTION_SIGNATURE_DOMAIN
+ authority._canonical(online_payload)
online_receipt = _kms_envelope(
online_private,
schema=authority.batch.ONLINE_TRANSACTION_RECEIPT_SCHEMA_VERSION,
prefix="rc06-online-transaction-",
domain=authority.batch.ONLINE_TRANSACTION_SIGNATURE_DOMAIN,
payload=online_payload,
request_id="fixture-kms-online",
)
online_receipt = {
"schema_version": authority.batch.ONLINE_TRANSACTION_RECEIPT_SCHEMA_VERSION,
"receipt_id": (
"rc06-online-transaction-"
+ authority.batch._sha256_bytes(
authority._canonical(online_payload) + online_signature
)[:24]
),
"payload": online_payload,
"signature": base64.b64encode(online_signature).decode("ascii"),
}
online_receipts_file = tmp_path / "online-transaction-receipts.json"
online_receipts_raw = authority._canonical(online_receipt)
online_receipts_file.write_bytes(online_receipts_raw)
runtime_payload = {
"schema_version": authority.batch.RUNTIME_PAYLOAD_SCHEMA_VERSION,
"batch_manifest_raw_sha256": manifest_sha,
"batch_manifest_payload_sha256": manifest["manifest_payload_sha256"],
"approval_receipt_sha256": approval_sha,
"legal_source_review_binding_sha256": challenge_payload[
"legal_source_review_binding_sha256"
],
"environment_id": "isolated-uat-fixture",
"tenant_id": "tenant-fixture",
"idempotency_key": challenge_payload["idempotency_key"],
"transaction_id": challenge_payload["transaction_id"],
"online_transaction_receipt_sha256": authority.batch._sha256_json(
online_receipt
),
"captured_at": "2026-09-27T04:01:00Z",
"generated_by": {
"type": "system",
"component": authority.batch.TRUSTED_ONLINE_APPLY_SERVICE_COMPONENT,
},
"transaction_mode": authority.batch.TRANSACTION_MODE,
"cross_database_atomicity": False,
"publication_item_count": 74,
"runtime_snapshot_dependencies": runtime_results,
"runtime_snapshot_dependency_count": 14,
"status": "published_and_resolved",
}
runtime_receipt = _kms_envelope(
online_private,
schema=authority.batch.RUNTIME_RECEIPT_SCHEMA_VERSION,
prefix="rc06-runtime-snapshot-",
domain=authority.batch.RUNTIME_SIGNATURE_DOMAIN,
payload=runtime_payload,
request_id="fixture-kms-runtime",
)
runtime_receipt_file = tmp_path / "runtime-snapshot-receipt.json"
runtime_receipt_raw = authority._canonical(runtime_receipt)
runtime_receipt_file.write_bytes(runtime_receipt_raw)
return {
"state": state,
"keys": keys,
@@ -328,11 +591,20 @@ def _formal_apply_materials(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) ->
"approval_sha": approval_sha,
"challenge": challenge,
"challenge_file": challenge_file,
"legal_request_file": legal_request_file,
"legal_request_sha": legal_request_sha,
"legal_decision_file": legal_decision_file,
"legal_decision_sha": legal_decision_sha,
"legal_keyring_file": legal_keyring_file,
"legal_keyring_sha": legal_keyring_sha,
"online_private": online_private,
"online_public_file": online_public_file,
"online_receipt": online_receipt,
"online_receipts_file": online_receipts_file,
"online_receipts_sha": authority.batch._sha256_bytes(online_receipts_raw),
"runtime_receipt": runtime_receipt,
"runtime_receipt_file": runtime_receipt_file,
"runtime_receipt_sha": authority.batch._sha256_bytes(runtime_receipt_raw),
}
@@ -494,6 +766,8 @@ def test_pin_preview_and_apply_update_only_public_hashes_and_epoch_in_temp_sourc
repo = _copy_pin_sources(tmp_path)
online_registration_file = tmp_path / "online-service-registration.json"
online_registration = _online_service_registration(online_registration_file)
legal_keyring_file = tmp_path / "scene-legal-keyring.json"
_legal_keyring(legal_keyring_file)
online_environments = ("isolated-uat-fixture",)
before = {
path.relative_to(repo): path.read_bytes()
@@ -508,15 +782,14 @@ def test_pin_preview_and_apply_update_only_public_hashes_and_epoch_in_temp_sourc
unprotect=_unprotect,
online_service_registration_file=online_registration_file,
online_environment_ids=online_environments,
legal_review_keyring_file=legal_keyring_file,
)
preview = authority.pin_plan_summary(plan)
assert preview["status"] == "preview_only_not_applied"
assert all(path.read_bytes() == before[path.relative_to(repo)] for path in repo.rglob("*.py"))
assert preview["values"]["online_apply_adapter_pinned"] is True
assert preview["values"]["online_apply_environment_ids"] == online_environments
for key, value in preview["values"].items():
if key not in {"online_apply_adapter_pinned", "online_apply_environment_ids"}:
assert len(str(value)) in {1, 64}
assert preview["values"]["scene_legal_review_keyring_pinned"] is True
applied = authority.apply_pin_plan(plan)
assert applied["status"] == "pins_applied_recoverable_transaction_not_frozen_not_signed"
@@ -534,7 +807,12 @@ def test_pin_preview_and_apply_update_only_public_hashes_and_epoch_in_temp_sourc
)
assert f'TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256 = "{keys[authority.ROLE_RC06_ROOT].public_sha256}"' in rc06
assert f'TRUSTED_APPLY_SERVICE_PUBLIC_KEY_SHA256 = "{keys[authority.ROLE_APPLY_SERVICE].public_sha256}"' in rc06
assert f'TRUSTED_ONLINE_APPLY_SERVICE_PUBLIC_KEY_SHA256 = "{online_registration["public_key_raw_sha256"]}"' in rc06
assert f'TRUSTED_ONLINE_APPLY_SERVICE_PROVIDER = "{online_registration["provider"]}"' in rc06
assert f'TRUSTED_ONLINE_APPLY_SERVICE_KEY_REFERENCE = "{online_registration["key_reference"]}"' in rc06
assert f'TRUSTED_ONLINE_APPLY_SERVICE_KEY_VERSION = "{online_registration["key_version"]}"' in rc06
assert f'TRUSTED_ONLINE_APPLY_SERVICE_ALGORITHM = "{online_registration["algorithm"]}"' in rc06
assert f'TRUSTED_ONLINE_APPLY_SERVICE_PUBLIC_KEY_SPKI_SHA256 = "{online_registration["public_key_spki_sha256"]}"' in rc06
assert f'TRUSTED_SCENE_LEGAL_REVIEW_KEYRING_SHA256 = "{authority.batch._sha256_bytes(legal_keyring_file.read_bytes())}"' in rc06
assert "FROZEN_APPROVAL_KEYRING_EPOCH = 1" in rc06
for text in (first, rollback):
assert f'TRUSTED_ROOT_PUBLIC_KEY_SHA256 = "{keys[authority.ROLE_RECOVERY_ROOT].public_sha256}"' in text
@@ -549,6 +827,7 @@ def test_pin_preview_and_apply_update_only_public_hashes_and_epoch_in_temp_sourc
unprotect=_unprotect,
online_service_registration_file=online_registration_file,
online_environment_ids=online_environments,
legal_review_keyring_file=legal_keyring_file,
)
assert all(change.original == change.proposed for change in repeated.changes)
assert authority.apply_pin_plan(repeated)["changed_file_count"] == 0
@@ -999,6 +1278,8 @@ def test_apply_payload_signing_is_bound_verified_atomic_and_idempotent(
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"],
runtime_snapshot_receipt_file=materials["runtime_receipt_file"],
runtime_snapshot_receipt_sha256=materials["runtime_receipt_sha"],
output_directory=output,
unprotect=_unprotect,
now=datetime(2026, 9, 27, 4, 1, tzinfo=UTC),
@@ -1016,10 +1297,18 @@ def test_apply_payload_signing_is_bound_verified_atomic_and_idempotent(
materials["keyring"],
materials["keys"][authority.ROLE_RC06_ROOT].public_key,
materials["keys"][authority.ROLE_APPLY_SERVICE].public_key,
materials["online_private"].public_key(),
materials["approval_sha"],
receipt,
) == receipt
materials["online_private"].public_key(),
materials["approval_sha"],
receipt,
legal_review_binding_sha256=materials["challenge"]["payload"][
"legal_source_review_binding_sha256"
],
legal_review_expires_at=materials["challenge"]["payload"][
"legal_source_review_expires_at"
],
environment_id="isolated-uat-fixture",
tenant_id="tenant-fixture",
) == receipt
repeated = authority.sign_apply_payloads(
materials["state"],
manifest_file=materials["manifest_file"],
@@ -1032,6 +1321,8 @@ def test_apply_payload_signing_is_bound_verified_atomic_and_idempotent(
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"],
runtime_snapshot_receipt_file=materials["runtime_receipt_file"],
runtime_snapshot_receipt_sha256=materials["runtime_receipt_sha"],
output_directory=output,
unprotect=_unprotect,
now=datetime(2026, 9, 27, 4, 1, tzinfo=UTC),
@@ -1055,6 +1346,8 @@ def test_apply_payload_signing_is_bound_verified_atomic_and_idempotent(
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"],
runtime_snapshot_receipt_file=materials["runtime_receipt_file"],
runtime_snapshot_receipt_sha256=materials["runtime_receipt_sha"],
output_directory=tmp_path / "replayed-to-another-output",
unprotect=_unprotect,
now=datetime(2026, 9, 27, 4, 2, tzinfo=UTC),
@@ -1074,7 +1367,14 @@ def test_apply_challenge_is_unique_per_authorization_context_and_recovers_publis
"approval_keyring_file": materials["keyring_file"],
"approval_keyring_sha256": materials["keyring_sha"],
"approval_root_public_key_file": materials["root_file"],
"legal_review_request_file": materials["legal_request_file"],
"legal_review_request_sha256": materials["legal_request_sha"],
"legal_review_decision_file": materials["legal_decision_file"],
"legal_review_decision_sha256": materials["legal_decision_sha"],
"legal_review_keyring_file": materials["legal_keyring_file"],
"legal_review_keyring_sha256": materials["legal_keyring_sha"],
"environment_id": "isolated-uat-fixture",
"tenant_id": "tenant-fixture",
"unprotect": _unprotect,
}
repeated = authority.issue_apply_challenge(
@@ -1178,6 +1478,8 @@ def test_apply_payload_signing_rejects_tamper_cross_approval_and_partial_bundle(
online_service_public_key_file=materials["online_public_file"],
online_transaction_receipts_file=tampered_file,
online_transaction_receipts_sha256=authority.batch._sha256_bytes(tampered_raw),
runtime_snapshot_receipt_file=materials["runtime_receipt_file"],
runtime_snapshot_receipt_sha256=materials["runtime_receipt_sha"],
output_directory=tampered_output,
unprotect=_unprotect,
now=datetime(2026, 9, 27, 4, 1, tzinfo=UTC),
@@ -1188,24 +1490,21 @@ def test_apply_payload_signing_rejects_tamper_cross_approval_and_partial_bundle(
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 = _kms_envelope(
materials["online_private"],
schema=authority.batch.ONLINE_TRANSACTION_RECEIPT_SCHEMA_VERSION,
prefix="rc06-online-transaction-",
domain=authority.batch.ONLINE_TRANSACTION_SIGNATURE_DOMAIN,
payload=challenge_tampered["payload"],
request_id="fixture-kms-challenge-tamper",
)
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"):
with pytest.raises(
authority.AuthorityError,
match="challenge.*(binding drift|unique authorization-context challenge)",
):
authority.sign_apply_payloads(
materials["state"],
manifest_file=materials["manifest_file"],
@@ -1220,6 +1519,8 @@ def test_apply_payload_signing_rejects_tamper_cross_approval_and_partial_bundle(
online_transaction_receipts_sha256=authority.batch._sha256_bytes(
challenge_tampered_raw
),
runtime_snapshot_receipt_file=materials["runtime_receipt_file"],
runtime_snapshot_receipt_sha256=materials["runtime_receipt_sha"],
output_directory=tmp_path / "tampered-challenge-output",
unprotect=_unprotect,
now=datetime(2026, 9, 27, 4, 1, tzinfo=UTC),
@@ -1244,7 +1545,10 @@ def test_apply_payload_signing_rejects_tamper_cross_approval_and_partial_bundle(
second_approval_file = tmp_path / "second-approval.json"
second_approval_raw = authority._canonical(second_approval)
second_approval_file.write_bytes(second_approval_raw)
with pytest.raises(authority.AuthorityError, match="not bound to the approval receipt"):
with pytest.raises(
authority.AuthorityError,
match="not bound to the approval receipt|cannot open apply challenge context",
):
authority.sign_apply_payloads(
materials["state"],
manifest_file=materials["manifest_file"],
@@ -1257,6 +1561,8 @@ def test_apply_payload_signing_rejects_tamper_cross_approval_and_partial_bundle(
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"],
runtime_snapshot_receipt_file=materials["runtime_receipt_file"],
runtime_snapshot_receipt_sha256=materials["runtime_receipt_sha"],
output_directory=tmp_path / "cross-approval-output",
unprotect=_unprotect,
now=datetime(2026, 9, 27, 4, 1, tzinfo=UTC),
@@ -1282,6 +1588,8 @@ def test_apply_payload_signing_rejects_tamper_cross_approval_and_partial_bundle(
online_service_public_key_file=materials["online_public_file"],
online_transaction_receipts_file=payload_list_file,
online_transaction_receipts_sha256=authority.batch._sha256_bytes(payload_list_raw),
runtime_snapshot_receipt_file=materials["runtime_receipt_file"],
runtime_snapshot_receipt_sha256=materials["runtime_receipt_sha"],
output_directory=partial_output,
unprotect=_unprotect,
now=datetime(2026, 9, 27, 4, 1, tzinfo=UTC),
@@ -1319,6 +1627,8 @@ def test_apply_challenge_consumption_recovers_after_finalize_interrupt(
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"],
runtime_snapshot_receipt_file=materials["runtime_receipt_file"],
runtime_snapshot_receipt_sha256=materials["runtime_receipt_sha"],
output_directory=output,
unprotect=_unprotect,
now=datetime(2026, 9, 27, 4, 1, tzinfo=UTC),
@@ -1350,6 +1660,8 @@ def test_apply_challenge_consumption_recovers_after_finalize_interrupt(
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"],
runtime_snapshot_receipt_file=materials["runtime_receipt_file"],
runtime_snapshot_receipt_sha256=materials["runtime_receipt_sha"],
output_directory=output,
unprotect=observed_unprotect,
now=datetime(2026, 9, 27, 4, 2, tzinfo=UTC),
@@ -1431,6 +1743,8 @@ def test_public_status_and_apply_validation_use_least_private_key_unsealing(
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"],
runtime_snapshot_receipt_file=materials["runtime_receipt_file"],
runtime_snapshot_receipt_sha256=materials["runtime_receipt_sha"],
output_directory=output,
unprotect=observed_unprotect,
now=datetime(2026, 9, 27, 4, 1, tzinfo=UTC),
@@ -1450,6 +1764,8 @@ def test_public_status_and_apply_validation_use_least_private_key_unsealing(
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"],
runtime_snapshot_receipt_file=materials["runtime_receipt_file"],
runtime_snapshot_receipt_sha256=materials["runtime_receipt_sha"],
output_directory=output,
unprotect=observed_unprotect,
now=datetime(2026, 9, 27, 6, 0, tzinfo=UTC),
@@ -1480,11 +1796,13 @@ def test_apply_signing_rejects_unprovisioned_custom_environment_and_expired_wind
"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"],
"runtime_snapshot_receipt_file": materials["runtime_receipt_file"],
"runtime_snapshot_receipt_sha256": materials["runtime_receipt_sha"],
"unprotect": observed_unprotect,
}
monkeypatch.setattr(
authority.batch, "TRUSTED_ONLINE_APPLY_SERVICE_PUBLIC_KEY_SHA256", ""
authority.batch, "TRUSTED_ONLINE_APPLY_SERVICE_PUBLIC_KEY_SPKI_SHA256", ""
)
with pytest.raises(authority.AuthorityError, match="trust is not provisioned"):
authority.sign_apply_payloads(
@@ -1496,27 +1814,32 @@ def test_apply_signing_rejects_unprovisioned_custom_environment_and_expired_wind
assert calls == []
monkeypatch.setattr(
authority.batch,
"TRUSTED_ONLINE_APPLY_SERVICE_PUBLIC_KEY_SHA256",
authority.batch._sha256_bytes(materials["online_public_file"].read_bytes()),
"TRUSTED_ONLINE_APPLY_SERVICE_PUBLIC_KEY_SPKI_SHA256",
authority.batch._sha256_bytes(
materials["online_private"].public_key().public_bytes(
serialization.Encoding.DER,
serialization.PublicFormat.SubjectPublicKeyInfo,
)
),
)
custom = copy.deepcopy(materials["online_receipt"])
custom["payload"]["environment_id"] = "attacker-environment"
custom_signature = materials["online_private"].sign(
authority.batch.ONLINE_TRANSACTION_SIGNATURE_DOMAIN
+ authority._canonical(custom["payload"])
)
custom["signature"] = base64.b64encode(custom_signature).decode("ascii")
custom["receipt_id"] = (
"rc06-online-transaction-"
+ authority.batch._sha256_bytes(
authority._canonical(custom["payload"]) + custom_signature
)[:24]
custom = _kms_envelope(
materials["online_private"],
schema=authority.batch.ONLINE_TRANSACTION_RECEIPT_SCHEMA_VERSION,
prefix="rc06-online-transaction-",
domain=authority.batch.ONLINE_TRANSACTION_SIGNATURE_DOMAIN,
payload=custom["payload"],
request_id="fixture-kms-custom-environment",
)
custom_file = tmp_path / "custom-environment-receipt.json"
custom_raw = authority._canonical(custom)
custom_file.write_bytes(custom_raw)
with pytest.raises(authority.AuthorityError, match="frozen allowlist"):
with pytest.raises(
authority.AuthorityError,
match="frozen allowlist|environment or tenant binding drift",
):
authority.sign_apply_payloads(
materials["state"],
output_directory=tmp_path / "custom-output",
@@ -1533,16 +1856,13 @@ def test_apply_signing_rejects_unprovisioned_custom_environment_and_expired_wind
future = copy.deepcopy(materials["online_receipt"])
future["payload"]["committed_at"] = "2026-09-27T04:45:00Z"
future_signature = materials["online_private"].sign(
authority.batch.ONLINE_TRANSACTION_SIGNATURE_DOMAIN
+ authority._canonical(future["payload"])
)
future["signature"] = base64.b64encode(future_signature).decode("ascii")
future["receipt_id"] = (
"rc06-online-transaction-"
+ authority.batch._sha256_bytes(
authority._canonical(future["payload"]) + future_signature
)[:24]
future = _kms_envelope(
materials["online_private"],
schema=authority.batch.ONLINE_TRANSACTION_RECEIPT_SCHEMA_VERSION,
prefix="rc06-online-transaction-",
domain=authority.batch.ONLINE_TRANSACTION_SIGNATURE_DOMAIN,
payload=future["payload"],
request_id="fixture-kms-future-commit",
)
future_file = tmp_path / "future-online-receipt.json"
future_raw = authority._canonical(future)
@@ -1565,23 +1885,14 @@ 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"],
output_directory=tmp_path / "expired-output",
now=datetime(2026, 9, 27, 5, 0, tzinfo=UTC),
**common,
)
assert calls == []
delayed = authority.sign_apply_payloads(
materials["state"],
output_directory=tmp_path / "post-challenge-countersign-output",
now=datetime(2026, 9, 27, 4, 6, tzinfo=UTC),
**common,
)
assert delayed["status"] == "apply_receipts_signed_and_verified_not_published"
assert calls == [authority.ROLE_APPLY_SERVICE]
def _pin_plan_fixture(tmp_path: Path, *, include_online: bool = True):
@@ -1595,6 +1906,8 @@ def _pin_plan_fixture(tmp_path: Path, *, include_online: bool = True):
now=datetime(2026, 9, 27, 2, 0, tzinfo=UTC),
)
repo = _copy_pin_sources(tmp_path)
legal_keyring_file = tmp_path / "scene-legal-keyring.json"
_legal_keyring(legal_keyring_file)
online_kwargs = {}
if include_online:
online_registration_file = tmp_path / "online-service-registration.json"
@@ -1609,6 +1922,7 @@ def _pin_plan_fixture(tmp_path: Path, *, include_online: bool = True):
Path(recovery["directory"]),
epoch=1,
unprotect=_unprotect,
legal_review_keyring_file=legal_keyring_file,
**online_kwargs,
)
return state, repo, plan
@@ -1872,22 +2186,8 @@ def test_pin_plan_can_freeze_online_service_key_and_exact_environment_allowlist(
) -> None:
state, repo, base_plan = _pin_plan_fixture(tmp_path, include_online=False)
recovery_directory = state / "keyrings" / "recovery-epoch-1"
online_private = Ed25519PrivateKey.generate()
online_raw = online_private.public_key().public_bytes(
serialization.Encoding.Raw,
serialization.PublicFormat.Raw,
)
registration = {
"schema_version": authority.ONLINE_SERVICE_REGISTRATION_SCHEMA,
"key_id": authority.batch.TRUSTED_ONLINE_APPLY_SERVICE_KEY_ID,
"algorithm": "Ed25519",
"component": authority.batch.TRUSTED_ONLINE_APPLY_SERVICE_COMPONENT,
"created_at": "2026-09-27T02:30:00Z",
"public_key_raw_base64": base64.b64encode(online_raw).decode("ascii"),
"public_key_raw_sha256": authority.batch._sha256_bytes(online_raw),
}
registration_file = tmp_path / "online-service-registration.json"
registration_file.write_bytes(authority._canonical(registration))
registration = _online_service_registration(registration_file)
plan = authority.plan_pin_updates(
repo,
state,
@@ -1896,6 +2196,7 @@ def test_pin_plan_can_freeze_online_service_key_and_exact_environment_allowlist(
unprotect=_unprotect,
online_service_registration_file=registration_file,
online_environment_ids=("uat-cn-north-1", "uat-cn-south-1"),
legal_review_keyring_file=tmp_path / "scene-legal-keyring.json",
)
assert plan.values["online_apply_adapter_pinned"] is True
assert plan.values["online_apply_environment_ids"] == (
@@ -1909,7 +2210,7 @@ def test_pin_plan_can_freeze_online_service_key_and_exact_environment_allowlist(
)
proposed = rc06_change.proposed.decode("utf-8")
assert (
f'TRUSTED_ONLINE_APPLY_SERVICE_PUBLIC_KEY_SHA256 = "{registration["public_key_raw_sha256"]}"'
f'TRUSTED_ONLINE_APPLY_SERVICE_PUBLIC_KEY_SPKI_SHA256 = "{registration["public_key_spki_sha256"]}"'
in proposed
)
assert (
@@ -1944,8 +2245,8 @@ def test_pin_apply_fails_closed_without_online_adapter_and_rejects_preseeded_tru
source = rc06.read_text(encoding="utf-8")
rc06.write_text(
source.replace(
'TRUSTED_ONLINE_APPLY_SERVICE_PUBLIC_KEY_SHA256 = ""',
'TRUSTED_ONLINE_APPLY_SERVICE_PUBLIC_KEY_SHA256 = "' + "f" * 64 + '"',
'TRUSTED_ONLINE_APPLY_SERVICE_PUBLIC_KEY_SPKI_SHA256 = ""',
'TRUSTED_ONLINE_APPLY_SERVICE_PUBLIC_KEY_SPKI_SHA256 = "' + "f" * 64 + '"',
),
encoding="utf-8",
)