diff --git a/yuqei-infra/scripts/build-scene-source-legal-review.py b/yuqei-infra/scripts/build-scene-source-legal-review.py new file mode 100644 index 00000000..d449763b --- /dev/null +++ b/yuqei-infra/scripts/build-scene-source-legal-review.py @@ -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()) diff --git a/yuqei-infra/tests/test_scene_source_legal_review.py b/yuqei-infra/tests/test_scene_source_legal_review.py new file mode 100644 index 00000000..1e9b3371 --- /dev/null +++ b/yuqei-infra/tests/test_scene_source_legal_review.py @@ -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)