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 b7f72a06..d11637b2 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 @@ -11,6 +11,11 @@ from pydantic import BaseModel, Field, SecretStr from pydantic_settings import BaseSettings, SettingsConfigDict from yuqei_contract_api.anti_abuse import anti_abuse_config_status, contract_endpoint_risk_policies +from yuqei_contract_api.rc06_publication_protocol import ( + Rc06ProtocolError, + load_pinned_ed25519, + public_key_raw, +) class ContractApiSettings(BaseSettings): @@ -796,6 +801,7 @@ def _check_rc06_publication_facade_config(settings: ContractApiSettings) -> Runt if bearer in reused: errors.append("facade bearer token must be independent from other service credentials") kms_credential_path = resolved_files.get("KMS credential") + kms_credential: str | None = None if bearer is not None and kms_credential_path is not None and kms_credential_path.is_file(): try: kms_credential = kms_credential_path.read_text(encoding="utf-8").strip() @@ -804,16 +810,41 @@ def _check_rc06_publication_facade_config(settings: ContractApiSettings) -> Runt else: if bearer == kms_credential: errors.append("facade bearer token must be independent from the KMS credential") - public_key_paths = [ + elif kms_credential_path is not None and kms_credential_path.is_file(): + try: + kms_credential = kms_credential_path.read_text(encoding="utf-8").strip() + except (OSError, UnicodeError): + errors.append("KMS credential cannot be read as UTF-8") + if kms_credential and kms_credential in { + 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(), + }: + errors.append("KMS credential must be independent from all service boundary 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") + approval_root_key = load_pinned_ed25519( + str(public_key_paths[0]), + settings.rc06_approval_root_public_key_sha256, + "approval root public key", + ) + apply_service_key = load_pinned_ed25519( + str(public_key_paths[1]), + settings.rc06_apply_service_public_key_sha256, + "apply service public key", + ) + except Rc06ProtocolError as error: + errors.append(str(error)) + else: + if public_key_raw(approval_root_key) == public_key_raw(apply_service_key): + errors.append( + "RC06 authority Ed25519 keys must be distinct across cryptographic purposes" + ) if settings.rc06_online_signing_algorithm not in { "RSA_PSS_SHA_256", "RSA_PKCS1_SHA_256", "ECDSA_SHA_256" }: 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 b166a0d0..8cf479ba 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 @@ -876,6 +876,14 @@ def _validate_approval_keyring( validated: list[dict[str, Any]] = [] seen: set[str] = set() seen_public_keys: set[bytes] = set() + reserved_public_keys = { + public_key_raw(trust.approval_root_public_key), + public_key_raw(trust.apply_service_public_key), + } + if len(reserved_public_keys) != 2: + raise Rc06ProtocolError( + "RC06 authority Ed25519 keys must be distinct across cryptographic purposes." + ) for value in keys: key = require_exact(value, "approval key", ( "key_id", "algorithm", "public_key_pem", "approver_id", "name", @@ -890,10 +898,12 @@ def _validate_approval_keyring( public_bytes = public_key_raw(public_key) if ( not isinstance(public_key, Ed25519PublicKey) - or public_bytes == public_key_raw(trust.approval_root_public_key) + or public_bytes in reserved_public_keys or public_bytes in seen_public_keys ): - raise Rc06ProtocolError("Approval key is invalid or reuses the offline root.") + raise Rc06ProtocolError( + "Approval key is invalid, duplicated, or reuses a key assigned to another purpose." + ) not_before = parse_time(key["not_before"], "approval key not_before") not_after = parse_time(key["not_after"], "approval key not_after") if not_before >= not_after: @@ -1057,6 +1067,38 @@ def _validate_legal( raw_pin = require_sha(request["legal_source_review_keyring_raw_sha256"], "legal keyring hash") if raw_pin != trust.legal_keyring_sha256 or sha256_json(keyring) != raw_pin: raise Rc06ProtocolError("Legal review keyring does not match the server pin.") + legal_keys = keyring["keys"] + if not isinstance(legal_keys, list) or not legal_keys: + raise Rc06ProtocolError("Legal review keyring contains no keys.") + approval_public_keys = { + public_key_raw(load_approval_public_key(key["public_key_pem"])) + for key in request["approval_keyring"]["keys"] + } + reserved_public_keys = { + public_key_raw(trust.approval_root_public_key), + public_key_raw(trust.apply_service_public_key), + *approval_public_keys, + } + for legal_key in legal_keys: + if not isinstance(legal_key, dict): + raise Rc06ProtocolError("Legal review key is invalid.") + if legal_key.get("status") != "trusted" or legal_key.get("revoked_at") is not None: + continue + try: + legal_public_raw = base64.b64decode( + require_text( + legal_key.get("public_key_raw_base64"), + "legal public key", + ), + validate=True, + ) + Ed25519PublicKey.from_public_bytes(legal_public_raw) + except ValueError as exc: + raise Rc06ProtocolError("Legal review public key is invalid.") from exc + if legal_public_raw in reserved_public_keys: + raise Rc06ProtocolError( + "RC06 authority Ed25519 keys must be distinct across cryptographic purposes." + ) legal_request = require_exact(request["legal_source_review_request"], "legal review request", ( "schema_version", "status", "candidate_commit", "created_at", "scope", "source_files", "occurrences", "guardrails", "request_sha256", diff --git a/yuqei-contract-platform/services/contract-api/tests/test_rc06_publication_deployment.py b/yuqei-contract-platform/services/contract-api/tests/test_rc06_publication_deployment.py index 4b00becc..d76fb90f 100644 --- a/yuqei-contract-platform/services/contract-api/tests/test_rc06_publication_deployment.py +++ b/yuqei-contract-platform/services/contract-api/tests/test_rc06_publication_deployment.py @@ -1,9 +1,12 @@ from __future__ import annotations +import hashlib from pathlib import Path import pytest import yaml +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from yuqei_contract_api.config import ( ContractApiSettings, @@ -27,7 +30,27 @@ def _write(path: Path, value: str) -> str: return str(path.resolve()) +def _write_bytes(path: Path, value: bytes) -> str: + path.write_bytes(value) + return str(path.resolve()) + + +def _public_material(tmp_path: Path, name: str) -> tuple[str, str]: + public_key = Ed25519PrivateKey.generate().public_key() + raw = public_key.public_bytes( + serialization.Encoding.Raw, + serialization.PublicFormat.Raw, + ) + pem = public_key.public_bytes( + serialization.Encoding.PEM, + serialization.PublicFormat.SubjectPublicKeyInfo, + ) + return _write_bytes(tmp_path / f"{name}.pem", pem), hashlib.sha256(raw).hexdigest() + + def _facade_settings(tmp_path: Path, **overrides) -> ContractApiSettings: + approval_root_path, approval_root_pin = _public_material(tmp_path, "approval-root") + apply_service_path, apply_service_pin = _public_material(tmp_path, "apply-service") values = { "environment": "production", "database_url": f"sqlite:///{tmp_path / 'must-not-be-created.sqlite3'}", @@ -50,16 +73,12 @@ def _facade_settings(tmp_path: Path, **overrides) -> ContractApiSettings: "rc06_frozen_candidate_commit": "a" * 40, "rc06_release_image_digest": "sha256:" + "b" * 64, "rc06_ai_release_image_digest": "sha256:" + "c" * 64, - "rc06_approval_root_public_key_file": _write( - tmp_path / "approval-root.pem", "approval-root-public-material" - ), - "rc06_approval_root_public_key_sha256": "d" * 64, + "rc06_approval_root_public_key_file": approval_root_path, + "rc06_approval_root_public_key_sha256": approval_root_pin, "rc06_approval_root_key_id": "yuqei-rc06-approval-root-v1", "rc06_approval_keyring_epoch": 1, - "rc06_apply_service_public_key_file": _write( - tmp_path / "apply-service.pem", "apply-service-public-material" - ), - "rc06_apply_service_public_key_sha256": "e" * 64, + "rc06_apply_service_public_key_file": apply_service_path, + "rc06_apply_service_public_key_sha256": apply_service_pin, "rc06_legal_review_keyring_sha256": "f" * 64, "rc06_online_signing_provider": "fixture-kms", "rc06_online_signing_key_reference": "kms://fixture/rc06-online", @@ -122,6 +141,103 @@ def test_enabled_facade_startup_checks_all_independent_trust_boundaries(tmp_path ) +@pytest.mark.parametrize("alternate_encoding", ["crlf-pem", "raw"]) +def test_enabled_facade_rejects_same_ed25519_key_across_root_and_apply_roles( + tmp_path: Path, + alternate_encoding: str, +) -> None: + public_key = Ed25519PrivateKey.generate().public_key() + raw = public_key.public_bytes( + serialization.Encoding.Raw, + serialization.PublicFormat.Raw, + ) + pem = public_key.public_bytes( + serialization.Encoding.PEM, + serialization.PublicFormat.SubjectPublicKeyInfo, + ) + alternate = pem.replace(b"\n", b"\r\n") if alternate_encoding == "crlf-pem" else raw + pin = hashlib.sha256(raw).hexdigest() + + with pytest.raises(RuntimeError, match="distinct across cryptographic purposes"): + validate_rc06_publication_startup( + _facade_settings( + tmp_path, + rc06_approval_root_public_key_file=_write_bytes( + tmp_path / "same-root.pem", pem + ), + rc06_approval_root_public_key_sha256=pin, + rc06_apply_service_public_key_file=_write_bytes( + tmp_path / f"same-apply-{alternate_encoding}.key", alternate + ), + rc06_apply_service_public_key_sha256=pin, + ) + ) + + +@pytest.mark.parametrize("unsupported_encoding", ["der", "openssh"]) +def test_enabled_facade_rejects_unsupported_ed25519_key_encodings( + tmp_path: Path, + unsupported_encoding: str, +) -> None: + public_key = Ed25519PrivateKey.generate().public_key() + raw = public_key.public_bytes( + serialization.Encoding.Raw, + serialization.PublicFormat.Raw, + ) + if unsupported_encoding == "der": + encoded = public_key.public_bytes( + serialization.Encoding.DER, + serialization.PublicFormat.SubjectPublicKeyInfo, + ) + else: + encoded = public_key.public_bytes( + serialization.Encoding.OpenSSH, + serialization.PublicFormat.OpenSSH, + ) + + with pytest.raises(RuntimeError, match="not an Ed25519 public key"): + validate_rc06_publication_startup( + _facade_settings( + tmp_path, + rc06_apply_service_public_key_file=_write_bytes( + tmp_path / f"unsupported-apply-service.{unsupported_encoding}", + encoded, + ), + rc06_apply_service_public_key_sha256=hashlib.sha256(raw).hexdigest(), + ) + ) + + +@pytest.mark.parametrize( + "setting_name", + ( + "web_proxy_shared_secret", + "rc06_publication_shared_secret", + "ai_platform_service_key", + "ai_platform_rc06_publication_key", + ), +) +def test_enabled_facade_rejects_kms_token_reused_as_service_credential( + tmp_path: Path, + setting_name: str, +) -> None: + reused = "reused-service-boundary-secret-00000000000000000000001" + + with pytest.raises(RuntimeError, match="KMS credential must be independent"): + validate_rc06_publication_startup( + _facade_settings( + tmp_path, + **{ + setting_name: reused, + "reliable_evidence_signing_credential_file": _write( + tmp_path / f"reused-kms-{setting_name}.credential", + reused, + ), + }, + ) + ) + + def test_production_compose_and_edge_expose_only_the_hardened_facade() -> None: compose_path = REPO_ROOT / "yuqei-infra/compose/docker-compose.production.yml" model = yaml.load(compose_path.read_text(encoding="utf-8"), Loader=_ComposeLoader) diff --git a/yuqei-contract-platform/services/contract-api/tests/test_rc06_publication_facade_client_integration.py b/yuqei-contract-platform/services/contract-api/tests/test_rc06_publication_facade_client_integration.py index d4b7727c..e9c33d26 100644 --- a/yuqei-contract-platform/services/contract-api/tests/test_rc06_publication_facade_client_integration.py +++ b/yuqei-contract-platform/services/contract-api/tests/test_rc06_publication_facade_client_integration.py @@ -25,6 +25,10 @@ from yuqei_ai_platform_api.rc06_publication import ( ) from yuqei_ai_platform_api.repository import SqlAlchemyAiPlatformRepository from yuqei_contract_api.config import ContractApiSettings +from yuqei_contract_api.persistence.models import ( + Rc06ApplyChallengeUseModel, + Rc06OnlinePublicationTransactionModel, +) from yuqei_contract_api.rc06_publication_facade import ( ConfiguredRc06Authority, Rc06FacadeHeaders, @@ -36,7 +40,10 @@ from yuqei_contract_api.rc06_publication_protocol import ( CHALLENGE_SIGNATURE_DOMAIN, LEGAL_SIGNATURE_DOMAIN, AuthorityTrust, + Rc06ProtocolError, canonical_json, + expected_publication_results, + expected_runtime_results, public_key_raw, publication_idempotency_key, publication_transaction_id, @@ -61,11 +68,17 @@ def _load_apply_fixture(): return module -def _resign_authority_material(material: dict) -> dict: +def _resign_authority_material( + material: dict, + *, + reviewer_private: Ed25519PrivateKey | None = None, + legal_private: Ed25519PrivateKey | None = None, + additional_legal_private: Ed25519PrivateKey | None = None, +) -> dict: """Replace the fixture's stub legal signature with fully verified material.""" result = deepcopy(material) - reviewer_private = Ed25519PrivateKey.generate() + reviewer_private = reviewer_private or Ed25519PrivateKey.generate() keyring = result["approval_keyring"] keyring["keys"][0]["public_key_pem"] = reviewer_private.public_key().public_bytes( serialization.Encoding.PEM, @@ -106,7 +119,7 @@ def _resign_authority_material(material: dict) -> dict: ) result["approval_raw"] = sha256_json(approval) - legal_private = Ed25519PrivateKey.generate() + legal_private = legal_private or Ed25519PrivateKey.generate() legal_request = result["legal_request"] source_path = "fixture-only/not-production.json" occurrences = [] @@ -145,6 +158,13 @@ def _resign_authority_material(material: dict) -> dict: legal_keyring["keys"][0]["public_key_raw_base64"] = base64.b64encode( public_key_raw(legal_private.public_key()) ).decode("ascii") + if additional_legal_private is not None: + additional_legal_key = deepcopy(legal_keyring["keys"][0]) + additional_legal_key["key_id"] = "fixture-additional-legal-reviewer" + additional_legal_key["public_key_raw_base64"] = base64.b64encode( + public_key_raw(additional_legal_private.public_key()) + ).decode("ascii") + legal_keyring["keys"].append(additional_legal_key) result["legal_keyring_raw"] = sha256_json(legal_keyring) legal_decision = result["legal_decision"] legal_decision["request_sha256"] = legal_request["request_sha256"] @@ -320,6 +340,173 @@ def _authority(material: dict) -> ConfiguredRc06Authority: return authority +def _online_apply_request(material: dict) -> dict: + return { + "schema_version": "yuqei.rc06-online-apply-request.v3", + "idempotency_key": material["idempotency_key"], + "transaction_id": material["transaction_id"], + "environment_id": material["environment_id"], + "tenant_id": material["tenant_id"], + "transaction_mode": "durable_saga", + "cross_database_atomicity": False, + "batch_manifest_raw_sha256": material["manifest_raw"], + "batch_manifest_payload_sha256": material["manifest"]["manifest_payload_sha256"], + "approval_receipt_sha256": material["approval_raw"], + "approval_keyring": material["approval_keyring"], + "approval_keyring_raw_sha256": material["approval_keyring_raw"], + "approval_keyring_canonical_sha256": sha256_json(material["approval_keyring"]), + "legal_source_review_binding_sha256": material["legal_binding"], + "legal_source_review_expires_at": material["legal_expires_at"], + "authority_challenge": material["authority_challenge"], + "manifest": material["manifest"], + "approval_receipt": material["approval"], + "legal_source_review_request": material["legal_request"], + "legal_source_review_decision": material["legal_decision"], + "legal_source_review_keyring": material["legal_keyring"], + "legal_source_review_keyring_raw_sha256": material["legal_keyring_raw"], + "expected_publication_item_results": expected_publication_results( + material["manifest"] + ), + "expected_publication_item_count": 74, + "expected_runtime_snapshot_dependencies": expected_runtime_results( + material["manifest"] + ), + "expected_runtime_snapshot_dependency_count": 14, + } + + +def test_authority_accepts_distinct_ed25519_keys_for_each_purpose( + monkeypatch: pytest.MonkeyPatch, +) -> None: + fixture = _load_apply_fixture() + material = _resign_authority_material( + fixture._material(monkeypatch, datetime.now(UTC).replace(microsecond=0)) + ) + + authorized = _authority(material).authorize( + _online_apply_request(material), now=material["now"] + ) + + assert authorized.legal_binding_sha256 == material["legal_binding"] + + +@pytest.mark.parametrize( + "collision", + ( + "root-apply", + "approval-root", + "approval-apply", + "approval-legal", + "legal-root", + "legal-apply", + "additional-legal-apply", + ), +) +def test_authority_rejects_ed25519_key_reuse_across_purposes( + monkeypatch: pytest.MonkeyPatch, + collision: str, +) -> None: + fixture = _load_apply_fixture() + base = fixture._material(monkeypatch, datetime.now(UTC).replace(microsecond=0)) + reviewer_private = None + legal_private = None + additional_legal_private = None + if collision == "root-apply": + base["apply_private"] = base["root_private"] + elif collision == "approval-root": + reviewer_private = base["root_private"] + elif collision == "approval-apply": + reviewer_private = base["apply_private"] + elif collision == "approval-legal": + reviewer_private = legal_private = Ed25519PrivateKey.generate() + elif collision == "legal-root": + legal_private = base["root_private"] + elif collision == "legal-apply": + legal_private = base["apply_private"] + elif collision == "additional-legal-apply": + additional_legal_private = base["apply_private"] + material = _resign_authority_material( + base, + reviewer_private=reviewer_private, + legal_private=legal_private, + additional_legal_private=additional_legal_private, + ) + + with pytest.raises( + Rc06ProtocolError, + match="another purpose|distinct across cryptographic purposes", + ): + _authority(material).authorize( + _online_apply_request(material), now=material["now"] + ) + + +def test_facade_apply_rejects_role_reuse_before_reservation_or_participant_writes( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + fixture = _load_apply_fixture() + base = fixture._material(monkeypatch, datetime.now(UTC).replace(microsecond=0)) + material = _resign_authority_material( + base, + reviewer_private=base["apply_private"], + ) + request = _online_apply_request(material) + repository = SqlAlchemyContractRepository( + f"sqlite:///{tmp_path / 'contract-role-isolation.sqlite3'}", + auto_create=True, + seed_demo_data=False, + file_storage_dir=tmp_path / "files", + ) + + class NeverCalled: + def __init__(self) -> None: + self.calls: list[str] = [] + + def __getattr__(self, name: str): + def fail(*_args, **_kwargs): + self.calls.append(name) + raise AssertionError(f"{name} must not run before authority validation") + + return fail + + signer = NeverCalled() + coordinator = NeverCalled() + service = Rc06PublicationFacadeService( + settings=ContractApiSettings( + environment="test", + rc06_publication_facade_enabled=True, + rc06_publication_environment_id=material["environment_id"], + rc06_publication_tenant_id=material["tenant_id"], + rc06_frozen_candidate_commit=material["manifest"]["candidate"]["commit"], + ), + repository=repository, + authority=_authority(material), + signer=signer, + coordinator=coordinator, + apply_service_public_key=material["apply_private"].public_key(), + ) + headers = Rc06FacadeHeaders( + idempotency_key=request["idempotency_key"], + manifest_sha256=request["batch_manifest_raw_sha256"], + approval_sha256=request["approval_receipt_sha256"], + legal_sha256=request["legal_source_review_binding_sha256"], + tenant_id=request["tenant_id"], + request_sha256=sha256_json(request), + ) + try: + with pytest.raises(Rc06ProtocolError, match="another purpose"): + service.apply(request, headers, now=material["now"]) + + assert signer.calls == [] + assert coordinator.calls == [] + with repository._session_factory() as session: + assert session.query(Rc06OnlinePublicationTransactionModel).count() == 0 + assert session.query(Rc06ApplyChallengeUseModel).count() == 0 + finally: + repository.engine.dispose() + + def test_real_client_validates_facade_status_and_recovers_lost_responses( tmp_path: Path, monkeypatch: pytest.MonkeyPatch,