diff --git a/project-docs/RC06-43项法源人工审核与专用签名操作说明-2026-10-07.md b/project-docs/RC06-43项法源人工审核与专用签名操作说明-2026-10-07.md new file mode 100644 index 00000000..6e446269 --- /dev/null +++ b/project-docs/RC06-43项法源人工审核与专用签名操作说明-2026-10-07.md @@ -0,0 +1,112 @@ +# RC-06 43 项法源人工审核与专用签名操作说明 + +日期：2026-10-07 + +## 1. 结论和边界 + +本工具只处理 30 个场景中的 43 次法源引用人工核验。它不会调用模型、不会发布资产、不会生成 RC-06 批次批准，也不会把任何人工判断自动填写为 `true`。 + +法源审核专用密钥与下列密钥必须分开，公钥不得相同： + +- `%LOCALAPPDATA%\Yuqei\ReleaseSigner` 中的 RC-06 批次审批密钥； +- RC-06 approval root、recovery root、apply-service 密钥； +- KMS 在线发布服务签名密钥。 + +默认专用状态目录为 `%LOCALAPPDATA%\Yuqei\SceneLegalReviewSigner`。加密私钥只保存在当前 Windows 用户和 `SYSTEM` 可访问的本机目录，不得上传仓库、聊天、邮件、共享盘或 CI artifact。 + +## 2. 冻结前一次性初始化 + +在干净、受控的 Windows 电脑上，从仓库根目录执行： + +```powershell +powershell.exe -NoLogo -NoProfile -ExecutionPolicy Bypass -File .\yuqei-infra\scripts\Launch-SceneLegalReviewSigner.ps1 initialize ` + --name "陈钰麒" ` + --organization "钰麒科技" ` + --qualification "钰麒科技内部合同法务审核负责人" ` + --credential-reference "钰麒科技内部任命/0001" +``` + +工具会遮罩询问并二次确认不少于 20 个字符的口令。口令不得通过命令行、环境变量、密码文件或聊天传递。 + +初始化结果中的 keyring 位于： + +```text +%LOCALAPPDATA%\Yuqei\SceneLegalReviewSigner\keys\scene-legal-review-keyring.json +``` + +系统操作人必须在候选冻结前先预览，再通过 `Launch-ReleaseAuthority.ps1 pin-source --legal-review-keyring <上述文件> --apply` 固定该 keyring 的规范字节 SHA-256。源码 pin 变化后才冻结唯一候选。不得把 ReleaseSigner 公钥复制进此 keyring。 + +## 3. 候选冻结后生成 43 项请求 + +只有唯一 commit/tag、source manifest、BUILD 和 RC-05 都绑定同一候选后，系统操作人才能生成正式请求： + +```powershell +python -B .\yuqei-infra\scripts\build-scene-source-legal-review.py prepare ` + --repo <冻结候选的只读 checkout> ` + --candidate-commit <40 位冻结 SHA> ` + --created-at <带时区的 RFC3339 时间> ` + --output <仓库外持久产物目录>\scene-legal-review-request.json +``` + +该请求固定 30 个场景、43 次引用、8 条法源记录和 7 个官方 URL。请求本身不是法律批准，也不授权发布。 + +随后生成供本人填写的模板： + +```powershell +powershell.exe -NoLogo -NoProfile -ExecutionPolicy Bypass -File .\yuqei-infra\scripts\Launch-SceneLegalReviewSigner.ps1 prepare-template ` + --repo <冻结候选的只读 checkout> ` + --request <请求文件> ` + --output-directory <仓库外的新目录> +``` + +输出包含 JSON 决定模板、UTF-8 CSV 对照表和摘要。43 行的两个判断、核验时间、状态 URL 和理由都保持空值。 + +## 4. 陈钰麒本人必须完成的内容 + +本人逐项打开对应官方页面并结合该场景确认： + +1. `current_effectiveness_verified`：法源当前是否有效； +2. `transaction_applicability_verified`：法源是否适用于该交易/文书场景； +3. `checked_at`：实际核验时间，必须带时区； +4. `authoritative_status_url`：实际核验的无凭据 `https://*.gov.cn` 官方页面； +5. `rationale`：不少于 20 个字符的具体判断理由。 + +只允许编辑决定模板中的上述五项。不得修改 occurrence ID、source hash、candidate commit、request hash、审核人身份、`reviewed_at`、`expires_at` 或 `signature`。 + +若任一项为 `false`、不确定、页面无法确认或理由不足，立即停止：保留为发布 blocker，不得为通过门禁而改成 `true`，也不得签署“全部通过”决定。 + +## 5. 专用签名和离线复验 + +43 项均由本人明确完成后执行： + +```powershell +powershell.exe -NoLogo -NoProfile -ExecutionPolicy Bypass -File .\yuqei-infra\scripts\Launch-SceneLegalReviewSigner.ps1 sign ` + --repo <冻结候选的只读 checkout> ` + --request <请求文件> ` + --completed-decision <本人完成的决定模板> ` + --output-directory <仓库外的新签名结果目录> ` + --valid-hours 12 +``` + +工具在输入口令后重新校验全部 43 项，取得当前时间，生成最长 24 小时且不超过专用密钥有效期的决定，使用独立 Ed25519 私钥签名并立即离线自验。输出目录存在时拒绝覆盖。 + +可在不接触私钥的机器上再次复验： + +```powershell +powershell.exe -NoLogo -NoProfile -ExecutionPolicy Bypass -File .\yuqei-infra\scripts\Launch-SceneLegalReviewSigner.ps1 verify ` + --repo <冻结候选的只读 checkout> ` + --request <请求文件> ` + --decision <签名结果目录>\scene-legal-review-decision.json ` + --keyring <固定的 keyring 文件> +``` + +有效输出仍固定为 `authorizes_publication_by_itself=false`。法源审核通过后，还必须在相邻有效期内完成独立的 RC-06 发布批次审批、在线 apply、74 项机器收据和 UAT。 + +## 6. 禁止事项 + +- 不得直接运行 Python 脚本完成正式密钥操作；使用随附 PowerShell 隔离启动器。 +- 不得复用、复制或导入 ReleaseSigner 私钥。 +- 不得把测试夹具、AI研究结果或空模板描述为人工法律审核。 +- 不得手工填写审核时间、到期时间或签名。 +- 不得覆盖既有模板或签名结果目录。 +- 不得在源码仍变化时生成正式请求或决定。 diff --git a/yuqei-infra/scripts/Launch-SceneLegalReviewSigner.ps1 b/yuqei-infra/scripts/Launch-SceneLegalReviewSigner.ps1 new file mode 100644 index 00000000..39f9facf --- /dev/null +++ b/yuqei-infra/scripts/Launch-SceneLegalReviewSigner.ps1 @@ -0,0 +1,93 @@ +[CmdletBinding()] +param( + [Parameter(Mandatory = $true, Position = 0)] + [ValidateSet("initialize", "status", "prepare-template", "sign", "verify")] + [string]$Command, + + [Parameter(ValueFromRemainingArguments = $true)] + [string[]]$CommandArguments +) + +$ErrorActionPreference = "Stop" +$env:PSModulePath = @( + (Join-Path $env:SystemRoot "System32\WindowsPowerShell\v1.0\Modules") + (Join-Path $env:ProgramFiles "WindowsPowerShell\Modules") +) -join ";" + +function Get-TrustedPythonCandidates { + $candidatePaths = [System.Collections.Generic.List[string]]::new() + foreach ($pattern in @( + (Join-Path $env:LOCALAPPDATA "Python\pythoncore-*\python.exe") + (Join-Path $env:LOCALAPPDATA "Programs\Python\Python*\python.exe") + (Join-Path $env:ProgramFiles "Python*\python.exe") + )) { + Get-ChildItem -Path $pattern -File -ErrorAction SilentlyContinue | + ForEach-Object { $candidatePaths.Add($_.FullName) } + } + foreach ($registryRoot in @( + "Registry::HKEY_CURRENT_USER\Software\Python\PythonCore" + "Registry::HKEY_LOCAL_MACHINE\Software\Python\PythonCore" + )) { + Get-ChildItem -LiteralPath $registryRoot -ErrorAction SilentlyContinue | + ForEach-Object { + $properties = Get-ItemProperty -LiteralPath (Join-Path $_.PSPath "InstallPath") -ErrorAction SilentlyContinue + if ($properties.ExecutablePath) { + $candidatePaths.Add([string]$properties.ExecutablePath) + } elseif ($properties.'(default)') { + $candidatePaths.Add((Join-Path ([string]$properties.'(default)') "python.exe")) + } + } + } + + $trusted = foreach ($candidatePath in $candidatePaths | Sort-Object -Unique) { + if (-not (Test-Path -LiteralPath $candidatePath -PathType Leaf)) { continue } + $item = Get-Item -LiteralPath $candidatePath -Force + if (($item.Attributes -band [System.IO.FileAttributes]::ReparsePoint) -ne 0) { continue } + if ($item.FullName -like "*\Microsoft\WindowsApps\*") { continue } + $version = [version]::new( + [Math]::Max(0, $item.VersionInfo.FileMajorPart), + [Math]::Max(0, $item.VersionInfo.FileMinorPart), + [Math]::Max(0, $item.VersionInfo.FileBuildPart), + [Math]::Max(0, $item.VersionInfo.FilePrivatePart) + ) + if ($version -lt [version]"3.11") { continue } + $signature = Get-AuthenticodeSignature -LiteralPath $item.FullName + if ( + $signature.Status -ne [System.Management.Automation.SignatureStatus]::Valid -or + $null -eq $signature.SignerCertificate -or + $signature.SignerCertificate.Subject -notlike "*CN=Python Software Foundation*" + ) { continue } + [pscustomobject]@{ Path = $item.FullName; Version = $version } + } + return $trusted | Sort-Object Version -Descending +} + +try { + $signerScript = Join-Path $PSScriptRoot "scene-source-legal-review-signer.py" + if (-not (Test-Path -LiteralPath $signerScript -PathType Leaf)) { + throw "The scene legal review signer program is missing" + } + Remove-Item Env:PYTHONPATH -ErrorAction SilentlyContinue + Remove-Item Env:PYTHONHOME -ErrorAction SilentlyContinue + Remove-Item Env:PYTHONSTARTUP -ErrorAction SilentlyContinue + Remove-Item Env:PYTHONUSERBASE -ErrorAction SilentlyContinue + $runtime = $null + $ErrorActionPreference = "Continue" + foreach ($candidate in @(Get-TrustedPythonCandidates)) { + $preflightOutput = @(& $candidate.Path -I -S -B $signerScript runtime-check 2>&1) + if ($LASTEXITCODE -eq 0) { + $runtime = $candidate + break + } + } + $ErrorActionPreference = "Stop" + if ($null -eq $runtime) { + throw "No trusted offline Python runtime has the required cryptography dependency" + } + & $runtime.Path -I -S -B $signerScript $Command @CommandArguments + exit $LASTEXITCODE +} catch { + Write-Host ("Scene legal review signer startup failed: " + $_.Exception.Message) -ForegroundColor Red + Write-Host "No legal-review key or signature was created." + exit 2 +} diff --git a/yuqei-infra/scripts/scene-source-legal-review-signer.py b/yuqei-infra/scripts/scene-source-legal-review-signer.py new file mode 100644 index 00000000..cc676fb2 --- /dev/null +++ b/yuqei-infra/scripts/scene-source-legal-review-signer.py @@ -0,0 +1,988 @@ +#!/usr/bin/env python3 +"""Prepare and sign the 43-item scene-source legal-review decision. + +This helper is deliberately separate from the RC-06 release-approval signer. +It creates a purpose-specific encrypted Ed25519 key, emits an unsigned template +whose human decisions are all null, signs only a fully and explicitly completed +decision, and immediately verifies the exact signed bytes offline. + +No command changes Git, publishes an asset, calls a model, or turns a human +decision into ``true``. User-facing commands must be launched through the +isolated PowerShell launcher shipped next to this file. +""" + +from __future__ import annotations + +import argparse +import base64 +import csv +import getpass +import importlib.util +import io +import json +import os +import sys +from datetime import UTC, datetime, timedelta +from pathlib import Path +from typing import Any, Mapping + + +SCRIPT_DIRECTORY = Path(__file__).resolve().parent +REPO_ROOT = SCRIPT_DIRECTORY.parents[1] +SECURE_IO_PATH = SCRIPT_DIRECTORY / "rc06-single-reviewer-signing.py" +LEGAL_REVIEW_PATH = SCRIPT_DIRECTORY / "build-scene-source-legal-review.py" + + +def _load_module(name: str, path: Path) -> Any: + spec = importlib.util.spec_from_file_location(name, path) + if spec is None or spec.loader is None: # pragma: no cover - damaged install + raise RuntimeError(f"cannot load required helper: {path.name}") + module = importlib.util.module_from_spec(spec) + sys.modules[name] = module + spec.loader.exec_module(module) + return module + + +# Loading secure_io first bootstraps only the selected interpreter's exact +# site-packages when the launcher invokes Python with -I -S -B. +secure_io = _load_module("yuqei_scene_legal_secure_io", SECURE_IO_PATH) +legal_review = _load_module("yuqei_scene_source_legal_review", LEGAL_REVIEW_PATH) +batch = secure_io.batch + +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric.ed25519 import ( + Ed25519PrivateKey, + Ed25519PublicKey, +) + + +REGISTRATION_SCHEMA = "yuqei.scene-source-legal-review-signer-registration.v1" +KEY_BUNDLE_SCHEMA = "yuqei.scene-source-legal-review-signer-key-bundle.v1" +SIGNING_RESULT_SCHEMA = "yuqei.scene-source-legal-review-signing-result.v1" +PRIVATE_KEY_FILENAME = "scene-legal-review-private.pem" # pragma: allowlist secret +PUBLIC_KEY_FILENAME = "scene-legal-review-public.pem" +REGISTRATION_FILENAME = "scene-legal-review-registration.json" +KEYRING_FILENAME = "scene-legal-review-keyring.json" +KEY_BUNDLE_DESCRIPTOR_FILENAME = "scene-legal-review-key-bundle.json" +DECISION_TEMPLATE_FILENAME = "scene-legal-review-decision-template.json" +CHECKLIST_FILENAME = "scene-legal-review-checklist.csv" +TEMPLATE_SUMMARY_FILENAME = "scene-legal-review-template-summary.json" +SIGNED_DECISION_FILENAME = "scene-legal-review-decision.json" +VERIFICATION_SUMMARY_FILENAME = "scene-legal-review-verification-summary.json" +SIGNING_RESULT_FILENAME = "scene-legal-review-signing-result.json" +IDENTITY_FIELDS = ("name", "organization", "qualification", "credential_reference") +ITEM_FIELDS = ( + "occurrence_id", + "source_record_sha256", + "current_effectiveness_verified", + "transaction_applicability_verified", + "checked_at", + "authoritative_status_url", + "rationale", +) +DECISION_FIELDS = { + "schema_version", + "request_sha256", + "candidate_commit", + "reviewed_at", + "expires_at", + "reviewer", + "items", + "signature", +} +PLACEHOLDERS = {"", "-", "n/a", "none", "pending", "unknown", "待确认", "待复核"} +MAX_SIGNED_DECISION_WINDOW = timedelta(hours=24) +MAX_KEY_VALIDITY = timedelta(days=400) +MAX_INPUT_BYTES = 4 * 1024 * 1024 + + +class LegalSignerError(RuntimeError): + """Fail-closed error safe to display to the human reviewer.""" + + +def _now_utc() -> datetime: + return datetime.now(UTC).replace(microsecond=0) + + +def _format_time(value: datetime) -> str: + if value.tzinfo is None or value.utcoffset() is None: + raise LegalSignerError("time value must include a timezone") + return value.astimezone(UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z") + + +def _parse_time(value: Any, label: str) -> datetime: + if not isinstance(value, str) or not value or len(value) > 64: + raise LegalSignerError(f"{label} must be an RFC3339 timestamp") + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError as exc: + raise LegalSignerError(f"{label} must be an RFC3339 timestamp") from exc + if parsed.tzinfo is None or parsed.utcoffset() is None: + raise LegalSignerError(f"{label} must include a timezone") + return parsed.astimezone(UTC) + + +def _canonical_json(value: Any) -> bytes: + return batch._canonical(value) + + +def _sha256_bytes(value: bytes) -> str: + return batch._sha256_bytes(value) + + +def _public_raw(key: Ed25519PublicKey) -> bytes: + return key.public_bytes(serialization.Encoding.Raw, serialization.PublicFormat.Raw) + + +def default_state_directory() -> Path: + if os.name == "nt": + base = os.environ.get("LOCALAPPDATA") + if not base: + raise LegalSignerError("LOCALAPPDATA is unavailable; legal signer storage is unsafe") + return Path(base) / "Yuqei" / "SceneLegalReviewSigner" + return Path.home() / ".local" / "share" / "yuqei" / "scene-legal-review-signer" + + +def default_release_signer_directory() -> Path: + if os.name == "nt": + base = os.environ.get("LOCALAPPDATA") + if not base: + raise LegalSignerError("LOCALAPPDATA is unavailable") + return Path(base) / "Yuqei" / "ReleaseSigner" + return Path.home() / ".local" / "share" / "yuqei" / "release-signer" + + +def _is_relative_to(path: Path, parent: Path) -> bool: + try: + path.relative_to(parent) + return True + except ValueError: + return False + + +def _normalize_state_directory( + state_directory: Path, + *, + release_signer_directory: Path | None = None, +) -> Path: + release_state = (release_signer_directory or default_release_signer_directory()).resolve( + strict=False + ) + try: + state = secure_io._assert_local_safe_path( + state_directory, "scene legal reviewer state directory", allow_existing=True + ) + except secure_io.SignerError as exc: + raise LegalSignerError(str(exc)) from exc + if ( + state == release_state + or _is_relative_to(state, release_state) + or _is_relative_to(release_state, state) + or any(part.casefold() == "releasesigner" for part in state.parts) + ): + raise LegalSignerError( + "scene legal reviewer state must be separate from the RC-06 ReleaseSigner state" + ) + return state + + +def _prepare_default_parent(state_directory: Path) -> None: + """Create only the fixed Yuqei container when it does not exist yet.""" + + parent = state_directory.parent + if parent.exists() or parent.is_symlink(): + return + if state_directory.resolve(strict=False) != default_state_directory().resolve(strict=False): + raise LegalSignerError("custom legal signer state parent must already exist") + grandparent = parent.parent + try: + secure_io._assert_local_safe_path( + grandparent, "legal signer state container parent", allow_existing=True + ) + except secure_io.SignerError as exc: + raise LegalSignerError(str(exc)) from exc + if not grandparent.is_dir() or grandparent.is_symlink(): + raise LegalSignerError("legal signer state container parent is unavailable") + try: + parent.mkdir(mode=0o700, exist_ok=False) + secure_io._harden_private_directory(parent) + except FileExistsError: + pass + except (OSError, secure_io.SignerError) as exc: + raise LegalSignerError(f"cannot create legal signer state container: {exc}") from exc + + +def _validate_identity(identity: Mapping[str, Any]) -> dict[str, str]: + if not isinstance(identity, Mapping) or set(identity) != set(IDENTITY_FIELDS): + raise LegalSignerError("legal reviewer identity must contain exactly four fields") + normalized: dict[str, str] = {} + for field in IDENTITY_FIELDS: + value = identity.get(field) + if ( + not isinstance(value, str) + or not value.strip() + or value != value.strip() + or len(value) > 512 + or value.casefold() in PLACEHOLDERS + ): + raise LegalSignerError(f"legal reviewer {field} is empty, invalid, or a placeholder") + normalized[field] = value + return normalized + + +def _read_json(path: Path, label: str) -> tuple[Any, str]: + try: + return batch.secure_read_json(path, label) + except batch.PublicationBatchError as exc: + raise LegalSignerError(str(exc)) from exc + + +def _release_signer_fingerprint(release_signer_directory: Path) -> str | None: + registration_path = release_signer_directory / "keys" / "reviewer-registration.json" + if not registration_path.exists() and not registration_path.is_symlink(): + return None + registration, _ = _read_json(registration_path, "RC-06 ReleaseSigner registration") + fingerprint = registration.get("public_key_raw_sha256") if isinstance(registration, dict) else None + if not isinstance(fingerprint, str) or batch.HEX64_RE.fullmatch(fingerprint) is None: + raise LegalSignerError("RC-06 ReleaseSigner registration fingerprint is invalid") + return fingerprint + + +def _reject_release_signer_key_reuse( + fingerprint: str, + release_signer_directory: Path, +) -> None: + release_fingerprint = _release_signer_fingerprint(release_signer_directory) + if release_fingerprint is not None and fingerprint == release_fingerprint: + raise LegalSignerError( + "scene legal review key must not reuse the RC-06 ReleaseSigner approval key" + ) + + +def _key_files() -> set[str]: + return { + PRIVATE_KEY_FILENAME, + PUBLIC_KEY_FILENAME, + REGISTRATION_FILENAME, + KEYRING_FILENAME, + KEY_BUNDLE_DESCRIPTOR_FILENAME, + } + + +def _load_public_material( + state_directory: Path, + *, + release_signer_directory: Path | None = None, +) -> tuple[dict[str, Any], dict[str, Any], Ed25519PublicKey]: + release_state = (release_signer_directory or default_release_signer_directory()).resolve( + strict=False + ) + state = _normalize_state_directory( + state_directory, release_signer_directory=release_state + ) + key_directory = state / "keys" + try: + secure_io._verify_complete_bundle( + key_directory, _key_files(), "scene legal reviewer key bundle" + ) + except secure_io.SignerError as exc: + raise LegalSignerError(str(exc)) from exc + registration, _ = _read_json( + key_directory / REGISTRATION_FILENAME, "scene legal reviewer registration" + ) + keyring, keyring_raw_sha256 = _read_json( + key_directory / KEYRING_FILENAME, "scene legal reviewer keyring" + ) + descriptor, _ = _read_json( + key_directory / KEY_BUNDLE_DESCRIPTOR_FILENAME, + "scene legal reviewer key bundle descriptor", + ) + expected_registration_fields = { + "schema_version", + "key_id", + "algorithm", + "reviewer", + "created_at", + "valid_from", + "valid_until", + "public_key_pem", + "public_key_raw_base64", + "public_key_raw_sha256", + "keyring_raw_sha256", + } + if not isinstance(registration, dict) or set(registration) != expected_registration_fields: + raise LegalSignerError("scene legal reviewer registration shape is invalid") + if registration["schema_version"] != REGISTRATION_SCHEMA or registration["algorithm"] != "Ed25519": + raise LegalSignerError("scene legal reviewer registration schema or algorithm is invalid") + identity = _validate_identity(registration["reviewer"]) + valid_from = _parse_time(registration["valid_from"], "legal key valid_from") + valid_until = _parse_time(registration["valid_until"], "legal key valid_until") + if valid_until <= valid_from or valid_until - valid_from > MAX_KEY_VALIDITY: + raise LegalSignerError("scene legal reviewer key validity window is invalid") + try: + public_key = serialization.load_pem_public_key( + registration["public_key_pem"].encode("ascii") + ) + except (AttributeError, UnicodeError, ValueError) as exc: + raise LegalSignerError("scene legal reviewer public key is invalid") from exc + if not isinstance(public_key, Ed25519PublicKey): + raise LegalSignerError("scene legal reviewer key must be Ed25519") + public_raw = _public_raw(public_key) + fingerprint = _sha256_bytes(public_raw) + if ( + registration["public_key_raw_base64"] + != base64.b64encode(public_raw).decode("ascii") + or registration["public_key_raw_sha256"] != fingerprint + or registration["key_id"] != f"scene-legal-review-{fingerprint[:20]}" + ): + raise LegalSignerError("scene legal reviewer public key identity drift") + try: + public_file = batch.secure_read_bytes( + key_directory / PUBLIC_KEY_FILENAME, + "scene legal reviewer public key file", + maximum_bytes=secure_io.MAX_SECRET_BYTES, + ) + except batch.PublicationBatchError as exc: + raise LegalSignerError(str(exc)) from exc + if public_file != registration["public_key_pem"].encode("ascii"): + raise LegalSignerError("scene legal reviewer public key file drift") + expected_descriptor = { + "schema_version": KEY_BUNDLE_SCHEMA, + "key_id": registration["key_id"], + "algorithm": "Ed25519", + "purpose": "scene_source_legal_review_only", + "created_at": registration["created_at"], + "encrypted_private_key": True, + "private_key_filename": PRIVATE_KEY_FILENAME, + "public_key_filename": PUBLIC_KEY_FILENAME, + "registration_filename": REGISTRATION_FILENAME, + "keyring_filename": KEYRING_FILENAME, + "public_key_raw_sha256": fingerprint, + } + if descriptor != expected_descriptor: + raise LegalSignerError("scene legal reviewer key bundle descriptor drift") + _reject_release_signer_key_reuse(fingerprint, release_state) + if keyring_raw_sha256 != registration["keyring_raw_sha256"]: + raise LegalSignerError("scene legal reviewer keyring hash drift") + expected_keyring = { + "schema_version": legal_review.KEYRING_SCHEMA, + "keys": [ + { + "key_id": registration["key_id"], + "status": "trusted", + "revoked_at": None, + "valid_from": registration["valid_from"], + "valid_until": registration["valid_until"], + "reviewer_name": identity["name"], + "reviewer_organization": identity["organization"], + "reviewer_qualification": identity["qualification"], + "credential_reference": identity["credential_reference"], + "public_key_raw_base64": registration["public_key_raw_base64"], + } + ], + } + if keyring != expected_keyring or _sha256_bytes(_canonical_json(keyring)) != keyring_raw_sha256: + raise LegalSignerError("scene legal reviewer keyring content is not canonical") + return registration, keyring, public_key + + +def _load_private_material( + state_directory: Path, + passphrase: str, + *, + release_signer_directory: Path | None = None, +) -> tuple[dict[str, Any], dict[str, Any], Ed25519PrivateKey]: + registration, keyring, public_key = _load_public_material( + state_directory, release_signer_directory=release_signer_directory + ) + key_directory = state_directory.resolve(strict=False) / "keys" + try: + private_raw = secure_io._read_secret_file( + key_directory / PRIVATE_KEY_FILENAME, + "scene legal reviewer encrypted private key", + ) + password = secure_io._encode_existing_passphrase(passphrase) + except secure_io.SignerError as exc: + raise LegalSignerError(str(exc)) from exc + if not private_raw.startswith(b"-----BEGIN ENCRYPTED PRIVATE KEY-----"): + raise LegalSignerError("scene legal reviewer private key must be encrypted PKCS#8") + try: + private_key = serialization.load_pem_private_key(private_raw, password=password) + except (TypeError, ValueError) as exc: + raise LegalSignerError("legal reviewer key password is incorrect or key is invalid") from exc + if not isinstance(private_key, Ed25519PrivateKey): + raise LegalSignerError("scene legal reviewer private key must be Ed25519") + if _public_raw(private_key.public_key()) != _public_raw(public_key): + raise LegalSignerError("scene legal reviewer private/public key mismatch") + return registration, keyring, private_key + + +def initialize_key( + state_directory: Path, + *, + identity: Mapping[str, Any], + passphrase: str, + created_at: datetime | None = None, + valid_days: int = 365, + release_signer_directory: Path | None = None, +) -> dict[str, Any]: + """Create one encrypted, purpose-specific key bundle without overwriting.""" + + normalized_identity = _validate_identity(identity) + if not isinstance(valid_days, int) or not 1 <= valid_days <= MAX_KEY_VALIDITY.days: + raise LegalSignerError("legal reviewer key validity must be 1 to 400 days") + release_state = (release_signer_directory or default_release_signer_directory()).resolve( + strict=False + ) + state = _normalize_state_directory( + state_directory, release_signer_directory=release_state + ) + _prepare_default_parent(state) + try: + password = secure_io._validate_passphrase(passphrase) + lock = secure_io._state_operation_lock(state) + except secure_io.SignerError as exc: + raise LegalSignerError(str(exc)) from exc + with lock: + key_directory = state / "keys" + if key_directory.exists() or key_directory.is_symlink(): + registration, keyring, _ = _load_private_material( + state, + passphrase, + release_signer_directory=release_state, + ) + if registration["reviewer"] != normalized_identity: + raise LegalSignerError( + "existing legal reviewer key belongs to a different exact identity" + ) + return { + "status": "already_initialized_and_verified", + "key_id": registration["key_id"], + "public_key_raw_sha256": registration["public_key_raw_sha256"], + "keyring_raw_sha256": registration["keyring_raw_sha256"], + "keyring_file": os.fspath(key_directory / KEYRING_FILENAME), + } + created = (created_at or _now_utc()).astimezone(UTC).replace(microsecond=0) + valid_until = created + timedelta(days=valid_days) + private_key = Ed25519PrivateKey.generate() + public_key = private_key.public_key() + public_raw = _public_raw(public_key) + fingerprint = _sha256_bytes(public_raw) + _reject_release_signer_key_reuse(fingerprint, release_state) + key_id = f"scene-legal-review-{fingerprint[:20]}" + public_pem = public_key.public_bytes( + serialization.Encoding.PEM, + serialization.PublicFormat.SubjectPublicKeyInfo, + ) + private_pem = private_key.private_bytes( + serialization.Encoding.PEM, + serialization.PrivateFormat.PKCS8, + serialization.BestAvailableEncryption(password), + ) + valid_from_text = _format_time(created) + valid_until_text = _format_time(valid_until) + public_raw_base64 = base64.b64encode(public_raw).decode("ascii") + keyring = { + "schema_version": legal_review.KEYRING_SCHEMA, + "keys": [ + { + "key_id": key_id, + "status": "trusted", + "revoked_at": None, + "valid_from": valid_from_text, + "valid_until": valid_until_text, + "reviewer_name": normalized_identity["name"], + "reviewer_organization": normalized_identity["organization"], + "reviewer_qualification": normalized_identity["qualification"], + "credential_reference": normalized_identity["credential_reference"], + "public_key_raw_base64": public_raw_base64, + } + ], + } + keyring_raw = _canonical_json(keyring) + registration = { + "schema_version": REGISTRATION_SCHEMA, + "key_id": key_id, + "algorithm": "Ed25519", + "reviewer": normalized_identity, + "created_at": valid_from_text, + "valid_from": valid_from_text, + "valid_until": valid_until_text, + "public_key_pem": public_pem.decode("ascii"), + "public_key_raw_base64": public_raw_base64, + "public_key_raw_sha256": fingerprint, + "keyring_raw_sha256": _sha256_bytes(keyring_raw), + } + bundle = { + "schema_version": KEY_BUNDLE_SCHEMA, + "key_id": key_id, + "algorithm": "Ed25519", + "purpose": "scene_source_legal_review_only", + "created_at": valid_from_text, + "encrypted_private_key": True, + "private_key_filename": PRIVATE_KEY_FILENAME, + "public_key_filename": PUBLIC_KEY_FILENAME, + "registration_filename": REGISTRATION_FILENAME, + "keyring_filename": KEYRING_FILENAME, + "public_key_raw_sha256": fingerprint, + } + try: + secure_io._write_bundle_atomically( + key_directory, + { + PRIVATE_KEY_FILENAME: private_pem, + PUBLIC_KEY_FILENAME: public_pem, + REGISTRATION_FILENAME: _canonical_json(registration), + KEYRING_FILENAME: keyring_raw, + KEY_BUNDLE_DESCRIPTOR_FILENAME: _canonical_json(bundle), + }, + private_directory=True, + ) + except secure_io.SignerError as exc: + raise LegalSignerError(str(exc)) from exc + loaded, _, _ = _load_private_material( + state, + passphrase, + release_signer_directory=release_state, + ) + if loaded != registration: + raise LegalSignerError("scene legal reviewer key self-verification failed") + return { + "status": "initialized_not_reviewed", + "key_id": key_id, + "public_key_raw_sha256": fingerprint, + "keyring_raw_sha256": registration["keyring_raw_sha256"], + "keyring_file": os.fspath(key_directory / KEYRING_FILENAME), + } + + +def _load_and_verify_request(repo: Path, request_path: Path) -> dict[str, Any]: + request, _ = _read_json(request_path, "scene legal review request") + try: + return legal_review.verify_request(request, repo.resolve(strict=True)) + except (OSError, legal_review.ReviewPackageError) as exc: + raise LegalSignerError(f"scene legal review request is invalid: {exc}") from exc + + +def _template_for_request( + request: Mapping[str, Any], identity: Mapping[str, str] +) -> dict[str, Any]: + return { + "schema_version": legal_review.DECISION_SCHEMA, + "request_sha256": request["request_sha256"], + "candidate_commit": request["candidate_commit"], + "reviewed_at": None, + "expires_at": None, + "reviewer": dict(identity), + "items": [ + { + "occurrence_id": occurrence["occurrence_id"], + "source_record_sha256": occurrence["source_record_sha256"], + "current_effectiveness_verified": None, + "transaction_applicability_verified": None, + "checked_at": None, + "authoritative_status_url": None, + "rationale": "", + } + for occurrence in request["occurrences"] + ], + "signature": None, + } + + +def _checklist_csv(request: Mapping[str, Any]) -> bytes: + output = io.StringIO(newline="") + fieldnames = [ + "occurrence_id", + "scene_id", + "template_id", + "source_title", + "source_url", + "source_scope", + "current_effectiveness_verified", + "transaction_applicability_verified", + "checked_at", + "authoritative_status_url", + "rationale", + ] + writer = csv.DictWriter(output, fieldnames=fieldnames, lineterminator="\n") + writer.writeheader() + for occurrence in request["occurrences"]: + source = occurrence["source"] + writer.writerow( + { + "occurrence_id": occurrence["occurrence_id"], + "scene_id": occurrence["scene_id"], + "template_id": occurrence["template_id"], + "source_title": source["title"], + "source_url": source["url"], + "source_scope": source["scope"], + "current_effectiveness_verified": "", + "transaction_applicability_verified": "", + "checked_at": "", + "authoritative_status_url": "", + "rationale": "", + } + ) + return output.getvalue().encode("utf-8-sig") + + +def prepare_template( + state_directory: Path, + *, + repo: Path, + request_path: Path, + output_directory: Path, + release_signer_directory: Path | None = None, +) -> dict[str, Any]: + registration, _, _ = _load_public_material( + state_directory, release_signer_directory=release_signer_directory + ) + request = _load_and_verify_request(repo, request_path) + template = _template_for_request(request, registration["reviewer"]) + summary = { + "status": "pending_43_item_qualified_human_review", + "candidate_commit": request["candidate_commit"], + "request_sha256": request["request_sha256"], + "reviewer": registration["reviewer"], + "occurrence_count": len(template["items"]), + "human_decisions_prefilled": 0, + "signature_created": False, + "authorizes_publication": False, + } + try: + secure_io._write_bundle_atomically( + output_directory, + { + DECISION_TEMPLATE_FILENAME: _canonical_json(template), + CHECKLIST_FILENAME: _checklist_csv(request), + TEMPLATE_SUMMARY_FILENAME: _canonical_json(summary), + }, + private_directory=True, + ) + except secure_io.SignerError as exc: + raise LegalSignerError(str(exc)) from exc + return {**summary, "output_directory": os.fspath(output_directory)} + + +def _validate_completed_decision( + request: Mapping[str, Any], + decision: Any, + identity: Mapping[str, str], + *, + reviewed_at: datetime, + expires_at: datetime, +) -> dict[str, Any]: + if not isinstance(decision, dict) or set(decision) != DECISION_FIELDS: + raise LegalSignerError("completed legal review decision shape is invalid") + if ( + decision["schema_version"] != legal_review.DECISION_SCHEMA + or decision["request_sha256"] != request["request_sha256"] + or decision["candidate_commit"] != request["candidate_commit"] + ): + raise LegalSignerError("completed legal review decision is not bound to the exact request") + if decision["reviewed_at"] is not None or decision["expires_at"] is not None: + raise LegalSignerError("reviewed_at and expires_at must remain null until the signing ceremony") + if decision["signature"] is not None: + raise LegalSignerError("completed legal review template already contains a signature") + if decision["reviewer"] != identity: + raise LegalSignerError("completed legal review identity does not match the dedicated key") + items = decision["items"] + occurrences = request["occurrences"] + if not isinstance(items, list) or len(items) != 43 or len(occurrences) != 43: + raise LegalSignerError("completed legal review decision must contain exactly 43 items") + request_created_at = _parse_time(request["created_at"], "legal review request created_at") + completed_items: list[dict[str, Any]] = [] + for index, (item, occurrence) in enumerate(zip(items, occurrences, strict=True)): + if not isinstance(item, dict) or set(item) != set(ITEM_FIELDS): + raise LegalSignerError(f"legal review item {index + 1} shape is invalid") + occurrence_id = occurrence["occurrence_id"] + if ( + item["occurrence_id"] != occurrence_id + or item["source_record_sha256"] != occurrence["source_record_sha256"] + ): + raise LegalSignerError(f"legal review item binding drift: {occurrence_id}") + for field in ( + "current_effectiveness_verified", + "transaction_applicability_verified", + ): + if item[field] is not True: + state = "false" if item[field] is False else "not explicitly completed" + raise LegalSignerError( + f"{occurrence_id} {field} is {state}; no release-qualifying signature was created" + ) + checked_at = _parse_time(item["checked_at"], f"{occurrence_id} checked_at") + if not request_created_at <= checked_at <= reviewed_at: + raise LegalSignerError(f"{occurrence_id} checked_at is outside the review window") + try: + legal_review._safe_official_url( + item["authoritative_status_url"], f"{occurrence_id} authoritative status URL" + ) + except legal_review.ReviewPackageError as exc: + raise LegalSignerError(str(exc)) from exc + rationale = item["rationale"] + if ( + not isinstance(rationale, str) + or rationale != rationale.strip() + or len(rationale) < 20 + or len(rationale) > 4096 + or rationale.casefold() in PLACEHOLDERS + ): + raise LegalSignerError(f"{occurrence_id} rationale is insufficient") + completed_items.append(dict(item)) + return { + **decision, + "reviewed_at": _format_time(reviewed_at), + "expires_at": _format_time(expires_at), + "items": completed_items, + } + + +def sign_completed_decision( + state_directory: Path, + *, + repo: Path, + request_path: Path, + completed_decision_path: Path, + output_directory: Path, + passphrase: str, + valid_hours: int = 12, + now: datetime | None = None, + release_signer_directory: Path | None = None, +) -> dict[str, Any]: + if not isinstance(valid_hours, int) or not 1 <= valid_hours <= 24: + raise LegalSignerError("signed legal review validity must be 1 to 24 hours") + reviewed_at = (now or _now_utc()).astimezone(UTC).replace(microsecond=0) + expires_at = reviewed_at + timedelta(hours=valid_hours) + request = _load_and_verify_request(repo, request_path) + completed, _ = _read_json(completed_decision_path, "completed legal review decision") + release_state = (release_signer_directory or default_release_signer_directory()).resolve( + strict=False + ) + state = _normalize_state_directory( + state_directory, release_signer_directory=release_state + ) + try: + lock = secure_io._state_operation_lock(state) + except secure_io.SignerError as exc: + raise LegalSignerError(str(exc)) from exc + with lock: + registration, keyring, private_key = _load_private_material( + state, + passphrase, + release_signer_directory=release_state, + ) + key_valid_from = _parse_time(registration["valid_from"], "legal key valid_from") + key_valid_until = _parse_time(registration["valid_until"], "legal key valid_until") + if not key_valid_from <= reviewed_at < expires_at <= key_valid_until: + raise LegalSignerError("legal reviewer key does not cover the signed decision window") + completed_payload = _validate_completed_decision( + request, + completed, + registration["reviewer"], + reviewed_at=reviewed_at, + expires_at=expires_at, + ) + unsigned = { + field: value + for field, value in completed_payload.items() + if field != "signature" + } + signature = private_key.sign(legal_review.SIGNATURE_DOMAIN + _canonical_json(unsigned)) + signed = { + **unsigned, + "signature": { + "algorithm": "Ed25519", + "key_id": registration["key_id"], + "value_base64": base64.b64encode(signature).decode("ascii"), + }, + } + try: + summary = legal_review.verify_signed_decision( + request, + signed, + keyring, + checked_at=_format_time(reviewed_at), + ) + except legal_review.ReviewPackageError as exc: + raise LegalSignerError(f"signed legal review self-verification failed: {exc}") from exc + signed_raw = _canonical_json(signed) + keyring_raw = _canonical_json(keyring) + result = { + "schema_version": SIGNING_RESULT_SCHEMA, + "status": "signed_and_offline_verified_not_published", + "candidate_commit": request["candidate_commit"], + "request_sha256": request["request_sha256"], + "decision_sha256": _sha256_bytes(signed_raw), + "keyring_raw_sha256": _sha256_bytes(keyring_raw), + "reviewer": registration["reviewer"], + "reviewed_occurrence_count": summary["reviewed_occurrence_count"], + "expires_at": summary["expires_at"], + "authorizes_publication": False, + } + try: + secure_io._write_bundle_atomically( + output_directory, + { + SIGNED_DECISION_FILENAME: signed_raw, + VERIFICATION_SUMMARY_FILENAME: _canonical_json(summary), + SIGNING_RESULT_FILENAME: _canonical_json(result), + }, + private_directory=True, + ) + except secure_io.SignerError as exc: + raise LegalSignerError(str(exc)) from exc + return {**result, "output_directory": os.fspath(output_directory)} + + +def verify_signed_decision( + *, + repo: Path, + request_path: Path, + decision_path: Path, + keyring_path: Path, + checked_at: datetime | None = None, +) -> dict[str, Any]: + request = _load_and_verify_request(repo, request_path) + decision, decision_raw_sha256 = _read_json(decision_path, "signed legal review decision") + keyring, keyring_raw_sha256 = _read_json(keyring_path, "scene legal reviewer keyring") + if _sha256_bytes(_canonical_json(keyring)) != keyring_raw_sha256: + raise LegalSignerError("scene legal reviewer keyring is not canonical") + now = (checked_at or _now_utc()).astimezone(UTC).replace(microsecond=0) + try: + summary = legal_review.verify_signed_decision( + request, + decision, + keyring, + checked_at=_format_time(now), + ) + except legal_review.ReviewPackageError as exc: + raise LegalSignerError(f"signed legal review verification failed: {exc}") from exc + return { + **summary, + "decision_raw_sha256": decision_raw_sha256, + "keyring_raw_sha256": keyring_raw_sha256, + } + + +def status( + state_directory: Path, + *, + release_signer_directory: Path | None = None, +) -> dict[str, Any]: + registration, _, _ = _load_public_material( + state_directory, release_signer_directory=release_signer_directory + ) + return { + "status": "initialized_not_a_legal_decision", + "key_id": registration["key_id"], + "reviewer": registration["reviewer"], + "valid_from": registration["valid_from"], + "valid_until": registration["valid_until"], + "public_key_raw_sha256": registration["public_key_raw_sha256"], + "keyring_raw_sha256": registration["keyring_raw_sha256"], + "keyring_file": os.fspath(state_directory / "keys" / KEYRING_FILENAME), + } + + +def _assert_trusted_runtime() -> None: + try: + secure_io._assert_trusted_runtime() + except secure_io.SignerError as exc: + raise LegalSignerError(str(exc)) from exc + + +def _parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--state-directory", type=Path, default=default_state_directory()) + commands = parser.add_subparsers(dest="command", required=True) + commands.add_parser("runtime-check") + initialize = commands.add_parser("initialize") + initialize.add_argument("--name", required=True) + initialize.add_argument("--organization", required=True) + initialize.add_argument("--qualification", required=True) + initialize.add_argument("--credential-reference", required=True) + initialize.add_argument("--valid-days", type=int, default=365) + commands.add_parser("status") + prepare = commands.add_parser("prepare-template") + prepare.add_argument("--repo", type=Path, required=True) + prepare.add_argument("--request", type=Path, required=True) + prepare.add_argument("--output-directory", type=Path, required=True) + sign = commands.add_parser("sign") + sign.add_argument("--repo", type=Path, required=True) + sign.add_argument("--request", type=Path, required=True) + sign.add_argument("--completed-decision", type=Path, required=True) + sign.add_argument("--output-directory", type=Path, required=True) + sign.add_argument("--valid-hours", type=int, default=12) + verify = commands.add_parser("verify") + verify.add_argument("--repo", type=Path, required=True) + verify.add_argument("--request", type=Path, required=True) + verify.add_argument("--decision", type=Path, required=True) + verify.add_argument("--keyring", type=Path, required=True) + verify.add_argument("--checked-at") + return parser + + +def main(argv: list[str] | None = None) -> int: + parser = _parser() + args = parser.parse_args(argv) + try: + _assert_trusted_runtime() + if args.command == "runtime-check": + result = {"status": "trusted_isolated_runtime", "python": sys.version.split()[0]} + elif args.command == "initialize": + first = getpass.getpass("法源审核专用密钥口令（不少于20个字符）：") + second = getpass.getpass("再次输入同一口令：") + if first != second: + raise LegalSignerError("两次口令不一致，未创建密钥") + result = initialize_key( + args.state_directory, + identity={ + "name": args.name, + "organization": args.organization, + "qualification": args.qualification, + "credential_reference": args.credential_reference, + }, + passphrase=first, + valid_days=args.valid_days, + ) + elif args.command == "status": + result = status(args.state_directory) + elif args.command == "prepare-template": + result = prepare_template( + args.state_directory, + repo=args.repo, + request_path=args.request, + output_directory=args.output_directory, + ) + elif args.command == "sign": + passphrase = getpass.getpass("法源审核专用密钥口令：") + result = sign_completed_decision( + args.state_directory, + repo=args.repo, + request_path=args.request, + completed_decision_path=args.completed_decision, + output_directory=args.output_directory, + passphrase=passphrase, + valid_hours=args.valid_hours, + ) + else: + checked_at = ( + _parse_time(args.checked_at, "checked_at") if args.checked_at else None + ) + result = verify_signed_decision( + repo=args.repo, + request_path=args.request, + decision_path=args.decision, + keyring_path=args.keyring, + checked_at=checked_at, + ) + print(json.dumps(result, ensure_ascii=False, sort_keys=True)) + return 0 + except LegalSignerError 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_signer.py b/yuqei-infra/tests/test_scene_source_legal_review_signer.py new file mode 100644 index 00000000..80157256 --- /dev/null +++ b/yuqei-infra/tests/test_scene_source_legal_review_signer.py @@ -0,0 +1,348 @@ +from __future__ import annotations + +import copy +import importlib.util +import json +import os +import stat +import subprocess +import sys +from datetime import UTC, datetime, timedelta +from pathlib import Path + +import pytest + + +ROOT = Path(__file__).resolve().parents[2] +SCRIPT = ROOT / "yuqei-infra/scripts/scene-source-legal-review-signer.py" +SPEC = importlib.util.spec_from_file_location("scene_source_legal_review_signer", SCRIPT) +assert SPEC and SPEC.loader +signer = importlib.util.module_from_spec(SPEC) +sys.modules[SPEC.name] = signer +SPEC.loader.exec_module(signer) + +IDENTITY = { + "name": "测试法务审核人", + "organization": "测试组织（非真实发布）", + "qualification": "测试内部合同法务审核负责人", + "credential_reference": "TEST-APPOINTMENT-NOT-A-REAL-CREDENTIAL", +} +PASSPHRASE = "only-for-tests-一段足够长且不可预测的口令-9472" + + +@pytest.fixture(scope="module") +def frozen_request() -> tuple[dict, datetime]: + now = datetime.now(UTC).replace(microsecond=0) + head = subprocess.run( + ["git", "rev-parse", "HEAD"], + cwd=ROOT, + check=True, + capture_output=True, + text=True, + encoding="utf-8", + ).stdout.strip() + request = signer.legal_review.build_request( + ROOT, + candidate_commit=head, + created_at=signer._format_time(now - timedelta(hours=2)), + ) + return request, now + + +def _write_json(path: Path, value: object) -> Path: + path.write_bytes(signer._canonical_json(value)) + return path + + +def _initialize( + tmp_path: Path, + now: datetime, + *, + identity: dict | None = None, +) -> tuple[Path, Path, dict]: + state = tmp_path / "legal-state" + release_state = tmp_path / "release-state" + result = signer.initialize_key( + state, + identity=identity or IDENTITY, + passphrase=PASSPHRASE, + created_at=now - timedelta(days=1), + release_signer_directory=release_state, + ) + return state, release_state, result + + +def _request_file(tmp_path: Path, request: dict) -> Path: + return _write_json(tmp_path / "legal-review-request.json", request) + + +def _completed_template(template: dict, request: dict, now: datetime) -> dict: + completed = copy.deepcopy(template) + checked_at = signer._format_time(now - timedelta(minutes=10)) + for item, occurrence in zip(completed["items"], request["occurrences"], strict=True): + item["current_effectiveness_verified"] = True + item["transaction_applicability_verified"] = True + item["checked_at"] = checked_at + item["authoritative_status_url"] = occurrence["source"]["url"] + item["rationale"] = ( + "测试夹具中的人工判断说明，仅验证签名协议，不代表真实法律结论：" + + occurrence["occurrence_id"] + ) + return completed + + +def test_template_leaves_all_43_human_decisions_explicitly_unset( + tmp_path: Path, + frozen_request: tuple[dict, datetime], +) -> None: + request, now = frozen_request + state, release_state, _ = _initialize(tmp_path, now) + request_path = _request_file(tmp_path, request) + output = tmp_path / "template-bundle" + + result = signer.prepare_template( + state, + repo=ROOT, + request_path=request_path, + output_directory=output, + release_signer_directory=release_state, + ) + + template = json.loads((output / signer.DECISION_TEMPLATE_FILENAME).read_text("utf-8")) + assert result["human_decisions_prefilled"] == 0 + assert result["signature_created"] is False + assert len(template["items"]) == 43 + assert template["reviewed_at"] is None + assert template["expires_at"] is None + assert template["signature"] is None + assert all(item["current_effectiveness_verified"] is None for item in template["items"]) + assert all(item["transaction_applicability_verified"] is None for item in template["items"]) + assert all(item["checked_at"] is None for item in template["items"]) + assert all(item["authoritative_status_url"] is None for item in template["items"]) + assert all(item["rationale"] == "" for item in template["items"]) + + +@pytest.mark.parametrize( + ("field", "value"), + ( + ("current_effectiveness_verified", None), + ("current_effectiveness_verified", False), + ("transaction_applicability_verified", None), + ("transaction_applicability_verified", False), + ), +) +def test_signing_refuses_any_unset_or_false_human_decision_without_output( + tmp_path: Path, + frozen_request: tuple[dict, datetime], + field: str, + value: object, +) -> None: + request, now = frozen_request + state, release_state, _ = _initialize(tmp_path, now) + request_path = _request_file(tmp_path, request) + template = signer._template_for_request(request, IDENTITY) + completed = _completed_template(template, request, now) + completed["items"][0][field] = value + completed_path = _write_json(tmp_path / "completed.json", completed) + output = tmp_path / "signed-bundle" + + with pytest.raises(signer.LegalSignerError, match="no release-qualifying signature"): + signer.sign_completed_decision( + state, + repo=ROOT, + request_path=request_path, + completed_decision_path=completed_path, + output_directory=output, + passphrase=PASSPHRASE, + now=now, + release_signer_directory=release_state, + ) + + assert not output.exists() + + +def test_exact_four_field_identity_is_bound_to_keyring_and_decision( + tmp_path: Path, + frozen_request: tuple[dict, datetime], +) -> None: + request, now = frozen_request + state, release_state, result = _initialize(tmp_path, now) + registration, keyring, _ = signer._load_public_material( + state, release_signer_directory=release_state + ) + + assert registration["reviewer"] == IDENTITY + assert keyring["keys"][0]["reviewer_name"] == IDENTITY["name"] + assert keyring["keys"][0]["reviewer_organization"] == IDENTITY["organization"] + assert keyring["keys"][0]["reviewer_qualification"] == IDENTITY["qualification"] + assert keyring["keys"][0]["credential_reference"] == IDENTITY["credential_reference"] + assert result["keyring_raw_sha256"] == registration["keyring_raw_sha256"] + assert ( + signer.batch.validate_scene_legal_review_keyring( + keyring, + result["keyring_raw_sha256"], + expected_sha256=result["keyring_raw_sha256"], + ) + == keyring + ) + + request_path = _request_file(tmp_path, request) + completed = _completed_template(signer._template_for_request(request, IDENTITY), request, now) + completed["reviewer"]["organization"] = "被篡改的组织" + completed_path = _write_json(tmp_path / "wrong-identity.json", completed) + with pytest.raises(signer.LegalSignerError, match="identity does not match"): + signer.sign_completed_decision( + state, + repo=ROOT, + request_path=request_path, + completed_decision_path=completed_path, + output_directory=tmp_path / "must-not-exist", + passphrase=PASSPHRASE, + now=now, + release_signer_directory=release_state, + ) + + +def test_completed_decision_signs_once_and_offline_verifies_exact_bytes( + tmp_path: Path, + frozen_request: tuple[dict, datetime], +) -> None: + request, now = frozen_request + state, release_state, _ = _initialize(tmp_path, now) + request_path = _request_file(tmp_path, request) + completed = _completed_template(signer._template_for_request(request, IDENTITY), request, now) + completed_path = _write_json(tmp_path / "completed.json", completed) + output = tmp_path / "signed-bundle" + + result = signer.sign_completed_decision( + state, + repo=ROOT, + request_path=request_path, + completed_decision_path=completed_path, + output_directory=output, + passphrase=PASSPHRASE, + valid_hours=12, + now=now, + release_signer_directory=release_state, + ) + verified = signer.verify_signed_decision( + repo=ROOT, + request_path=request_path, + decision_path=output / signer.SIGNED_DECISION_FILENAME, + keyring_path=state / "keys" / signer.KEYRING_FILENAME, + checked_at=now, + ) + + assert result["status"] == "signed_and_offline_verified_not_published" + assert result["authorizes_publication"] is False + assert result["reviewed_occurrence_count"] == 43 + assert verified["status"] == "verified_qualified_human_review" + assert verified["authorizes_publication_by_itself"] is False + with pytest.raises(signer.LegalSignerError, match="already exists|refusing to overwrite"): + signer.sign_completed_decision( + state, + repo=ROOT, + request_path=request_path, + completed_decision_path=completed_path, + output_directory=output, + passphrase=PASSPHRASE, + valid_hours=12, + now=now, + release_signer_directory=release_state, + ) + + +def test_tampered_signed_decision_fails_offline_signature_verification( + tmp_path: Path, + frozen_request: tuple[dict, datetime], +) -> None: + request, now = frozen_request + state, release_state, _ = _initialize(tmp_path, now) + request_path = _request_file(tmp_path, request) + completed = _completed_template(signer._template_for_request(request, IDENTITY), request, now) + completed_path = _write_json(tmp_path / "completed.json", completed) + output = tmp_path / "signed-bundle" + signer.sign_completed_decision( + state, + repo=ROOT, + request_path=request_path, + completed_decision_path=completed_path, + output_directory=output, + passphrase=PASSPHRASE, + now=now, + release_signer_directory=release_state, + ) + tampered = json.loads((output / signer.SIGNED_DECISION_FILENAME).read_text("utf-8")) + tampered["items"][0]["rationale"] += "篡改" + tampered_path = _write_json(tmp_path / "tampered.json", tampered) + + with pytest.raises(signer.LegalSignerError, match="signature verification failed"): + signer.verify_signed_decision( + repo=ROOT, + request_path=request_path, + decision_path=tampered_path, + keyring_path=state / "keys" / signer.KEYRING_FILENAME, + checked_at=now, + ) + + +def test_release_signer_state_and_public_key_reuse_are_rejected( + tmp_path: Path, + frozen_request: tuple[dict, datetime], +) -> None: + _, now = frozen_request + release_state = tmp_path / "release-state" + with pytest.raises(signer.LegalSignerError, match="separate"): + signer.initialize_key( + release_state, + identity=IDENTITY, + passphrase=PASSPHRASE, + created_at=now, + release_signer_directory=release_state, + ) + + state, _, result = _initialize(tmp_path, now) + registration_directory = release_state / "keys" + registration_directory.mkdir(parents=True) + _write_json( + registration_directory / "reviewer-registration.json", + {"public_key_raw_sha256": result["public_key_raw_sha256"]}, + ) + with pytest.raises(signer.LegalSignerError, match="must not reuse"): + signer.status(state, release_signer_directory=release_state) + + +def test_private_key_bundle_is_owner_only_encrypted_and_atomically_published( + tmp_path: Path, + frozen_request: tuple[dict, datetime], + monkeypatch: pytest.MonkeyPatch, +) -> None: + _, now = frozen_request + success_root = tmp_path / "success" + success_root.mkdir() + state, release_state, _ = _initialize(success_root, now) + private_path = state / "keys" / signer.PRIVATE_KEY_FILENAME + raw = private_path.read_bytes() + assert raw.startswith(b"-----BEGIN ENCRYPTED PRIVATE KEY-----") + signer.secure_io._assert_private_file_security(private_path, "test legal private key") + if os.name != "nt": + assert stat.S_IMODE(private_path.stat().st_mode) & 0o077 == 0 + + failed_root = tmp_path / "failure" + failed_root.mkdir() + failed_state = failed_root / "legal-state" + failed_release = failed_root / "release-state" + + def fail_publish(_source: Path, _destination: Path) -> None: + raise signer.secure_io.SignerError("synthetic publication failure") + + monkeypatch.setattr(signer.secure_io, "_publish_directory_no_replace", fail_publish) + with pytest.raises(signer.LegalSignerError, match="synthetic publication failure"): + signer.initialize_key( + failed_state, + identity=IDENTITY, + passphrase=PASSPHRASE, + created_at=now, + release_signer_directory=failed_release, + ) + assert not (failed_state / "keys").exists()