"""Task 3 参照资产可信消费门的真实文件边界测试。""" from __future__ import annotations import hashlib import importlib import json import os import shutil import socket import stat import sys import tempfile import weakref from pathlib import Path import pytest CHEAP_WORKER = Path(__file__).resolve().parents[1] REPO_ROOT = CHEAP_WORKER.parent sys.path.insert(0, str(CHEAP_WORKER)) import artifact_snapshot # noqa: E402 RELEASE_REF = "contracts/play-loop/reference-asset-release.initial.json" REGISTRY_REF = "contracts/play-loop/reference-asset-registry.v2.initial.json" POLICY_REF = "contracts/play-loop/reference-asset-consumption-policy.initial.json" MANIFEST_REF = "game-runtime/games/shanhai-xingji/reference/consumption-manifest.json" BUNDLE_REF = "game-runtime/games/shanhai-xingji/dist/shanhai-bundle.js" POLICY_ID = "survivor-gold-v1" EXPECTED_RELEASE_HASH = "5856d607dd973c64dbafc448ccff87cc825c917ede48b6072a52421569e6ccd0" EXPECTED_REGISTRY_HASH = "57c82adcef617fdb929d056733a395f7c4133636eb16bd38926637857b331b78" EXPECTED_POLICY_HASH = "695b6143fec3c7b30d7be5f5ce088ccb1c1add97899dae9f167bd3a9524d8bbf" EXPECTED_MANIFEST_HASH = "0f600598cfb842b213f7d54ae1be1d07fe277a3e66b3e4ab3ad21b73dea320d9" EXPECTED_BUNDLE_HASH = "1c760811ec435fe0f3b5ba79aa8c4fcc44119e019b3e1240ce2c51e55e25870b" def _gate(): """把缺少生产模块转成可记录的 RED，而不是 pytest collection error。""" try: return importlib.import_module("reference_asset_gate") except ModuleNotFoundError as exc: pytest.fail(f"reference_asset_gate 尚未实现: {exc}") def _copy_into(root: Path, relative: str) -> None: """把仓内冻结输入复制到临时可信仓根，保留相对路径和原始字节。""" target = root / relative target.parent.mkdir(parents=True, exist_ok=True) shutil.copyfile(REPO_ROOT / relative, target) def _fixture_root(tmp_path: Path) -> Path: """建立真实 U2 release、registry、policy、bundle、manifest 和 entries 快照。""" root = tmp_path / "trusted-root" root.mkdir(parents=True) for relative in (RELEASE_REF, REGISTRY_REF, POLICY_REF, MANIFEST_REF, BUNDLE_REF): _copy_into(root, relative) manifest = json.loads((REPO_ROOT / MANIFEST_REF).read_bytes()) for entry in manifest["entries"]: _copy_into(root, entry["path"]) return root def _verify(root: Path, *, declarations=None, mode="frozen_preflight", expected_release_hash=EXPECTED_RELEASE_HASH): """按生产入口调用可信消费门。""" return _gate().verify_policy( POLICY_ID, release_ref=RELEASE_REF, expected_release_hash=expected_release_hash, trusted_root=root, mode=mode, declarations=declarations, ) def _error_code(callable_): """读取专用异常稳定 code，同时禁止测试依赖异常正文。""" with pytest.raises(Exception) as caught: callable_() assert hasattr(caught.value, "code") return caught.value.code, str(caught.value) def _canonical(value: object) -> bytes: """按消费 manifest 的 canonical JSON 字节口径序列化。""" return json.dumps( value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False, ).encode("utf-8") def _rebind_release(root: Path, *, registry_hash=None, policy_hash=None) -> str: """让测试显式信任变异后的上游文件，以便继续覆盖 schema/语义门。""" path = root / RELEASE_REF release = json.loads(path.read_bytes()) if registry_hash is not None: release["registryHash"] = registry_hash if policy_hash is not None: release["policyHash"] = policy_hash raw = _canonical(release) path.write_bytes(raw) return hashlib.sha256(raw).hexdigest() def _rebind_manifest_chain(root: Path) -> str: """同步临时 manifest、active record 和 release 的原始字节 hash。""" manifest_hash = hashlib.sha256((root / MANIFEST_REF).read_bytes()).hexdigest() registry_path = root / REGISTRY_REF registry = json.loads(registry_path.read_bytes()) record = next(row for row in registry["records"] if row["recordId"] == POLICY_ID.replace("survivor-gold-v1", "gac-shanhai-xingji")) record["consumptionManifestHash"] = manifest_hash registry_raw = _canonical(registry) registry_path.write_bytes(registry_raw) return _rebind_release(root, registry_hash=hashlib.sha256(registry_raw).hexdigest()) def _write_sized_file(path: Path, size: int, fill: bytes = b"\x00") -> str: """写入测试文件并返回独立计算的 SHA-256；大文件用稀疏零文件控制磁盘占用。""" path.parent.mkdir(parents=True, exist_ok=True) if fill == b"\x00": with path.open("wb") as handle: handle.truncate(size) digest = hashlib.sha256() block = b"\x00" * min(1024 * 1024, size or 1) remaining = size while remaining: chunk = block[:min(len(block), remaining)] digest.update(chunk) remaining -= len(chunk) return digest.hexdigest() content = (fill * ((size + len(fill) - 1) // len(fill)))[:size] path.write_bytes(content) return hashlib.sha256(content).hexdigest() def _build_batch_policy( root: Path, name: str, entry_specs: list[tuple[str, int, bytes]], *, record_id: str | None = None, ) -> dict: """构造一套 release/registry/policy，供多记录原子验证真实走完整 fd 链。""" record_id = record_id or f"record-{name}" policy_id = f"policy-{name}" consumer_ref = "batch-test@reference-assets/2" asset_root = f"assets/{name}" artifact_ref = f"{asset_root}/bundle.js" manifest_ref = f"{asset_root}/manifest.json" registry_ref = f"contracts/{name}-registry.json" policy_ref = f"contracts/{name}-policy.json" release_ref = f"contracts/{name}-release.json" bundle = f"bundle-{name}".encode("utf-8") (root / artifact_ref).parent.mkdir(parents=True, exist_ok=True) (root / artifact_ref).write_bytes(bundle) entries = [] for relative, size, fill in entry_specs: sha256 = _write_sized_file(root / relative, size, fill) entries.append({"path": relative, "size": size, "sha256": sha256}) entries.sort(key=lambda entry: entry["path"].encode("utf-8")) manifest = { "schemaVersion": "ReferenceAssetConsumptionManifest/1", "manifestId": f"manifest-{name}", "canonicalization": "reference-asset-consumption-manifest/1", "entries": entries, } manifest_raw = _canonical(manifest) (root / manifest_ref).write_bytes(manifest_raw) record = { "schemaVersion": "ReferenceAssetRecord/2", "recordId": record_id, "role": "generation_exemplar", "lifecycleStatus": "active", "assetRef": asset_root, "assetVersion": "1.0.0", "artifactHash": hashlib.sha256(bundle).hexdigest(), "consumerRef": consumer_ref, "designRef": [ entry["path"] for entry in entries if not entry["path"].startswith(asset_root + "/") ] or None, "evidenceRefs": [], "signedBy": "test", "signedAt": "2026-07-27", "artifactRef": artifact_ref, "consumptionManifestRef": manifest_ref, "consumptionManifestHash": hashlib.sha256(manifest_raw).hexdigest(), } registry = { "schemaVersion": "ReferenceAssetRegistry/2", "registryVersion": f"{name}.1", "sourceOfTruth": "test", "records": [record], } registry_raw = _canonical(registry) (root / registry_ref).parent.mkdir(parents=True, exist_ok=True) (root / registry_ref).write_bytes(registry_raw) policy = { "schemaVersion": "ReferenceAssetConsumptionPolicy/1", "policyId": policy_id, "recordId": record_id, "role": record["role"], "consumerRef": consumer_ref, "route": f"route-{name}", "autoSelect": False, "mode": "frozen_preflight", } policy_raw = _canonical(policy) (root / policy_ref).write_bytes(policy_raw) release = { "schemaVersion": "ReferenceAssetRelease/1", "releaseId": f"release-{name}", "registryRef": registry_ref, "registryHash": hashlib.sha256(registry_raw).hexdigest(), "policyRef": policy_ref, "policyHash": hashlib.sha256(policy_raw).hexdigest(), "verifierVersion": "reference-asset-verifier/1.0.0", "trustedRootId": "wanxiang-reference-assets-root-v1", } release_raw = _canonical(release) (root / release_ref).write_bytes(release_raw) return { "policy_id": policy_id, "release_ref": release_ref, "expected_release_hash": hashlib.sha256(release_raw).hexdigest(), "trusted_root": root, "mode": "frozen_preflight", } def test_capture_selected_files_reads_real_files_and_returns_immutable_snapshot(tmp_path): """选择性捕获应以相对 NFC 路径读取真实文件并返回不可变映射。""" root = tmp_path / "root" (root / "dir").mkdir(parents=True) (root / "dir" / "a.txt").write_bytes(b"a\x00b") (root / "z.txt").write_bytes(b"z") capture = getattr(artifact_snapshot, "capture_selected_files", None) assert capture is not None snapshot = capture(root, ["z.txt", "dir/a.txt"], {"max_files": 2, "max_bytes": 64}) assert dict(snapshot.files) == {"dir/a.txt": b"a\x00b", "z.txt": b"z"} with pytest.raises(TypeError): snapshot.files["new.txt"] = b"nope" assert snapshot.file_count == 2 assert snapshot.total_bytes == 4 def test_verify_policy_returns_frozen_files_records_receipts_and_snapshot_hash(tmp_path): """冻结输入匹配时应一次返回受保护文件、约束记录、回执和 canonical snapshot hash。""" result = _verify(_fixture_root(tmp_path)) assert result.snapshot_hash == result.canonical_snapshot_hash assert result.files assert result.constraint_records[0]["recordId"] == "gac-shanhai-xingji" assert result.receipts[0]["expectedRegistryHash"] == EXPECTED_REGISTRY_HASH assert result.receipts[0]["observedRegistryHash"] == EXPECTED_REGISTRY_HASH assert result.receipts[0]["expected"]["artifactHash"] == EXPECTED_BUNDLE_HASH with pytest.raises(TypeError): result.files["new.txt"] = b"nope" with pytest.raises(TypeError): result.constraint_records[0]["recordId"] = "tampered" def test_snapshot_hash_matches_domain_vector(tmp_path): """snapshot hash 必须使用域标签、UTF-8 路径长度和原始内容长度向量。""" root = _fixture_root(tmp_path) result = _verify(root) expected = hashlib.sha256() expected.update(b"reference-asset-consumption-snapshot/1\n") for path in sorted(result.files, key=lambda value: value.encode("utf-8")): path_bytes = path.encode("utf-8") content = result.files[path] expected.update(len(path_bytes).to_bytes(8, "big")) expected.update(path_bytes) expected.update(len(content).to_bytes(8, "big")) expected.update(content) assert result.snapshot_hash == expected.hexdigest() @pytest.mark.parametrize( ("label", "mutate", "expected_code"), [ ("release hash", lambda root: None, "reference_registry_untrusted"), ("registry hash", lambda root: (root / REGISTRY_REF).write_bytes( (root / REGISTRY_REF).read_bytes() + b" "), "reference_registry_untrusted"), ("registry schema", lambda root: _mutate_json(root / REGISTRY_REF, {"schemaVersion": "bad"}), "reference_registry_invalid"), ("registry duplicate recordId", lambda root: _duplicate_registry_record(root), "reference_registry_invalid"), ("policy hash", lambda root: (root / POLICY_REF).write_bytes( (root / POLICY_REF).read_bytes() + b" "), "reference_registry_untrusted"), ("policy identity", lambda root: _mutate_json(root / POLICY_REF, {"policyId": "other"}), "reference_policy_missing"), ("policy mode", lambda root: _mutate_json(root / POLICY_REF, {"mode": "live"}), "reference_declaration_mismatch"), ("bundle hash", lambda root: (root / BUNDLE_REF).write_bytes(b"tampered bundle"), "reference_artifact_hash_mismatch"), ("manifest hash", lambda root: (root / MANIFEST_REF).write_bytes( (root / MANIFEST_REF).read_bytes() + b" "), "reference_manifest_hash_mismatch"), ("entry hash", lambda root: _tamper_manifest_entry(root), "reference_entry_hash_mismatch"), ], ) def test_frozen_input_tampering_is_rejected_with_stable_code(tmp_path, label, mutate, expected_code): """release、registry、policy、bundle、manifest 和 entry 漂移不得形成消费快照。""" root = _fixture_root(tmp_path) if label == "release hash": code, message = _error_code(lambda: _gate().verify_policy( POLICY_ID, release_ref=RELEASE_REF, expected_release_hash="0" * 64, trusted_root=root, mode="frozen_preflight", )) else: mutate(root) expected_hash = EXPECTED_RELEASE_HASH if label in {"registry schema", "registry duplicate recordId"}: expected_hash = _rebind_release( root, registry_hash=hashlib.sha256((root / REGISTRY_REF).read_bytes()).hexdigest(), ) elif label in {"policy identity", "policy mode"}: expected_hash = _rebind_release( root, policy_hash=hashlib.sha256((root / POLICY_REF).read_bytes()).hexdigest(), ) code, message = _error_code(lambda: _verify(root, expected_release_hash=expected_hash)) assert code == expected_code assert str(tmp_path) not in message assert b"tampered" not in message.encode("utf-8") @pytest.mark.parametrize( "invalid_record_id", [ "bad\nrecord", "bad\x00record", ("x" * 2049) + "/", ], ids=["newline", "control", "overlong"], ) def test_invalid_record_id_error_does_not_echo_untrusted_value(tmp_path, invalid_record_id): """recordId 未通过 identifier 校验时，异常正文只能保留稳定 code。""" root = _fixture_root(tmp_path) registry_path = root / REGISTRY_REF registry = json.loads(registry_path.read_bytes()) record = next(row for row in registry["records"] if row["recordId"] == "gac-shanhai-xingji") record["recordId"] = invalid_record_id registry_raw = _canonical(registry) registry_path.write_bytes(registry_raw) expected_hash = _rebind_release(root, registry_hash=hashlib.sha256(registry_raw).hexdigest()) code, message = _error_code(lambda: _verify(root, expected_release_hash=expected_hash)) assert code == "reference_registry_invalid" assert message == "code=reference_registry_invalid" assert invalid_record_id not in message assert "\n" not in message assert str(tmp_path) not in message def test_valid_record_id_remains_in_later_registry_error_context(tmp_path): """recordId 已合法时，后续结构错误仍可携带稳定上下文。""" root = _fixture_root(tmp_path) registry_path = root / REGISTRY_REF registry = json.loads(registry_path.read_bytes()) record = next(row for row in registry["records"] if row["recordId"] == "gac-shanhai-xingji") record["schemaVersion"] = "bad" registry_raw = _canonical(registry) registry_path.write_bytes(registry_raw) expected_hash = _rebind_release(root, registry_hash=hashlib.sha256(registry_raw).hexdigest()) code, message = _error_code(lambda: _verify(root, expected_release_hash=expected_hash)) assert code == "reference_registry_invalid" assert message == "code=reference_registry_invalid recordId=gac-shanhai-xingji" def _mutate_json(path: Path, changes: dict) -> None: """对临时 JSON 输入做最小字段变异。""" value = json.loads(path.read_bytes()) value.update(changes) path.write_bytes(_canonical(value)) def _duplicate_registry_record(root: Path) -> None: """复制 active recordId，命中 registry 跨记录唯一性语义。""" path = root / REGISTRY_REF value = json.loads(path.read_bytes()) value["records"].append(dict(value["records"][-1])) path.write_bytes(_canonical(value)) def _tamper_manifest_entry(root: Path) -> None: """只改清单覆盖的真实文件，保持 manifest/registry 双 hash 仍匹配。""" manifest = json.loads((root / MANIFEST_REF).read_bytes()) path = root / manifest["entries"][0]["path"] path.write_bytes(path.read_bytes() + b"tampered") @pytest.mark.parametrize( ("relative", "expected_code"), [ ("/absolute.txt", "reference_path_escape"), ("../escape.txt", "reference_path_escape"), ("dir/../../escape.txt", "reference_path_escape"), ("dir//file.txt", "reference_path_invalid"), ("dir/./file.txt", "reference_path_invalid"), ("dir\\file.txt", "reference_path_invalid"), ("", "reference_path_invalid"), ("e\u0301.txt", "reference_path_invalid"), ], ) def test_selected_capture_rejects_unsafe_paths(tmp_path, relative, expected_code): """选择性读取器在打开任何对象前拒绝越界、非 NFC 和歧义路径。""" root = tmp_path / "root" root.mkdir() code, _ = _error_code(lambda: artifact_snapshot.capture_selected_files( root, [relative], {"max_files": 2, "max_bytes": 64}, )) assert code == expected_code def test_selected_capture_rejects_normalization_collision_before_io(tmp_path): """NFC 规范化后重复的两个名字不能让读取顺序产生歧义。""" root = tmp_path / "root" root.mkdir() (root / "é.txt").write_bytes(b"one") code, _ = _error_code(lambda: artifact_snapshot.capture_selected_files( root, ["é.txt", "e\u0301.txt"], None, )) assert code == "reference_path_invalid" @pytest.mark.parametrize( ("setup", "expected_code"), [ ("root_symlink", "reference_symlink"), ("ancestor_symlink", "reference_symlink"), ("leaf_symlink", "reference_symlink"), ("missing", "reference_missing"), ("directory", "reference_not_regular"), ("fifo", "reference_not_regular"), ("socket", "reference_not_regular"), ], ) def test_selected_capture_fails_closed_for_object_types_and_symlinks(tmp_path, setup, expected_code): """根、祖先、末级 symlink 及特殊文件均不得进入内存快照。""" root = tmp_path / "root" root.mkdir() relative = "target.txt" if setup == "root_symlink": target = tmp_path / "real-root" target.mkdir() (target / relative).write_bytes(b"x") root.rmdir() root.symlink_to(target, target_is_directory=True) root_for_read = root else: root_for_read = root if setup == "ancestor_symlink": target = tmp_path / "real-dir" target.mkdir() (target / "file.txt").write_bytes(b"x") (root / "dir").symlink_to(target, target_is_directory=True) relative = "dir/file.txt" elif setup == "leaf_symlink": (root / "real.txt").write_bytes(b"x") (root / relative).symlink_to(root / "real.txt") elif setup == "directory": (root / relative).mkdir() elif setup == "fifo": if not hasattr(os, "mkfifo"): pytest.skip("平台没有 FIFO") os.mkfifo(root / relative) elif setup == "socket": relative = "s" shutil.rmtree(root) socket_parent = "/private/tmp" if Path("/private/tmp").is_dir() else "/tmp" root = Path(tempfile.mkdtemp(prefix="rasset-", dir=socket_parent)) root_for_read = root listener = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) listener.bind(str(root / relative)) elif setup == "missing": pass if setup != "socket": listener = None try: code, _ = _error_code(lambda: artifact_snapshot.capture_selected_files( root_for_read, [relative], None, )) assert code == expected_code finally: if setup == "socket": listener.close() (root / relative).unlink(missing_ok=True) @pytest.mark.parametrize( ("limits", "expected_code"), [ ({"max_files": 1}, "reference_oversize"), ({"max_bytes": 2}, "reference_oversize"), ({"max_file_bytes": 2}, "reference_oversize"), ], ) def test_selected_capture_enforces_file_count_and_logical_size_limits(tmp_path, limits, expected_code): """文件数、总记录和单文件上限在读前按逻辑大小 fail-closed。""" root = tmp_path / "root" root.mkdir() (root / "a.txt").write_bytes(b"abc") paths = ["a.txt", "missing.txt"] if limits.get("max_files") else ["a.txt"] code, _ = _error_code(lambda: artifact_snapshot.capture_selected_files(root, paths, limits)) assert code == expected_code def test_manifest_must_be_canonical_and_entries_must_be_sorted_and_bounded(tmp_path): """非 canonical、重复/乱序、超项数和超清单字节均拒绝且不读取 entry。""" root = _fixture_root(tmp_path) manifest_path = root / MANIFEST_REF manifest = json.loads(manifest_path.read_bytes()) manifest["entries"] = list(reversed(manifest["entries"])) manifest_path.write_bytes(json.dumps(manifest, ensure_ascii=False).encode("utf-8")) code, _ = _error_code(lambda: _verify(root)) assert code == "reference_manifest_hash_mismatch" def test_manifest_path_outside_protected_asset_roots_is_rejected_before_entry_io(tmp_path): """清单只能读取 active 记录的 assetRef 与批准 designRef 覆盖范围。""" root = _fixture_root(tmp_path) manifest_path = root / MANIFEST_REF manifest = json.loads(manifest_path.read_bytes()) manifest["entries"][0]["path"] = "contracts/play-loop/reference-asset-release.initial.json" manifest["entries"].sort(key=lambda entry: entry["path"].encode("utf-8")) manifest_path.write_bytes(_canonical(manifest)) expected_hash = _rebind_manifest_chain(root) code, _ = _error_code(lambda: _verify(root, expected_release_hash=expected_hash)) assert code == "reference_path_escape" def test_declarations_must_match_policy_and_empty_live_mode_is_rejected(tmp_path): """声明缺失/冲突及 frozen policy 误用于 live 均在读取 bundle 前拒绝。""" root = _fixture_root(tmp_path) for declarations in ( {}, {"referenceAssetRecordIds": ["other"], "consumerRef": "generation-runtime@reference-assets/2"}, {"referenceAssetRecordIds": ["gac-shanhai-xingji"], "consumerRef": "other"}, ): code, _ = _error_code(lambda: _verify(root, declarations=declarations)) assert code == "reference_declaration_mismatch" code, _ = _error_code(lambda: _verify(root, declarations=[], mode="live")) assert code == "reference_declaration_mismatch" def test_unknown_policy_is_rejected_before_placeholder_asset_io(tmp_path, monkeypatch): """未知 policy 必须在任何 bundle/manifest/entry 占位读取前终止。""" root = _fixture_root(tmp_path) gate = _gate() calls = [] original = artifact_snapshot.capture_selected_files def spy(*args, **kwargs): calls.append(args[1] if len(args) > 1 else kwargs.get("paths")) return original(*args, **kwargs) monkeypatch.setattr(artifact_snapshot, "capture_selected_files", spy) code, _ = _error_code(lambda: gate.verify_policy( "unknown-policy", release_ref=RELEASE_REF, expected_release_hash=EXPECTED_RELEASE_HASH, trusted_root=root, mode="frozen_preflight", )) assert code == "reference_policy_missing" assert calls == [] @pytest.mark.parametrize("entrypoint", ["single", "batch"]) def test_auto_select_policy_is_rejected_before_target_asset_io(tmp_path, monkeypatch, entrypoint): """autoSelect=true 的禁用策略在两条入口均不得读取目标资产。""" root = _fixture_root(tmp_path) _mutate_json(root / POLICY_REF, {"autoSelect": True}) expected_hash = _rebind_release( root, policy_hash=hashlib.sha256((root / POLICY_REF).read_bytes()).hexdigest(), ) manifest = json.loads((root / MANIFEST_REF).read_bytes()) target_paths = {BUNDLE_REF, MANIFEST_REF, *(entry["path"] for entry in manifest["entries"])} calls = [] original = artifact_snapshot.capture_selected_files def spy(*args, **kwargs): """记录真实选择性读取请求，保留原读取器验证其余信任链。""" paths = tuple(args[1] if len(args) > 1 else kwargs["paths"]) calls.extend(paths) return original(*args, **kwargs) monkeypatch.setattr(artifact_snapshot, "capture_selected_files", spy) if entrypoint == "single": call = lambda: _verify(root, expected_release_hash=expected_hash) else: call = lambda: _gate().verify_policies([{ "policy_id": POLICY_ID, "release_ref": RELEASE_REF, "expected_release_hash": expected_hash, "trusted_root": root, "mode": "frozen_preflight", }]) code, _ = _error_code(call) assert code == "reference_policy_missing" assert target_paths.isdisjoint(calls) def test_non_active_record_rejects_without_reading_placeholder_asset(tmp_path, monkeypatch): """registry 中非 active 记录即使声明了路径，也不能触发占位资产 I/O。""" root = _fixture_root(tmp_path) registry_path = root / REGISTRY_REF registry = json.loads(registry_path.read_bytes()) active = next(row for row in registry["records"] if row["recordId"] == "gac-shanhai-xingji") active["lifecycleStatus"] = "migration_pending" active["consumerRef"] = None active["designRef"] = None active["artifactRef"] = None active["consumptionManifestRef"] = None active["consumptionManifestHash"] = None registry_raw = _canonical(registry) registry_path.write_bytes(registry_raw) expected_hash = _rebind_release(root, registry_hash=hashlib.sha256(registry_raw).hexdigest()) calls = [] gate = _gate() original = artifact_snapshot.capture_selected_files def spy(*args, **kwargs): calls.append(tuple(args[1])) return original(*args, **kwargs) monkeypatch.setattr(artifact_snapshot, "capture_selected_files", spy) code, _ = _error_code(lambda: _verify(root, expected_release_hash=expected_hash)) assert code == "reference_registry_invalid" assert all(BUNDLE_REF not in call and MANIFEST_REF not in call for call in calls) def test_all_or_nothing_does_not_expose_files_or_receipts_after_entry_failure(tmp_path): """任一 entry 失败时 verify_policy 只抛异常，不返回任何半成品对象。""" root = _fixture_root(tmp_path) manifest = json.loads((root / MANIFEST_REF).read_bytes()) bad_entry = manifest["entries"][-1] (root / bad_entry["path"]).write_bytes(b"changed during setup") code, _ = _error_code(lambda: _verify(root)) assert code == "reference_entry_hash_mismatch" def test_cwd_does_not_change_trusted_root_resolution(tmp_path, monkeypatch): """releaseRef 始终相对 trusted_root 解析，调用 cwd 不参与身份和结果。""" root = _fixture_root(tmp_path) first = _verify(root) other = tmp_path / "other-cwd" other.mkdir() monkeypatch.chdir(other) second = _verify(root) assert first.snapshot_hash == second.snapshot_hash assert dict(first.files) == dict(second.files) def test_release_registry_policy_and_manifest_missing_map_to_stable_codes(tmp_path): """受信链各层缺失不能落到裸 FileNotFoundError 或泄露系统路径。""" for relative, expected_code in ( (RELEASE_REF, "reference_missing"), (REGISTRY_REF, "reference_missing"), (POLICY_REF, "reference_policy_missing"), (BUNDLE_REF, "reference_missing"), (MANIFEST_REF, "reference_missing"), ): root = _fixture_root(tmp_path / relative.replace("/", "_")) (root / relative).unlink() code, message = _error_code(lambda: _verify(root)) assert code == expected_code assert str(root) not in message def test_reading_file_changed_during_read_is_rejected(tmp_path, monkeypatch): """同一 fd 读中发生截断或改写必须命中 changed_during_read。""" root = tmp_path / "root" root.mkdir() target = root / "file.txt" target.write_bytes(b"original") original_read = os.read changed = False def rewrite_after_first_read(fd, size): nonlocal changed chunk = original_read(fd, size) if chunk and not changed: changed = True target.write_bytes(b"rewritten") return chunk monkeypatch.setattr(artifact_snapshot.os, "read", rewrite_after_first_read) code, _ = _error_code(lambda: artifact_snapshot.capture_selected_files(root, ["file.txt"], None)) assert code == "reference_changed_during_read" def test_sparse_file_uses_logical_size_for_budget(tmp_path): """稀疏文件不能借助物理占用小绕过逻辑大小预算。""" root = tmp_path / "root" root.mkdir() target = root / "sparse.bin" with target.open("wb") as handle: handle.seek(16 * 1024 * 1024) handle.write(b"x") code, _ = _error_code(lambda: artifact_snapshot.capture_selected_files( root, ["sparse.bin"], {"max_file_bytes": 16 * 1024 * 1024}, )) assert code == "reference_oversize" def test_verify_policy_rejects_active_missing_dual_identity_and_retired_partial_identity(tmp_path): """active 双身份缺失和 retired 半身份均不得进入可信消费。""" for field in ("artifactRef", "consumptionManifestRef", "consumptionManifestHash", "consumerRef"): root = _fixture_root(tmp_path / field) registry_path = root / REGISTRY_REF registry = json.loads(registry_path.read_bytes()) record = next(row for row in registry["records"] if row["recordId"] == "gac-shanhai-xingji") record[field] = None registry_raw = _canonical(registry) registry_path.write_bytes(registry_raw) expected_hash = _rebind_release(root, registry_hash=hashlib.sha256(registry_raw).hexdigest()) code, _ = _error_code(lambda: _verify(root, expected_release_hash=expected_hash)) assert code == "reference_registry_invalid" def test_registry_allows_non_active_record_with_optional_identity_fields_omitted(tmp_path): """candidate/migration_pending/无历史 retired 可按 Record/2 省略可选身份字段。""" root = _fixture_root(tmp_path) registry_path = root / REGISTRY_REF registry = json.loads(registry_path.read_bytes()) pending = next(row for row in registry["records"] if row["recordId"] == "_template-puzzle") for field in ( "consumerRef", "designRef", "signedBy", "signedAt", "artifactRef", "consumptionManifestRef", "consumptionManifestHash", ): pending.pop(field, None) registry_raw = _canonical(registry) registry_path.write_bytes(registry_raw) expected_hash = _rebind_release(root, registry_hash=hashlib.sha256(registry_raw).hexdigest()) result = _verify(root, expected_release_hash=expected_hash) assert result.constraint_records[0]["recordId"] == "gac-shanhai-xingji" def test_registry_rejects_retired_record_with_partial_historical_identity(tmp_path): """retired 一旦保留任一历史签认字段，就必须保留完整双身份闭包。""" root = _fixture_root(tmp_path) registry_path = root / REGISTRY_REF registry = json.loads(registry_path.read_bytes()) retired = next(row for row in registry["records"] if row["recordId"] == "_template-puzzle") retired["lifecycleStatus"] = "retired" retired["consumerRef"] = "historical-consumer" for field in ("signedBy", "signedAt", "artifactRef", "consumptionManifestRef", "consumptionManifestHash"): retired.pop(field, None) registry_raw = _canonical(registry) registry_path.write_bytes(registry_raw) expected_hash = _rebind_release(root, registry_hash=hashlib.sha256(registry_raw).hexdigest()) code, _ = _error_code(lambda: _verify(root, expected_release_hash=expected_hash)) assert code == "reference_registry_invalid" def test_selected_capture_rejects_files_over_default_single_record_limit(tmp_path): """默认单文件和单记录上限分别为 16 MiB 与 64 MiB。""" root = tmp_path / "root" root.mkdir() target = root / "large.bin" with target.open("wb") as handle: handle.seek(16 * 1024 * 1024) handle.write(b"x") code, _ = _error_code(lambda: artifact_snapshot.capture_selected_files(root, ["large.bin"], None)) assert code == "reference_oversize" def test_custom_limits_cannot_relax_selected_capture_hard_caps(tmp_path): """调用方给更大 limits 时，16 MiB 单文件、512 文件和 64 MiB 单记录硬帽仍生效。""" root = tmp_path / "root" root.mkdir() oversized = root / "oversized.bin" with oversized.open("wb") as handle: handle.truncate(17 * 1024 * 1024) code, _ = _error_code(lambda: artifact_snapshot.capture_selected_files( root, ["oversized.bin"], {"max_file_bytes": 20 * 1024 * 1024, "max_record_bytes": 80 * 1024 * 1024}, )) assert code == "reference_oversize" many_paths = [] for index in range(513): relative = f"many/{index:03d}.txt" target = root / relative target.parent.mkdir(parents=True, exist_ok=True) target.write_bytes(b"x") many_paths.append(relative) code, _ = _error_code(lambda: artifact_snapshot.capture_selected_files( root, many_paths, {"max_files": 1000}, )) assert code == "reference_oversize" record_paths = [] for index in range(5): relative = f"record/{index}.bin" target = root / relative target.parent.mkdir(parents=True, exist_ok=True) with target.open("wb") as handle: handle.truncate(16 * 1024 * 1024 if index < 4 else 1) record_paths.append(relative) code, _ = _error_code(lambda: artifact_snapshot.capture_selected_files( root, record_paths, {"max_file_bytes": 20 * 1024 * 1024, "max_record_bytes": 80 * 1024 * 1024}, )) assert code == "reference_oversize" def test_verified_result_uses_deep_read_only_containers_and_json_copy(tmp_path): """内建基类旁路不能篡改验证结果，显式 JSON 副本仍符合 receipt schema。""" gate = _gate() result = _verify(_fixture_root(tmp_path)) record = result.records[0] receipt = result.receipts[0] assert not isinstance(record, dict) assert not isinstance(receipt, dict) assert isinstance(record["designRef"], tuple) with pytest.raises(TypeError): dict.__setitem__(record, "recordId", "tampered") with pytest.raises(TypeError): dict.update(receipt, {"recordId": "tampered"}) with pytest.raises(TypeError): list.extend(record["designRef"], ["tampered"]) receipt_json = gate.to_json_value(receipt) assert isinstance(receipt_json, dict) assert json.loads(json.dumps(receipt_json, ensure_ascii=False)) == receipt_json validate_path = REPO_ROOT / "contracts/play-loop/validate.py" spec = importlib.util.spec_from_file_location("play_loop_validate_for_receipt", validate_path) validator = importlib.util.module_from_spec(spec) assert spec.loader is not None spec.loader.exec_module(validator) schema = validator._load(REPO_ROOT / "contracts/play-loop/reference-asset-verification-receipt.schema.json") assert validator.validate(schema, receipt_json, schema) == [] def test_files_is_a_pure_path_to_bytes_mapping_with_separate_root_index(tmp_path): """files 的迭代、成员关系和取值必须都只表达逻辑路径到 bytes。""" result = _verify(_fixture_root(tmp_path)) assert all(isinstance(path, str) and isinstance(result.files[path], bytes) for path in result.files) assert "gac-shanhai-xingji" not in result.files with pytest.raises(KeyError): result.files["gac-shanhai-xingji"] assert result.reference_roots["gac-shanhai-xingji"] with pytest.raises(TypeError): result.reference_roots["other"] = ("assets",) @pytest.mark.parametrize( ("mutation", "expected_code"), [ ("canonical-unsorted", "reference_path_invalid"), ("duplicate", "reference_path_invalid"), ("513", "reference_oversize"), ("over-1mib", "reference_oversize"), ], ) def test_manifest_semantics_are_checked_after_rebinding_upstream_hashes(tmp_path, mutation, expected_code): """每类 manifest 负例都重绑上游 hash，确保命中自身语义而非陈旧 hash。""" root = _fixture_root(tmp_path) manifest_path = root / MANIFEST_REF manifest = json.loads(manifest_path.read_bytes()) if mutation == "canonical-unsorted": manifest["entries"] = list(reversed(manifest["entries"])) elif mutation == "duplicate": manifest["entries"].append(dict(manifest["entries"][0])) elif mutation == "513": manifest["entries"] = [ {"path": f"game-runtime/games/shanhai-xingji/src/generated-{index:03d}.js", "size": 0, "sha256": hashlib.sha256(b"").hexdigest()} for index in range(513) ] else: manifest["padding"] = "x" * (1024 * 1024) manifest_path.write_bytes(_canonical(manifest)) expected_hash = _rebind_manifest_chain(root) code, _ = _error_code(lambda: _verify(root, expected_release_hash=expected_hash)) assert code == expected_code @pytest.mark.parametrize("field", ["role", "consumerRef"]) def test_active_record_role_and_consumer_ref_drift_are_rejected(tmp_path, field): """active record 与 release 绑定 policy 的 role/consumerRef 漂移必须 fail-closed。""" root = _fixture_root(tmp_path) registry_path = root / REGISTRY_REF registry = json.loads(registry_path.read_bytes()) record = next(row for row in registry["records"] if row["recordId"] == "gac-shanhai-xingji") record[field] = "generation_exemplar" if field == "role" else "drifted-consumer" registry_raw = _canonical(registry) registry_path.write_bytes(registry_raw) expected_hash = _rebind_release(root, registry_hash=hashlib.sha256(registry_raw).hexdigest()) code, _ = _error_code(lambda: _verify(root, expected_release_hash=expected_hash)) assert code == "reference_declaration_mismatch" def test_inode_replacement_after_stat_is_rejected(tmp_path, monkeypatch): """初次 lstat 后、open 前替换目录项 inode，必须命中 changed_during_read。""" root = tmp_path / "root" root.mkdir() target = root / "file.txt" replacement = root / "replacement.txt" target.write_bytes(b"original") replacement.write_bytes(b"replacement") original_open = os.open replaced = False def replace_before_open(path, flags, *args, **kwargs): nonlocal replaced if path == "file.txt" and kwargs.get("dir_fd") is not None and not replaced: replaced = True os.replace(replacement, target) return original_open(path, flags, *args, **kwargs) monkeypatch.setattr(artifact_snapshot.os, "open", replace_before_open) code, _ = _error_code(lambda: artifact_snapshot.capture_selected_files(root, ["file.txt"], None)) assert code == "reference_changed_during_read" def test_unreadable_file_maps_to_stable_code_when_permissions_are_enforced(tmp_path): """平台执行权限位时，不可读普通文件必须映射为稳定 unreadable。""" if hasattr(os, "geteuid") and os.geteuid() == 0: pytest.skip("root 会绕过普通权限位") root = tmp_path / "root" root.mkdir() target = root / "file.txt" target.write_bytes(b"secret") target.chmod(0) try: code, _ = _error_code(lambda: artifact_snapshot.capture_selected_files(root, ["file.txt"], None)) assert code == "reference_unreadable" finally: target.chmod(stat.S_IRUSR | stat.S_IWUSR) def test_unprotected_legacy_path_is_rejected_before_target_io(tmp_path, monkeypatch): """manifest 指向未受保护旧路径时，不得对该旧路径发起选择性读取。""" root = _fixture_root(tmp_path) legacy_path = "contracts/play-loop/reference-asset-registry.initial.json" manifest_path = root / MANIFEST_REF manifest = json.loads(manifest_path.read_bytes()) manifest["entries"][0]["path"] = legacy_path manifest["entries"].sort(key=lambda entry: entry["path"].encode("utf-8")) manifest_path.write_bytes(_canonical(manifest)) expected_hash = _rebind_manifest_chain(root) calls = [] original = artifact_snapshot.capture_selected_files def spy(*args, **kwargs): paths = tuple(args[1] if len(args) > 1 else kwargs["paths"]) calls.append(paths) return original(*args, **kwargs) monkeypatch.setattr(artifact_snapshot, "capture_selected_files", spy) code, _ = _error_code(lambda: _verify(root, expected_release_hash=expected_hash)) assert code == "reference_path_escape" assert all(legacy_path not in paths for paths in calls) def test_verify_policies_atomically_merges_multiple_records_and_deduplicates_identical_paths(tmp_path): """batch 全部通过后才合并，多记录文件与相同路径同字节按逻辑路径去重。""" first_root = tmp_path / "first" second_root = tmp_path / "second" first_root.mkdir() second_root.mkdir() shared = "assets/shared/common.bin" first = _build_batch_policy(first_root, "first", [(shared, 3, b"x"), ("assets/first/a.bin", 2, b"a")]) second = _build_batch_policy(second_root, "second", [(shared, 3, b"x"), ("assets/second/b.bin", 4, b"b")]) result = _gate().verify_policies([first, second]) assert [record["recordId"] for record in result.records] == ["record-first", "record-second"] assert len(result.receipts) == 2 assert dict(result.files) == { "assets/first/a.bin": b"aa", "assets/second/b.bin": b"bbbb", shared: b"xxx", } assert set(result.reference_roots) == {"record-first", "record-second"} def test_verify_policies_accepts_exact_128_mib_of_deduplicated_logical_bytes(tmp_path): """两个各 64 MiB 的记录合并后，去重逻辑字节恰为 128 MiB 时允许。""" root = tmp_path / "root" root.mkdir() chunk = 16 * 1024 * 1024 first_entries = [(f"assets/large-first/{index}.bin", chunk, b"\x00") for index in range(4)] second_entries = [(f"assets/large-second/{index}.bin", chunk, b"\x00") for index in range(4)] first = _build_batch_policy(root, "large-first", first_entries) second = _build_batch_policy(root, "large-second", second_entries) result = _gate().verify_policies([first, second]) assert sum(len(content) for content in result.files.values()) == 128 * 1024 * 1024 def test_verify_policies_rejects_deduplicated_total_over_128_mib(tmp_path): """所有单记录均不超 64 MiB 时，合并唯一逻辑路径超过 128 MiB 仍拒绝。""" root = tmp_path / "root" root.mkdir() chunk = 16 * 1024 * 1024 first = _build_batch_policy( root, "limit-first", [(f"assets/limit-first/{index}.bin", chunk, b"\x00") for index in range(4)], ) second = _build_batch_policy( root, "limit-second", [(f"assets/limit-second/{index}.bin", chunk, b"\x00") for index in range(4)], ) third = _build_batch_policy(root, "limit-third", [("assets/limit-third/extra.bin", 1, b"x")]) code, _ = _error_code(lambda: _gate().verify_policies([first, second, third])) assert code == "reference_oversize" def test_verify_policies_does_not_construct_partial_result_when_later_record_fails(tmp_path, monkeypatch): """第二条记录失败时，公开结果对象一次也不能被构造。""" root = tmp_path / "root" root.mkdir() first = _build_batch_policy(root, "atomic-first", [("assets/atomic-first/a.bin", 1, b"a")]) second = _build_batch_policy(root, "atomic-second", [("assets/atomic-second/b.bin", 1, b"b")]) (root / "assets/atomic-second/b.bin").write_bytes(b"tampered") gate = _gate() original_type = gate.VerifiedReferenceAssets constructions = [] def track_construction(*args, **kwargs): constructions.append((args, kwargs)) return original_type(*args, **kwargs) monkeypatch.setattr(gate, "VerifiedReferenceAssets", track_construction) code, _ = _error_code(lambda: gate.verify_policies([first, second])) assert code == "reference_entry_hash_mismatch" assert constructions == [] def test_verify_policies_rejects_same_logical_path_with_conflicting_bytes(tmp_path): """不同可信根出现同逻辑路径但内容冲突时，batch 必须 fail-closed。""" first_root = tmp_path / "first" second_root = tmp_path / "second" first_root.mkdir() second_root.mkdir() shared = "assets/shared/conflict.bin" first = _build_batch_policy(first_root, "conflict-first", [(shared, 3, b"a")]) second = _build_batch_policy(second_root, "conflict-second", [(shared, 3, b"b")]) code, _ = _error_code(lambda: _gate().verify_policies([first, second])) assert code == "reference_entry_hash_mismatch" def test_verify_policies_releases_each_stage_before_staging_next_record(tmp_path, monkeypatch): """batch 合并应在下一条记录读取前释放上一条未合并 stage。""" root = tmp_path / "root" root.mkdir() requests = [ _build_batch_policy( root, f"stream-{index}", [(f"assets/stream-{index}/item.bin", 1, bytes([65 + index]))], ) for index in range(4) ] gate = _gate() original = gate._stage_one_policy alive_stages = [] peak_alive = 0 def tracked_stage(*args, **kwargs): """用弱引用统计未合并 stage 的真实生命周期，不改变 stage 行为。""" nonlocal peak_alive item = original(*args, **kwargs) def release(reference): """阶段对象释放时从生命周期计数中移除对应弱引用。""" alive_stages.remove(reference) reference = weakref.ref(item, release) alive_stages.append(reference) peak_alive = max(peak_alive, len(alive_stages)) return item monkeypatch.setattr(gate, "_stage_one_policy", tracked_stage) result = gate.verify_policies(requests) assert len(result.records) == len(requests) assert peak_alive <= 2