From 3920b6e98533c4857a36cd2a49c55aa4555add1e Mon Sep 17 00:00:00 2001 From: yuqei Date: Tue, 6 Oct 2026 23:55:03 +0800 Subject: [PATCH] fix(rc06): harden publication facade recovery --- ...261006_0082_add_rc06_publication_facade.py | 18 +- .../src/yuqei_contract_api/config.py | 63 +++- .../src/yuqei_contract_api/main.py | 7 +- .../yuqei_contract_api/persistence/models.py | 19 +- .../yuqei_contract_api/rc06_publication.py | 92 ++++-- .../rc06_publication_facade.py | 290 +++++++++++++++--- .../rc06_publication_protocol.py | 117 ++++++- .../src/yuqei_contract_api/schema_check.py | 25 ++ .../tests/test_rc06_migrations.py | 87 ++++++ .../tests/test_rc06_publication_facade.py | 215 ++++++++++++- 10 files changed, 835 insertions(+), 98 deletions(-) diff --git a/yuqei-contract-platform/services/contract-api/migrations/versions/20261006_0082_add_rc06_publication_facade.py b/yuqei-contract-platform/services/contract-api/migrations/versions/20261006_0082_add_rc06_publication_facade.py index c450ba43..49a246e5 100644 --- a/yuqei-contract-platform/services/contract-api/migrations/versions/20261006_0082_add_rc06_publication_facade.py +++ b/yuqei-contract-platform/services/contract-api/migrations/versions/20261006_0082_add_rc06_publication_facade.py @@ -16,6 +16,15 @@ depends_on = None def upgrade() -> None: + bind = op.get_bind() + saga_count = bind.execute( + sa.text("SELECT COUNT(*) FROM rc06_publication_sagas") + ).scalar_one() + if saga_count: + raise RuntimeError( + "RC06 migration 0082 refuses to attach facade authority provenance " + "to pre-existing saga rows; retire or explicitly adjudicate them first." + ) with op.batch_alter_table("rc06_publication_sagas") as batch: batch.add_column(sa.Column("ai_compensation_status_json", sa.Text(), nullable=True)) @@ -121,6 +130,7 @@ def upgrade() -> None: ["transaction_id"], ["rc06_online_publication_transactions.transaction_id"], ondelete="NO ACTION", + name="fk_rc06_apply_challenge_transaction", ), sa.UniqueConstraint("nonce_sha256", name="ux_rc06_apply_challenge_nonce"), sa.UniqueConstraint("transaction_id", name="ux_rc06_apply_challenge_transaction"), @@ -135,13 +145,19 @@ def upgrade() -> None: def downgrade() -> None: bind = op.get_bind() + compensation_count = bind.execute( + sa.text( + "SELECT COUNT(*) FROM rc06_publication_sagas " + "WHERE ai_compensation_status_json IS NOT NULL" + ) + ).scalar_one() online_count = bind.execute( sa.text("SELECT COUNT(*) FROM rc06_online_publication_transactions") ).scalar_one() challenge_count = bind.execute( sa.text("SELECT COUNT(*) FROM rc06_apply_challenge_uses") ).scalar_one() - if online_count or challenge_count: + if online_count or challenge_count or compensation_count: raise RuntimeError( "RC06 migration 0082 refuses to erase durable publication authority, " "receipt, countersign, or challenge-consumption evidence." diff --git a/yuqei-contract-platform/services/contract-api/src/yuqei_contract_api/config.py b/yuqei-contract-platform/services/contract-api/src/yuqei_contract_api/config.py index 63db6920..d2410d96 100644 --- a/yuqei-contract-platform/services/contract-api/src/yuqei_contract_api/config.py +++ b/yuqei-contract-platform/services/contract-api/src/yuqei_contract_api/config.py @@ -2,6 +2,7 @@ from email.headerregistry import Address from functools import lru_cache import json from pathlib import Path +import os import re from typing import Literal from urllib.parse import urlparse @@ -748,13 +749,59 @@ def _check_rc06_publication_facade_config(settings: ContractApiSettings) -> Runt ): if re.fullmatch(r"[a-f0-9]{64}", value) is None: errors.append(f"{label} is invalid") - for label, value in ( + file_values = ( ("approval root public key", settings.rc06_approval_root_public_key_file), ("apply service public key", settings.rc06_apply_service_public_key_file), ("facade bearer token", settings.rc06_publication_bearer_token_file), - ): - if not value or not Path(value).is_absolute(): + ("KMS credential", settings.reliable_evidence_signing_credential_file), + ("KMS CA", settings.reliable_evidence_signing_ca_file), + ) + resolved_files: dict[str, Path] = {} + for label, value in file_values: + path = Path(value) if value else None + if path is None or not path.is_absolute(): errors.append(f"{label} path must be absolute") + continue + resolved_files[label] = path + if not path.is_file(): + errors.append(f"{label} file is unavailable") + continue + try: + # Windows does not expose POSIX group/world write bits faithfully; + # Docker/Linux production does, so enforce them where meaningful. + if os.name != "nt" and path.stat().st_mode & 0o022: + errors.append(f"{label} file must not be group/world writable") + except OSError: + errors.append(f"{label} file metadata is unavailable") + if len(set(resolved_files.values())) != len(resolved_files): + errors.append("facade trust and credential files must use distinct paths") + bearer_path = resolved_files.get("facade bearer token") + if bearer_path is not None and bearer_path.is_file(): + try: + bearer = bearer_path.read_text(encoding="utf-8").strip() + except (OSError, UnicodeError): + errors.append("facade bearer token cannot be read as UTF-8") + else: + if len(bearer.encode("utf-8")) < 32: + errors.append("facade bearer token must contain at least 32 bytes") + reused = { + settings.web_proxy_shared_secret.strip(), + settings.rc06_publication_shared_secret.strip(), + settings.ai_platform_service_key.strip(), + settings.ai_platform_rc06_publication_key.strip(), + } + if bearer in reused: + errors.append("facade bearer token must be independent from other service credentials") + public_key_paths = [ + resolved_files.get("approval root public key"), + resolved_files.get("apply service public key"), + ] + if all(path is not None and path.is_file() for path in public_key_paths): + try: + if public_key_paths[0].read_bytes() == public_key_paths[1].read_bytes(): + errors.append("approval root and apply service keys must be distinct") + except OSError: + errors.append("facade public key files cannot be read") if settings.rc06_online_signing_algorithm not in { "RSA_PSS_SHA_256", "RSA_PKCS1_SHA_256", "ECDSA_SHA_256" }: @@ -777,6 +824,16 @@ def _check_rc06_publication_facade_config(settings: ContractApiSettings) -> Runt ) +def validate_rc06_publication_startup(settings: ContractApiSettings) -> None: + """Fail process startup when an enabled RC06 facade is not deployable.""" + + if not settings.rc06_publication_facade_enabled: + return + result = _check_rc06_publication_facade_config(settings) + if result.status != "ok": + raise RuntimeError(result.message) + + def _check_worker_config(settings: ContractApiSettings) -> RuntimeConfigCheck: if not settings.review_worker_name.strip(): return RuntimeConfigCheck( diff --git a/yuqei-contract-platform/services/contract-api/src/yuqei_contract_api/main.py b/yuqei-contract-platform/services/contract-api/src/yuqei_contract_api/main.py index e22f22f6..cea6967f 100644 --- a/yuqei-contract-platform/services/contract-api/src/yuqei_contract_api/main.py +++ b/yuqei-contract-platform/services/contract-api/src/yuqei_contract_api/main.py @@ -38,7 +38,11 @@ from yuqei_contract_api.rc06_publication_facade import ( is_rc06_publication_facade_path, register_rc06_publication_facade_routes, ) -from yuqei_contract_api.config import ContractApiSettings, get_settings +from yuqei_contract_api.config import ( + ContractApiSettings, + get_settings, + validate_rc06_publication_startup, +) from yuqei_contract_api.logging import ( clear_request_tenant_id, configure_logging, @@ -71,6 +75,7 @@ def create_app( rc06_ai_client: AiRc06Client | None = None, ) -> FastAPI: resolved_settings = settings or get_settings() + validate_rc06_publication_startup(resolved_settings) configure_logging(resolved_settings.log_level) repository = contract_repository or build_contract_repository( resolved_settings.database_url, diff --git a/yuqei-contract-platform/services/contract-api/src/yuqei_contract_api/persistence/models.py b/yuqei-contract-platform/services/contract-api/src/yuqei_contract_api/persistence/models.py index c3622ad4..f6335f57 100644 --- a/yuqei-contract-platform/services/contract-api/src/yuqei_contract_api/persistence/models.py +++ b/yuqei-contract-platform/services/contract-api/src/yuqei_contract_api/persistence/models.py @@ -1776,6 +1776,12 @@ class Rc06OnlinePublicationTransactionModel(Base): "idempotency_key", name="ux_rc06_online_publication_idempotency", ), + UniqueConstraint( + "challenge_id", name="ux_rc06_online_publication_challenge_id" + ), + UniqueConstraint( + "challenge_sha256", name="ux_rc06_online_publication_challenge_sha256" + ), CheckConstraint( "state in ('authorized', 'saga_prepared', 'saga_committing', 'ai_committed', " "'contract_committed', 'evidence_pending', 'online_primary_signed', " @@ -1799,8 +1805,8 @@ class Rc06OnlinePublicationTransactionModel(Base): approval_receipt_sha256: Mapped[str] = mapped_column(String(64), nullable=False) legal_source_review_binding_sha256: Mapped[str] = mapped_column(String(64), nullable=False) legal_source_review_expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) - challenge_id: Mapped[str] = mapped_column(String(64), nullable=False, unique=True) - challenge_sha256: Mapped[str] = mapped_column(String(64), nullable=False, unique=True) + challenge_id: Mapped[str] = mapped_column(String(64), nullable=False) + challenge_sha256: Mapped[str] = mapped_column(String(64), nullable=False) state: Mapped[str] = mapped_column(String(64), nullable=False, index=True) request_json: Mapped[str] = mapped_column(Text, nullable=False) prepare_request_json: Mapped[str] = mapped_column(Text, nullable=False) @@ -1844,6 +1850,7 @@ class Rc06ApplyChallengeUseModel(Base): __table_args__ = ( UniqueConstraint("nonce_sha256", name="ux_rc06_apply_challenge_nonce"), UniqueConstraint("transaction_id", name="ux_rc06_apply_challenge_transaction"), + UniqueConstraint("challenge_sha256", name="ux_rc06_apply_challenge_sha256"), CheckConstraint( "outcome in ('reserved', 'pending_countersign', 'finalized', 'rolled_back', " "'compensation_failed', 'no_go')", @@ -1853,12 +1860,16 @@ class Rc06ApplyChallengeUseModel(Base): challenge_id: Mapped[str] = mapped_column(String(64), primary_key=True) nonce_sha256: Mapped[str] = mapped_column(String(64), nullable=False) - challenge_sha256: Mapped[str] = mapped_column(String(64), nullable=False, unique=True) + challenge_sha256: Mapped[str] = mapped_column(String(64), nullable=False) tenant_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True) idempotency_key: Mapped[str] = mapped_column(String(160), nullable=False) transaction_id: Mapped[str] = mapped_column( String(64), - ForeignKey("rc06_online_publication_transactions.transaction_id", ondelete="NO ACTION"), + ForeignKey( + "rc06_online_publication_transactions.transaction_id", + ondelete="NO ACTION", + name="fk_rc06_apply_challenge_transaction", + ), nullable=False, ) request_sha256: Mapped[str] = mapped_column(String(64), nullable=False) diff --git a/yuqei-contract-platform/services/contract-api/src/yuqei_contract_api/rc06_publication.py b/yuqei-contract-platform/services/contract-api/src/yuqei_contract_api/rc06_publication.py index 961f8029..ffb5e50f 100644 --- a/yuqei-contract-platform/services/contract-api/src/yuqei_contract_api/rc06_publication.py +++ b/yuqei-contract-platform/services/contract-api/src/yuqei_contract_api/rc06_publication.py @@ -469,29 +469,7 @@ class Rc06PublicationCoordinator: self._commit_templates(transaction_id) row = self._load_saga(transaction_id, tenant_id=tenant_id) if row.state == "contract_committed": - ai_status = self._ai.status(transaction_id) - if ai_status.get("state") != "sealed": - self._assert_ai_state( - ai_status, - transaction_id, - self._frozen_request(row), - expected_state="committed", - ) - capability = self._require_compensation_capability(row) - try: - ai_status = self._ai.seal(transaction_id, capability) - except Exception: - # The seal may have committed even when its HTTP response - # was lost. Reconcile by the durable participant state. - ai_status = self._ai.status(transaction_id) - self._assert_ai_state( - ai_status, - transaction_id, - self._frozen_request(row), - expected_state="sealed", - ) - self._record_ai_sealed(transaction_id, ai_status) - self._finalize(transaction_id, ai_status) + self._finish_contract_committed(transaction_id, row) return self._verify_final(transaction_id) except KeyError: raise @@ -511,6 +489,74 @@ class Rc06PublicationCoordinator: raise raise Rc06PublicationError(f"RC06 commit failed: {type(error).__name__}") from error + def recover_after_contract_commit( + self, + transaction_id: str, + *, + tenant_id: str | None = None, + ) -> Rc06SagaStatus: + """Finish readback/sealing only after Contract writes are durable. + + This entry point deliberately cannot advance prepared or AI-only state. + It is used by the facade when the short-lived apply authority has + expired but the no-rollback Contract boundary is already proven. + """ + + try: + row = self._load_saga(transaction_id, tenant_id=tenant_id) + if row.state == "committed": + return self._verify_final(transaction_id) + if row.state != "contract_committed": + raise Rc06Conflict( + f"RC06 post-commit recovery cannot run from {row.state}." + ) + self._finish_contract_committed(transaction_id, row) + return self._verify_final(transaction_id) + except KeyError: + raise + except Exception as error: + try: + if self._load_saga(transaction_id, tenant_id=tenant_id).state == "contract_committed": + self._record_error(transaction_id, error) + except Exception: + pass + if isinstance(error, Rc06Conflict): + raise + raise Rc06PublicationError( + f"RC06 post-commit recovery failed: {type(error).__name__}" + ) from error + + def _finish_contract_committed( + self, + transaction_id: str, + row: Rc06PublicationSagaModel, + ) -> None: + if row.state != "contract_committed" or row.contract_committed_at is None: + raise Rc06Conflict("RC06 Contract durable commit boundary is not proven.") + ai_status = self._ai.status(transaction_id) + if ai_status.get("state") != "sealed": + self._assert_ai_state( + ai_status, + transaction_id, + self._frozen_request(row), + expected_state="committed", + ) + capability = self._require_compensation_capability(row) + try: + ai_status = self._ai.seal(transaction_id, capability) + except Exception: + # The seal may have committed even when its HTTP response was + # lost. Reconcile by the durable participant state. + ai_status = self._ai.status(transaction_id) + self._assert_ai_state( + ai_status, + transaction_id, + self._frozen_request(row), + expected_state="sealed", + ) + self._record_ai_sealed(transaction_id, ai_status) + self._finalize(transaction_id, ai_status) + def compensate(self, transaction_id: str, *, tenant_id: str | None = None) -> Rc06SagaStatus: compensation_capability: str with _coordinator_lock, self._session_factory.begin() as session: diff --git a/yuqei-contract-platform/services/contract-api/src/yuqei_contract_api/rc06_publication_facade.py b/yuqei-contract-platform/services/contract-api/src/yuqei_contract_api/rc06_publication_facade.py index 24f83ad9..702df1f7 100644 --- a/yuqei-contract-platform/services/contract-api/src/yuqei_contract_api/rc06_publication_facade.py +++ b/yuqei-contract-platform/services/contract-api/src/yuqei_contract_api/rc06_publication_facade.py @@ -13,6 +13,7 @@ import hashlib import hmac import json import re +from contextlib import contextmanager from dataclasses import dataclass from datetime import UTC, datetime from pathlib import Path @@ -113,6 +114,9 @@ class Rc06FacadeSigner(Protocol): class Rc06FacadeCoordinator(Protocol): def prepare(self, payload: Rc06PrepareRequest, *, tenant_id: str) -> Rc06SagaStatus: ... def commit(self, transaction_id: str, *, tenant_id: str | None = None) -> Rc06SagaStatus: ... + def recover_after_contract_commit( + self, transaction_id: str, *, tenant_id: str | None = None + ) -> Rc06SagaStatus: ... def compensate(self, transaction_id: str, *, tenant_id: str | None = None) -> Rc06SagaStatus: ... def status(self, transaction_id: str, *, tenant_id: str | None = None) -> Rc06SagaStatus: ... @@ -383,27 +387,38 @@ class Rc06PublicationFacadeService: checked_at = (now or datetime.now(UTC)).astimezone(UTC) request_sha256 = sha256_json(request) self._assert_request_headers(request, headers, request_sha256=request_sha256) - material, row = self._authorize_and_reserve(request, request_sha256, checked_at) - if row.state == "rolled_back": - raise Rc06FacadeConflict("RC06 transaction was already rolled back.") - if row.state == "finalized": - return self._apply_response(row) - try: - saga = self._drive_saga(row.transaction_id, material) - return self._drive_receipts(row.transaction_id, material, saga) - except (Rc06FacadeError, Rc06FacadeConflict, Rc06ProtocolError, Rc06PublicationError, Rc06Conflict) as error: - self._record_error(row.transaction_id, error) - raise - except Exception as error: - self._record_error(row.transaction_id, error) - raise Rc06FacadeError(f"RC06 online apply failed: {type(error).__name__}") from error + transaction_id = request.get("transaction_id") + with self._transaction_singleflight(transaction_id): + material, row, authority_current = self._authorize_and_reserve( + request, request_sha256, checked_at + ) + if row.state == "rolled_back": + raise Rc06FacadeConflict("RC06 transaction was already rolled back.") + if row.state in {"online_committed_pending_countersign", "finalized"}: + outcome = "finalized" if row.state == "finalized" else "pending_countersign" + at = row.finalized_at or row.pending_countersign_at or row.updated_at + self._consume_challenge(row.challenge_id, outcome, _utc(at)) + return self._apply_response(row) + try: + saga = self._drive_saga( + row.transaction_id, + material, + authority_current=authority_current, + ) + return self._drive_receipts(row.transaction_id, material, saga) + except (Rc06FacadeError, Rc06FacadeConflict, Rc06ProtocolError, Rc06PublicationError, Rc06Conflict) as error: + self._record_error(row.transaction_id, error) + raise + except Exception as error: + self._record_error(row.transaction_id, error) + raise Rc06FacadeError(f"RC06 online apply failed: {type(error).__name__}") from error def _authorize_and_reserve( self, request: dict[str, Any], request_sha256: str, now: datetime, - ) -> tuple[AuthorizedMaterial, Rc06OnlinePublicationTransactionModel]: + ) -> tuple[AuthorizedMaterial, Rc06OnlinePublicationTransactionModel, bool]: idempotency_key = request.get("idempotency_key") transaction_id = request.get("transaction_id") if ( @@ -425,7 +440,17 @@ class Rc06PublicationFacadeService: if existing.transaction_id != transaction_id or existing.request_sha256 != request_sha256: raise Rc06FacadeConflict("Idempotency key was already used for different RC06 material.") replay_time = _utc(existing.created_at) - material = self._authority.authorize(request, now=replay_time or now) + authority_current = True + try: + material = self._authority.authorize(request, now=now) + except Rc06ProtocolError: + if replay_time is None: + raise + # An exact durable replay still needs a full signature/content + # validation. The original reservation instant is used only for + # that static check; it never grants authority for new writes. + material = self._authority.authorize(request, now=replay_time) + authority_current = False prepare = self._prepare_request(material).model_dump(mode="json") challenge = material.challenge challenge_payload = challenge["payload"] @@ -453,7 +478,7 @@ class Rc06PublicationFacadeService: ): raise Rc06FacadeConflict("RC06 durable reservation material drifted.") session.expunge(existing) - return material, existing + return material, existing, authority_current if session.get(Rc06OnlinePublicationTransactionModel, transaction_id) is not None: raise Rc06FacadeConflict("RC06 transaction id was already used.") created = Rc06OnlinePublicationTransactionModel( @@ -523,7 +548,7 @@ class Rc06PublicationFacadeService: )) session.flush() session.expunge(created) - return material, created + return material, created, authority_current except IntegrityError as error: raise Rc06FacadeConflict("RC06 challenge, nonce, transaction, or idempotency key was already consumed.") from error @@ -559,16 +584,52 @@ class Rc06PublicationFacadeService: runtime_identities=runtime, ) - def _drive_saga(self, transaction_id: str, material: AuthorizedMaterial) -> Rc06SagaStatus: + def _drive_saga( + self, + transaction_id: str, + material: AuthorizedMaterial, + *, + authority_current: bool, + ) -> Rc06SagaStatus: row = self._load(transaction_id) prepare = Rc06PrepareRequest.model_validate(_decode(row.prepare_request_json, {})) try: status = self._coordinator.status(transaction_id, tenant_id=self._tenant_id) except KeyError: + if not authority_current: + raise Rc06FacadeConflict( + "RC06 authority window expired before the participant saga was prepared." + ) status = self._coordinator.prepare(prepare, tenant_id=self._tenant_id) self._record_saga_status(transaction_id, status) if status.state == "prepared": self._set_state(transaction_id, "saga_prepared") + if not authority_current: + if status.state in {"prepared", "ai_committed"}: + self._set_state( + transaction_id, + self._facade_state(status.state), + error="authority window expired before Contract durable commit", + ) + raise Rc06FacadeConflict( + "RC06 authority window expired before Contract durable commit; " + "explicit rollback is required." + ) + if status.state == "contract_committed": + self._set_state( + transaction_id, + "contract_committed", + contract_committed_at=status.contract_committed_at, + ) + status = self._coordinator.recover_after_contract_commit( + transaction_id, + tenant_id=self._tenant_id, + ) + self._record_saga_status(transaction_id, status) + elif status.state != "committed": + raise Rc06FacadeConflict( + "RC06 expired authority may only recover an already committed transaction." + ) if status.state in {"prepared", "ai_committed", "contract_committed"}: self._set_state(transaction_id, self._facade_state(status.state, committing=True)) try: @@ -676,9 +737,10 @@ class Rc06PublicationFacadeService: online_receipt = _decode(row.online_receipt_json, None) if not isinstance(online_receipt, dict): raise Rc06FacadeError("RC06 online receipt persistence is incomplete.") - captured_at = datetime.now(UTC) - if captured_at < committed_at: - captured_at = committed_at + # The coordinator persists this timestamp with the exact 74/14 + # readback. Reuse it after a process restart; recovery time must not + # masquerade as the original runtime snapshot capture time. + captured_at = committed_at if captured_at >= authority_expires_at or captured_at >= approval_expires_at: self._set_state(transaction_id, "evidence_pending", error="runtime capture outside authority window") raise Rc06FacadeError("RC06 runtime evidence could not be captured inside its authority window.") @@ -859,19 +921,25 @@ class Rc06PublicationFacadeService: raise Rc06FacadeConflict("Persisted runtime receipt envelope drifted.") if receipt is None: now = datetime.now(UTC) - self._update( - transaction_id, - state="online_committed_pending_countersign", - runtime_receipt_json=_encode(expected), - runtime_receipt_sha256=sha256_json(expected), - pending_countersign_at=now, - error_message=None, - failure_code=None, - failure_at=None, - failure_evidence_sha256=None, - ) with _facade_lock, self._session_factory.begin() as session: - use = session.get(Rc06ApplyChallengeUseModel, self._load(transaction_id).challenge_id) + self._lock_transaction(session, transaction_id) + current = session.get(Rc06OnlinePublicationTransactionModel, transaction_id) + if current is None or current.tenant_id != self._tenant_id: + raise KeyError(transaction_id) + self._assert_frozen_row(current) + persisted = _decode(current.runtime_receipt_json, None) + if persisted is not None and persisted != expected: + raise Rc06FacadeConflict("Persisted runtime receipt envelope drifted.") + current.state = "online_committed_pending_countersign" + current.runtime_receipt_json = _encode(expected) + current.runtime_receipt_sha256 = sha256_json(expected) + current.pending_countersign_at = now + current.error_message = None + current.failure_code = None + current.failure_at = None + current.failure_evidence_sha256 = None + current.updated_at = now + use = session.get(Rc06ApplyChallengeUseModel, current.challenge_id) if use is None: raise Rc06FacadeError("RC06 challenge consumption ledger is missing.") use.outcome = "pending_countersign" @@ -910,8 +978,47 @@ class Rc06PublicationFacadeService: }, request_id def status(self, transaction_id: str, headers: Rc06FacadeHeaders) -> dict[str, Any]: + with self._transaction_singleflight(transaction_id): + return self._status_singleflight(transaction_id, headers) + + def _status_singleflight( + self, + transaction_id: str, + headers: Rc06FacadeHeaders, + ) -> dict[str, Any]: row = self._load(transaction_id) self._assert_status_headers(row, headers) + if row.state not in { + "online_committed_pending_countersign", "finalized", "rolled_back", + }: + try: + saga = self._coordinator.status(transaction_id, tenant_id=self._tenant_id) + except (KeyError, Rc06PublicationError, Rc06Conflict): + saga = None + if saga is not None: + self._record_saga_status(transaction_id, saga) + if saga.state in {"contract_committed", "committed"}: + current = self._load(transaction_id) + if current.contract_committed_at is None: + self._set_state( + transaction_id, + "contract_committed", + contract_committed_at=saga.contract_committed_at, + error="participant commit is durable; facade evidence is pending", + ) + self._record_error( + transaction_id, + Rc06FacadeError( + "participant commit is durable; facade evidence is pending" + ), + ) + elif saga.state in {"prepared", "ai_committed"}: + self._set_state( + transaction_id, + self._facade_state(saga.state), + error=saga.error, + ) + row = self._load(transaction_id) request = _decode(row.request_json, {}) committed = row.state in {"online_committed_pending_countersign", "finalized"} rolled_back = row.state == "rolled_back" @@ -1010,6 +1117,15 @@ class Rc06PublicationFacadeService: transaction_id: str, request: dict[str, Any], headers: Rc06FacadeHeaders, + ) -> dict[str, Any]: + with self._transaction_singleflight(transaction_id): + return self._rollback_singleflight(transaction_id, request, headers) + + def _rollback_singleflight( + self, + transaction_id: str, + request: dict[str, Any], + headers: Rc06FacadeHeaders, ) -> dict[str, Any]: row = self._load(transaction_id) self._assert_status_headers(row, headers) @@ -1034,6 +1150,11 @@ class Rc06PublicationFacadeService: raise Rc06FacadeConflict("RC06 rollback request binding drifted.") persisted = _decode(row.rollback_response_json, None) if persisted is not None: + self._consume_challenge( + row.challenge_id, + "rolled_back", + _utc(row.rolled_back_at or row.updated_at), + ) return persisted if row.state in { "contract_committed", "evidence_pending", "online_primary_signed", @@ -1136,8 +1257,10 @@ class Rc06PublicationFacadeService: response["rollback_evidence_sha256"] = sha256_json(response) response_sha256 = sha256_json(response) now = datetime.now(UTC) - self._update( + self._update_and_consume( transaction_id, + outcome="rolled_back", + consumed_at=now, state="rolled_back", rollback_response_json=_encode(response), rollback_evidence_sha256=response["rollback_evidence_sha256"], @@ -1148,7 +1271,6 @@ class Rc06PublicationFacadeService: failure_at=None, failure_evidence_sha256=None, ) - self._consume_challenge(row.challenge_id, "rolled_back", now) return response def finalize( @@ -1156,6 +1278,15 @@ class Rc06PublicationFacadeService: transaction_id: str, request: dict[str, Any], headers: Rc06FacadeHeaders, + ) -> dict[str, Any]: + with self._transaction_singleflight(transaction_id): + return self._finalize_singleflight(transaction_id, request, headers) + + def _finalize_singleflight( + self, + transaction_id: str, + request: dict[str, Any], + headers: Rc06FacadeHeaders, ) -> dict[str, Any]: row = self._load(transaction_id) self._assert_status_headers(row, headers) @@ -1185,6 +1316,11 @@ class Rc06PublicationFacadeService: if row.state == "finalized": if row.apply_receipt_sha256 != request["apply_receipt_sha256"]: raise Rc06FacadeConflict("RC06 transaction was finalized with different countersign evidence.") + self._consume_challenge( + row.challenge_id, + "finalized", + _utc(row.finalized_at or row.updated_at), + ) return self._finalize_response(row) if row.state != "online_committed_pending_countersign": raise Rc06FacadeConflict("RC06 transaction is not awaiting offline countersign.") @@ -1192,8 +1328,10 @@ class Rc06PublicationFacadeService: now = datetime.now(UTC) if now < countersigned_at: now = countersigned_at - self._update( + self._update_and_consume( transaction_id, + outcome="finalized", + consumed_at=now, state="finalized", apply_receipt_json=_encode(request["apply_receipt"]), apply_receipt_sha256=request["apply_receipt_sha256"], @@ -1203,7 +1341,6 @@ class Rc06PublicationFacadeService: failure_at=None, failure_evidence_sha256=None, ) - self._consume_challenge(row.challenge_id, "finalized", now) return self._finalize_response(self._load(transaction_id)) def _validate_apply_receipt( @@ -1367,6 +1504,33 @@ class Rc06PublicationFacadeService: setattr(row, key, value) row.updated_at = datetime.now(UTC) + def _update_and_consume( + self, + transaction_id: str, + *, + outcome: str, + consumed_at: datetime, + **values: Any, + ) -> None: + with _facade_lock, self._session_factory.begin() as session: + self._lock_transaction(session, transaction_id) + row = session.get(Rc06OnlinePublicationTransactionModel, transaction_id) + if row is None or row.tenant_id != self._tenant_id: + raise KeyError(transaction_id) + self._assert_frozen_row(row) + use = session.get(Rc06ApplyChallengeUseModel, row.challenge_id) + if use is None: + raise Rc06FacadeError("RC06 challenge consumption ledger is missing.") + if use.outcome == "finalized" and outcome != "finalized": + raise Rc06FacadeConflict("Finalized RC06 challenge outcome is immutable.") + for key, value in values.items(): + if not hasattr(row, key): + raise Rc06FacadeError(f"Unsupported RC06 durable field: {key}") + setattr(row, key, value) + row.updated_at = consumed_at + use.outcome = outcome + use.consumed_at = consumed_at + def _set_state( self, transaction_id: str, @@ -1391,7 +1555,13 @@ class Rc06PublicationFacadeService: error_message=status.error, ) - def _record_error(self, transaction_id: str, error: Exception) -> None: + def _record_error( + self, + transaction_id: str, + error: Exception, + *, + outcome: str = "no_go", + ) -> None: try: row = self._load(transaction_id) state = row.state @@ -1426,15 +1596,16 @@ class Rc06PublicationFacadeService: ), "saga_status_sha256": sha256_json(saga_status) if saga_status is not None else None, } - self._update( + self._update_and_consume( transaction_id, + outcome=outcome, + consumed_at=failed_at, state=state, error_message=message[:4000], failure_code=code, failure_at=failed_at, failure_evidence_sha256=sha256_json(evidence), ) - self._consume_challenge(row.challenge_id, "no_go", failed_at) else: self._set_state(transaction_id, state, error=message) except Exception: @@ -1457,9 +1628,11 @@ class Rc06PublicationFacadeService: *, outcome: str = "no_go", ) -> None: - self._record_error(row.transaction_id, Rc06FacadeError(error)) - if outcome != "no_go": - self._consume_challenge(row.challenge_id, outcome, datetime.now(UTC)) + self._record_error( + row.transaction_id, + Rc06FacadeError(error), + outcome=outcome, + ) @staticmethod def _facade_state(saga_state: str, *, committing: bool = False) -> str: @@ -1508,6 +1681,35 @@ class Rc06PublicationFacadeService: elif dialect == "sqlite": session.execute(text("BEGIN IMMEDIATE")) + @contextmanager + def _transaction_singleflight(self, transaction_id: Any): + """Serialize one facade transaction across workers while it mutates. + + PostgreSQL uses a session advisory lock on a namespace distinct from + the short transaction locks used by journal writes, so nested journal + sessions cannot deadlock on their own coordinator. SQLite tests use + the process lock because SQLite has no cross-connection advisory lock. + """ + + identity = transaction_id if isinstance(transaction_id, str) else "invalid" + with self._session_factory() as session: + if session.get_bind().dialect.name != "postgresql": + with _facade_lock: + yield + return + key = int.from_bytes( + hashlib.sha256( + f"rc06-online-singleflight:{identity}".encode() + ).digest()[:8], + "big", + signed=True, + ) + session.execute(text("SELECT pg_advisory_lock(:key)"), {"key": key}) + try: + yield + finally: + session.execute(text("SELECT pg_advisory_unlock(:key)"), {"key": key}) + def is_rc06_publication_facade_path(path: str) -> bool: base = "/internal/rc06/publication-transactions" diff --git a/yuqei-contract-platform/services/contract-api/src/yuqei_contract_api/rc06_publication_protocol.py b/yuqei-contract-platform/services/contract-api/src/yuqei_contract_api/rc06_publication_protocol.py index 2f5e03c3..ca95ccd6 100644 --- a/yuqei-contract-platform/services/contract-api/src/yuqei_contract_api/rc06_publication_protocol.py +++ b/yuqei-contract-platform/services/contract-api/src/yuqei_contract_api/rc06_publication_protocol.py @@ -15,6 +15,7 @@ from dataclasses import dataclass from datetime import UTC, datetime, timedelta from pathlib import Path from typing import Any, Mapping +from urllib.parse import urlparse from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives import serialization @@ -72,6 +73,8 @@ EXPECTED_ASSETS = 60 EXPECTED_MODULES = 14 EXPECTED_RUNTIME = 14 EXPECTED_GOLDEN = 31 +LEGAL_PLACEHOLDERS = {"", "-", "n/a", "none", "pending", "unknown", "待确认", "待复核"} +KEY_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$") class Rc06ProtocolError(RuntimeError): @@ -131,6 +134,45 @@ def require_text(value: Any, label: str, *, maximum: int = 4096) -> str: return value +def load_approval_public_key(value: Any) -> Ed25519PublicKey: + """Parse the exact signed PEM string without rejecting its standard LF.""" + + if not isinstance(value, str) or not value or len(value) > 16384 or "\x00" in value: + raise Rc06ProtocolError("Approval public key is invalid.") + lines = value.splitlines() + if ( + len(lines) < 3 + or lines[0] != "-----BEGIN PUBLIC KEY-----" + or lines[-1] != "-----END PUBLIC KEY-----" + or value.startswith(("\r", "\n", " ", "\t")) + or value.rstrip("\r\n") != value.rstrip("\r\n ") + ): + raise Rc06ProtocolError("Approval public key is invalid.") + try: + encoded = value.encode("ascii") + public_key = serialization.load_pem_public_key(encoded) + except (UnicodeError, ValueError) as exc: + raise Rc06ProtocolError("Approval public key is invalid.") from exc + if not isinstance(public_key, Ed25519PublicKey): + raise Rc06ProtocolError("Approval public key is invalid.") + return public_key + + +def require_official_url(value: Any, label: str) -> str: + text_value = require_text(value, label, maximum=2048) + parsed = urlparse(text_value) + host = (parsed.hostname or "").lower() + if ( + parsed.scheme != "https" + or not host + or parsed.username is not None + or parsed.password is not None + or not (host == "gov.cn" or host.endswith(".gov.cn")) + ): + raise Rc06ProtocolError(f"{label} must use a credential-free official HTTPS URL.") + return text_value + + def require_exact(value: Any, label: str, fields: tuple[str, ...]) -> dict[str, Any]: if not isinstance(value, dict) or set(value) != set(fields): raise Rc06ProtocolError(f"{label} fields are invalid.") @@ -458,12 +500,7 @@ def _validate_approval_keyring( raise Rc06ProtocolError("Approval key is duplicated, revoked, or unsupported.") if {"approver_id": key["approver_id"], "name": key["name"]} != SOLE_REVIEWER: raise Rc06ProtocolError("Approval key does not belong to the sole release reviewer.") - try: - public_key = serialization.load_pem_public_key(require_text( - key["public_key_pem"], "approval public key", maximum=16384 - ).encode("ascii")) - except (UnicodeError, ValueError) as exc: - raise Rc06ProtocolError("Approval public key is invalid.") from exc + public_key = load_approval_public_key(key["public_key_pem"]) if not isinstance(public_key, Ed25519PublicKey) or public_key_raw(public_key) == public_key_raw( trust.approval_root_public_key ): @@ -568,6 +605,7 @@ def _validate_approval( approved_at=approved_at, manifest_raw_sha256=request["batch_manifest_raw_sha256"], ) + keyring_expires_at = parse_time(keyring["expires_at"], "approval keyring expires_at") if ( payload["approval_keyring_epoch"] != keyring["epoch"] or payload["approval_keyring_payload_sha256"] != keyring["keyring_payload_sha256"] @@ -584,7 +622,11 @@ def _validate_approval( ): raise Rc06ProtocolError("Approval receipt material drift.") matches = [key for key in trusted_keys if key["key_id"] == payload["key_id"]] - if len(matches) != 1 or expires_at > matches[0]["not_after_parsed"]: + if ( + expires_at > keyring_expires_at + or len(matches) != 1 + or expires_at > matches[0]["not_after_parsed"] + ): raise Rc06ProtocolError("Approval signing key is not uniquely valid for the receipt window.") try: signature = base64.b64decode(require_text(receipt["signature"], "approval signature"), validate=True) @@ -614,19 +656,40 @@ def _validate_legal( request_sha = require_sha(legal_request["request_sha256"], "legal review request hash") if request_sha != sha256_json({key: child for key, child in legal_request.items() if key != "request_sha256"}): raise Rc06ProtocolError("Legal review request embedded hash drift.") - scope = legal_request["scope"] + parse_time(legal_request["created_at"], "legal review request created_at") + scope = require_exact(legal_request["scope"], "legal review scope", ( + "scene_count", "reference_occurrence_count", "distinct_source_record_count", + "distinct_url_count", + )) occurrences = legal_request["occurrences"] if ( legal_request["schema_version"] != LEGAL_REQUEST_SCHEMA or legal_request["status"] != "pending_qualified_human_review" or legal_request["candidate_commit"] != manifest["candidate"]["commit"] - or not isinstance(scope, dict) or scope.get("scene_count") != 30 or scope.get("reference_occurrence_count") != 43 + or scope.get("distinct_source_record_count") != 8 + or scope.get("distinct_url_count") != 7 or not isinstance(occurrences, list) or len(occurrences) != 43 ): raise Rc06ProtocolError("Legal review request does not cover exact 30 scenes and 43 occurrences.") + if ( + not isinstance(legal_request["source_files"], list) + or not legal_request["source_files"] + or any( + not isinstance(item, str) or not item.strip() or item != item.strip() + for item in legal_request["source_files"] + ) + ): + raise Rc06ProtocolError("Legal review source file inventory is empty or invalid.") + if legal_request["guardrails"] != { + "not_legal_approval": True, + "authorizes_publication": False, + "ai_must_not_supply_human_decisions": True, + "all_occurrences_require_qualified_human_review": True, + }: + raise Rc06ProtocolError("Legal review guardrails drift.") occurrence_by_id: dict[str, dict[str, Any]] = {} for occurrence in occurrences: if not isinstance(occurrence, dict): @@ -645,12 +708,24 @@ def _validate_legal( raise Rc06ProtocolError("Legal review decision binding drift.") reviewed_at = parse_time(decision.get("reviewed_at"), "legal reviewed_at") expires_at = parse_time(decision.get("expires_at"), "legal expires_at") - if not reviewed_at <= now <= expires_at or expires_at - reviewed_at > MAX_LEGAL_WINDOW: + approval_expires_at = parse_time( + request["approval_receipt"]["payload"]["expires_at"], + "approval expires_at", + ) + if ( + not reviewed_at <= now <= expires_at + or expires_at <= reviewed_at + or expires_at - reviewed_at > MAX_LEGAL_WINDOW + or expires_at > approval_expires_at + ): raise Rc06ProtocolError("Legal review decision is outside its validity window.") reviewer = decision.get("reviewer") - if not isinstance(reviewer, dict) or any(not isinstance(reviewer.get(field), str) or not reviewer[field].strip() for field in ( - "name", "organization", "qualification", "credential_reference" - )): + if not isinstance(reviewer, dict) or any( + not isinstance(reviewer.get(field), str) + or not reviewer[field].strip() + or reviewer[field].strip().casefold() in LEGAL_PLACEHOLDERS + for field in ("name", "organization", "qualification", "credential_reference") + ): raise Rc06ProtocolError("Legal reviewer identity is incomplete.") items = decision.get("items") if not isinstance(items, list) or len(items) != 43: @@ -670,14 +745,17 @@ def _validate_legal( ): raise Rc06ProtocolError("Legal review item is not affirmatively verified.") parse_time(item.get("checked_at"), "legal item checked_at") - require_text(item.get("authoritative_status_url"), "legal authoritative URL", maximum=4096) - if len(require_text(item.get("rationale"), "legal rationale", maximum=4096)) < 20: + require_official_url(item.get("authoritative_status_url"), "legal authoritative URL") + rationale = require_text(item.get("rationale"), "legal rationale", maximum=4096) + if len(rationale) < 20 or rationale.casefold() in LEGAL_PLACEHOLDERS: raise Rc06ProtocolError("Legal review rationale is insufficient.") seen.add(occurrence_id) signature = decision.get("signature") if not isinstance(signature, dict) or signature.get("algorithm") != "Ed25519": raise Rc06ProtocolError("Legal review signature is missing.") key_id = require_text(signature.get("key_id"), "legal signature key id", maximum=128) + if KEY_ID.fullmatch(key_id) is None: + raise Rc06ProtocolError("Legal review signature key id is invalid.") matches = [key for key in keyring["keys"] if isinstance(key, dict) and key.get("key_id") == key_id] if len(matches) != 1: raise Rc06ProtocolError("Legal review key is not uniquely trusted.") @@ -761,9 +839,18 @@ def _validate_challenge( raise Rc06ProtocolError("Apply challenge binding drift.") issued_at = parse_time(payload["issued_at"], "challenge issued_at") expires_at = parse_time(payload["expires_at"], "challenge expires_at") + approval_payload = request["approval_receipt"]["payload"] + approval_approved_at = parse_time( + approval_payload["approved_at"], "approval approved_at" + ) + approval_expires_at = parse_time( + approval_payload["expires_at"], "approval expires_at" + ) if ( expires_at <= issued_at or expires_at - issued_at > MAX_CHALLENGE_WINDOW + or issued_at < approval_approved_at + or expires_at > approval_expires_at or expires_at > parse_time(legal_expires_at, "legal expires_at") or not issued_at <= now < expires_at ): diff --git a/yuqei-contract-platform/services/contract-api/src/yuqei_contract_api/schema_check.py b/yuqei-contract-platform/services/contract-api/src/yuqei_contract_api/schema_check.py index bb3f8a0b..e238e0bb 100644 --- a/yuqei-contract-platform/services/contract-api/src/yuqei_contract_api/schema_check.py +++ b/yuqei-contract-platform/services/contract-api/src/yuqei_contract_api/schema_check.py @@ -221,27 +221,47 @@ RC06_COLUMNS_BY_TABLE = { "candidate_commit", "contract_release_image_digest", "ai_release_image_digest", + "batch_manifest_raw_sha256", + "batch_manifest_payload_sha256", + "approval_receipt_sha256", + "legal_source_review_binding_sha256", + "legal_source_review_expires_at", "challenge_id", "challenge_sha256", "state", "request_json", "prepare_request_json", + "saga_status_json", "participant_evidence_json", + "online_payload_json", + "online_primary_signature_json", + "online_attestation_json", "online_primary_kms_request_id", "online_attestation_kms_request_id", "online_receipt_json", + "online_receipt_sha256", + "runtime_payload_json", + "runtime_primary_signature_json", + "runtime_attestation_json", "runtime_primary_kms_request_id", "runtime_attestation_kms_request_id", "runtime_receipt_json", + "runtime_receipt_sha256", "rollback_response_json", + "rollback_evidence_sha256", "rollback_response_sha256", "apply_receipt_json", + "apply_receipt_sha256", + "error_message", "failure_code", "failure_at", "failure_evidence_sha256", + "created_at", + "updated_at", "contract_committed_at", "pending_countersign_at", "finalized_at", + "rolled_back_at", ), "rc06_apply_challenge_uses": ( "challenge_id", @@ -267,11 +287,15 @@ RC06_CONSTRAINTS_BY_TABLE = { ), "rc06_online_publication_transactions": ( "ux_rc06_online_publication_idempotency", + "ux_rc06_online_publication_challenge_id", + "ux_rc06_online_publication_challenge_sha256", "ck_rc06_online_publication_state", ), "rc06_apply_challenge_uses": ( "ux_rc06_apply_challenge_nonce", "ux_rc06_apply_challenge_transaction", + "ux_rc06_apply_challenge_sha256", + "fk_rc06_apply_challenge_transaction", "ck_rc06_apply_challenge_outcome", ), } @@ -749,6 +773,7 @@ def _check_rc06_schema(inspector, table_names: set[str], *, dialect: str) -> Run for item in ( inspector.get_unique_constraints(table_name, schema=schema) + inspector.get_check_constraints(table_name, schema=schema) + + inspector.get_foreign_keys(table_name, schema=schema) + inspector.get_indexes(table_name, schema=schema) ) if item.get("name") diff --git a/yuqei-contract-platform/services/contract-api/tests/test_rc06_migrations.py b/yuqei-contract-platform/services/contract-api/tests/test_rc06_migrations.py index 33a0e1a2..e0a15176 100644 --- a/yuqei-contract-platform/services/contract-api/tests/test_rc06_migrations.py +++ b/yuqei-contract-platform/services/contract-api/tests/test_rc06_migrations.py @@ -39,6 +39,34 @@ def _old_head(connection) -> None: ) +def _insert_hardened_saga(connection, *, transaction_id: str, compensation=None) -> None: + now = datetime.now(UTC) + connection.execute(sa.text( + "INSERT INTO rc06_publication_sagas (" + "transaction_id,tenant_id,idempotency_key,request_sha256,candidate_commit," + "contract_candidate_commit,contract_release_image_digest,ai_candidate_commit," + "ai_release_image_digest,candidate_tag,batch_manifest_raw_sha256," + "batch_manifest_payload_sha256,approval_receipt_sha256,state,request_json," + "ai_compensation_status_json,created_at,updated_at) VALUES (" + ":transaction_id,'tenant',:idempotency,:request_sha,:commit,:commit,:contract_image," + ":commit,:ai_image,'candidate',:raw,:payload,:approval,'prepared','{}'," + ":compensation,:created_at,:updated_at)" + ), { + "transaction_id": transaction_id, + "idempotency": f"idempotency-{transaction_id}", + "request_sha": "1" * 64, + "commit": "2" * 40, + "contract_image": "sha256:" + "3" * 64, + "ai_image": "sha256:" + "4" * 64, + "raw": "5" * 64, + "payload": "6" * 64, + "approval": "7" * 64, + "compensation": compensation, + "created_at": now, + "updated_at": now, + }) + + def test_empty_old_head_upgrades_and_round_trips_without_rewriting_0080(tmp_path): engine = sa.create_engine(f"sqlite+pysqlite:///{(tmp_path / 'contract-migration.sqlite3').as_posix()}") migration = _load("20261006_0081_harden_rc06_publication_saga.py") @@ -200,3 +228,62 @@ def test_rc06_facade_migration_round_trip_and_refuses_evidence_erasure(tmp_path) _run(connection, facade, "downgrade") assert "rc06_online_publication_transactions" in sa.inspect(connection).get_table_names() engine.dispose() + + +def test_rc06_facade_upgrade_rejects_preexisting_saga_without_authority_provenance(tmp_path): + engine = sa.create_engine(f"sqlite+pysqlite:///{(tmp_path / 'contract-facade-old-saga.sqlite3').as_posix()}") + hardening = _load("20261006_0081_harden_rc06_publication_saga.py") + facade = _load("20261006_0082_add_rc06_publication_facade.py") + with engine.begin() as connection: + _old_head(connection) + _run(connection, hardening, "upgrade") + # 0082 owns the compensation column, so insert a valid 0081 saga first. + now = datetime.now(UTC) + connection.execute(sa.text( + "INSERT INTO rc06_publication_sagas (" + "transaction_id,tenant_id,idempotency_key,request_sha256,candidate_commit," + "contract_candidate_commit,contract_release_image_digest,ai_candidate_commit," + "ai_release_image_digest,candidate_tag,batch_manifest_raw_sha256," + "batch_manifest_payload_sha256,approval_receipt_sha256,state,request_json," + "created_at,updated_at) VALUES (" + "'rc06-existing-saga','tenant','existing-idempotency',:request_sha,:commit," + ":commit,:contract_image,:commit,:ai_image,'candidate',:raw,:payload,:approval," + "'prepared','{}',:created_at,:updated_at)" + ), { + "request_sha": "1" * 64, + "commit": "2" * 40, + "contract_image": "sha256:" + "3" * 64, + "ai_image": "sha256:" + "4" * 64, + "raw": "5" * 64, + "payload": "6" * 64, + "approval": "7" * 64, + "created_at": now, + "updated_at": now, + }) + with pytest.raises(RuntimeError, match="authority provenance"): + _run(connection, facade, "upgrade") + assert "ai_compensation_status_json" not in { + item["name"] for item in sa.inspect(connection).get_columns("rc06_publication_sagas") + } + engine.dispose() + + +def test_rc06_facade_downgrade_refuses_ai_compensation_evidence(tmp_path): + engine = sa.create_engine(f"sqlite+pysqlite:///{(tmp_path / 'contract-facade-compensation.sqlite3').as_posix()}") + hardening = _load("20261006_0081_harden_rc06_publication_saga.py") + facade = _load("20261006_0082_add_rc06_publication_facade.py") + with engine.begin() as connection: + _old_head(connection) + _run(connection, hardening, "upgrade") + _run(connection, facade, "upgrade") + _insert_hardened_saga( + connection, + transaction_id="rc06-compensation-evidence", + compensation='{"schema_version":"rc06-ai-compensation-evidence/v1"}', + ) + with pytest.raises(RuntimeError, match="refuses to erase"): + _run(connection, facade, "downgrade") + assert "ai_compensation_status_json" in { + item["name"] for item in sa.inspect(connection).get_columns("rc06_publication_sagas") + } + engine.dispose() diff --git a/yuqei-contract-platform/services/contract-api/tests/test_rc06_publication_facade.py b/yuqei-contract-platform/services/contract-api/tests/test_rc06_publication_facade.py index 3167e3c4..499f3947 100644 --- a/yuqei-contract-platform/services/contract-api/tests/test_rc06_publication_facade.py +++ b/yuqei-contract-platform/services/contract-api/tests/test_rc06_publication_facade.py @@ -1,7 +1,10 @@ from __future__ import annotations import base64 +import time +from concurrent.futures import ThreadPoolExecutor from datetime import UTC, datetime, timedelta +from threading import Event import pytest from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey @@ -23,6 +26,7 @@ from yuqei_contract_api.rc06_publication_protocol import ( FINALIZE_REQUEST_SCHEMA, ROLLBACK_REQUEST_SCHEMA, AuthorizedMaterial, + Rc06ProtocolError, canonical_json, iso_time, publication_transaction_id, @@ -50,14 +54,30 @@ class FakeAuthority: return self.material +class ExpiringFakeAuthority(FakeAuthority): + def __init__(self, material: AuthorizedMaterial, *, expires_at: datetime): + super().__init__(material) + self.expires_at = expires_at + + def authorize(self, request, *, now): + assert request == self.material.request + self.checked_at.append(now) + if now >= self.expires_at: + raise Rc06ProtocolError("authority window expired") + return self.material + + class FakeSigner: - def __init__(self, *, fail_once_at: int | None = None): + def __init__(self, *, fail_once_at: int | None = None, delay_seconds: float = 0): self.calls: list[tuple[bytes, dict, bool]] = [] self.fail_once_at = fail_once_at self.failed = False + self.delay_seconds = delay_seconds def sign(self, *, domain, payload, include_request_id): self.calls.append((domain, dict(payload), include_request_id)) + if self.delay_seconds: + time.sleep(self.delay_seconds) if len(self.calls) == self.fail_once_at and not self.failed: self.failed = True raise Rc06FacadeError("injected KMS response loss") @@ -80,10 +100,15 @@ class FakeCoordinator: self.state: Rc06SagaStatus | None = None self.fail_commit = fail_commit self.compensate_calls = 0 + self.prepare_calls = 0 + self.commit_calls = 0 + self.recovery_calls = 0 def _status(self, state: str) -> Rc06SagaStatus: now = datetime.now(UTC) committed = state == "committed" + ai_committed = state in {"ai_committed", "contract_committed", "committed"} + contract_committed = state in {"contract_committed", "committed"} return Rc06SagaStatus( transaction_id=self.material.request["transaction_id"], state=state, @@ -112,24 +137,32 @@ class FakeCoordinator: runtime_snapshot_count=14 if committed else 0, no_go=not committed, cross_database_atomicity=False, - ai_committed_at=now if committed else None, - contract_committed_at=now if committed else None, + ai_committed_at=now if ai_committed else None, + contract_committed_at=now if contract_committed else None, committed_at=now if committed else None, ) def prepare(self, payload, *, tenant_id): + self.prepare_calls += 1 assert tenant_id == TENANT assert payload.contract_release_image_digest == CONTRACT_IMAGE self.state = self._status("prepared") return self.state def commit(self, transaction_id, *, tenant_id=None): + self.commit_calls += 1 assert tenant_id == TENANT if self.fail_commit: raise Rc06FacadeError("injected pre-Contract commit failure") self.state = self._status("committed") return self.state + def recover_after_contract_commit(self, transaction_id, *, tenant_id=None): + assert tenant_id == TENANT + self.recovery_calls += 1 + self.state = self._status("committed") + return self.state + def status(self, transaction_id, *, tenant_id=None): if self.state is None: raise KeyError(transaction_id) @@ -137,8 +170,37 @@ class FakeCoordinator: def compensate(self, transaction_id, *, tenant_id=None): self.compensate_calls += 1 + previous_state = self.state.state if self.state is not None else "prepared" self.state = self._status("compensated") - self.state.compensated_from_state = "prepared" + self.state.compensated_from_state = ( + "committed" if previous_state == "ai_committed" else "prepared" + ) + if previous_state == "ai_committed": + self.state.compensation_item_results = [ + { + "kind": "prompt_module", + "item_id": item["item_id"], + "pre_apply_content_sha256": None, + "post_compensation_content_sha256": None, + "action": "removed_new", + "status": "verified", + } + for item in self.material.expected_publication_results + if item["kind"] == "prompt_module" + ] + self.state.runtime_snapshot_compensation_results = [ + { + "kind": item["kind"], + "runtime_key": item["runtime_key"], + "version": item["version"], + "pre_apply_content_sha256": None, + "post_compensation_content_sha256": None, + "action": "removed_new", + "status": "verified", + } + for item in self.material.expected_runtime_results + ] + self.state.removed_item_count = 28 self.state.compensated_at = datetime.now(UTC) self.state.compensation_evidence_sha256 = "d" * 64 return self.state @@ -240,7 +302,14 @@ def _material(now: datetime | None = None): ) -def _service(tmp_path, *, fail_commit=False, fail_sign_once_at=None): +def _service( + tmp_path, + *, + fail_commit=False, + fail_sign_once_at=None, + sign_delay_seconds: float = 0, + authority_expires_at: datetime | None = None, +): repository = SqlAlchemyContractRepository( f"sqlite:///{tmp_path / 'contract.sqlite3'}", auto_create=True, @@ -248,7 +317,10 @@ def _service(tmp_path, *, fail_commit=False, fail_sign_once_at=None): file_storage_dir=tmp_path / "files", ) material = _material() - signer = FakeSigner(fail_once_at=fail_sign_once_at) + signer = FakeSigner( + fail_once_at=fail_sign_once_at, + delay_seconds=sign_delay_seconds, + ) coordinator = FakeCoordinator(material, fail_commit=fail_commit) apply_private = Ed25519PrivateKey.generate() token = tmp_path / "rc06-token" @@ -263,10 +335,15 @@ def _service(tmp_path, *, fail_commit=False, fail_sign_once_at=None): rc06_release_image_digest=CONTRACT_IMAGE, rc06_ai_release_image_digest=AI_IMAGE, ) + authority = ( + ExpiringFakeAuthority(material, expires_at=authority_expires_at) + if authority_expires_at is not None + else FakeAuthority(material) + ) service = Rc06PublicationFacadeService( settings=settings, repository=repository, - authority=FakeAuthority(material), + authority=authority, signer=signer, coordinator=coordinator, apply_service_public_key=apply_private.public_key(), @@ -492,3 +569,127 @@ def test_post_contract_kms_failure_is_no_go_recoverable_and_never_compensated(tm assert coordinator.compensate_calls == 0 finally: repository.engine.dispose() + + +def test_two_facade_instances_singleflight_the_four_kms_stages(tmp_path): + repository, first, signer, coordinator, apply_private, material, headers = _service( + tmp_path, + sign_delay_seconds=0.02, + ) + second = Rc06PublicationFacadeService( + settings=first._settings, # noqa: SLF001 - independent worker regression + repository=repository, + authority=first._authority, # noqa: SLF001 + signer=signer, + coordinator=coordinator, + apply_service_public_key=apply_private.public_key(), + ) + start = Event() + + def invoke(service): + start.wait(timeout=5) + return service.apply(material.request, headers) + + try: + with ThreadPoolExecutor(max_workers=2) as executor: + futures = [executor.submit(invoke, service) for service in (first, second)] + start.set() + responses = [future.result(timeout=20) for future in futures] + assert responses[0] == responses[1] + assert len(signer.calls) == 4 + assert coordinator.prepare_calls == 1 + assert coordinator.commit_calls == 1 + finally: + repository.engine.dispose() + + +def test_expired_authority_cannot_prepare_or_commit_a_reserved_transaction(tmp_path): + initial = datetime.now(UTC).replace(microsecond=0) + expires_at = initial + timedelta(minutes=5) + repository, service, _, coordinator, _, material, headers = _service( + tmp_path, + authority_expires_at=expires_at, + ) + try: + service._authorize_and_reserve( # noqa: SLF001 - recovery boundary regression + material.request, + sha256_json(material.request), + initial, + ) + with pytest.raises(Rc06FacadeConflict, match="authority window expired"): + service.apply(material.request, headers, now=expires_at + timedelta(seconds=1)) + assert coordinator.prepare_calls == 0 + assert coordinator.commit_calls == 0 + assert service.status(material.request["transaction_id"], headers)["state"] == "prepared" + finally: + repository.engine.dispose() + + +def test_expired_authority_leaves_ai_only_commit_for_explicit_rollback(tmp_path): + initial = datetime.now(UTC).replace(microsecond=0) + expires_at = initial + timedelta(minutes=5) + repository, service, _, coordinator, _, material, headers = _service( + tmp_path, + authority_expires_at=expires_at, + ) + try: + service._authorize_and_reserve( # noqa: SLF001 - recovery boundary regression + material.request, + sha256_json(material.request), + initial, + ) + coordinator.state = coordinator._status("ai_committed") # noqa: SLF001 + with pytest.raises(Rc06FacadeConflict, match="explicit rollback"): + service.apply(material.request, headers, now=expires_at + timedelta(seconds=1)) + assert coordinator.commit_calls == 0 + assert coordinator.compensate_calls == 0 + + rollback_request = { + "schema_version": ROLLBACK_REQUEST_SCHEMA, + "idempotency_key": material.request["idempotency_key"], + "transaction_id": material.request["transaction_id"], + "batch_manifest_raw_sha256": material.request["batch_manifest_raw_sha256"], + "approval_receipt_sha256": material.request["approval_receipt_sha256"], + "legal_source_review_binding_sha256": material.legal_binding_sha256, + "environment_id": ENVIRONMENT, + "tenant_id": TENANT, + "transaction_mode": "durable_saga", + "cross_database_atomicity": False, + } + rollback = service.rollback( + material.request["transaction_id"], rollback_request, headers + ) + assert rollback["state"] == "rolled_back" + assert len(rollback["compensation_item_results"]) == 14 + assert len(rollback["runtime_snapshot_compensation_results"]) == 14 + assert coordinator.compensate_calls == 1 + finally: + repository.engine.dispose() + + +def test_expired_authority_can_only_finish_proven_contract_commit(tmp_path): + initial = datetime.now(UTC).replace(microsecond=0) + expires_at = initial + timedelta(minutes=5) + repository, service, signer, coordinator, _, material, headers = _service( + tmp_path, + authority_expires_at=expires_at, + ) + try: + service._authorize_and_reserve( # noqa: SLF001 - recovery boundary regression + material.request, + sha256_json(material.request), + initial, + ) + coordinator.state = coordinator._status("contract_committed") # noqa: SLF001 + response = service.apply( + material.request, + headers, + now=expires_at + timedelta(seconds=1), + ) + assert response["publication_item_count"] == 74 + assert coordinator.prepare_calls == 0 + assert coordinator.commit_calls == 0 + assert coordinator.recovery_calls == 1 + assert len(signer.calls) == 4 + finally: + repository.engine.dispose()