fix(rc06): isolate authority key purposes

2026-10-07 02:16:15 +08:00
parent 379c6aa116
commit eefafc23e5
4 changed files with 395 additions and 19 deletions
@@ -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"
    }:
@@ -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",
@@ -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)
@@ -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,