feat(release): bind scene source legal review
This commit is contained in:
@@ -0,0 +1,461 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Prepare and verify the 30-scene, 43-reference legal review request.
|
||||
|
||||
The tool never decides whether a source is current or applicable. It binds the
|
||||
exact source records to a candidate and accepts a completed review only when a
|
||||
trusted, qualified human reviewer signed every occurrence.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import base64
|
||||
import hashlib
|
||||
import json
|
||||
import re
|
||||
import subprocess
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from pathlib import Path
|
||||
from typing import Any, Mapping
|
||||
from urllib.parse import urlparse
|
||||
|
||||
from cryptography.exceptions import InvalidSignature
|
||||
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
|
||||
|
||||
|
||||
REQUEST_SCHEMA = "yuqei.scene-source-legal-review-request.v1"
|
||||
DECISION_SCHEMA = "yuqei.scene-source-legal-review-decision.v1"
|
||||
KEYRING_SCHEMA = "yuqei.scene-source-legal-review-keyring.v1"
|
||||
SIGNATURE_DOMAIN = b"YUQEI_SCENE_SOURCE_LEGAL_REVIEW_V1\0"
|
||||
PACK_PATHS = (
|
||||
"yuqei-contract-platform/services/contract-api/src/yuqei_contract_api/"
|
||||
"template_assets/v1_contract_templates/scene-packs/batch-a.json",
|
||||
"yuqei-contract-platform/services/contract-api/src/yuqei_contract_api/"
|
||||
"template_assets/v1_contract_templates/scene-packs/batch-b.json",
|
||||
)
|
||||
EXPECTED_SCENES = 30
|
||||
EXPECTED_OCCURRENCES = 43
|
||||
EXPECTED_RECORDS = 8
|
||||
EXPECTED_URLS = 7
|
||||
COMMIT_RE = re.compile(r"[0-9a-f]{40}")
|
||||
SHA256_RE = re.compile(r"[0-9a-f]{64}")
|
||||
KEY_ID_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,127}")
|
||||
PLACEHOLDERS = {"", "-", "n/a", "none", "pending", "unknown", "待确认", "待复核"}
|
||||
MAX_REVIEW_WINDOW = timedelta(days=7)
|
||||
|
||||
|
||||
class ReviewPackageError(RuntimeError):
|
||||
"""The review request or signed decision is unsafe or incomplete."""
|
||||
|
||||
|
||||
def canonical_json(value: Any) -> bytes:
|
||||
try:
|
||||
return json.dumps(
|
||||
value,
|
||||
ensure_ascii=False,
|
||||
sort_keys=True,
|
||||
separators=(",", ":"),
|
||||
allow_nan=False,
|
||||
).encode("utf-8")
|
||||
except (TypeError, ValueError) as exc:
|
||||
raise ReviewPackageError(f"value is not canonical JSON: {exc}") from exc
|
||||
|
||||
|
||||
def sha256_bytes(value: bytes) -> str:
|
||||
return hashlib.sha256(value).hexdigest()
|
||||
|
||||
|
||||
def sha256_json(value: Any) -> str:
|
||||
return sha256_bytes(canonical_json(value))
|
||||
|
||||
|
||||
def _reject_duplicate_pairs(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
|
||||
result: dict[str, Any] = {}
|
||||
for key, value in pairs:
|
||||
if key in result:
|
||||
raise ReviewPackageError(f"duplicate JSON key is forbidden: {key}")
|
||||
result[key] = value
|
||||
return result
|
||||
|
||||
|
||||
def load_json(path: Path) -> Any:
|
||||
try:
|
||||
return json.loads(
|
||||
path.read_text(encoding="utf-8"),
|
||||
object_pairs_hook=_reject_duplicate_pairs,
|
||||
parse_constant=lambda value: (_ for _ in ()).throw(
|
||||
ReviewPackageError(f"non-finite JSON value is forbidden: {value}")
|
||||
),
|
||||
)
|
||||
except (OSError, UnicodeError, json.JSONDecodeError) as exc:
|
||||
raise ReviewPackageError(f"cannot read JSON {path}: {exc}") from exc
|
||||
|
||||
|
||||
def _required_text(value: Any, label: str, *, maximum: int = 4096) -> str:
|
||||
if not isinstance(value, str) or not value.strip() or len(value) > maximum:
|
||||
raise ReviewPackageError(f"{label} must be a non-empty string")
|
||||
return value.strip()
|
||||
|
||||
|
||||
def _parse_timestamp(value: Any, label: str) -> datetime:
|
||||
text = _required_text(value, label, maximum=64)
|
||||
try:
|
||||
parsed = datetime.fromisoformat(text.replace("Z", "+00:00"))
|
||||
except ValueError as exc:
|
||||
raise ReviewPackageError(f"{label} must be an RFC3339 timestamp") from exc
|
||||
if parsed.tzinfo is None or parsed.utcoffset() is None:
|
||||
raise ReviewPackageError(f"{label} must include a timezone")
|
||||
return parsed.astimezone(timezone.utc)
|
||||
|
||||
|
||||
def _safe_official_url(value: Any, label: str) -> str:
|
||||
text = _required_text(value, label, maximum=2048)
|
||||
parsed = urlparse(text)
|
||||
host = (parsed.hostname or "").lower()
|
||||
if parsed.scheme != "https" or not host or parsed.username or parsed.password:
|
||||
raise ReviewPackageError(f"{label} must be a credential-free HTTPS URL")
|
||||
if not (host == "gov.cn" or host.endswith(".gov.cn")):
|
||||
raise ReviewPackageError(f"{label} must use an official gov.cn host")
|
||||
return text
|
||||
|
||||
|
||||
def _source_identity(source: Mapping[str, Any]) -> dict[str, Any]:
|
||||
required = (
|
||||
"title",
|
||||
"url",
|
||||
"publication_date",
|
||||
"enactment_date",
|
||||
"accessed_on",
|
||||
"scope",
|
||||
"current_effectiveness_verified",
|
||||
)
|
||||
if set(source) != set(required):
|
||||
raise ReviewPackageError("scene source record fields do not match the frozen schema")
|
||||
if source["current_effectiveness_verified"] is not False:
|
||||
raise ReviewPackageError("review preparation requires an explicitly unverified source record")
|
||||
_safe_official_url(source["url"], "scene source URL")
|
||||
for field in required[:-1]:
|
||||
_required_text(source[field], f"scene source {field}")
|
||||
return {field: source[field] for field in required}
|
||||
|
||||
|
||||
def verify_candidate_sources(repo: Path, candidate_commit: str) -> None:
|
||||
"""Require every reviewed source byte to come from the checked-out commit."""
|
||||
|
||||
try:
|
||||
head = subprocess.run(
|
||||
["git", "rev-parse", "--verify", "HEAD^{commit}"],
|
||||
cwd=repo,
|
||||
check=True,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
encoding="utf-8",
|
||||
).stdout.strip()
|
||||
except (OSError, subprocess.CalledProcessError) as exc:
|
||||
raise ReviewPackageError("review repository is not a readable Git checkout") from exc
|
||||
if head != candidate_commit:
|
||||
raise ReviewPackageError("candidate commit is not the checked-out HEAD")
|
||||
for relative in PACK_PATHS:
|
||||
try:
|
||||
subprocess.run(
|
||||
["git", "cat-file", "-e", f"{candidate_commit}:{relative}"],
|
||||
cwd=repo,
|
||||
check=True,
|
||||
capture_output=True,
|
||||
)
|
||||
compared = subprocess.run(
|
||||
["git", "diff", "--quiet", candidate_commit, "--", relative],
|
||||
cwd=repo,
|
||||
check=False,
|
||||
capture_output=True,
|
||||
)
|
||||
except OSError as exc:
|
||||
raise ReviewPackageError(f"candidate source is not readable from Git: {relative}") from exc
|
||||
if compared.returncode == 1:
|
||||
raise ReviewPackageError(f"candidate source has uncommitted drift: {relative}")
|
||||
if compared.returncode != 0:
|
||||
raise ReviewPackageError(f"candidate source comparison failed: {relative}")
|
||||
|
||||
|
||||
def collect_occurrences(
|
||||
repo: Path, *, candidate_commit: str | None = None
|
||||
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
|
||||
occurrences: list[dict[str, Any]] = []
|
||||
files: list[dict[str, Any]] = []
|
||||
scene_ids: set[str] = set()
|
||||
template_ids: set[str] = set()
|
||||
record_hashes: set[str] = set()
|
||||
urls: set[str] = set()
|
||||
for relative in PACK_PATHS:
|
||||
path = repo / relative
|
||||
if candidate_commit is None:
|
||||
raw = path.read_bytes()
|
||||
else:
|
||||
try:
|
||||
raw = subprocess.run(
|
||||
["git", "show", f"{candidate_commit}:{relative}"],
|
||||
cwd=repo,
|
||||
check=True,
|
||||
capture_output=True,
|
||||
).stdout
|
||||
except (OSError, subprocess.CalledProcessError) as exc:
|
||||
raise ReviewPackageError(f"candidate source is not readable from Git: {relative}") from exc
|
||||
payload = load_json(path)
|
||||
if not isinstance(payload, dict):
|
||||
raise ReviewPackageError(f"scene pack must be an object: {relative}")
|
||||
sources = payload.get("source_references")
|
||||
scenes = payload.get("scenes")
|
||||
if not isinstance(sources, dict) or not isinstance(scenes, list):
|
||||
raise ReviewPackageError(f"scene pack sources/scenes are invalid: {relative}")
|
||||
normalized_sources: dict[str, dict[str, Any]] = {}
|
||||
for source_key, source in sources.items():
|
||||
if not isinstance(source_key, str) or not isinstance(source, dict):
|
||||
raise ReviewPackageError(f"scene pack source is invalid: {relative}")
|
||||
normalized_sources[source_key] = _source_identity(source)
|
||||
files.append({"path": relative, "raw_sha256": sha256_bytes(raw)})
|
||||
for scene in scenes:
|
||||
if not isinstance(scene, dict):
|
||||
raise ReviewPackageError(f"scene row is invalid: {relative}")
|
||||
scene_id = _required_text(scene.get("id"), "scene id", maximum=128)
|
||||
template_id = _required_text(scene.get("template_id"), "template id", maximum=191)
|
||||
source_keys = scene.get("sources")
|
||||
if not isinstance(source_keys, list) or not source_keys:
|
||||
raise ReviewPackageError(f"scene {scene_id} has no source references")
|
||||
if scene_id in scene_ids or template_id in template_ids:
|
||||
raise ReviewPackageError(f"duplicate scene/template identity: {scene_id}/{template_id}")
|
||||
scene_ids.add(scene_id)
|
||||
template_ids.add(template_id)
|
||||
for source_key in source_keys:
|
||||
if source_key not in normalized_sources:
|
||||
raise ReviewPackageError(f"scene {scene_id} references unknown source {source_key}")
|
||||
source = normalized_sources[source_key]
|
||||
record_sha = sha256_json(source)
|
||||
record_hashes.add(record_sha)
|
||||
urls.add(source["url"])
|
||||
occurrences.append(
|
||||
{
|
||||
"occurrence_id": f"{scene_id}:{template_id}:{source_key}",
|
||||
"scene_id": scene_id,
|
||||
"template_id": template_id,
|
||||
"source_key": source_key,
|
||||
"source_pack_path": relative,
|
||||
"source_record_sha256": record_sha,
|
||||
"source": source,
|
||||
"required_human_decisions": [
|
||||
"current_effectiveness_verified",
|
||||
"transaction_applicability_verified",
|
||||
],
|
||||
}
|
||||
)
|
||||
occurrences.sort(key=lambda item: item["occurrence_id"])
|
||||
files.sort(key=lambda item: item["path"])
|
||||
actual = (len(scene_ids), len(occurrences), len(record_hashes), len(urls))
|
||||
expected = (EXPECTED_SCENES, EXPECTED_OCCURRENCES, EXPECTED_RECORDS, EXPECTED_URLS)
|
||||
if actual != expected:
|
||||
raise ReviewPackageError(f"scene source baseline drift: expected {expected}, got {actual}")
|
||||
return occurrences, files
|
||||
|
||||
|
||||
def build_request(repo: Path, *, candidate_commit: str, created_at: str) -> dict[str, Any]:
|
||||
if not COMMIT_RE.fullmatch(candidate_commit):
|
||||
raise ReviewPackageError("candidate commit must be a 40-character lowercase Git SHA")
|
||||
_parse_timestamp(created_at, "created_at")
|
||||
verify_candidate_sources(repo, candidate_commit)
|
||||
occurrences, files = collect_occurrences(repo, candidate_commit=candidate_commit)
|
||||
request: dict[str, Any] = {
|
||||
"schema_version": REQUEST_SCHEMA,
|
||||
"status": "pending_qualified_human_review",
|
||||
"candidate_commit": candidate_commit,
|
||||
"created_at": created_at,
|
||||
"scope": {
|
||||
"scene_count": EXPECTED_SCENES,
|
||||
"reference_occurrence_count": EXPECTED_OCCURRENCES,
|
||||
"distinct_source_record_count": EXPECTED_RECORDS,
|
||||
"distinct_url_count": EXPECTED_URLS,
|
||||
},
|
||||
"source_files": files,
|
||||
"occurrences": occurrences,
|
||||
"guardrails": {
|
||||
"not_legal_approval": True,
|
||||
"authorizes_publication": False,
|
||||
"ai_must_not_supply_human_decisions": True,
|
||||
"all_occurrences_require_qualified_human_review": True,
|
||||
},
|
||||
}
|
||||
request["request_sha256"] = sha256_json(request)
|
||||
return request
|
||||
|
||||
|
||||
def verify_request(request: Any, repo: Path) -> dict[str, Any]:
|
||||
if not isinstance(request, dict) or request.get("schema_version") != REQUEST_SCHEMA:
|
||||
raise ReviewPackageError("review request schema is invalid")
|
||||
embedded = request.get("request_sha256")
|
||||
if not isinstance(embedded, str) or not SHA256_RE.fullmatch(embedded):
|
||||
raise ReviewPackageError("review request SHA-256 is invalid")
|
||||
projection = {key: value for key, value in request.items() if key != "request_sha256"}
|
||||
if sha256_json(projection) != embedded:
|
||||
raise ReviewPackageError("review request embedded hash drift")
|
||||
candidate = request.get("candidate_commit")
|
||||
if not isinstance(candidate, str) or not COMMIT_RE.fullmatch(candidate):
|
||||
raise ReviewPackageError("review request candidate is invalid")
|
||||
rebuilt = build_request(repo, candidate_commit=candidate, created_at=request["created_at"])
|
||||
if rebuilt != request:
|
||||
raise ReviewPackageError("review request differs from current candidate source records")
|
||||
return request
|
||||
|
||||
|
||||
def _trusted_key(keyring: Any, key_id: str, checked_at: datetime) -> dict[str, Any]:
|
||||
if not isinstance(keyring, dict) or keyring.get("schema_version") != KEYRING_SCHEMA:
|
||||
raise ReviewPackageError("legal reviewer keyring schema is invalid")
|
||||
keys = keyring.get("keys")
|
||||
if not isinstance(keys, list):
|
||||
raise ReviewPackageError("legal reviewer keyring keys are invalid")
|
||||
matches = [item for item in keys if isinstance(item, dict) and item.get("key_id") == key_id]
|
||||
if len(matches) != 1:
|
||||
raise ReviewPackageError("review signature key is not uniquely trusted")
|
||||
key = matches[0]
|
||||
if key.get("status") != "trusted" or key.get("revoked_at") is not None:
|
||||
raise ReviewPackageError("review signature key is not active")
|
||||
valid_from = _parse_timestamp(key.get("valid_from"), "review key valid_from")
|
||||
valid_until = _parse_timestamp(key.get("valid_until"), "review key valid_until")
|
||||
if not valid_from <= checked_at <= valid_until:
|
||||
raise ReviewPackageError("review signature key is outside its validity window")
|
||||
return key
|
||||
|
||||
|
||||
def verify_signed_decision(
|
||||
request: dict[str, Any], decision: Any, keyring: Any, *, checked_at: str
|
||||
) -> dict[str, Any]:
|
||||
now = _parse_timestamp(checked_at, "checked_at")
|
||||
if not isinstance(decision, dict) or decision.get("schema_version") != DECISION_SCHEMA:
|
||||
raise ReviewPackageError("legal review decision schema is invalid")
|
||||
if decision.get("request_sha256") != request["request_sha256"]:
|
||||
raise ReviewPackageError("legal review decision is not bound to the exact request")
|
||||
if decision.get("candidate_commit") != request["candidate_commit"]:
|
||||
raise ReviewPackageError("legal review decision candidate does not match")
|
||||
reviewed_at = _parse_timestamp(decision.get("reviewed_at"), "reviewed_at")
|
||||
expires_at = _parse_timestamp(decision.get("expires_at"), "expires_at")
|
||||
if not reviewed_at <= now <= expires_at:
|
||||
raise ReviewPackageError("legal review decision is not currently valid")
|
||||
if expires_at - reviewed_at <= timedelta(0) or expires_at - reviewed_at > MAX_REVIEW_WINDOW:
|
||||
raise ReviewPackageError("legal review validity must be greater than zero and at most seven days")
|
||||
reviewer = decision.get("reviewer")
|
||||
if not isinstance(reviewer, dict):
|
||||
raise ReviewPackageError("legal reviewer identity is missing")
|
||||
for field in ("name", "organization", "qualification", "credential_reference"):
|
||||
value = _required_text(reviewer.get(field), f"reviewer {field}", maximum=512)
|
||||
if value.casefold() in PLACEHOLDERS:
|
||||
raise ReviewPackageError(f"reviewer {field} is a placeholder")
|
||||
expected = {item["occurrence_id"]: item for item in request["occurrences"]}
|
||||
items = decision.get("items")
|
||||
if not isinstance(items, list) or len(items) != EXPECTED_OCCURRENCES:
|
||||
raise ReviewPackageError("legal review decision must contain exactly 43 items")
|
||||
actual: dict[str, dict[str, Any]] = {}
|
||||
for item in items:
|
||||
if not isinstance(item, dict):
|
||||
raise ReviewPackageError("legal review item must be an object")
|
||||
occurrence_id = _required_text(item.get("occurrence_id"), "occurrence_id", maximum=512)
|
||||
if occurrence_id in actual or occurrence_id not in expected:
|
||||
raise ReviewPackageError(f"unexpected or duplicate legal review item: {occurrence_id}")
|
||||
if item.get("source_record_sha256") != expected[occurrence_id]["source_record_sha256"]:
|
||||
raise ReviewPackageError(f"source record hash mismatch: {occurrence_id}")
|
||||
if item.get("current_effectiveness_verified") is not True:
|
||||
raise ReviewPackageError(f"current effectiveness is not approved: {occurrence_id}")
|
||||
if item.get("transaction_applicability_verified") is not True:
|
||||
raise ReviewPackageError(f"transaction applicability is not approved: {occurrence_id}")
|
||||
_parse_timestamp(item.get("checked_at"), f"{occurrence_id} checked_at")
|
||||
_safe_official_url(item.get("authoritative_status_url"), f"{occurrence_id} status URL")
|
||||
rationale = _required_text(item.get("rationale"), f"{occurrence_id} rationale", maximum=4096)
|
||||
if len(rationale) < 20 or rationale.casefold() in PLACEHOLDERS:
|
||||
raise ReviewPackageError(f"legal review rationale is insufficient: {occurrence_id}")
|
||||
actual[occurrence_id] = item
|
||||
if set(actual) != set(expected):
|
||||
raise ReviewPackageError("legal review decision item identities are incomplete")
|
||||
signature = decision.get("signature")
|
||||
if not isinstance(signature, dict) or signature.get("algorithm") != "Ed25519":
|
||||
raise ReviewPackageError("a trusted Ed25519 legal review signature is required")
|
||||
key_id = _required_text(signature.get("key_id"), "signature key_id", maximum=128)
|
||||
if not KEY_ID_RE.fullmatch(key_id):
|
||||
raise ReviewPackageError("signature key_id is invalid")
|
||||
key = _trusted_key(keyring, key_id, reviewed_at)
|
||||
for decision_field, key_field in (
|
||||
("name", "reviewer_name"),
|
||||
("organization", "reviewer_organization"),
|
||||
("qualification", "reviewer_qualification"),
|
||||
("credential_reference", "credential_reference"),
|
||||
):
|
||||
if reviewer[decision_field] != key.get(key_field):
|
||||
raise ReviewPackageError(f"reviewer {decision_field} does not match the trusted key")
|
||||
try:
|
||||
public_raw = base64.b64decode(
|
||||
_required_text(key.get("public_key_raw_base64"), "review public key"), validate=True
|
||||
)
|
||||
signature_raw = base64.b64decode(
|
||||
_required_text(signature.get("value_base64"), "review signature"), validate=True
|
||||
)
|
||||
public_key = Ed25519PublicKey.from_public_bytes(public_raw)
|
||||
signed_payload = {field: value for field, value in decision.items() if field != "signature"}
|
||||
public_key.verify(signature_raw, SIGNATURE_DOMAIN + canonical_json(signed_payload))
|
||||
except (ValueError, InvalidSignature) as exc:
|
||||
raise ReviewPackageError("legal review signature verification failed") from exc
|
||||
return {
|
||||
"status": "verified_qualified_human_review",
|
||||
"candidate_commit": request["candidate_commit"],
|
||||
"request_sha256": request["request_sha256"],
|
||||
"reviewer_name": reviewer["name"],
|
||||
"reviewed_occurrence_count": len(actual),
|
||||
"all_current_effectiveness_verified": True,
|
||||
"all_transaction_applicability_verified": True,
|
||||
"expires_at": decision["expires_at"],
|
||||
"authorizes_publication_by_itself": False,
|
||||
}
|
||||
|
||||
|
||||
def _write_json(path: Path, payload: Any) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
temporary = path.with_name(f".{path.name}.tmp")
|
||||
temporary.write_bytes(json.dumps(payload, ensure_ascii=False, indent=2).encode("utf-8") + b"\n")
|
||||
temporary.replace(path)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
subparsers = parser.add_subparsers(dest="command", required=True)
|
||||
prepare = subparsers.add_parser("prepare")
|
||||
prepare.add_argument("--repo", type=Path, required=True)
|
||||
prepare.add_argument("--candidate-commit", required=True)
|
||||
prepare.add_argument("--created-at", required=True)
|
||||
prepare.add_argument("--output", type=Path, required=True)
|
||||
validate = subparsers.add_parser("validate")
|
||||
validate.add_argument("--repo", type=Path, required=True)
|
||||
validate.add_argument("--request", type=Path, required=True)
|
||||
validate.add_argument("--decision", type=Path, required=True)
|
||||
validate.add_argument("--trusted-keyring", type=Path, required=True)
|
||||
validate.add_argument("--checked-at", required=True)
|
||||
args = parser.parse_args()
|
||||
try:
|
||||
if args.command == "prepare":
|
||||
payload = build_request(
|
||||
args.repo.resolve(),
|
||||
candidate_commit=args.candidate_commit,
|
||||
created_at=args.created_at,
|
||||
)
|
||||
_write_json(args.output.resolve(), payload)
|
||||
print(json.dumps({"status": "prepared", "request_sha256": payload["request_sha256"]}))
|
||||
return 0
|
||||
request = verify_request(load_json(args.request), args.repo.resolve())
|
||||
result = verify_signed_decision(
|
||||
request,
|
||||
load_json(args.decision),
|
||||
load_json(args.trusted_keyring),
|
||||
checked_at=args.checked_at,
|
||||
)
|
||||
print(json.dumps(result, ensure_ascii=False, sort_keys=True))
|
||||
return 0
|
||||
except ReviewPackageError as exc:
|
||||
parser.error(str(exc))
|
||||
return 2
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,155 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import copy
|
||||
import importlib.util
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from cryptography.hazmat.primitives import serialization
|
||||
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[2]
|
||||
SCRIPT = ROOT / "yuqei-infra/scripts/build-scene-source-legal-review.py"
|
||||
SPEC = importlib.util.spec_from_file_location("scene_source_legal_review", SCRIPT)
|
||||
assert SPEC and SPEC.loader
|
||||
review = importlib.util.module_from_spec(SPEC)
|
||||
SPEC.loader.exec_module(review)
|
||||
|
||||
CANDIDATE = subprocess.run(
|
||||
["git", "rev-parse", "HEAD"],
|
||||
cwd=ROOT,
|
||||
check=True,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
encoding="utf-8",
|
||||
).stdout.strip()
|
||||
CREATED_AT = "2026-10-06T09:00:00+08:00"
|
||||
REVIEWED_AT = "2026-10-06T10:00:00+08:00"
|
||||
EXPIRES_AT = "2026-10-13T10:00:00+08:00"
|
||||
|
||||
|
||||
def _request():
|
||||
return review.build_request(ROOT, candidate_commit=CANDIDATE, created_at=CREATED_AT)
|
||||
|
||||
|
||||
def _signed_material(request):
|
||||
private_key = Ed25519PrivateKey.generate()
|
||||
public_raw = private_key.public_key().public_bytes(
|
||||
encoding=serialization.Encoding.Raw,
|
||||
format=serialization.PublicFormat.Raw,
|
||||
)
|
||||
reviewer = {
|
||||
"name": "Named Qualified Reviewer",
|
||||
"organization": "Example Legal Department",
|
||||
"qualification": "PRC qualified legal reviewer",
|
||||
"credential_reference": "credential-reference-001",
|
||||
}
|
||||
decision = {
|
||||
"schema_version": review.DECISION_SCHEMA,
|
||||
"request_sha256": request["request_sha256"],
|
||||
"candidate_commit": request["candidate_commit"],
|
||||
"reviewed_at": REVIEWED_AT,
|
||||
"expires_at": EXPIRES_AT,
|
||||
"reviewer": reviewer,
|
||||
"items": [
|
||||
{
|
||||
"occurrence_id": occurrence["occurrence_id"],
|
||||
"source_record_sha256": occurrence["source_record_sha256"],
|
||||
"current_effectiveness_verified": True,
|
||||
"transaction_applicability_verified": True,
|
||||
"checked_at": REVIEWED_AT,
|
||||
"authoritative_status_url": occurrence["source"]["url"],
|
||||
"rationale": "The named qualified reviewer checked the current official source and this exact scene scope.",
|
||||
}
|
||||
for occurrence in request["occurrences"]
|
||||
],
|
||||
}
|
||||
signature = private_key.sign(review.SIGNATURE_DOMAIN + review.canonical_json(decision))
|
||||
decision["signature"] = {
|
||||
"algorithm": "Ed25519",
|
||||
"key_id": "legal-reviewer-key-001",
|
||||
"value_base64": base64.b64encode(signature).decode("ascii"),
|
||||
}
|
||||
keyring = {
|
||||
"schema_version": review.KEYRING_SCHEMA,
|
||||
"keys": [
|
||||
{
|
||||
"key_id": "legal-reviewer-key-001",
|
||||
"status": "trusted",
|
||||
"revoked_at": None,
|
||||
"valid_from": "2026-10-01T00:00:00+08:00",
|
||||
"valid_until": "2026-10-31T23:59:59+08:00",
|
||||
"public_key_raw_base64": base64.b64encode(public_raw).decode("ascii"),
|
||||
"reviewer_name": reviewer["name"],
|
||||
"reviewer_organization": reviewer["organization"],
|
||||
"reviewer_qualification": reviewer["qualification"],
|
||||
"credential_reference": reviewer["credential_reference"],
|
||||
}
|
||||
],
|
||||
}
|
||||
return decision, keyring
|
||||
|
||||
|
||||
def test_request_covers_exact_scene_source_baseline_without_approving_it():
|
||||
request = _request()
|
||||
|
||||
assert request["scope"] == {
|
||||
"scene_count": 30,
|
||||
"reference_occurrence_count": 43,
|
||||
"distinct_source_record_count": 8,
|
||||
"distinct_url_count": 7,
|
||||
}
|
||||
assert len(request["occurrences"]) == 43
|
||||
assert len({item["scene_id"] for item in request["occurrences"]}) == 30
|
||||
assert all(item["source"]["current_effectiveness_verified"] is False for item in request["occurrences"])
|
||||
assert request["guardrails"]["not_legal_approval"] is True
|
||||
assert request["guardrails"]["authorizes_publication"] is False
|
||||
assert review.verify_request(request, ROOT) == request
|
||||
|
||||
|
||||
def test_request_rejects_source_or_embedded_hash_tampering():
|
||||
request = _request()
|
||||
request["occurrences"][0]["source"]["scope"] = "tampered"
|
||||
|
||||
with pytest.raises(review.ReviewPackageError, match="embedded hash drift"):
|
||||
review.verify_request(request, ROOT)
|
||||
|
||||
|
||||
def test_request_rejects_a_candidate_other_than_checked_out_head():
|
||||
with pytest.raises(review.ReviewPackageError, match="checked-out HEAD"):
|
||||
review.build_request(ROOT, candidate_commit="1" * 40, created_at=CREATED_AT)
|
||||
|
||||
|
||||
def test_decision_requires_a_signature_from_the_trusted_human_key():
|
||||
request = _request()
|
||||
decision, keyring = _signed_material(request)
|
||||
decision.pop("signature")
|
||||
|
||||
with pytest.raises(review.ReviewPackageError, match="Ed25519"):
|
||||
review.verify_signed_decision(request, decision, keyring, checked_at=REVIEWED_AT)
|
||||
|
||||
|
||||
def test_signed_complete_decision_verifies_but_does_not_publish():
|
||||
request = _request()
|
||||
decision, keyring = _signed_material(request)
|
||||
|
||||
result = review.verify_signed_decision(request, decision, keyring, checked_at=REVIEWED_AT)
|
||||
|
||||
assert result["status"] == "verified_qualified_human_review"
|
||||
assert result["reviewed_occurrence_count"] == 43
|
||||
assert result["all_current_effectiveness_verified"] is True
|
||||
assert result["all_transaction_applicability_verified"] is True
|
||||
assert result["authorizes_publication_by_itself"] is False
|
||||
|
||||
|
||||
def test_signed_decision_fails_closed_when_one_scene_is_not_applicable():
|
||||
request = _request()
|
||||
decision, keyring = _signed_material(request)
|
||||
changed = copy.deepcopy(decision)
|
||||
changed["items"][0]["transaction_applicability_verified"] = False
|
||||
|
||||
with pytest.raises(review.ReviewPackageError, match="transaction applicability is not approved"):
|
||||
review.verify_signed_decision(request, changed, keyring, checked_at=REVIEWED_AT)
|
||||
Reference in New Issue
Block a user