feat: prepare named legal production approval
Build and Test Quality Gate / Contract Web typecheck and build (pull_request) Successful in 6m10s
Build and Test Quality Gate / Python tests and Contract API migration (pull_request) Failing after 23m2s
Security Quality Gate / npm audit, Bandit, pip-audit, detect-secrets (pull_request) Successful in 7m20s
Build and Test Quality Gate / Contract Web typecheck and build (pull_request) Successful in 6m10s
Build and Test Quality Gate / Python tests and Contract API migration (pull_request) Failing after 23m2s
Security Quality Gate / npm audit, Bandit, pip-audit, detect-secrets (pull_request) Successful in 7m20s
This commit is contained in:
@@ -0,0 +1,142 @@
|
||||
# V2 Named Legal Offline Signing Runbook
|
||||
|
||||
Date: 2026-08-11
|
||||
|
||||
## Boundary
|
||||
|
||||
This procedure automates evidence binding and cryptography. It does not make a legal
|
||||
decision, appoint a reviewer, or turn technical success into legal approval. A named
|
||||
legal reviewer must inspect the exact materials and personally authorize the signature.
|
||||
|
||||
The legal private key and its password must stay on the reviewer's controlled machine.
|
||||
Never place either file under the repository, `.runtime`, CI artifacts, chat, email, or
|
||||
the shared staging host. The helper reads the key only during `sign` and never copies or
|
||||
prints it.
|
||||
|
||||
## Inputs
|
||||
|
||||
- Exact shared-staging technical package: 10 legal QA results, 8 contract-review
|
||||
results, 10 knowledge-card IDs, and 18 distinct audit bindings.
|
||||
- Contract product golden summary generated by `run-contract-product-golden.py`. The
|
||||
helper reruns all 31 deterministic cases and refuses a stale or mismatched summary.
|
||||
- The reserved public HTTPS evidence URL on port 443:
|
||||
`https://api.contact.yuqei.com/provider/legal/provider-legal-review-evidence.json`.
|
||||
- The reviewer's real name, stable approver ID, role, organization, key ID, approval
|
||||
time, and expiry time. Approval validity cannot exceed 30 days.
|
||||
|
||||
The current eligible shared-staging package was created on 2026-08-11 for tenant
|
||||
`default`, Provider configuration `provider-914ba619fe4b`, Provider `ccswitch`, and
|
||||
model `gpt-5.6-sol`. Its exact SHA-256 is
|
||||
`ed1830430e8049397800cc75e9c24bac85a093a268bb669cf9d8db862adac9f2`. Do not use the
|
||||
older failed 2026-08-10 package or any reformatted copy.
|
||||
|
||||
The canonical technical/product evidence is already published at the reserved URL.
|
||||
Its exact size is 33,582 bytes and its SHA-256 is
|
||||
`d3affadde5f3665a619c0424cc9e80c1cc9ee9af4eb157cba3cd558dabbbd7e4`. A legal
|
||||
reviewer must still inspect those bytes and the bound source materials; publication is
|
||||
not approval.
|
||||
|
||||
## 1. Prepare Exact Evidence And Payload
|
||||
|
||||
Run from the repository root. Replace every angle-bracket value. The output remains
|
||||
unsigned and is not legal approval.
|
||||
|
||||
```powershell
|
||||
python yuqei-infra/scripts/prepare-provider-legal-approval.py prepare `
|
||||
--review-package-file .runtime/production-closure/provider/provider-technical-review-package.json `
|
||||
--golden-summary-file .runtime/v2-contract-product-golden/final-20260810/contract-product-golden-summary.json `
|
||||
--evidence-ref https://api.contact.yuqei.com/provider/legal/provider-legal-review-evidence.json `
|
||||
--key-id <legal-key-id> `
|
||||
--approver-id <legal-work-email> `
|
||||
--reviewer "<real-reviewer-name>" `
|
||||
--role "<legal-role>" `
|
||||
--organization "<organization>" `
|
||||
--approved-at <ISO-8601-with-timezone> `
|
||||
--expires-at <ISO-8601-with-timezone> `
|
||||
--evidence-output .runtime/production-closure/provider/provider-legal-review-evidence.json `
|
||||
--payload-output .runtime/production-closure/provider/provider-legal-approval-payload.json
|
||||
```
|
||||
|
||||
Record the three SHA-256 values printed by the command. Upload the evidence output as
|
||||
exact bytes, without a newline, reformatting, compression, redirect, authentication,
|
||||
query string, or content transformation. The finalize service fetches it again and
|
||||
rejects any byte-level hash difference.
|
||||
|
||||
Before signing, the reviewer must inspect the technical package, all 10 knowledge
|
||||
cards, 18 Provider outputs, all 31 product-golden cases, both generated files, their
|
||||
recorded hashes, and the final HTTPS object returned by the evidence URL.
|
||||
|
||||
## 2. Keep And Publish The Legal Key Correctly
|
||||
|
||||
Generate an encrypted Ed25519 key only on the reviewer's controlled machine. The exact
|
||||
key-management procedure may instead use the organization's approved key service.
|
||||
|
||||
```powershell
|
||||
openssl genpkey -algorithm ED25519 -aes-256-cbc -out legal-approval-private.pem
|
||||
openssl pkey -in legal-approval-private.pem -pubout -out legal-approval-public.pem
|
||||
```
|
||||
|
||||
Create a one-line password file outside the repository with access limited to the
|
||||
reviewer. Then create a candidate public keyring. Security/operations must independently
|
||||
approve and install this public keyring on shared staging; generation is not trust.
|
||||
|
||||
```powershell
|
||||
python yuqei-infra/scripts/prepare-provider-legal-approval.py keyring `
|
||||
--public-key-file legal-approval-public.pem `
|
||||
--key-id <legal-key-id> `
|
||||
--approver-id <legal-work-email> `
|
||||
--reviewer "<real-reviewer-name>" `
|
||||
--role "<legal-role>" `
|
||||
--organization "<organization>" `
|
||||
--not-before <ISO-8601-with-timezone> `
|
||||
--not-after <ISO-8601-with-timezone> `
|
||||
--output .runtime/production-closure/provider/legal-trusted-keyring.candidate.json
|
||||
```
|
||||
|
||||
## 3. Reviewer Signs The Exact Payload
|
||||
|
||||
The explicit confirmation string is an intentional-action guard. It is not a substitute
|
||||
for the reviewer's legal decision.
|
||||
|
||||
```powershell
|
||||
python yuqei-infra/scripts/prepare-provider-legal-approval.py sign `
|
||||
--payload-file .runtime/production-closure/provider/provider-legal-approval-payload.json `
|
||||
--evidence-file .runtime/production-closure/provider/provider-legal-review-evidence.json `
|
||||
--review-package-file .runtime/production-closure/provider/provider-technical-review-package.json `
|
||||
--golden-summary-file .runtime/v2-contract-product-golden/final-20260810/contract-product-golden-summary.json `
|
||||
--private-key-file legal-approval-private.pem `
|
||||
--private-key-password-file <protected-password-file> `
|
||||
--confirmation I_APPROVE_THIS_EXACT_PAYLOAD `
|
||||
--output .runtime/production-closure/provider/named-legal-approval.json
|
||||
```
|
||||
|
||||
## 4. Verify Before Finalize
|
||||
|
||||
Use the keyring actually installed as trusted input, not merely the candidate file.
|
||||
|
||||
```powershell
|
||||
python yuqei-infra/scripts/prepare-provider-legal-approval.py verify `
|
||||
--approval-file .runtime/production-closure/provider/named-legal-approval.json `
|
||||
--keyring-file <installed-legal-trusted-keyring.json> `
|
||||
--evidence-file .runtime/production-closure/provider/provider-legal-review-evidence.json `
|
||||
--review-package-file .runtime/production-closure/provider/provider-technical-review-package.json `
|
||||
--golden-summary-file .runtime/v2-contract-product-golden/final-20260810/contract-product-golden-summary.json
|
||||
```
|
||||
|
||||
Only a `status=verified` result is eligible for the existing shared-staging `finalize`
|
||||
flow. `finalize` still must fetch the public HTTPS evidence, validate the host allowlist,
|
||||
register the immutable acceptance manifest, and issue the separate promotion receipt.
|
||||
|
||||
## Fail-Closed Conditions
|
||||
|
||||
- Technical package is not the exact canonical shared-staging package.
|
||||
- Any of the 18 Provider results is missing, duplicated, or failed.
|
||||
- The 31-case product suite is stale, mismatched, incomplete, or technically failed.
|
||||
- Evidence or unsigned payload bytes were reformatted or changed.
|
||||
- Identity differs between the payload and trusted keyring.
|
||||
- Key is unknown, duplicated, revoked, outside its validity window, or not Ed25519.
|
||||
- Approval is future-dated, expired, longer than policy, tampered, or cross-tenant.
|
||||
- Evidence URL is not HTTPS port 443 or contains credentials, query, or fragment.
|
||||
|
||||
Do not run `finalize`, create an `acceptance_run_id`, or claim product legal approval
|
||||
after any failure.
|
||||
@@ -0,0 +1,68 @@
|
||||
# AI 合同平台 V2 生产外部门禁闭环
|
||||
|
||||
日期:2026-08-11
|
||||
|
||||
## 当前结论
|
||||
|
||||
六项外部门禁已收敛为一个必须由真实自然人完成的入口:具名法务本人对精确技术包和产品黄金套件作出批准并使用本人控制的 Ed25519 私钥签名。AI 不得代签,也不得把技术通过写成法务批准。
|
||||
|
||||
在该签名到位前,正式生产保持 No-Go:`yuqei-v2-production` 容器数为 0,`api.contact.yuqei.com` 根路径返回 503,未切换 DNS 或业务流量。签名到位后,余下 finalize、receipt pin、production preflight、备份恢复、双节点 canary、撤销和 kill switch 演练均已有确定命令和基础设施,不再需要新增产品代码。
|
||||
|
||||
## 六项状态
|
||||
|
||||
| 门禁 | 状态 | 已完成证据 | 剩余动作 |
|
||||
| --- | --- | --- | --- |
|
||||
| 1. 具名法务签署 | 外部阻塞 | 固定包 SHA-256 `ed1830430e8049397800cc75e9c24bac85a093a268bb669cf9d8db862adac9f2`;18 条 Provider 结果、10 张知识卡、31 条合同生成/条款推荐黄金用例均完成技术对账;离线签署/验签工具和 runbook 已完成 | 法务本人提供真实身份字段,审阅并亲自签名 |
|
||||
| 2. acceptance receipt / promotion | 等待门禁 1 | receipt Ed25519 私钥只保存在共享预发布主机,公钥 keyring 已放入 production 验证目录;AI 镜像 digest 已固定 | 运行 `finalize`,生成真实 `acceptance_run_id`、receipt v2 和 promotion bundle |
|
||||
| 3. production env / TLS / DNS / 备份 | 准备完成,等待 receipt | `api.contact.yuqei.com` DNS 有效;专用 Let's Encrypt 证书有效至 2026-11-09;production env draft 权限 `0600`;preflight `21/24`,仅 receipt 三项失败;Compose config 通过;阿里云 NFSv4 异地挂载 70GB 可用 | 写入 receipt 和 SHA pin,将 draft 原子提升为正式 env;启动后执行首份备份及隔离恢复演练 |
|
||||
| 4. 双节点 canary | 准备完成,等待 receipt 和主节点 | `192.168.3.90` 与阿里云 `47.94.178.212` 为两个独立主机,通过 WireGuard 互联;第二节点已按 digest 拉取 Contract API 和 AI API 镜像并注销 Registry 凭据 | 主节点迁移完成后启动第二节点只读 AI canary,运行统一指纹、撤销 SLA 和 kill switch 验证 |
|
||||
| 5. support access | 生产首发闭环为禁用 | 运行 OpenAPI 中 `support-access`、`support_access`、`execute` 路径为 0;执行器恒为 false,默认 kill switch 为 true;无破坏性身份写 API | 首发不开放该能力。未来启用必须另行接入真实 IdP/MFA/会话设备绑定并取得隐私和法务具名批准 |
|
||||
| 6. 容量 | 完成 | 两轮只清理未引用 build cache;BuildKit 可回收量为 0;生产 PostgreSQL、MinIO、合同文件卷绑定到 `/home` 独立 LV(198GB 可用);异地备份 70GB 可用 | 上线后纳入磁盘告警;不得清理业务卷或 Registry 包 |
|
||||
|
||||
## 已发布的不可变镜像
|
||||
|
||||
- Contract API:`git.yuqei.com/yuqei/ai-contract-platform-contract-api@sha256:a6969d90680f866232f1ad2c09607fd63f226476028b6c0f34c3175e7407549f`
|
||||
- AI Platform API:`git.yuqei.com/yuqei/ai-contract-platform-ai-api@sha256:2924e68c337bc59be6bc45e6183f8dcf239d80151c4ca076ace4feee644f5d67`
|
||||
- AI worker:`git.yuqei.com/yuqei/ai-contract-platform-worker@sha256:a72d40a5a000e525c1b1d30fb2f2a214db8a32492308198783323be0817dbecc`
|
||||
- Contract Web:`git.yuqei.com/yuqei/ai-contract-platform-web@sha256:8132c515502668b9348a16dbb2c56154a65582af5c4c8efbae887538f0b59e33`
|
||||
- Release ID:`c24f7a4f1ee041d92feca96bf147e423e58da17c`
|
||||
|
||||
生产必须按 digest 拉取,不得只使用 tag 或本地 image ID。
|
||||
|
||||
## 法务 evidence
|
||||
|
||||
公开只读地址:
|
||||
|
||||
`https://api.contact.yuqei.com/provider/legal/provider-legal-review-evidence.json`
|
||||
|
||||
- HTTP 200
|
||||
- TLS 验证通过
|
||||
- 重定向数 0
|
||||
- 精确大小 33,582 bytes
|
||||
- SHA-256 `d3affadde5f3665a619c0424cc9e80c1cc9ee9af4eb157cba3cd558dabbbd7e4`
|
||||
|
||||
公网重新下载的文件与本地 canonical evidence 逐字节哈希一致。公开 evidence 只证明技术材料固定,不表示已经取得法务批准。
|
||||
|
||||
## 用户最小配合项
|
||||
|
||||
请提供一位实际承担审批责任的法务人员的:
|
||||
|
||||
1. 真实姓名;
|
||||
2. 稳定的工作 ID,优先使用工作邮箱;
|
||||
3. 职务;
|
||||
4. 组织全称;
|
||||
5. 法务公钥 `key_id`,建议 `provider-legal-2026q3`;
|
||||
6. 批准有效窗口,最长 30 天。
|
||||
|
||||
法务本人需在受控电脑上生成并保管 Ed25519 私钥,按 `V2-named-legal-offline-signing-runbook-2026-08-11.md` 审阅、签署。私钥和口令不得上传仓库、聊天、共享预发布或 CI artifact。安全/运维只接收法务公钥并将其加入 trusted keyring。
|
||||
|
||||
## 签名后的固定自动顺序
|
||||
|
||||
1. 验证法务签名、trusted keyring 和公网 evidence exact hash。
|
||||
2. 在共享预发布执行 `finalize`,生成真实 `acceptance_run_id`、receipt v2 和 promotion bundle。
|
||||
3. 将 receipt、公钥 keyring 和 receipt SHA-256 pin 安装到 production,保持 kill switch 为 true 运行正式 preflight。
|
||||
4. preflight 全通过后拉取镜像,启动主节点但不接公网流量,完成迁移、备份和隔离恢复演练。
|
||||
5. 启动阿里云第二只读节点,验证两节点同一 release/image/receipt/key 指纹。
|
||||
6. 演练 receipt 撤销传播 SLA 和 kill switch;随后恢复到 fail-closed 状态。
|
||||
7. 由用户单独授权后才修改 API 上游和切换流量;support access 和身份 execute API 继续关闭。
|
||||
|
||||
+6
-1
@@ -87,11 +87,16 @@ class ProviderLegalApprovalKeyring(StrictModel):
|
||||
def sign_legal_approval(
|
||||
payload: ProviderLegalApprovalPayload,
|
||||
private_key_pem: bytes,
|
||||
*,
|
||||
private_key_password: bytes | None = None,
|
||||
) -> SignedProviderLegalApproval:
|
||||
problems = validate_legal_approval_payload(payload)
|
||||
if problems:
|
||||
raise ValueError("legal approval payload is invalid: " + "; ".join(problems))
|
||||
key = serialization.load_pem_private_key(private_key_pem, password=None)
|
||||
key = serialization.load_pem_private_key(
|
||||
private_key_pem,
|
||||
password=private_key_password,
|
||||
)
|
||||
if not isinstance(key, Ed25519PrivateKey):
|
||||
raise ValueError("legal approval signing key must be an Ed25519 private key")
|
||||
canonical = canonical_legal_approval_payload(payload)
|
||||
|
||||
@@ -114,6 +114,30 @@ def test_named_legal_approval_signs_and_verifies_idempotently() -> None:
|
||||
assert approval.approval_id.startswith("provider-legal-approval-")
|
||||
|
||||
|
||||
def test_named_legal_approval_supports_encrypted_private_key() -> None:
|
||||
private = Ed25519PrivateKey.generate()
|
||||
password = b"test-password-not-a-production-secret"
|
||||
encrypted_pem = private.private_bytes(
|
||||
serialization.Encoding.PEM,
|
||||
serialization.PrivateFormat.PKCS8,
|
||||
serialization.BestAvailableEncryption(password),
|
||||
)
|
||||
public_pem = private.public_key().public_bytes(
|
||||
serialization.Encoding.PEM,
|
||||
serialization.PublicFormat.SubjectPublicKeyInfo,
|
||||
).decode("ascii")
|
||||
_unused, key = _key_material()
|
||||
key = key.model_copy(update={"public_key_pem": public_pem})
|
||||
|
||||
approval = sign_legal_approval(
|
||||
_payload(),
|
||||
encrypted_pem,
|
||||
private_key_password=password,
|
||||
)
|
||||
|
||||
assert _verify(approval, ProviderLegalApprovalKeyring(version=1, keys=[key])) == []
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("field", "value", "problem"),
|
||||
[
|
||||
|
||||
@@ -0,0 +1,736 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Prepare, sign, and verify an offline named-legal Provider approval."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import csv
|
||||
import importlib.util
|
||||
import ipaddress
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
from datetime import UTC, datetime
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from urllib.parse import urlsplit
|
||||
|
||||
from cryptography.hazmat.primitives import serialization
|
||||
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
|
||||
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[2]
|
||||
AI_PLATFORM_SRC = REPO_ROOT / "yuqei-ai-platform" / "services" / "ai-platform-api" / "src"
|
||||
if str(AI_PLATFORM_SRC) not in sys.path:
|
||||
sys.path.insert(0, str(AI_PLATFORM_SRC))
|
||||
|
||||
from yuqei_ai_platform_api.provider_legal_approval import ( # noqa: E402
|
||||
LEGAL_APPROVAL_ALGORITHM,
|
||||
LEGAL_APPROVAL_VERSION,
|
||||
LegalApprovalAuditBinding,
|
||||
ProviderLegalApprovalKeyring,
|
||||
ProviderLegalApprovalPayload,
|
||||
ProviderLegalApprovalPublicKey,
|
||||
SignedProviderLegalApproval,
|
||||
canonical_legal_approval_payload,
|
||||
load_legal_approval_file,
|
||||
load_legal_keyring_file,
|
||||
sign_legal_approval,
|
||||
validate_legal_approval_payload,
|
||||
verify_legal_approval,
|
||||
)
|
||||
from yuqei_ai_platform_api.provider_acceptance import expected_acceptance_inventory # noqa: E402
|
||||
from yuqei_ai_platform_api.provider_review_package import ( # noqa: E402
|
||||
ProviderTechnicalReviewPackage,
|
||||
canonical_json,
|
||||
canonical_review_package,
|
||||
review_package_sha256,
|
||||
sha256_bytes,
|
||||
)
|
||||
|
||||
|
||||
EVIDENCE_VERSION = "provider-legal-review-evidence-v1"
|
||||
EVIDENCE_SIGNATURE_REQUIREMENT = "detached_named_legal_signature_required"
|
||||
SIGN_CONFIRMATION = "I_APPROVE_THIS_EXACT_PAYLOAD"
|
||||
MAX_INPUT_BYTES = 10 * 1024 * 1024
|
||||
MAX_KEY_BYTES = 64 * 1024
|
||||
GOLDEN_SCRIPT = REPO_ROOT / "yuqei-infra" / "scripts" / "run-contract-product-golden.py"
|
||||
PHASE13_ROOT = (
|
||||
REPO_ROOT
|
||||
/ "yuqei-ai-platform"
|
||||
/ "services"
|
||||
/ "ai-platform-api"
|
||||
/ "src"
|
||||
/ "yuqei_ai_platform_api"
|
||||
/ "golden_assets"
|
||||
/ "v1_phase13_legal_knowledge_template_library"
|
||||
)
|
||||
GENERATION_INVENTORY = PHASE13_ROOT / "inventories" / "generation-golden-sample-inventory.csv"
|
||||
|
||||
|
||||
def _strict_json(raw: bytes, *, label: str) -> Any:
|
||||
def reject_duplicates(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
|
||||
result: dict[str, Any] = {}
|
||||
for key, value in pairs:
|
||||
if key in result:
|
||||
raise ValueError(f"{label} contains duplicate JSON key: {key}")
|
||||
result[key] = value
|
||||
return result
|
||||
|
||||
try:
|
||||
return json.loads(raw.decode("utf-8"), object_pairs_hook=reject_duplicates)
|
||||
except (UnicodeDecodeError, json.JSONDecodeError) as error:
|
||||
raise ValueError(f"{label} is not valid UTF-8 JSON") from error
|
||||
|
||||
|
||||
def _read_bounded(path: Path, *, label: str, limit: int = MAX_INPUT_BYTES) -> bytes:
|
||||
resolved = path.resolve(strict=True)
|
||||
if not resolved.is_file():
|
||||
raise ValueError(f"{label} is not a regular file")
|
||||
with resolved.open("rb") as stream:
|
||||
raw = stream.read(limit + 1)
|
||||
if len(raw) > limit:
|
||||
raise ValueError(f"{label} exceeds the size limit")
|
||||
return raw
|
||||
|
||||
|
||||
def _write_atomic(path: Path, raw: bytes, *, protected_paths: list[Path] | None = None) -> None:
|
||||
target = path.resolve()
|
||||
if target.suffix.lower() != ".json":
|
||||
raise ValueError("output must be a JSON file")
|
||||
protected = {item.resolve() for item in (protected_paths or [])}
|
||||
if target in protected:
|
||||
raise ValueError("output must not overwrite an input or key file")
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
descriptor, temporary = tempfile.mkstemp(prefix=f".{target.name}.", dir=target.parent)
|
||||
try:
|
||||
with os.fdopen(descriptor, "wb") as stream:
|
||||
stream.write(raw)
|
||||
stream.flush()
|
||||
os.fsync(stream.fileno())
|
||||
os.replace(temporary, target)
|
||||
except Exception:
|
||||
try:
|
||||
os.unlink(temporary)
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
raise
|
||||
|
||||
|
||||
def load_exact_review_package(path: Path) -> tuple[ProviderTechnicalReviewPackage, bytes]:
|
||||
raw = _read_bounded(path, label="technical review package")
|
||||
try:
|
||||
package = ProviderTechnicalReviewPackage.model_validate(
|
||||
_strict_json(raw, label="technical review package")
|
||||
)
|
||||
except (TypeError, ValueError):
|
||||
raise ValueError("technical review package is invalid") from None
|
||||
canonical = canonical_review_package(package)
|
||||
if raw != canonical:
|
||||
raise ValueError("technical review package is not in its exact canonical form")
|
||||
if len(package.cases) != 18 or any(item.technical_status != "passed" for item in package.cases):
|
||||
raise ValueError("technical review package must contain 18 passing results")
|
||||
_validate_provider_package_inventory(package)
|
||||
return package, raw
|
||||
|
||||
|
||||
def _validate_provider_package_inventory(package: ProviderTechnicalReviewPackage) -> None:
|
||||
expected = expected_acceptance_inventory()
|
||||
inventory_checks = (
|
||||
(package.qa_sample_ids, expected["qa_sample_ids"], "legal QA"),
|
||||
(package.review_sample_ids, expected["review_sample_ids"], "contract review"),
|
||||
(package.knowledge_card_ids, expected["knowledge_card_ids"], "knowledge card"),
|
||||
)
|
||||
for actual, required, label in inventory_checks:
|
||||
if len(actual) != len(set(actual)) or set(actual) != set(required):
|
||||
raise ValueError(
|
||||
f"technical review package {label} inventory does not match bundled assets"
|
||||
)
|
||||
if package.suite_manifest_sha256 != expected["suite_manifest_sha256"]:
|
||||
raise ValueError("technical review package suite manifest does not match bundled assets")
|
||||
if package.knowledge_manifest_sha256 != expected["knowledge_manifest_sha256"]:
|
||||
raise ValueError(
|
||||
"technical review package knowledge manifest does not match bundled assets"
|
||||
)
|
||||
|
||||
expected_cases = {
|
||||
*(("legal_qa", item) for item in package.qa_sample_ids),
|
||||
*(("contract_review", item) for item in package.review_sample_ids),
|
||||
}
|
||||
actual_cases = {(item.suite, item.sample_id) for item in package.cases}
|
||||
audit_ids = [item.audit_run_id for item in package.cases]
|
||||
if actual_cases != expected_cases or len(set(audit_ids)) != 18:
|
||||
raise ValueError("technical review package case or audit inventory is incomplete")
|
||||
for item in package.cases:
|
||||
if item.provider != package.provider or item.model != package.model:
|
||||
raise ValueError("technical review package case Provider identity is inconsistent")
|
||||
if item.output_sha256 != sha256_bytes(canonical_json(item.output)):
|
||||
raise ValueError("technical review package output hash is inconsistent")
|
||||
|
||||
expected_assets: dict[str, tuple[str, int]] = {}
|
||||
for asset_path in sorted(item for item in PHASE13_ROOT.rglob("*") if item.is_file()):
|
||||
raw = asset_path.read_bytes()
|
||||
expected_assets[asset_path.relative_to(PHASE13_ROOT).as_posix()] = (
|
||||
sha256_bytes(raw),
|
||||
len(raw),
|
||||
)
|
||||
actual_assets = {item.path: (item.sha256, item.size_bytes) for item in package.assets}
|
||||
if len(actual_assets) != len(package.assets) or actual_assets != expected_assets:
|
||||
raise ValueError("technical review package asset inventory does not match bundled assets")
|
||||
|
||||
|
||||
def _load_golden_module() -> Any:
|
||||
name = "yuqei_contract_product_golden_for_legal_review"
|
||||
module = sys.modules.get(name)
|
||||
if module is not None:
|
||||
return module
|
||||
spec = importlib.util.spec_from_file_location(name, GOLDEN_SCRIPT)
|
||||
if spec is None or spec.loader is None:
|
||||
raise RuntimeError("contract product golden runner could not be loaded")
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
sys.modules[name] = module
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
def _load_golden_summary(path: Path) -> tuple[dict[str, Any], bytes]:
|
||||
raw = _read_bounded(path, label="contract product golden summary")
|
||||
parsed = _strict_json(raw, label="contract product golden summary")
|
||||
if not isinstance(parsed, dict):
|
||||
raise ValueError("contract product golden summary must be an object")
|
||||
return parsed, raw
|
||||
|
||||
|
||||
def _safe_asset(relative: str) -> tuple[str, str, int]:
|
||||
root = PHASE13_ROOT.resolve()
|
||||
candidate = (root / relative).resolve(strict=True)
|
||||
try:
|
||||
candidate.relative_to(root)
|
||||
except ValueError as error:
|
||||
raise ValueError("contract product golden asset escapes its root") from error
|
||||
if not candidate.is_file():
|
||||
raise ValueError("contract product golden asset is not a regular file")
|
||||
raw = candidate.read_bytes()
|
||||
return relative.replace("\\", "/"), sha256_bytes(raw), len(raw)
|
||||
|
||||
|
||||
def build_product_golden_evidence(summary_file: Path) -> dict[str, Any]:
|
||||
supplied_summary, summary_raw = _load_golden_summary(summary_file)
|
||||
golden = _load_golden_module()
|
||||
results = golden.run_product_golden()
|
||||
actual_summary = golden.build_summary(results)
|
||||
if supplied_summary != actual_summary:
|
||||
raise ValueError("contract product golden summary does not match a fresh deterministic run")
|
||||
counts = actual_summary.get("counts", {})
|
||||
if (
|
||||
actual_summary.get("status") not in {"needs_legal_review", "passed"}
|
||||
or counts.get("total") != 31
|
||||
or counts.get("contract_generation") != 23
|
||||
or counts.get("clause_recommendation") != 8
|
||||
or counts.get("technical_passed") != 31
|
||||
or counts.get("technical_failed") != 0
|
||||
or actual_summary.get("problems") != []
|
||||
):
|
||||
raise ValueError("contract product golden suite has not passed all 31 technical cases")
|
||||
|
||||
result_by_id = {item.sample_id: item for item in results}
|
||||
if len(result_by_id) != 31:
|
||||
raise ValueError("contract product golden results are incomplete or duplicated")
|
||||
with GENERATION_INVENTORY.open("r", encoding="utf-8-sig", newline="") as stream:
|
||||
rows = list(csv.DictReader(stream))
|
||||
if len(rows) != 31 or {row["sample_id"] for row in rows} != set(result_by_id):
|
||||
raise ValueError("contract product golden inventory does not match the executed results")
|
||||
|
||||
cases: list[dict[str, Any]] = []
|
||||
for row in rows:
|
||||
result = result_by_id[row["sample_id"]]
|
||||
if result.technical_status != "passed" or result.problems:
|
||||
raise ValueError("contract product golden evidence contains a failed case")
|
||||
variable_path, variable_hash, variable_size = _safe_asset(row["variable_payload_path"])
|
||||
document_path, document_hash, document_size = _safe_asset(row["expected_document_path"])
|
||||
cases.append(
|
||||
{
|
||||
"sample_id": result.sample_id,
|
||||
"suite": result.suite,
|
||||
"template_id": row["template_id"],
|
||||
"technical_status": result.technical_status,
|
||||
"legal_status_before_signature": result.legal_status,
|
||||
"placeholder_count": result.placeholder_count,
|
||||
"variable_asset": {
|
||||
"path": variable_path,
|
||||
"sha256": variable_hash,
|
||||
"size_bytes": variable_size,
|
||||
},
|
||||
"expected_document_asset": {
|
||||
"path": document_path,
|
||||
"sha256": document_hash,
|
||||
"size_bytes": document_size,
|
||||
},
|
||||
"expected_review_rule_ids": sorted(
|
||||
golden.split_ids(row["expected_review_rule_ids"])
|
||||
),
|
||||
"expected_source_citations": sorted(
|
||||
golden.split_ids(row["expected_source_citations"])
|
||||
),
|
||||
}
|
||||
)
|
||||
cases.sort(key=lambda item: (item["suite"], item["sample_id"]))
|
||||
inventory_raw = GENERATION_INVENTORY.read_bytes()
|
||||
return {
|
||||
"summary_file_sha256": sha256_bytes(summary_raw),
|
||||
"inventory_file_sha256": sha256_bytes(inventory_raw),
|
||||
"summary": actual_summary,
|
||||
"cases": cases,
|
||||
}
|
||||
|
||||
|
||||
def build_review_evidence(
|
||||
review_package_file: Path,
|
||||
golden_summary_file: Path,
|
||||
) -> tuple[dict[str, Any], ProviderTechnicalReviewPackage]:
|
||||
package, package_raw = load_exact_review_package(review_package_file)
|
||||
audit_bindings = sorted(
|
||||
(
|
||||
{
|
||||
"suite": item.suite,
|
||||
"sample_id": item.sample_id,
|
||||
"audit_run_id": item.audit_run_id,
|
||||
}
|
||||
for item in package.cases
|
||||
),
|
||||
key=lambda item: (item["suite"], item["sample_id"], item["audit_run_id"]),
|
||||
)
|
||||
evidence = {
|
||||
"evidence_version": EVIDENCE_VERSION,
|
||||
"signature_requirement": EVIDENCE_SIGNATURE_REQUIREMENT,
|
||||
"provider_technical_review": {
|
||||
"package_sha256": sha256_bytes(package_raw),
|
||||
"package_version": package.package_version,
|
||||
"tenant_id": package.tenant_id,
|
||||
"source_environment": package.source_environment,
|
||||
"provider_config_id": package.provider_config_id,
|
||||
"provider": package.provider,
|
||||
"model": package.model,
|
||||
"qa_sample_ids": sorted(package.qa_sample_ids),
|
||||
"review_sample_ids": sorted(package.review_sample_ids),
|
||||
"knowledge_card_ids": sorted(package.knowledge_card_ids),
|
||||
"audit_bindings": audit_bindings,
|
||||
"suite_manifest_sha256": package.suite_manifest_sha256,
|
||||
"knowledge_manifest_sha256": package.knowledge_manifest_sha256,
|
||||
},
|
||||
"contract_product_golden": build_product_golden_evidence(golden_summary_file),
|
||||
"decision_boundary": (
|
||||
"This evidence proves technical completeness only. Legal approval exists only after a "
|
||||
"named reviewer signs the separately generated canonical approval payload."
|
||||
),
|
||||
}
|
||||
return evidence, package
|
||||
|
||||
|
||||
def validate_exact_evidence(
|
||||
evidence_file: Path,
|
||||
review_package_file: Path,
|
||||
golden_summary_file: Path,
|
||||
) -> tuple[dict[str, Any], ProviderTechnicalReviewPackage, bytes]:
|
||||
raw = _read_bounded(evidence_file, label="legal review evidence")
|
||||
parsed = _strict_json(raw, label="legal review evidence")
|
||||
expected, package = build_review_evidence(review_package_file, golden_summary_file)
|
||||
expected_raw = canonical_json(expected)
|
||||
if raw != expected_raw or parsed != expected:
|
||||
raise ValueError(
|
||||
"legal review evidence is not the exact canonical evidence for these inputs"
|
||||
)
|
||||
return expected, package, raw
|
||||
|
||||
|
||||
def _parse_time(value: str, *, label: str) -> datetime:
|
||||
try:
|
||||
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
|
||||
except ValueError as error:
|
||||
raise ValueError(f"{label} is not a valid ISO-8601 timestamp") from error
|
||||
if parsed.tzinfo is None or parsed.utcoffset() is None:
|
||||
raise ValueError(f"{label} must include a timezone")
|
||||
return parsed.astimezone(UTC)
|
||||
|
||||
|
||||
def _validate_evidence_ref(value: str) -> str:
|
||||
parsed = urlsplit(value.strip())
|
||||
hostname = parsed.hostname.lower().rstrip(".") if parsed.hostname else ""
|
||||
try:
|
||||
literal = ipaddress.ip_address(hostname)
|
||||
except ValueError:
|
||||
literal = None
|
||||
private_literal = bool(
|
||||
literal
|
||||
and (
|
||||
literal.is_private
|
||||
or literal.is_loopback
|
||||
or literal.is_link_local
|
||||
or literal.is_multicast
|
||||
or literal.is_unspecified
|
||||
or literal.is_reserved
|
||||
)
|
||||
)
|
||||
if (
|
||||
parsed.scheme.lower() != "https"
|
||||
or not hostname
|
||||
or parsed.username
|
||||
or parsed.password
|
||||
or parsed.port not in (None, 443)
|
||||
or parsed.query
|
||||
or parsed.fragment
|
||||
or hostname == "localhost"
|
||||
or private_literal
|
||||
):
|
||||
raise ValueError(
|
||||
"evidence reference must be a public-style HTTPS URL on port 443 "
|
||||
"without credentials, query, or fragment"
|
||||
)
|
||||
return value.strip()
|
||||
|
||||
|
||||
def build_approval_payload(
|
||||
*,
|
||||
package: ProviderTechnicalReviewPackage,
|
||||
evidence_ref: str,
|
||||
evidence_raw: bytes,
|
||||
key_id: str,
|
||||
approver_id: str,
|
||||
reviewer: str,
|
||||
role: str,
|
||||
organization: str,
|
||||
approved_at: datetime,
|
||||
expires_at: datetime,
|
||||
) -> ProviderLegalApprovalPayload:
|
||||
payload = ProviderLegalApprovalPayload(
|
||||
approval_version=LEGAL_APPROVAL_VERSION,
|
||||
algorithm=LEGAL_APPROVAL_ALGORITHM,
|
||||
key_id=key_id,
|
||||
decision="approved",
|
||||
tenant_id=package.tenant_id,
|
||||
approver_id=approver_id,
|
||||
reviewer=reviewer,
|
||||
role=role,
|
||||
organization=organization,
|
||||
approved_at=approved_at,
|
||||
expires_at=expires_at,
|
||||
provider_config_id=package.provider_config_id,
|
||||
provider=package.provider,
|
||||
model=package.model,
|
||||
qa_sample_ids=package.qa_sample_ids,
|
||||
review_sample_ids=package.review_sample_ids,
|
||||
knowledge_card_ids=package.knowledge_card_ids,
|
||||
audit_bindings=[
|
||||
LegalApprovalAuditBinding(
|
||||
suite=item.suite,
|
||||
sample_id=item.sample_id,
|
||||
audit_run_id=item.audit_run_id,
|
||||
)
|
||||
for item in package.cases
|
||||
],
|
||||
suite_manifest_sha256=package.suite_manifest_sha256,
|
||||
knowledge_manifest_sha256=package.knowledge_manifest_sha256,
|
||||
technical_evidence_sha256=review_package_sha256(package),
|
||||
evidence_ref=_validate_evidence_ref(evidence_ref),
|
||||
evidence_sha256=sha256_bytes(evidence_raw),
|
||||
)
|
||||
problems = validate_legal_approval_payload(payload)
|
||||
if problems:
|
||||
raise ValueError("legal approval payload is invalid: " + "; ".join(problems))
|
||||
return payload
|
||||
|
||||
|
||||
def verify_offline_approval(
|
||||
*,
|
||||
approval_file: Path,
|
||||
keyring_file: Path,
|
||||
evidence_file: Path,
|
||||
review_package_file: Path,
|
||||
golden_summary_file: Path,
|
||||
checked_at: datetime | None = None,
|
||||
) -> SignedProviderLegalApproval:
|
||||
_evidence, package, evidence_raw = validate_exact_evidence(
|
||||
evidence_file, review_package_file, golden_summary_file
|
||||
)
|
||||
approval = load_legal_approval_file(approval_file)
|
||||
keyring = load_legal_keyring_file(keyring_file)
|
||||
problems = verify_legal_approval(
|
||||
approval,
|
||||
keyring=keyring,
|
||||
tenant_id=package.tenant_id,
|
||||
provider_config_id=package.provider_config_id,
|
||||
provider=package.provider,
|
||||
model=package.model,
|
||||
technical_evidence_sha256=review_package_sha256(package),
|
||||
now=checked_at,
|
||||
)
|
||||
payload = approval.payload
|
||||
expected_bindings = {
|
||||
(item.suite, item.sample_id, item.audit_run_id) for item in package.cases
|
||||
}
|
||||
actual_bindings = {
|
||||
(item.suite, item.sample_id, item.audit_run_id) for item in payload.audit_bindings
|
||||
}
|
||||
comparisons = (
|
||||
(set(payload.qa_sample_ids), set(package.qa_sample_ids), "legal QA inventory"),
|
||||
(set(payload.review_sample_ids), set(package.review_sample_ids), "review inventory"),
|
||||
(
|
||||
set(payload.knowledge_card_ids),
|
||||
set(package.knowledge_card_ids),
|
||||
"knowledge-card inventory",
|
||||
),
|
||||
(actual_bindings, expected_bindings, "audit bindings"),
|
||||
)
|
||||
problems.extend(
|
||||
f"legal approval {label} does not match the package"
|
||||
for actual, expected, label in comparisons
|
||||
if actual != expected
|
||||
)
|
||||
if payload.suite_manifest_sha256 != package.suite_manifest_sha256:
|
||||
problems.append("legal approval suite asset hash does not match the package")
|
||||
if payload.knowledge_manifest_sha256 != package.knowledge_manifest_sha256:
|
||||
problems.append("legal approval knowledge asset hash does not match the package")
|
||||
if payload.evidence_sha256 != sha256_bytes(evidence_raw):
|
||||
problems.append("legal approval evidence hash does not match the local canonical evidence")
|
||||
_validate_evidence_ref(payload.evidence_ref)
|
||||
if problems:
|
||||
raise ValueError("offline legal approval verification failed: " + "; ".join(problems))
|
||||
return approval
|
||||
|
||||
|
||||
def _identity_arguments(parser: argparse.ArgumentParser) -> None:
|
||||
parser.add_argument("--key-id", required=True)
|
||||
parser.add_argument("--approver-id", required=True)
|
||||
parser.add_argument("--reviewer", required=True)
|
||||
parser.add_argument("--role", required=True)
|
||||
parser.add_argument("--organization", required=True)
|
||||
|
||||
|
||||
def _parse_args() -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
commands = parser.add_subparsers(dest="command", required=True)
|
||||
|
||||
prepare = commands.add_parser("prepare", help="create canonical evidence and unsigned payload")
|
||||
prepare.add_argument("--review-package-file", type=Path, required=True)
|
||||
prepare.add_argument("--golden-summary-file", type=Path, required=True)
|
||||
prepare.add_argument("--evidence-ref", required=True)
|
||||
prepare.add_argument("--approved-at", required=True)
|
||||
prepare.add_argument("--expires-at", required=True)
|
||||
prepare.add_argument("--evidence-output", type=Path, required=True)
|
||||
prepare.add_argument("--payload-output", type=Path, required=True)
|
||||
_identity_arguments(prepare)
|
||||
|
||||
keyring = commands.add_parser("keyring", help="create a candidate trusted public keyring")
|
||||
keyring.add_argument("--public-key-file", type=Path, required=True)
|
||||
keyring.add_argument("--not-before", required=True)
|
||||
keyring.add_argument("--not-after", required=True)
|
||||
keyring.add_argument("--output", type=Path, required=True)
|
||||
_identity_arguments(keyring)
|
||||
|
||||
sign = commands.add_parser("sign", help="sign a canonical payload with a legal-owned key")
|
||||
sign.add_argument("--payload-file", type=Path, required=True)
|
||||
sign.add_argument("--evidence-file", type=Path, required=True)
|
||||
sign.add_argument("--review-package-file", type=Path, required=True)
|
||||
sign.add_argument("--golden-summary-file", type=Path, required=True)
|
||||
sign.add_argument("--private-key-file", type=Path, required=True)
|
||||
sign.add_argument("--private-key-password-file", type=Path)
|
||||
sign.add_argument("--confirmation", required=True)
|
||||
sign.add_argument("--output", type=Path, required=True)
|
||||
|
||||
verify = commands.add_parser(
|
||||
"verify", help="verify all offline evidence and signature bindings"
|
||||
)
|
||||
verify.add_argument("--approval-file", type=Path, required=True)
|
||||
verify.add_argument("--keyring-file", type=Path, required=True)
|
||||
verify.add_argument("--evidence-file", type=Path, required=True)
|
||||
verify.add_argument("--review-package-file", type=Path, required=True)
|
||||
verify.add_argument("--golden-summary-file", type=Path, required=True)
|
||||
verify.add_argument("--checked-at")
|
||||
return parser.parse_args()
|
||||
|
||||
|
||||
def _run_prepare(args: argparse.Namespace) -> dict[str, Any]:
|
||||
evidence, package = build_review_evidence(args.review_package_file, args.golden_summary_file)
|
||||
evidence_raw = canonical_json(evidence)
|
||||
payload = build_approval_payload(
|
||||
package=package,
|
||||
evidence_ref=args.evidence_ref,
|
||||
evidence_raw=evidence_raw,
|
||||
key_id=args.key_id,
|
||||
approver_id=args.approver_id,
|
||||
reviewer=args.reviewer,
|
||||
role=args.role,
|
||||
organization=args.organization,
|
||||
approved_at=_parse_time(args.approved_at, label="approved-at"),
|
||||
expires_at=_parse_time(args.expires_at, label="expires-at"),
|
||||
)
|
||||
payload_raw = canonical_legal_approval_payload(payload)
|
||||
protected = [args.review_package_file, args.golden_summary_file]
|
||||
_write_atomic(args.evidence_output, evidence_raw, protected_paths=protected)
|
||||
_write_atomic(
|
||||
args.payload_output,
|
||||
payload_raw,
|
||||
protected_paths=[*protected, args.evidence_output],
|
||||
)
|
||||
return {
|
||||
"status": "prepared_not_signed",
|
||||
"technical_package_sha256": review_package_sha256(package),
|
||||
"evidence_sha256": sha256_bytes(evidence_raw),
|
||||
"payload_sha256": sha256_bytes(payload_raw),
|
||||
"provider_results": 18,
|
||||
"knowledge_cards": 10,
|
||||
"product_golden_results": 31,
|
||||
}
|
||||
|
||||
|
||||
def _run_keyring(args: argparse.Namespace) -> dict[str, Any]:
|
||||
public_raw = _read_bounded(args.public_key_file, label="legal public key", limit=MAX_KEY_BYTES)
|
||||
key = serialization.load_pem_public_key(public_raw)
|
||||
if not isinstance(key, Ed25519PublicKey):
|
||||
raise ValueError("legal public key must be an Ed25519 key")
|
||||
canonical_public = key.public_bytes(
|
||||
serialization.Encoding.PEM,
|
||||
serialization.PublicFormat.SubjectPublicKeyInfo,
|
||||
).decode("ascii")
|
||||
not_before = _parse_time(args.not_before, label="not-before")
|
||||
not_after = _parse_time(args.not_after, label="not-after")
|
||||
if not_after <= not_before:
|
||||
raise ValueError("legal public key not-after must follow not-before")
|
||||
keyring = ProviderLegalApprovalKeyring(
|
||||
version=1,
|
||||
keys=[
|
||||
ProviderLegalApprovalPublicKey(
|
||||
key_id=args.key_id,
|
||||
algorithm=LEGAL_APPROVAL_ALGORITHM,
|
||||
public_key_pem=canonical_public,
|
||||
approver_id=args.approver_id,
|
||||
reviewer=args.reviewer,
|
||||
role=args.role,
|
||||
organization=args.organization,
|
||||
not_before=not_before,
|
||||
not_after=not_after,
|
||||
revoked=False,
|
||||
)
|
||||
],
|
||||
)
|
||||
raw = canonical_json(keyring.model_dump(mode="json"))
|
||||
_write_atomic(args.output, raw, protected_paths=[args.public_key_file])
|
||||
return {
|
||||
"status": "candidate_keyring_created",
|
||||
"key_id": args.key_id,
|
||||
"sha256": sha256_bytes(raw),
|
||||
}
|
||||
|
||||
|
||||
def _run_sign(args: argparse.Namespace) -> dict[str, Any]:
|
||||
if args.confirmation != SIGN_CONFIRMATION:
|
||||
raise ValueError(f"signing requires --confirmation {SIGN_CONFIRMATION}")
|
||||
payload_raw = _read_bounded(args.payload_file, label="legal approval payload")
|
||||
payload = ProviderLegalApprovalPayload.model_validate(
|
||||
_strict_json(payload_raw, label="legal approval payload")
|
||||
)
|
||||
if payload_raw != canonical_legal_approval_payload(payload):
|
||||
raise ValueError("legal approval payload is not in its exact canonical form")
|
||||
_evidence, package, evidence_raw = validate_exact_evidence(
|
||||
args.evidence_file,
|
||||
args.review_package_file,
|
||||
args.golden_summary_file,
|
||||
)
|
||||
expected_payload = build_approval_payload(
|
||||
package=package,
|
||||
evidence_ref=payload.evidence_ref,
|
||||
evidence_raw=evidence_raw,
|
||||
key_id=payload.key_id,
|
||||
approver_id=payload.approver_id,
|
||||
reviewer=payload.reviewer,
|
||||
role=payload.role,
|
||||
organization=payload.organization,
|
||||
approved_at=payload.approved_at,
|
||||
expires_at=payload.expires_at,
|
||||
)
|
||||
if payload != expected_payload:
|
||||
raise ValueError(
|
||||
"legal approval payload does not match the exact review package and golden evidence"
|
||||
)
|
||||
private_raw = _read_bounded(
|
||||
args.private_key_file,
|
||||
label="legal private key",
|
||||
limit=MAX_KEY_BYTES,
|
||||
)
|
||||
password: bytes | None = None
|
||||
protected = [
|
||||
args.payload_file,
|
||||
args.evidence_file,
|
||||
args.review_package_file,
|
||||
args.golden_summary_file,
|
||||
args.private_key_file,
|
||||
]
|
||||
if args.private_key_password_file:
|
||||
password = _read_bounded(
|
||||
args.private_key_password_file,
|
||||
label="legal private key password",
|
||||
limit=MAX_KEY_BYTES,
|
||||
).rstrip(b"\r\n")
|
||||
if not password:
|
||||
raise ValueError("legal private key password file is empty")
|
||||
protected.append(args.private_key_password_file)
|
||||
approval = sign_legal_approval(payload, private_raw, private_key_password=password)
|
||||
approval_raw = canonical_json(approval.model_dump(mode="json"))
|
||||
_write_atomic(args.output, approval_raw, protected_paths=protected)
|
||||
return {
|
||||
"status": "signed",
|
||||
"approval_id": approval.approval_id,
|
||||
"key_id": approval.payload.key_id,
|
||||
"approval_sha256": sha256_bytes(approval_raw),
|
||||
}
|
||||
|
||||
|
||||
def _run_verify(args: argparse.Namespace) -> dict[str, Any]:
|
||||
checked_at = _parse_time(args.checked_at, label="checked-at") if args.checked_at else None
|
||||
approval = verify_offline_approval(
|
||||
approval_file=args.approval_file,
|
||||
keyring_file=args.keyring_file,
|
||||
evidence_file=args.evidence_file,
|
||||
review_package_file=args.review_package_file,
|
||||
golden_summary_file=args.golden_summary_file,
|
||||
checked_at=checked_at,
|
||||
)
|
||||
return {
|
||||
"status": "verified",
|
||||
"approval_id": approval.approval_id,
|
||||
"key_id": approval.payload.key_id,
|
||||
"evidence_sha256": approval.payload.evidence_sha256,
|
||||
"technical_package_sha256": approval.payload.technical_evidence_sha256,
|
||||
}
|
||||
|
||||
|
||||
def main() -> int:
|
||||
if hasattr(sys.stdout, "reconfigure"):
|
||||
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
|
||||
args = _parse_args()
|
||||
try:
|
||||
if args.command == "prepare":
|
||||
result = _run_prepare(args)
|
||||
elif args.command == "keyring":
|
||||
result = _run_keyring(args)
|
||||
elif args.command == "sign":
|
||||
result = _run_sign(args)
|
||||
else:
|
||||
result = _run_verify(args)
|
||||
except Exception as error:
|
||||
print(
|
||||
json.dumps(
|
||||
{"status": "failed", "error": f"{type(error).__name__}: {error}"},
|
||||
ensure_ascii=True,
|
||||
separators=(",", ":"),
|
||||
),
|
||||
file=sys.stderr,
|
||||
)
|
||||
return 2
|
||||
print(json.dumps(result, ensure_ascii=True, separators=(",", ":")))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,288 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
import json
|
||||
import sys
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
from cryptography.hazmat.primitives import serialization
|
||||
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
|
||||
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[2]
|
||||
AI_PLATFORM_SRC = REPO_ROOT / "yuqei-ai-platform" / "services" / "ai-platform-api" / "src"
|
||||
if str(AI_PLATFORM_SRC) not in sys.path:
|
||||
sys.path.insert(0, str(AI_PLATFORM_SRC))
|
||||
|
||||
|
||||
def _load_helper():
|
||||
path = REPO_ROOT / "yuqei-infra" / "scripts" / "prepare-provider-legal-approval.py"
|
||||
spec = importlib.util.spec_from_file_location("provider_legal_signing_helper", path)
|
||||
assert spec and spec.loader
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
sys.modules[spec.name] = module
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
helper = _load_helper()
|
||||
|
||||
from yuqei_ai_platform_api.provider_acceptance import expected_acceptance_inventory # noqa: E402
|
||||
from yuqei_ai_platform_api.provider_legal_approval import ( # noqa: E402
|
||||
LEGAL_APPROVAL_ALGORITHM,
|
||||
ProviderLegalApprovalKeyring,
|
||||
ProviderLegalApprovalPublicKey,
|
||||
sign_legal_approval,
|
||||
)
|
||||
from yuqei_ai_platform_api.provider_review_package import ( # noqa: E402
|
||||
ProviderReviewAsset,
|
||||
ProviderReviewCase,
|
||||
ProviderTechnicalReviewPackage,
|
||||
canonical_json,
|
||||
canonical_review_package,
|
||||
sha256_bytes,
|
||||
)
|
||||
|
||||
|
||||
NOW = datetime(2026, 8, 11, 8, tzinfo=UTC)
|
||||
IDENTITY = {
|
||||
"key_id": "named-legal-2026q3",
|
||||
"approver_id": "legal.reviewer@example.test",
|
||||
"reviewer": "Named Legal Reviewer",
|
||||
"role": "Senior Legal Counsel",
|
||||
"organization": "Example Legal",
|
||||
}
|
||||
|
||||
|
||||
def _review_package() -> ProviderTechnicalReviewPackage:
|
||||
inventory = expected_acceptance_inventory()
|
||||
assets = []
|
||||
for path in sorted(item for item in helper.PHASE13_ROOT.rglob("*") if item.is_file()):
|
||||
raw = path.read_bytes()
|
||||
assets.append(
|
||||
ProviderReviewAsset(
|
||||
path=path.relative_to(helper.PHASE13_ROOT).as_posix(),
|
||||
sha256=sha256_bytes(raw),
|
||||
size_bytes=len(raw),
|
||||
)
|
||||
)
|
||||
cases = []
|
||||
case_ids = [
|
||||
*(("legal_qa", item) for item in inventory["qa_sample_ids"]),
|
||||
*(("contract_review", item) for item in inventory["review_sample_ids"]),
|
||||
]
|
||||
for index, (suite, sample_id) in enumerate(case_ids):
|
||||
output = (
|
||||
{"answer": f"answer for {sample_id}", "citations": []}
|
||||
if suite == "legal_qa"
|
||||
else {"summary": f"summary for {sample_id}", "risk_items": [], "citations": []}
|
||||
)
|
||||
cases.append(
|
||||
ProviderReviewCase(
|
||||
suite=suite,
|
||||
sample_id=sample_id,
|
||||
audit_run_id=f"run-{index + 1:012d}",
|
||||
audit_snapshot_sha256=f"{index + 1:064x}",
|
||||
technical_status="passed",
|
||||
provider="provider-under-review",
|
||||
model="legal-model-under-review",
|
||||
latency_ms=100,
|
||||
output=output,
|
||||
output_sha256=sha256_bytes(canonical_json(output)),
|
||||
)
|
||||
)
|
||||
return ProviderTechnicalReviewPackage(
|
||||
package_version="provider-technical-review-package-v1",
|
||||
tenant_id="default",
|
||||
source_environment="shared_staging",
|
||||
created_at=NOW,
|
||||
provider_config_id="provider-config-under-review",
|
||||
provider="provider-under-review",
|
||||
model="legal-model-under-review",
|
||||
qa_sample_ids=inventory["qa_sample_ids"],
|
||||
review_sample_ids=inventory["review_sample_ids"],
|
||||
knowledge_card_ids=inventory["knowledge_card_ids"],
|
||||
suite_manifest_sha256=inventory["suite_manifest_sha256"],
|
||||
knowledge_manifest_sha256=inventory["knowledge_manifest_sha256"],
|
||||
assets=assets,
|
||||
cases=cases,
|
||||
)
|
||||
|
||||
|
||||
def _input_files(tmp_path: Path) -> tuple[Path, Path]:
|
||||
package_file = tmp_path / "technical-package.json"
|
||||
package_file.write_bytes(canonical_review_package(_review_package()))
|
||||
golden = helper._load_golden_module()
|
||||
summary = golden.build_summary(golden.run_product_golden())
|
||||
summary_file = tmp_path / "contract-product-golden-summary.json"
|
||||
summary_file.write_text(
|
||||
json.dumps(summary, ensure_ascii=False, indent=2) + "\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
return package_file, summary_file
|
||||
|
||||
|
||||
def _key_material(password: bytes):
|
||||
private = Ed25519PrivateKey.generate()
|
||||
private_pem = private.private_bytes(
|
||||
serialization.Encoding.PEM,
|
||||
serialization.PrivateFormat.PKCS8,
|
||||
serialization.BestAvailableEncryption(password),
|
||||
)
|
||||
public_pem = private.public_key().public_bytes(
|
||||
serialization.Encoding.PEM,
|
||||
serialization.PublicFormat.SubjectPublicKeyInfo,
|
||||
).decode("ascii")
|
||||
return private_pem, public_pem
|
||||
|
||||
|
||||
def test_prepares_signs_and_verifies_exact_package_and_31_case_suite(tmp_path: Path) -> None:
|
||||
package_file, summary_file = _input_files(tmp_path)
|
||||
evidence, package = helper.build_review_evidence(package_file, summary_file)
|
||||
evidence_raw = canonical_json(evidence)
|
||||
evidence_file = tmp_path / "legal-review-evidence.json"
|
||||
evidence_file.write_bytes(evidence_raw)
|
||||
payload = helper.build_approval_payload(
|
||||
package=package,
|
||||
evidence_ref="https://evidence.example.test/provider/legal/review.json",
|
||||
evidence_raw=evidence_raw,
|
||||
approved_at=NOW,
|
||||
expires_at=NOW + timedelta(days=7),
|
||||
**IDENTITY,
|
||||
)
|
||||
|
||||
password = b"test-password-not-a-production-secret"
|
||||
private_pem, public_pem = _key_material(password)
|
||||
approval = sign_legal_approval(payload, private_pem, private_key_password=password)
|
||||
approval_file = tmp_path / "named-legal-approval.json"
|
||||
approval_file.write_bytes(canonical_json(approval.model_dump(mode="json")))
|
||||
keyring = ProviderLegalApprovalKeyring(
|
||||
version=1,
|
||||
keys=[
|
||||
ProviderLegalApprovalPublicKey(
|
||||
algorithm=LEGAL_APPROVAL_ALGORITHM,
|
||||
public_key_pem=public_pem,
|
||||
not_before=NOW - timedelta(days=1),
|
||||
not_after=NOW + timedelta(days=30),
|
||||
revoked=False,
|
||||
**IDENTITY,
|
||||
)
|
||||
],
|
||||
)
|
||||
keyring_file = tmp_path / "legal-keyring.json"
|
||||
keyring_file.write_bytes(canonical_json(keyring.model_dump(mode="json")))
|
||||
|
||||
verified = helper.verify_offline_approval(
|
||||
approval_file=approval_file,
|
||||
keyring_file=keyring_file,
|
||||
evidence_file=evidence_file,
|
||||
review_package_file=package_file,
|
||||
golden_summary_file=summary_file,
|
||||
checked_at=NOW + timedelta(minutes=1),
|
||||
)
|
||||
|
||||
assert verified.approval_id == approval.approval_id
|
||||
assert evidence["signature_requirement"] == "detached_named_legal_signature_required"
|
||||
assert len(evidence["provider_technical_review"]["audit_bindings"]) == 18
|
||||
assert len(evidence["provider_technical_review"]["knowledge_card_ids"]) == 10
|
||||
assert len(evidence["contract_product_golden"]["cases"]) == 31
|
||||
assert evidence["contract_product_golden"]["summary"]["status"] == "needs_legal_review"
|
||||
|
||||
|
||||
def test_rejects_reformatted_or_tampered_evidence_and_stale_golden_summary(tmp_path: Path) -> None:
|
||||
package_file, summary_file = _input_files(tmp_path)
|
||||
evidence, _package = helper.build_review_evidence(package_file, summary_file)
|
||||
evidence_file = tmp_path / "evidence.json"
|
||||
evidence_file.write_text(json.dumps(evidence, indent=2), encoding="utf-8")
|
||||
|
||||
with pytest.raises(ValueError, match="exact canonical evidence"):
|
||||
helper.validate_exact_evidence(evidence_file, package_file, summary_file)
|
||||
|
||||
stale = json.loads(summary_file.read_text(encoding="utf-8"))
|
||||
stale["counts"]["technical_passed"] = 30
|
||||
summary_file.write_text(json.dumps(stale), encoding="utf-8")
|
||||
with pytest.raises(ValueError, match="fresh deterministic run"):
|
||||
helper.build_review_evidence(package_file, summary_file)
|
||||
|
||||
|
||||
def test_sign_command_requires_explicit_confirmation_and_never_overwrites_key(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
package_file, summary_file = _input_files(tmp_path)
|
||||
evidence, package = helper.build_review_evidence(package_file, summary_file)
|
||||
payload = helper.build_approval_payload(
|
||||
package=package,
|
||||
evidence_ref="https://evidence.example.test/provider/legal/review.json",
|
||||
evidence_raw=canonical_json(evidence),
|
||||
approved_at=NOW,
|
||||
expires_at=NOW + timedelta(days=7),
|
||||
**IDENTITY,
|
||||
)
|
||||
payload_file = tmp_path / "payload.json"
|
||||
payload_file.write_bytes(helper.canonical_legal_approval_payload(payload))
|
||||
evidence_file = tmp_path / "evidence.json"
|
||||
evidence_file.write_bytes(canonical_json(evidence))
|
||||
password = b"test-password-not-a-production-secret"
|
||||
private_pem, _public_pem = _key_material(password)
|
||||
private_file = tmp_path / "private.pem"
|
||||
password_file = tmp_path / "password.txt"
|
||||
private_file.write_bytes(private_pem)
|
||||
password_file.write_bytes(password + b"\n")
|
||||
original_key_hash = sha256_bytes(private_file.read_bytes())
|
||||
|
||||
arguments = SimpleNamespace(
|
||||
payload_file=payload_file,
|
||||
evidence_file=evidence_file,
|
||||
review_package_file=package_file,
|
||||
golden_summary_file=summary_file,
|
||||
private_key_file=private_file,
|
||||
private_key_password_file=password_file,
|
||||
confirmation="not-approved",
|
||||
output=tmp_path / "approval.json",
|
||||
)
|
||||
with pytest.raises(ValueError, match="signing requires"):
|
||||
helper._run_sign(arguments)
|
||||
arguments.confirmation = helper.SIGN_CONFIRMATION
|
||||
result = helper._run_sign(arguments)
|
||||
|
||||
assert result["status"] == "signed"
|
||||
assert sha256_bytes(private_file.read_bytes()) == original_key_hash
|
||||
assert password_file.read_bytes() == password + b"\n"
|
||||
assert b"PRIVATE KEY" not in arguments.output.read_bytes()
|
||||
|
||||
|
||||
def test_rejects_package_whose_asset_or_output_binding_was_recomputed_incorrectly(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
package = _review_package()
|
||||
package.cases[0].output_sha256 = "f" * 64
|
||||
package_file = tmp_path / "technical-package.json"
|
||||
package_file.write_bytes(canonical_review_package(package))
|
||||
|
||||
with pytest.raises(ValueError, match="output hash is inconsistent"):
|
||||
helper.load_exact_review_package(package_file)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"evidence_ref",
|
||||
(
|
||||
"http://evidence.example.test/review.json",
|
||||
"https://127.0.0.1/review.json",
|
||||
"https://192.168.3.90/review.json",
|
||||
"https://evidence.example.test:8443/review.json",
|
||||
"https://evidence.example.test/review.json?token=secret",
|
||||
),
|
||||
)
|
||||
def test_rejects_non_public_or_mutable_evidence_references(evidence_ref: str) -> None:
|
||||
with pytest.raises(ValueError, match="public-style HTTPS URL"):
|
||||
helper.build_approval_payload(
|
||||
package=_review_package(),
|
||||
evidence_ref=evidence_ref,
|
||||
evidence_raw=b"real canonical evidence bytes",
|
||||
approved_at=NOW,
|
||||
expires_at=NOW + timedelta(days=7),
|
||||
**IDENTITY,
|
||||
)
|
||||
Reference in New Issue
Block a user