1981a6be01
增加不可变源码检出、两阶段 finalize/activate、Runtime 同源对象读取与 iframe 安全边界，并用 fail-closed CI 门固定契约。Git 仅提交平台实现、契约、迁移和 SoT，不包含具体游戏源码、素材、bundle 或验收证据。
743 lines
36 KiB
Python
743 lines
36 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""ReferenceAsset/2 trusted-consumption contract tests.
|
||
|
||
本测试只验证 Task 1 新增的七条版本线、严格样本和旧契约隔离。测试通过
|
||
validate.py 的真实 CLI 与同一 stdlib 校验器执行，不把 schema 源文本当作行为。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import hashlib
|
||
import os
|
||
import re
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
from pathlib import Path
|
||
|
||
HERE = Path(__file__).resolve().parent
|
||
sys.path.insert(0, str(HERE))
|
||
import validate as V # noqa: E402 —— 与 CLI 使用同一校验器
|
||
|
||
|
||
CONTRACTS = (
|
||
("reference-asset-record-v2", "reference-asset-record-v2.schema.json"),
|
||
("reference-asset-registry-v2", "reference-asset-registry-v2.schema.json"),
|
||
("reference-asset-consumption-manifest", "reference-asset-consumption-manifest.schema.json"),
|
||
("reference-asset-consumption-policy", "reference-asset-consumption-policy.schema.json"),
|
||
("reference-asset-release", "reference-asset-release.schema.json"),
|
||
("reference-asset-verification-receipt", "reference-asset-verification-receipt.schema.json"),
|
||
("acceptance-provenance-v4", "acceptance-provenance-v4.schema.json"),
|
||
)
|
||
|
||
_PASS: list[tuple[str, str]] = []
|
||
_FAIL: list[tuple[str, str]] = []
|
||
|
||
REPO_ROOT = HERE.parent.parent
|
||
REGISTRY_V1 = HERE / "reference-asset-registry.initial.json"
|
||
REGISTRY_V2 = HERE / "reference-asset-registry.v2.initial.json"
|
||
POLICY_INITIAL = HERE / "reference-asset-consumption-policy.initial.json"
|
||
RELEASE_INITIAL = HERE / "reference-asset-release.initial.json"
|
||
MANIFEST = REPO_ROOT / "game-runtime/games/shanhai-xingji/reference/consumption-manifest.json"
|
||
GENERATOR = HERE / "build_reference_asset_consumption_manifest.py"
|
||
GAME_ROOT = REPO_ROOT / "game-runtime/games/shanhai-xingji"
|
||
EXPECTED_BUNDLE_HASH = "17b9073c767faf7990e0bf4563a86d55ffa81121f11e7c8ad37c8b8ce72e8cbd"
|
||
EXPECTED_MANIFEST_REF = "game-runtime/games/shanhai-xingji/reference/consumption-manifest.json"
|
||
EXPECTED_V1_REGISTRY_HASH = "cefa7c136700c7badaa5870ddbac230e4a93a4342a8254a0d130a471a2ad7b9d"
|
||
V1_BYTES_AT_TEST_START = REGISTRY_V1.read_bytes() if REGISTRY_V1.is_file() else None
|
||
|
||
|
||
def check(name: str, ok: bool, detail: str = "") -> None:
|
||
"""打印单条断言并保留失败证据，便于 RED/GREEN 原始输出复核。"""
|
||
(_PASS if ok else _FAIL).append((name, detail))
|
||
line = ("PASS " if ok else "FAIL ") + name
|
||
if not ok and detail:
|
||
line += f"\n 证据：{detail}"
|
||
print(line)
|
||
|
||
|
||
def sample(path: Path):
|
||
"""按 validate.py 的声明式样本口径展开正负样本。"""
|
||
return V._load_sample_instance(path.resolve())
|
||
|
||
|
||
def errors_of(schema_file: str, instance) -> list:
|
||
"""执行结构层和语义层完整校验，返回空列表表示通过。"""
|
||
schema = V._load((HERE / schema_file).resolve())
|
||
errors = V.validate(schema, instance, schema)
|
||
if not errors:
|
||
errors += V._semantic_validate(schema, instance)
|
||
return errors
|
||
|
||
|
||
def errors_of_path(schema_file: str, path: Path) -> list:
|
||
"""按真实样本文件校验，保留原始字节约束和声明式负样本展开。"""
|
||
schema = V._load((HERE / schema_file).resolve())
|
||
return V._validate_file(schema, path.resolve())
|
||
|
||
|
||
def repo_ref(path: Path) -> str:
|
||
"""把临时测试文件转换为仓根相对引用，模拟契约中的 ref。"""
|
||
return path.absolute().relative_to(HERE.parent.parent).as_posix()
|
||
|
||
|
||
def write_json(path: Path, value: dict) -> None:
|
||
"""写入测试专用跨文件 fixture；生产文件仍由仓内样本提供。"""
|
||
path.write_bytes((json.dumps(value, ensure_ascii=False, indent=2) + "\n").encode("utf-8"))
|
||
|
||
|
||
def expected_manifest_paths(registry: dict, game_root: Path = GAME_ROOT) -> list[str]:
|
||
"""按 Task 2 的固定白名单独立推导清单路径，不复用生成器实现。"""
|
||
paths = {
|
||
(game_root / name).relative_to(REPO_ROOT).as_posix()
|
||
for name in ("README.md", "index.html", "entry.js")
|
||
}
|
||
for pattern in ("src/*.js", "assets/atlas/*.json"):
|
||
paths.update(path.relative_to(REPO_ROOT).as_posix() for path in game_root.glob(pattern))
|
||
paths.add((game_root / "assets/manifest.json").relative_to(REPO_ROOT).as_posix())
|
||
active = next(
|
||
record for record in registry.get("records", [])
|
||
if record.get("recordId") == "gac-shanhai-xingji"
|
||
)
|
||
paths.update(active.get("designRef") or [])
|
||
return sorted(paths, key=lambda value: value.encode("utf-8"))
|
||
|
||
|
||
def run_manifest_generator(
|
||
repo_root: Path,
|
||
output: Path,
|
||
source_mode: str | None = None,
|
||
) -> subprocess.CompletedProcess:
|
||
"""运行生产清单生成命令，保留 stdout/stderr 供 RED/GREEN 证据使用。"""
|
||
command = [
|
||
sys.executable,
|
||
str(GENERATOR),
|
||
"--repo-root",
|
||
str(repo_root),
|
||
"--output",
|
||
str(output),
|
||
]
|
||
if source_mode is not None:
|
||
command.extend(("--mode", source_mode))
|
||
return subprocess.run(
|
||
command,
|
||
cwd=REPO_ROOT,
|
||
capture_output=True,
|
||
text=True,
|
||
check=False,
|
||
)
|
||
|
||
|
||
def copy_manifest_inputs(root: Path, source_paths: list[str]) -> None:
|
||
"""复制固定消费范围，构造只含测试输入的临时归档。"""
|
||
# 测试归档只复制批准清单和 /1 registry，不把真实工作区的其它文件带入候选范围。
|
||
for relative in source_paths:
|
||
target = root / relative
|
||
target.parent.mkdir(parents=True, exist_ok=True)
|
||
shutil.copyfile(REPO_ROOT / relative, target)
|
||
registry_target = root / "contracts/play-loop/reference-asset-registry.initial.json"
|
||
registry_target.parent.mkdir(parents=True, exist_ok=True)
|
||
shutil.copyfile(REGISTRY_V1, registry_target)
|
||
|
||
|
||
def matches_v1_trust_anchor(path: Path) -> bool:
|
||
"""只接受与批准的 /1 原始字节完全相同的 registry。"""
|
||
return path.is_file() and hashlib.sha256(path.read_bytes()).hexdigest() == EXPECTED_V1_REGISTRY_HASH
|
||
|
||
|
||
def test_task2_artifact_files_and_manifest_scope() -> None:
|
||
"""Task 2 数据文件必须存在，且清单只能覆盖批准的确定性文件集。"""
|
||
required = (REGISTRY_V2, POLICY_INITIAL, RELEASE_INITIAL, MANIFEST, GENERATOR)
|
||
missing = [str(path.relative_to(REPO_ROOT)) for path in required if not path.is_file()]
|
||
check("Task 2 生成器与四份数据文件已创建", not missing, ", ".join(missing))
|
||
if missing:
|
||
return
|
||
|
||
registry = sample(REGISTRY_V2)
|
||
policy = sample(POLICY_INITIAL)
|
||
release = sample(RELEASE_INITIAL)
|
||
manifest = sample(MANIFEST)
|
||
check("Registry/2 数据通过 strict schema", not errors_of("reference-asset-registry-v2.schema.json", registry))
|
||
check("消费 policy 数据通过 strict schema", not errors_of("reference-asset-consumption-policy.schema.json", policy))
|
||
check("可信 release 数据通过 strict schema", not errors_of("reference-asset-release.schema.json", release))
|
||
check(
|
||
"消费 manifest 数据通过 strict schema",
|
||
not errors_of("reference-asset-consumption-manifest.schema.json", manifest),
|
||
)
|
||
actual = MANIFEST.read_bytes()
|
||
check(
|
||
"仓内 manifest 原始字节符合 canonical JSON",
|
||
actual == canonical_manifest_bytes(manifest),
|
||
f"actual_sha={hashlib.sha256(actual).hexdigest()}",
|
||
)
|
||
|
||
expected = expected_manifest_paths(sample(REGISTRY_V1))
|
||
actual_paths = [entry.get("path") for entry in manifest.get("entries", [])]
|
||
check("manifest 精确覆盖批准文件集", actual_paths == expected, f"actual={actual_paths}")
|
||
check(
|
||
"manifest entries 按 UTF-8 字节序排列",
|
||
actual_paths == sorted(actual_paths, key=lambda value: value.encode("utf-8")),
|
||
str(actual_paths),
|
||
)
|
||
forbidden = ("/evidence/", "/dist/", "/test/", "/scripts/")
|
||
check(
|
||
"manifest 不含 evidence/dist/test/scripts 文件",
|
||
not any(any(f"/{part}/" in f"/{path}" for part in ("evidence", "dist", "test", "scripts"))
|
||
for path in actual_paths),
|
||
str(actual_paths),
|
||
)
|
||
entry_by_path = {entry["path"]: entry for entry in manifest.get("entries", [])}
|
||
hash_errors = []
|
||
for path in expected:
|
||
source = REPO_ROOT / path
|
||
if not source.is_file():
|
||
hash_errors.append(f"缺文件:{path}")
|
||
continue
|
||
raw = source.read_bytes()
|
||
entry = entry_by_path.get(path)
|
||
if entry is None:
|
||
hash_errors.append(f"缺 entry:{path}")
|
||
continue
|
||
if entry.get("size") != len(raw) or entry.get("sha256") != hashlib.sha256(raw).hexdigest():
|
||
hash_errors.append(path)
|
||
check("manifest entries 保存真实 size/hash", not hash_errors, ", ".join(hash_errors))
|
||
|
||
|
||
def test_task2_manifest_is_deterministic_and_detects_drift() -> None:
|
||
"""同一输入必须同字节，任一纳入源文件变化必须改变 entry 与 manifest hash。"""
|
||
source_registry = sample(REGISTRY_V1) if REGISTRY_V1.is_file() else {}
|
||
source_paths = expected_manifest_paths(source_registry) if source_registry else []
|
||
with tempfile.TemporaryDirectory(dir=REPO_ROOT) as directory:
|
||
root = Path(directory)
|
||
copy_manifest_inputs(root, source_paths)
|
||
# clean-archive 没有 Git 元数据，仍要按明确的临时输出命名排除候选文件。
|
||
for relative in (
|
||
"game-runtime/games/shanhai-xingji/src/generated-output.js",
|
||
"game-runtime/games/shanhai-xingji/assets/atlas/cache.generated.json",
|
||
):
|
||
temporary = root / relative
|
||
temporary.parent.mkdir(parents=True, exist_ok=True)
|
||
temporary.write_bytes(b"temporary output")
|
||
default_path = root / "default-worktree-manifest.json"
|
||
default = run_manifest_generator(root, default_path)
|
||
check(
|
||
"默认 worktree 模式拒绝无 Git 元数据归档",
|
||
default.returncode != 0 and "code=reference_git_metadata_unavailable" in default.stderr,
|
||
(default.stdout + default.stderr)[-2000:],
|
||
)
|
||
first_path = root / "first-manifest.json"
|
||
second_path = root / "second-manifest.json"
|
||
first = run_manifest_generator(root, first_path, "clean-archive")
|
||
second = run_manifest_generator(root, second_path, "clean-archive")
|
||
detail = (first.stdout + first.stderr + second.stdout + second.stderr)[-4000:]
|
||
check("显式 clean-archive manifest 生成器可执行", first.returncode == 0 and second.returncode == 0, detail)
|
||
if first.returncode != 0 or second.returncode != 0:
|
||
return
|
||
first_bytes = first_path.read_bytes()
|
||
second_bytes = second_path.read_bytes()
|
||
check("同一输入连续生成 manifest 字节一致", first_bytes == second_bytes)
|
||
check("clean-archive 重建结果与仓内 manifest 逐字一致", first_bytes == MANIFEST.read_bytes())
|
||
before = sample(first_path)
|
||
before_paths = {entry["path"] for entry in before["entries"]}
|
||
check(
|
||
"clean-archive 排除明确的临时输出命名",
|
||
not {
|
||
"game-runtime/games/shanhai-xingji/src/generated-output.js",
|
||
"game-runtime/games/shanhai-xingji/assets/atlas/cache.generated.json",
|
||
} & before_paths,
|
||
str(sorted(before_paths)),
|
||
)
|
||
before_hash = hashlib.sha256(first_bytes).hexdigest()
|
||
before_entry_hash = next(
|
||
entry["sha256"] for entry in before["entries"] if entry["path"].endswith("/entry.js")
|
||
)
|
||
entry_source = root / "game-runtime/games/shanhai-xingji/entry.js"
|
||
entry_source.write_bytes(entry_source.read_bytes() + b"\n// task-2 drift fixture")
|
||
drift_path = root / "drift-manifest.json"
|
||
drift = run_manifest_generator(root, drift_path, "clean-archive")
|
||
check("修改纳入源文件后生成器仍可执行", drift.returncode == 0, (drift.stdout + drift.stderr)[-2000:])
|
||
if drift.returncode != 0:
|
||
return
|
||
after = sample(drift_path)
|
||
after_hash = hashlib.sha256(drift_path.read_bytes()).hexdigest()
|
||
after_entry_hash = next(
|
||
entry["sha256"] for entry in after["entries"] if entry["path"].endswith("/entry.js")
|
||
)
|
||
check("纳入源文件漂移改变 entry hash", before_entry_hash != after_entry_hash)
|
||
check("纳入源文件漂移改变 manifest hash", before_hash != after_hash)
|
||
|
||
|
||
def test_task2_manifest_excludes_gitignored_candidates_in_worktree() -> None:
|
||
"""真实 worktree 必须按 Git ignore 语义和临时命名共同排除候选文件。"""
|
||
source_registry = sample(REGISTRY_V1) if REGISTRY_V1.is_file() else {}
|
||
source_paths = expected_manifest_paths(source_registry) if source_registry else []
|
||
with tempfile.TemporaryDirectory(dir=REPO_ROOT) as directory:
|
||
root = Path(directory)
|
||
copy_manifest_inputs(root, source_paths)
|
||
shutil.copyfile(REPO_ROOT / ".gitignore", root / ".gitignore")
|
||
with (root / ".gitignore").open("ab") as ignore_file:
|
||
ignore_file.write(
|
||
b"\n/game-runtime/games/shanhai-xingji/src/debug.js\n"
|
||
b"/game-runtime/games/shanhai-xingji/src/cache.js\n"
|
||
b"/game-runtime/games/shanhai-xingji/assets/atlas/generated.json\n"
|
||
)
|
||
initialized = subprocess.run(
|
||
["git", "init", "--quiet"],
|
||
cwd=root,
|
||
capture_output=True,
|
||
text=True,
|
||
check=False,
|
||
)
|
||
check("临时 worktree 可初始化 Git 元数据", initialized.returncode == 0, initialized.stderr[-1000:])
|
||
if initialized.returncode != 0:
|
||
return
|
||
|
||
# 这组 /2 判例复验的是迁移前已经入库的历史金标。仓根现已默认忽略所有具体游戏，
|
||
# 因此临时 worktree 必须显式复刻“批准范围已被 Git 跟踪”，新增候选仍保持未跟踪并受 ignore 约束。
|
||
tracked = subprocess.run(
|
||
["git", "add", "-f", "--", *source_paths],
|
||
cwd=root,
|
||
capture_output=True,
|
||
text=True,
|
||
check=False,
|
||
)
|
||
check("临时 worktree 复刻历史金标已跟踪状态", tracked.returncode == 0, tracked.stderr[-1000:])
|
||
if tracked.returncode != 0:
|
||
return
|
||
|
||
ignored_candidates = (
|
||
"game-runtime/games/shanhai-xingji/src/debug.js",
|
||
"game-runtime/games/shanhai-xingji/src/cache.js",
|
||
"game-runtime/games/shanhai-xingji/assets/atlas/generated.json",
|
||
)
|
||
temporary_candidates = (
|
||
"game-runtime/games/shanhai-xingji/src/entry.tmp.js",
|
||
"game-runtime/games/shanhai-xingji/assets/atlas/atlas.generated.json",
|
||
)
|
||
for relative in ignored_candidates + temporary_candidates:
|
||
candidate = root / relative
|
||
candidate.parent.mkdir(parents=True, exist_ok=True)
|
||
candidate.write_bytes(b"candidate output")
|
||
|
||
ignored_checks = [
|
||
subprocess.run(
|
||
["git", "check-ignore", "--quiet", "--", relative],
|
||
cwd=root,
|
||
capture_output=True,
|
||
check=False,
|
||
).returncode
|
||
for relative in ignored_candidates
|
||
]
|
||
check("临时 worktree 的 ignored 候选命中 Git 规则", ignored_checks == [0, 0, 0], str(ignored_checks))
|
||
|
||
output = root / "worktree-manifest.json"
|
||
result = run_manifest_generator(root, output)
|
||
check("真实 worktree manifest 生成器可执行", result.returncode == 0, (result.stdout + result.stderr)[-3000:])
|
||
if result.returncode != 0:
|
||
return
|
||
actual_paths = {entry["path"] for entry in json.loads(output.read_bytes()) ["entries"]}
|
||
check(
|
||
"真实 worktree 排除 gitignored 与临时输出候选",
|
||
not (set(ignored_candidates) | set(temporary_candidates)) & actual_paths,
|
||
str(sorted(actual_paths)),
|
||
)
|
||
check(
|
||
"真实 worktree 保持固定批准范围",
|
||
sorted(actual_paths, key=lambda value: value.encode("utf-8")) == source_paths,
|
||
str(sorted(actual_paths)),
|
||
)
|
||
|
||
|
||
def test_task2_v1_trust_anchor_rejects_pre_mutation() -> None:
|
||
"""批准 hash 必须拒绝测试启动前已经发生的 /1 registry 变异。"""
|
||
observed_hash = hashlib.sha256(REGISTRY_V1.read_bytes()).hexdigest() if REGISTRY_V1.is_file() else "missing"
|
||
check("/1 registry 原始字节匹配批准冻结 hash", observed_hash == EXPECTED_V1_REGISTRY_HASH, observed_hash)
|
||
with tempfile.TemporaryDirectory(dir=REPO_ROOT) as directory:
|
||
mutated = Path(directory) / "reference-asset-registry.initial.json"
|
||
mutated.write_bytes(REGISTRY_V1.read_bytes() + b" ")
|
||
mutated_hash = hashlib.sha256(mutated.read_bytes()).hexdigest()
|
||
check(
|
||
"预先变异的 /1 registry 被批准 hash 门拒绝",
|
||
not matches_v1_trust_anchor(mutated),
|
||
mutated_hash,
|
||
)
|
||
|
||
|
||
def test_task2_registry_policy_release_and_real_hashes() -> None:
|
||
"""Registry/2、policy、release 必须闭合真实 hash，且不能改变 /1 或 bundle。"""
|
||
required = (REGISTRY_V2, POLICY_INITIAL, RELEASE_INITIAL, MANIFEST)
|
||
missing = [str(path.relative_to(REPO_ROOT)) for path in required if not path.is_file()]
|
||
check("Task 2 跨文件数据均可读取", not missing, ", ".join(missing))
|
||
if missing:
|
||
return
|
||
|
||
registry_v1 = sample(REGISTRY_V1)
|
||
registry_v2 = sample(REGISTRY_V2)
|
||
policy = sample(POLICY_INITIAL)
|
||
release = sample(RELEASE_INITIAL)
|
||
manifest = sample(MANIFEST)
|
||
records = registry_v2.get("records", [])
|
||
by_id = {record.get("recordId"): record for record in records}
|
||
check("Registry/2 机械迁移全部 15 条记录", len(records) == 15 and len(by_id) == 15, str(len(records)))
|
||
check(
|
||
"Registry/2 所有记录均为 Record/2",
|
||
all(record.get("schemaVersion") == "ReferenceAssetRecord/2" for record in records),
|
||
)
|
||
migration_errors = []
|
||
for old in registry_v1.get("records", []):
|
||
new = by_id.get(old.get("recordId"))
|
||
if new is None:
|
||
migration_errors.append(f"缺记录:{old.get('recordId')}")
|
||
continue
|
||
for field, value in old.items():
|
||
if field == "schemaVersion":
|
||
continue
|
||
if old.get("recordId") == "gac-shanhai-xingji" and field == "consumerRef":
|
||
if new.get(field) != "generation-runtime@reference-assets/2":
|
||
migration_errors.append(f"active consumerRef:{new.get(field)}")
|
||
elif new.get(field) != value:
|
||
migration_errors.append(f"{old.get('recordId')}.{field}")
|
||
for field in ("artifactRef", "consumptionManifestRef", "consumptionManifestHash"):
|
||
if field not in new:
|
||
migration_errors.append(f"缺字段:{old.get('recordId')}.{field}")
|
||
elif old.get("recordId") != "gac-shanhai-xingji" and new.get(field) is not None:
|
||
migration_errors.append(f"非 active 非 null:{old.get('recordId')}.{field}")
|
||
check("Registry/2 保留 /1 历史字段并显式迁移 null", not migration_errors, ", ".join(migration_errors))
|
||
|
||
active = by_id.get("gac-shanhai-xingji", {})
|
||
manifest_hash = hashlib.sha256(MANIFEST.read_bytes()).hexdigest()
|
||
check(
|
||
"active 记录补齐真实双身份",
|
||
active.get("artifactRef") == "game-runtime/games/shanhai-xingji/dist/shanhai-bundle.js"
|
||
and active.get("consumptionManifestRef") == EXPECTED_MANIFEST_REF
|
||
and active.get("consumptionManifestHash") == manifest_hash,
|
||
json.dumps(active, ensure_ascii=False),
|
||
)
|
||
check(
|
||
"policy 固定为 survivor-gold frozen preflight",
|
||
policy == {
|
||
"schemaVersion": "ReferenceAssetConsumptionPolicy/1",
|
||
"policyId": "survivor-gold-v1",
|
||
"recordId": "gac-shanhai-xingji",
|
||
"role": "game_content_gold",
|
||
"consumerRef": "generation-runtime@reference-assets/2",
|
||
"route": "survivor-gold",
|
||
"autoSelect": False,
|
||
"mode": "frozen_preflight",
|
||
},
|
||
json.dumps(policy, ensure_ascii=False),
|
||
)
|
||
release_errors = []
|
||
if release.get("registryRef") != "contracts/play-loop/reference-asset-registry.v2.initial.json":
|
||
release_errors.append("registryRef")
|
||
if release.get("policyRef") != "contracts/play-loop/reference-asset-consumption-policy.initial.json":
|
||
release_errors.append("policyRef")
|
||
if release.get("registryHash") != hashlib.sha256(REGISTRY_V2.read_bytes()).hexdigest():
|
||
release_errors.append("registryHash")
|
||
if release.get("policyHash") != hashlib.sha256(POLICY_INITIAL.read_bytes()).hexdigest():
|
||
release_errors.append("policyHash")
|
||
if release.get("verifierVersion") != "reference-asset-verifier/1.0.0":
|
||
release_errors.append("verifierVersion")
|
||
if not release.get("trustedRootId") or "/" in release["trustedRootId"] or "\\" in release["trustedRootId"]:
|
||
release_errors.append("trustedRootId")
|
||
check("release 绑定 registry/policy 原始字节真实 hash", not release_errors, ", ".join(release_errors))
|
||
bundle = GAME_ROOT / "dist/shanhai-bundle.js"
|
||
check(
|
||
"当前 bundle hash 保持批准值",
|
||
bundle.is_file() and hashlib.sha256(bundle.read_bytes()).hexdigest() == EXPECTED_BUNDLE_HASH,
|
||
hashlib.sha256(bundle.read_bytes()).hexdigest() if bundle.is_file() else "missing",
|
||
)
|
||
check(
|
||
"/1 registry 原始字节在本任务内保持不变",
|
||
V1_BYTES_AT_TEST_START is not None and REGISTRY_V1.read_bytes() == V1_BYTES_AT_TEST_START,
|
||
hashlib.sha256(REGISTRY_V1.read_bytes()).hexdigest(),
|
||
)
|
||
|
||
|
||
def test_receipt_rejects_registry_hash_drift() -> None:
|
||
"""release 伪造 registryHash 时，即使 receipt 跟随声明也必须失败。"""
|
||
receipt = sample(HERE / "samples/reference-asset-verification-receipt/valid/01-matching-hashes.json")
|
||
release = sample(HERE / "samples/reference-asset-release/valid/01-trusted-release.json")
|
||
registry_ref = repo_ref(HERE / "samples/reference-asset-registry-v2/valid/01-trusted-release-registry.json")
|
||
policy_ref = repo_ref(HERE / "samples/reference-asset-consumption-policy/valid/01-survivor-gold-v1.json")
|
||
with tempfile.TemporaryDirectory(dir=HERE.parent.parent) as directory:
|
||
release_path = Path(directory) / "release-registry-drift.json"
|
||
release["registryRef"] = registry_ref
|
||
release["registryHash"] = "0" * 64
|
||
release["policyRef"] = policy_ref
|
||
release["policyHash"] = receipt["policyHash"]
|
||
write_json(release_path, release)
|
||
receipt["releaseRef"] = repo_ref(release_path)
|
||
receipt["expectedRegistryHash"] = "0" * 64
|
||
receipt["observedRegistryHash"] = "0" * 64
|
||
errors = errors_of("reference-asset-verification-receipt.schema.json", receipt)
|
||
check("receipt 拒绝 registry 原始字节 hash 漂移", bool(errors), "; ".join(errors))
|
||
|
||
|
||
def test_receipt_rejects_policy_hash_drift() -> None:
|
||
"""release 伪造 policyHash 时，即使 receipt 跟随声明也必须失败。"""
|
||
receipt = sample(HERE / "samples/reference-asset-verification-receipt/valid/01-matching-hashes.json")
|
||
release = sample(HERE / "samples/reference-asset-release/valid/01-trusted-release.json")
|
||
release["registryRef"] = repo_ref(
|
||
HERE / "samples/reference-asset-registry-v2/valid/01-trusted-release-registry.json",
|
||
)
|
||
release["registryHash"] = receipt["expectedRegistryHash"]
|
||
release["policyRef"] = repo_ref(
|
||
HERE / "samples/reference-asset-consumption-policy/valid/01-survivor-gold-v1.json",
|
||
)
|
||
with tempfile.TemporaryDirectory(dir=HERE.parent.parent) as directory:
|
||
release_path = Path(directory) / "release-policy-drift.json"
|
||
release["policyHash"] = "0" * 64
|
||
write_json(release_path, release)
|
||
receipt["releaseRef"] = repo_ref(release_path)
|
||
receipt["policyHash"] = "0" * 64
|
||
errors = errors_of("reference-asset-verification-receipt.schema.json", receipt)
|
||
check("receipt 拒绝 policy 原始字节 hash 漂移", bool(errors), "; ".join(errors))
|
||
|
||
|
||
def test_receipt_requires_typed_registry_and_active_record() -> None:
|
||
"""receipt 不得把 schema 冒充 registry，也不得消费非 active record。"""
|
||
base_receipt = sample(HERE / "samples/reference-asset-verification-receipt/valid/01-matching-hashes.json")
|
||
base_release = sample(HERE / "samples/reference-asset-release/valid/01-trusted-release.json")
|
||
policy_ref = repo_ref(HERE / "samples/reference-asset-consumption-policy/valid/01-survivor-gold-v1.json")
|
||
policy_hash = hashlib.sha256((HERE / "samples/reference-asset-consumption-policy/valid/01-survivor-gold-v1.json").read_bytes()).hexdigest()
|
||
registry_schema = HERE / "reference-asset-registry-v2.schema.json"
|
||
cases = []
|
||
cases.append(("registry schema", repo_ref(registry_schema), hashlib.sha256(registry_schema.read_bytes()).hexdigest()))
|
||
|
||
registry = sample(HERE / "samples/reference-asset-registry-v2/valid/01-trusted-release-registry.json")
|
||
registry["records"][0]["lifecycleStatus"] = "candidate"
|
||
cases.append(("non-active record", registry, None))
|
||
|
||
failures = []
|
||
with tempfile.TemporaryDirectory(dir=HERE.parent.parent) as directory:
|
||
for label, registry_value, declared_hash in cases:
|
||
release = dict(base_release)
|
||
release["policyRef"] = policy_ref
|
||
release["policyHash"] = policy_hash
|
||
receipt = dict(base_receipt)
|
||
if isinstance(registry_value, str):
|
||
release["registryRef"] = registry_value
|
||
release["registryHash"] = declared_hash
|
||
else:
|
||
registry_path = Path(directory) / f"{label.replace(' ', '-')}.json"
|
||
write_json(registry_path, registry_value)
|
||
release["registryRef"] = repo_ref(registry_path)
|
||
release["registryHash"] = hashlib.sha256(registry_path.read_bytes()).hexdigest()
|
||
release_path = Path(directory) / f"release-{label.replace(' ', '-')}.json"
|
||
write_json(release_path, release)
|
||
receipt["releaseRef"] = repo_ref(release_path)
|
||
receipt["expectedRegistryHash"] = release["registryHash"]
|
||
receipt["observedRegistryHash"] = release["registryHash"]
|
||
receipt["policyHash"] = policy_hash
|
||
errors = errors_of("reference-asset-verification-receipt.schema.json", receipt)
|
||
if not errors:
|
||
failures.append(label)
|
||
check("receipt 拒绝错误 registry 类型和非 active record", not failures, ", ".join(failures))
|
||
|
||
|
||
def test_provenance_reads_and_validates_receipt_files() -> None:
|
||
"""provenance 必须读取 receipt 原始文件并闭合 schema、hash 与声明身份。"""
|
||
provenance = sample(HERE / "samples/acceptance-provenance-v4/valid/01-with-verification-receipt.json")
|
||
receipt_path = HERE / "samples/reference-asset-verification-receipt/valid/01-matching-hashes.json"
|
||
receipt_ref = repo_ref(receipt_path)
|
||
receipt_hash = hashlib.sha256(receipt_path.read_bytes()).hexdigest()
|
||
cases = []
|
||
|
||
missing = json.loads(json.dumps(provenance))
|
||
missing["referenceAssetVerificationReceipts"][0] = {"ref": "contracts/play-loop/missing-receipt.json", "hash": "0" * 64}
|
||
cases.append(("missing receipt", missing))
|
||
|
||
wrong_type = json.loads(json.dumps(provenance))
|
||
schema_path = HERE / "reference-asset-verification-receipt.schema.json"
|
||
wrong_type["referenceAssetVerificationReceipts"][0] = {
|
||
"ref": repo_ref(schema_path),
|
||
"hash": hashlib.sha256(schema_path.read_bytes()).hexdigest(),
|
||
}
|
||
cases.append(("non-receipt JSON", wrong_type))
|
||
|
||
hash_drift = json.loads(json.dumps(provenance))
|
||
hash_drift["referenceAssetVerificationReceipts"][0] = {"ref": receipt_ref, "hash": "0" * 64}
|
||
cases.append(("receipt hash drift", hash_drift))
|
||
|
||
identity_drift = json.loads(json.dumps(provenance))
|
||
identity_drift["referenceAssetVerificationReceipts"][0] = {"ref": receipt_ref, "hash": receipt_hash}
|
||
identity_drift["referenceAssetRecordIds"] = ["other-record"]
|
||
identity_drift["consumedReferenceAssets"][0]["recordId"] = "other-record"
|
||
cases.append(("receipt identity drift", identity_drift))
|
||
|
||
failures = []
|
||
for label, candidate in cases:
|
||
errors = errors_of("acceptance-provenance-v4.schema.json", candidate)
|
||
if not errors:
|
||
failures.append(label)
|
||
check("provenance 拒绝 receipt 文件闭包负例", not failures, ", ".join(failures))
|
||
|
||
|
||
def test_provenance_rejects_non_regular_and_symlink_receipts() -> None:
|
||
"""provenance receipt ref 的末级对象必须是普通文件且不能是 symlink。"""
|
||
provenance = sample(HERE / "samples/acceptance-provenance-v4/valid/01-with-verification-receipt.json")
|
||
with tempfile.TemporaryDirectory(dir=HERE.parent.parent) as directory:
|
||
directory_path = Path(directory)
|
||
non_regular = json.loads(json.dumps(provenance))
|
||
non_regular["referenceAssetVerificationReceipts"][0] = {
|
||
"ref": repo_ref(directory_path),
|
||
"hash": "0" * 64,
|
||
}
|
||
symlink_path = directory_path / "receipt-link.json"
|
||
os.symlink(
|
||
HERE / "samples/reference-asset-verification-receipt/valid/01-matching-hashes.json",
|
||
symlink_path,
|
||
)
|
||
symlink = json.loads(json.dumps(provenance))
|
||
symlink["referenceAssetVerificationReceipts"][0] = {
|
||
"ref": repo_ref(symlink_path),
|
||
"hash": "0" * 64,
|
||
}
|
||
directory_errors = errors_of("acceptance-provenance-v4.schema.json", non_regular)
|
||
symlink_errors = errors_of("acceptance-provenance-v4.schema.json", symlink)
|
||
check("provenance 拒绝非普通 receipt 文件", bool(directory_errors), "; ".join(directory_errors))
|
||
check(
|
||
"provenance 拒绝末级 symlink receipt",
|
||
any("末级引用不得是 symlink" in error for error in symlink_errors),
|
||
"; ".join(symlink_errors),
|
||
)
|
||
|
||
|
||
def canonical_manifest_bytes(instance: dict) -> bytes:
|
||
"""按批准设计的项目自有 canonical JSON 字节规则生成比较值。"""
|
||
return json.dumps(
|
||
instance,
|
||
ensure_ascii=False,
|
||
sort_keys=True,
|
||
separators=(",", ":"),
|
||
allow_nan=False,
|
||
).encode("utf-8")
|
||
|
||
|
||
def test_schema_samples() -> None:
|
||
"""每条新契约的 valid 必须通过、invalid 必须被拦截。"""
|
||
for contract, schema_file in CONTRACTS:
|
||
schema_path = HERE / schema_file
|
||
if not schema_path.is_file():
|
||
check(f"{contract} schema 已创建", False, f"文件不存在：{schema_path}")
|
||
continue
|
||
check(f"{contract} schema 可加载", True)
|
||
schema = V._load(schema_path)
|
||
valid_dir = HERE / "samples" / contract / "valid"
|
||
invalid_dir = HERE / "samples" / contract / "invalid"
|
||
valid_files = sorted(valid_dir.glob("*.json"))
|
||
invalid_files = sorted(invalid_dir.glob("*.json"))
|
||
check(f"{contract} 有正负样本", bool(valid_files) and bool(invalid_files),
|
||
f"valid={len(valid_files)} invalid={len(invalid_files)}")
|
||
for path in valid_files:
|
||
errors = errors_of_path(schema_file, path)
|
||
check(f"{contract} 正样本通过：{path.name}", not errors, "; ".join(errors))
|
||
for path in invalid_files:
|
||
errors = errors_of_path(schema_file, path)
|
||
check(f"{contract} 负样本被拦：{path.name}", bool(errors), "负样本竟然通过")
|
||
|
||
|
||
def test_manifest_canonical_bytes() -> None:
|
||
"""manifest 正样本必须逐字等于 canonical JSON，原始字节变体必须被拦截。"""
|
||
schema_file = "reference-asset-consumption-manifest.schema.json"
|
||
valid_dir = HERE / "samples/reference-asset-consumption-manifest/valid"
|
||
for path in sorted(valid_dir.glob("*.json")):
|
||
instance = V._load_sample_instance(path.resolve())
|
||
expected = canonical_manifest_bytes(instance)
|
||
actual = path.read_bytes()
|
||
check(
|
||
f"manifest 正样本原始字节 canonical：{path.name}",
|
||
actual == expected,
|
||
f"actual_sha={hashlib.sha256(actual).hexdigest()} "
|
||
f"expected_sha={hashlib.sha256(expected).hexdigest()}",
|
||
)
|
||
|
||
invalid_dir = HERE / "samples/reference-asset-consumption-manifest/invalid"
|
||
for name in (
|
||
"09-bom.json",
|
||
"10-trailing-newline.json",
|
||
"11-key-order.json",
|
||
"12-non-ascii-escape.json",
|
||
):
|
||
path = invalid_dir / name
|
||
errors = errors_of_path(schema_file, path)
|
||
check(f"manifest 原始字节变体被拦：{name}", bool(errors), "原始非 canonical 字节竟然通过")
|
||
|
||
|
||
def test_validate_suite_registration() -> None:
|
||
"""validate.py suite 必须真正登记七条新契约，而非只在测试中手工调用。"""
|
||
result = subprocess.run(
|
||
[sys.executable, str(HERE / "validate.py"), "--suite", str(HERE / "samples")],
|
||
cwd=HERE,
|
||
capture_output=True,
|
||
text=True,
|
||
check=False,
|
||
)
|
||
output = result.stdout + result.stderr
|
||
check("validate.py suite 可执行", result.returncode in (0, 1),
|
||
f"returncode={result.returncode}\n{output[-4000:]}")
|
||
for contract, _ in CONTRACTS:
|
||
heading = f"══ {contract}（"
|
||
start = output.find(heading)
|
||
next_heading = output.find("\n══ ", start + 1) if start >= 0 else -1
|
||
block = output[start:next_heading if next_heading >= 0 else None]
|
||
registered = start >= 0 and bool(
|
||
re.search(r"小计：正样本 (?P<v>\d+)/(?P=v) 过，负样本 (?P<i>\d+)/(?P=i) 拦", block)
|
||
)
|
||
check(f"validate.py suite 已登记且分段自洽：{contract}", registered,
|
||
block[-2000:] if block else output[-2000:])
|
||
|
||
|
||
def test_version_isolation() -> None:
|
||
"""旧 /1 仍按旧 schema 通过，新 /2 不得被旧 schema 接受。"""
|
||
old_record_path = HERE / "samples/reference-asset-record/valid/03-active-game-content-gold.json"
|
||
old_record_schema = V._load(HERE / "reference-asset-record.schema.json")
|
||
old_record = sample(old_record_path)
|
||
old_errors = V.validate(old_record_schema, old_record, old_record_schema)
|
||
check("旧 ReferenceAssetRecord/1 正样本仍按 /1 通过", not old_errors, "; ".join(old_errors))
|
||
|
||
new_record_path = HERE / "samples/reference-asset-record-v2/valid/01-active-trusted-gold.json"
|
||
new_record = sample(new_record_path)
|
||
old_schema_errors = V.validate(old_record_schema, new_record, old_record_schema)
|
||
check("新 ReferenceAssetRecord/2 不被旧 /1 schema 接受", bool(old_schema_errors),
|
||
"新 /2 样本竟被旧 schema 接受")
|
||
|
||
v2_registry_invalid = HERE / "samples/reference-asset-registry-v2/invalid/04-v1-record-cannot-enter.json"
|
||
v2_errors = errors_of("reference-asset-registry-v2.schema.json", sample(v2_registry_invalid))
|
||
check("ReferenceAssetRegistry/2 拒绝嵌入 ReferenceAssetRecord/1", bool(v2_errors),
|
||
"/1 record 竟进入 /2 registry")
|
||
|
||
|
||
def main() -> int:
|
||
"""运行所有断言并以非零状态报告失败。"""
|
||
test_schema_samples()
|
||
test_manifest_canonical_bytes()
|
||
test_receipt_rejects_registry_hash_drift()
|
||
test_receipt_rejects_policy_hash_drift()
|
||
test_receipt_requires_typed_registry_and_active_record()
|
||
test_provenance_reads_and_validates_receipt_files()
|
||
test_provenance_rejects_non_regular_and_symlink_receipts()
|
||
test_validate_suite_registration()
|
||
# schema 尚未创建时跳过会再次抛 FileNotFoundError 的隔离断言，保持 RED 是可读失败。
|
||
if all((HERE / schema_file).is_file() for _, schema_file in CONTRACTS):
|
||
test_version_isolation()
|
||
test_task2_v1_trust_anchor_rejects_pre_mutation()
|
||
test_task2_artifact_files_and_manifest_scope()
|
||
test_task2_manifest_is_deterministic_and_detects_drift()
|
||
test_task2_manifest_excludes_gitignored_candidates_in_worktree()
|
||
test_task2_registry_policy_release_and_real_hashes()
|
||
print(f"\n══ 汇总：{len(_PASS)} 过 / {len(_FAIL)} 败 ══")
|
||
if _FAIL:
|
||
for name, detail in _FAIL:
|
||
print(f" ✗ {name}: {detail.splitlines()[0] if detail else ''}")
|
||
return 1
|
||
print("全部通过：/2 strict schema、suite 登记和 /1 版本隔离均符合预期。")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|