feat: add governed public policy promotion
Build and Test Quality Gate / Contract Web typecheck and build (pull_request) Successful in 4m4s
Build and Test Quality Gate / Python tests and Contract API migration (pull_request) Failing after 9m18s
Security Quality Gate / npm audit, Bandit, pip-audit, detect-secrets (pull_request) Successful in 8m20s

This commit is contained in:
2026-08-17 19:52:07 +08:00
parent 41e5a85567
commit 801a9b04bc
9 changed files with 1101 additions and 1 deletions
@@ -0,0 +1,186 @@
# V2 公开政策生产 Promotion 运维手册
日期:2026-08-17
## 边界
本手册只描述已验签公开政策的受控状态切换。它不替代法务审批、运营审批、容量审批或公网切流审批。候选 keyring、手工修改数据库、HTTP 写接口、政策私钥和共享预发布凭据都不能用于生产激活。
`CONTRACT_API_PUBLIC_POLICY_EXECUTOR_ENABLED` 必须保持 `false`。生产读取、注册协议和既有用户重新同意由 Contract API 提供;真正的 promotion 和 rollback 只通过本机受控 CLI 执行,并且不暴露 HTTP 路由。
## 前置条件
1. Contract API 数据库已经迁移到 `20260812_0046`,`public_policy_state` 存在且状态指针可验证。
2. 四份政策候选文件、`manifest.json`、具名法务审批文件和 trusted keyring 已由安全/运维安装到生产主机。
3. 材料目录和文件均为绝对路径,文件所有者可读,未包含 symlink,目录总量不超过 2 MiB。
4. `CONTRACT_API_PUBLIC_POLICY_RELEASE_DIGEST` 等于当前 Contract API 不可变镜像 digest。
5. 生产只读 preflight 通过。
6. 已取得单独的运营激活授权文件。该文件只用于计算 SHA-256,内容不得进入仓库、聊天、日志或 CI artifact。
7. 已记录 expected generation、current set ID、trace ID、idempotency key 和授权文件 hash。
推荐主机布局:
```text
/etc/yuqei/public-policy/2026-08-12/
manifest.json
user-terms.md
privacy-policy.md
service-agreement.md
ai-use-notice.md
named-policy-legal-approval.json
policy-legal-keyring.trusted.json
```
生产环境模板中的绑定:
```text
PUBLIC_POLICY_CANDIDATE_DIR_HOST=/etc/yuqei/public-policy/2026-08-12
PUBLIC_POLICY_LEGAL_APPROVAL_FILE_HOST=/etc/yuqei/public-policy/2026-08-12/named-policy-legal-approval.json
PUBLIC_POLICY_TRUSTED_KEYRING_FILE_HOST=/etc/yuqei/public-policy/2026-08-12/policy-legal-keyring.trusted.json
CONTRACT_API_PUBLIC_POLICY_RELEASE_DIGEST=sha256:<contract-api-image-digest>
CONTRACT_API_PUBLIC_POLICY_EXECUTOR_ENABLED=false
```
## 1. 只读预检
在生产主机执行:
```bash
python3 yuqei-infra/deploy/preflight-production.py \
--env-file /etc/yuqei/production.env \
--artifact-dir /var/lib/yuqei/preflight/public-policy
```
必须确认以下检查为 `pass`:
- `public_policy_deployment`
- `public_policy_credential_isolation`
- `public_policy_candidate_dir`
- `public_policy_manifest`
- `public_policy_legal_approval_file`
- `public_policy_trusted_keyring_file`
- 相关 owner-only 权限检查
任一失败时停止,不得通过改名、复制、手工改库或跳过 preflight 继续。
## 2. 查看当前状态
下面的 `COMPOSE_ARGS` 按生产部署手册替换为真实 project、env 和两个 Compose 文件参数:
```bash
COMPOSE_ARGS=(
--project-name yuqei-v2-production
--env-file /etc/yuqei/production.env
-f yuqei-infra/compose/docker-compose.yml
-f yuqei-infra/compose/docker-compose.production.yml
)
docker compose "${COMPOSE_ARGS[@]}" run --rm --no-deps contract-api \
python -m yuqei_contract_api.public_policy_promotion_cli status
```
输出不包含政策正文,只包含状态指针、版本、生效时间、公开路径和每个文件的 SHA-256。
## 3. Promotion dry-run
```bash
docker compose "${COMPOSE_ARGS[@]}" run --rm --no-deps contract-api \
python -m yuqei_contract_api.public_policy_promotion_cli preflight
```
显式 promotion dry-run 还必须提供工单上的 expected state:
```bash
docker compose "${COMPOSE_ARGS[@]}" run --rm --no-deps contract-api \
python -m yuqei_contract_api.public_policy_promotion_cli \
promote \
--expected-generation 0 \
--expected-current-set-id none \
--idempotency-key policy-promotion-20260812-001 \
--trace-id ops-policy-promotion-20260812-001
```
Dry-run 不写数据库,也不需要授权文件。
## 4. 受控 Promotion
授权文件通过一次性只读挂载进入容器。不要把该挂载写入生产 Compose 文件:
```bash
AUTHORIZATION_FILE=/run/yuqei/operations/policy-promotion-20260812-001.txt
docker compose "${COMPOSE_ARGS[@]}" run --rm --no-deps \
--volume "$AUTHORIZATION_FILE:/run/yuqei/public-policy/activation-authorization.txt:ro" \
contract-api \
python -m yuqei_contract_api.public_policy_promotion_cli \
promote \
--apply \
--confirmation I_AUTHORIZE_THIS_EXACT_PUBLIC_POLICY_PROMOTION \
--authorization-file /run/yuqei/public-policy/activation-authorization.txt \
--expected-generation 0 \
--expected-current-set-id none \
--idempotency-key policy-promotion-20260812-001 \
--trace-id ops-policy-promotion-20260812-001
```
命令输出中的 `authorization_sha256`、generation、set ID、event ID 和 state hash 必须写入运维证据。重复执行同一命令必须使用完全相同的材料;换任何材料必须换 idempotency key 并重新审批。
## 5. 生效后回归
到达批准的 `effective_at` 后验证:
1. `GET /api/v1/public-policies/current` 返回四份政策。
2. 四份政策页面返回 approved 字节,页面版本和生效时间正确。
3. 新注册用户必须绑定四个 effective 协议版本。
4. 既有用户下次登录进入重新同意流程。
5. 同意提交记录版本、时间和主体。
6. 审计和 append-only 政策事件链完整。
7. Contract Web 不出现未发布、接口故障与真实空态混淆。
8. 生产 health、schema-check、worker、Provider readiness 和核心合同流程保持通过。
## 6. 受控 Rollback
Rollback 只允许指向曾经 active 的历史 policy set,不能伪造或引用不存在的 set。先执行 dry-run:
```bash
docker compose "${COMPOSE_ARGS[@]}" run --rm --no-deps contract-api \
python -m yuqei_contract_api.public_policy_promotion_cli \
rollback \
--target-set-id <historical-policy-set-id> \
--expected-generation <ticket-generation> \
--expected-current-set-id <ticket-current-set-id> \
--idempotency-key policy-rollback-<ticket-id> \
--trace-id ops-policy-rollback-<ticket-id>
```
确认目标后,使用单独的 rollback 授权文件:
```bash
ROLLBACK_AUTHORIZATION=/run/yuqei/operations/policy-rollback-<ticket-id>.txt
docker compose "${COMPOSE_ARGS[@]}" run --rm --no-deps \
--volume "$ROLLBACK_AUTHORIZATION:/run/yuqei/public-policy/rollback-authorization.txt:ro" \
contract-api \
python -m yuqei_contract_api.public_policy_promotion_cli \
rollback \
--apply \
--confirmation I_AUTHORIZE_THIS_EXACT_PUBLIC_POLICY_ROLLBACK \
--authorization-file /run/yuqei/public-policy/rollback-authorization.txt \
--target-set-id <historical-policy-set-id> \
--expected-generation <ticket-generation> \
--expected-current-set-id <ticket-current-set-id> \
--idempotency-key policy-rollback-<ticket-id> \
--trace-id ops-policy-rollback-<ticket-id>
```
Rollback 也是 append-only 事件,不删除历史政策、同意记录或审计。执行后必须重复第 5 节回归,并通知法务和运营具名复核。
## 7. 明确禁止
- 禁止把法务私钥、运营授权文件或任何 policy signing key 放入仓库、生产环境文件或 CI artifact。
- 禁止使用 candidate keyring 直接激活。
- 禁止把 trusted keyring 的 `policy_activation_authorized=false` 手工改为 true。
- 禁止直接 UPDATE/DELETE `public_policy_*` 表。
- 禁止新增 HTTP promotion、rollback 或 execute 路由。
- 禁止在 preflight、容量、备份、法务或运营审批失败时启动 edge 或切换 DNS。
- 禁止把 dry-run 或技术验签输出宣称为法务批准或公网切流授权。
@@ -21,6 +21,7 @@ dependencies = [
[project.scripts]
yuqei-import-v1-legal-fulltext = "yuqei_contract_api.v1_legal_fulltext_import:main"
yuqei-bootstrap-identity-admin = "yuqei_contract_api.identity_bootstrap:main"
yuqei-public-policy-promotion = "yuqei_contract_api.public_policy_promotion_cli:main"
[build-system]
requires = ["hatchling"]
@@ -0,0 +1,346 @@
from __future__ import annotations
import argparse
import hashlib
import json
import os
import stat
from datetime import UTC, datetime
from pathlib import Path
from typing import Any
from sqlalchemy import create_engine
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import sessionmaker
from yuqei_contract_api.config import ContractApiSettings
from yuqei_contract_api.public_policies import (
PublicPolicyError,
PublicPolicyRepository,
build_promotion_preflight,
)
from yuqei_contract_api.sqlalchemy_engine import build_engine_kwargs, ensure_sqlite_parent_dir
PROMOTION_CONFIRMATION = "I_AUTHORIZE_THIS_EXACT_PUBLIC_POLICY_PROMOTION"
ROLLBACK_CONFIRMATION = "I_AUTHORIZE_THIS_EXACT_PUBLIC_POLICY_ROLLBACK"
AUTHORIZATION_FILE_MAX_BYTES = 64 * 1024
def main(
argv: list[str] | None = None,
*,
repository: PublicPolicyRepository | None = None,
settings: ContractApiSettings | None = None,
checked_at: datetime | None = None,
) -> int:
parser = build_parser()
args = parser.parse_args(argv)
resolved_settings = settings or ContractApiSettings()
if resolved_settings.public_policy_executor_enabled:
print(json.dumps({
"status": "error",
"error": "The public policy executor flag must remain false; use this named CLI in an approved operations window.",
}, ensure_ascii=False))
return 1
if args.command in {"promote", "rollback"} and resolved_settings.environment == "local":
print(json.dumps({
"status": "error",
"error": "Public policy writes require test, staging, or production environment settings.",
}, ensure_ascii=False))
return 1
engine = None
try:
if repository is None:
database_url = args.database_url or resolved_settings.database_url
ensure_sqlite_parent_dir(database_url)
engine = create_engine(database_url, **build_engine_kwargs(database_url))
session_factory = sessionmaker(bind=engine, expire_on_commit=False, future=True)
repository = PublicPolicyRepository(session_factory)
if args.command == "status":
payload = _status_payload(repository)
elif args.command == "preflight":
payload = _preflight_payload(repository, resolved_settings, checked_at=checked_at)
elif args.command == "promote":
payload = _promote_payload(
args,
repository,
resolved_settings,
checked_at=checked_at,
)
elif args.command == "rollback":
payload = _rollback_payload(args, repository, checked_at=checked_at)
else: # pragma: no cover - argparse rejects this branch.
raise PublicPolicyError(f"Unsupported command: {args.command}")
print(json.dumps(payload, ensure_ascii=False, indent=2))
return 0
except (PublicPolicyError, ValueError, OSError) as exc:
print(json.dumps({"status": "error", "error": str(exc)}, ensure_ascii=False))
return 1
except SQLAlchemyError:
print(json.dumps({"status": "error", "error": "Public policy repository operation failed."}, ensure_ascii=False))
return 1
finally:
if engine is not None:
engine.dispose()
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
description="Governed, non-HTTP public policy state operations."
)
parser.add_argument("--database-url", default=None, help="Override CONTRACT_API_DATABASE_URL.")
subparsers = parser.add_subparsers(dest="command", required=True)
subparsers.add_parser("status", help="Read the current public policy state without document bodies.")
subparsers.add_parser("preflight", help="Run the read-only verified-policy promotion preflight.")
promote = subparsers.add_parser("promote", help="Promote one exact verified policy bundle.")
_add_write_arguments(promote, confirmation=PROMOTION_CONFIRMATION)
promote.add_argument(
"--expected-generation",
type=int,
required=True,
help="Generation observed on the approved operations ticket.",
)
promote.add_argument(
"--expected-current-set-id",
required=True,
help="Current policy set ID from the ticket; use 'none' for the baseline state.",
)
rollback = subparsers.add_parser("rollback", help="Append a rollback pointer to a historical policy set.")
_add_write_arguments(rollback, confirmation=ROLLBACK_CONFIRMATION)
rollback.add_argument(
"--target-set-id",
required=True,
help="Previously active policy set to restore.",
)
rollback.add_argument(
"--expected-generation",
type=int,
required=True,
help="Generation observed on the approved rollback ticket.",
)
rollback.add_argument(
"--expected-current-set-id",
required=True,
help="Current policy set ID from the ticket.",
)
return parser
def _add_write_arguments(parser: argparse.ArgumentParser, *, confirmation: str) -> None:
parser.add_argument("--apply", action="store_true", help="Write the state change. Omit for a read-only dry run.")
parser.add_argument(
"--confirmation",
default="",
help=f"Required when --apply is used: {confirmation}",
)
parser.add_argument(
"--authorization-file",
type=Path,
help="Absolute owner-only file whose exact bytes authorize this operation. Its content is never printed.",
)
parser.add_argument("--idempotency-key", required=True, help="Stable operation key from the approval record.")
parser.add_argument("--trace-id", required=True, help="Stable operations trace ID.")
parser.add_argument(
"--checked-at",
default=None,
help="Optional timezone-aware ISO-8601 time. Defaults to the current UTC time.",
)
def _status_payload(repository: PublicPolicyRepository) -> dict[str, Any]:
state = repository.state()
catalog = repository.current_catalog()
return {
"command": "status",
"status": "ok",
"state": state.model_dump(mode="json"),
"current_catalog": None if catalog is None else {
"version": catalog.version,
"effective_at": catalog.effective_at,
"existing_user_reconsent": catalog.existing_user_reconsent,
"documents": [
{
"policy_id": document.policy_id,
"version": document.version,
"public_path": document.public_path,
"sha256": document.sha256,
"size_bytes": document.size_bytes,
}
for document in catalog.documents
],
},
}
def _preflight_payload(
repository: PublicPolicyRepository,
settings: ContractApiSettings,
*,
checked_at: datetime | None,
) -> dict[str, Any]:
result = build_promotion_preflight(settings, repository, checked_at=checked_at)
return {
"command": "preflight",
**result.model_dump(mode="json"),
}
def _promote_payload(
args: argparse.Namespace,
repository: PublicPolicyRepository,
settings: ContractApiSettings,
*,
checked_at: datetime | None,
) -> dict[str, Any]:
_require_configured_absolute_policy_paths(settings)
now = _resolve_now(args, checked_at)
state = _require_expected_state(
repository,
expected_generation=args.expected_generation,
expected_current_set_id=args.expected_current_set_id,
)
preflight = build_promotion_preflight(settings, repository, checked_at=now)
if not args.apply:
return {
"command": "promote",
"mode": "dry_run",
**preflight.model_dump(mode="json"),
}
_require_write_intent(args, confirmation=PROMOTION_CONFIRMATION)
authorization_sha256 = _authorization_sha256(args.authorization_file)
if preflight.current_state != state:
raise PublicPolicyError("Public policy state changed while this command was running.")
snapshot = repository.promote_verified_bundle(
preflight=preflight,
policy_dir=Path(settings.public_policy_candidate_dir),
authorization_sha256=authorization_sha256,
idempotency_key=args.idempotency_key,
trace_id=args.trace_id,
now=now,
approval_file=Path(settings.public_policy_legal_approval_file),
keyring_file=Path(settings.public_policy_trusted_keyring_file),
release_digest=settings.public_policy_release_digest,
)
return {
"command": "promote",
"status": "promoted",
"authorization_sha256": authorization_sha256,
"trace_id": args.trace_id,
"state": snapshot.model_dump(mode="json"),
}
def _rollback_payload(
args: argparse.Namespace,
repository: PublicPolicyRepository,
*,
checked_at: datetime | None,
) -> dict[str, Any]:
now = _resolve_now(args, checked_at)
state = _require_expected_state(
repository,
expected_generation=args.expected_generation,
expected_current_set_id=args.expected_current_set_id,
)
if not args.apply:
return {
"command": "rollback",
"mode": "dry_run",
"target_set_id": args.target_set_id,
"state": state.model_dump(mode="json"),
}
_require_write_intent(args, confirmation=ROLLBACK_CONFIRMATION)
authorization_sha256 = _authorization_sha256(args.authorization_file)
snapshot = repository.rollback_verified_set(
expected_state=state,
target_set_id=args.target_set_id,
authorization_sha256=authorization_sha256,
idempotency_key=args.idempotency_key,
trace_id=args.trace_id,
now=now,
)
return {
"command": "rollback",
"status": "rolled_back",
"authorization_sha256": authorization_sha256,
"trace_id": args.trace_id,
"state": snapshot.model_dump(mode="json"),
}
def _require_expected_state(
repository: PublicPolicyRepository,
*,
expected_generation: int,
expected_current_set_id: str,
) -> Any:
state = repository.state()
expected_set_id = None if expected_current_set_id.lower() == "none" else expected_current_set_id
if state.generation != expected_generation or state.current_set_id != expected_set_id:
raise PublicPolicyError(
"Current public policy state does not match the approved expected state."
)
return state
def _require_write_intent(args: argparse.Namespace, *, confirmation: str) -> None:
if not args.apply:
raise PublicPolicyError("A write was requested without --apply.")
if args.confirmation != confirmation:
raise PublicPolicyError(f"Confirmation does not match the exact operation: {confirmation}")
if args.authorization_file is None:
raise PublicPolicyError("An absolute owner-only authorization file is required.")
def _authorization_sha256(path: Path | None) -> str:
if path is None:
raise PublicPolicyError("An authorization file is required.")
try:
resolved = path.resolve(strict=True)
if not resolved.is_absolute() or not resolved.is_file():
raise PublicPolicyError("Authorization file must be an absolute file.")
if resolved.stat().st_size <= 0 or resolved.stat().st_size > AUTHORIZATION_FILE_MAX_BYTES:
raise PublicPolicyError("Authorization file must be non-empty and no larger than 64 KiB.")
if os.name != "nt" and stat.S_IMODE(resolved.stat().st_mode) & 0o077:
raise PublicPolicyError("Authorization file must be owner-only.")
return hashlib.sha256(resolved.read_bytes()).hexdigest()
except PublicPolicyError:
raise
except OSError as exc:
raise PublicPolicyError("Authorization file is unavailable.") from exc
def _resolve_now(args: argparse.Namespace, fallback: datetime | None) -> datetime:
if fallback is not None:
return fallback
raw = getattr(args, "checked_at", None)
if not raw:
return datetime.now(UTC)
try:
value = datetime.fromisoformat(raw)
except ValueError as exc:
raise PublicPolicyError("checked_at must be a valid ISO-8601 timestamp.") from exc
if value.tzinfo is None:
raise PublicPolicyError("checked_at must include a timezone.")
return value.astimezone(UTC)
def _require_configured_absolute_policy_paths(settings: ContractApiSettings) -> None:
labels = {
"candidate directory": Path(settings.public_policy_candidate_dir),
"legal approval file": Path(settings.public_policy_legal_approval_file),
"trusted keyring file": Path(settings.public_policy_trusted_keyring_file),
}
for label, path in labels.items():
if not path.is_absolute():
raise PublicPolicyError(f"Configured public policy {label} must be absolute.")
if __name__ == "__main__":
raise SystemExit(main())
@@ -17,6 +17,11 @@ from yuqei_contract_api.persistence.models import (
PublicPolicyStateEventModel,
PublicPolicyStateModel,
)
from yuqei_contract_api.public_policy_promotion_cli import (
PROMOTION_CONFIRMATION,
ROLLBACK_CONFIRMATION,
main as public_policy_cli_main,
)
from yuqei_contract_api.public_policies import (
BASELINE_STATE_HASH,
PublicPolicyError,
@@ -476,3 +481,228 @@ def test_rollback_rejects_missing_and_current_targets(tmp_path: Path, monkeypatc
trace_id="trace-current-target",
now=datetime.now(UTC),
)
def _cli_settings(policy_dir: Path, approval_file: Path, keyring_file: Path) -> ContractApiSettings:
return ContractApiSettings(
environment="test",
public_policy_candidate_dir=str(policy_dir.resolve()),
public_policy_legal_approval_file=str(approval_file.resolve()),
public_policy_trusted_keyring_file=str(keyring_file.resolve()),
public_policy_release_digest=RELEASE_DIGEST,
)
def test_public_policy_cli_dry_run_and_promotion_require_explicit_authorization(
tmp_path: Path,
monkeypatch,
capsys,
) -> None:
repository, _ = _repository(tmp_path)
policy_dir = tmp_path / "candidate"
policy_dir.mkdir()
effective_at = "2026-08-11T00:00:00+08:00"
approval_file, keyring_file, documents = _evidence(policy_dir, effective_at=effective_at)
_mock_verifier(monkeypatch, version="2026-08-12", effective_at=effective_at, documents=documents)
settings = _cli_settings(policy_dir, approval_file, keyring_file)
dry_run = public_policy_cli_main(
[
"--database-url", "sqlite+pysqlite:///:memory:",
"promote",
"--expected-generation", "0",
"--expected-current-set-id", "none",
"--idempotency-key", "policy-cli-v1",
"--trace-id", "policy-cli-trace-v1",
],
repository=repository,
settings=settings,
checked_at=datetime(2026, 8, 13, tzinfo=UTC),
)
dry_run_payload = json.loads(capsys.readouterr().out)
assert dry_run == 0
assert dry_run_payload["mode"] == "dry_run"
assert dry_run_payload["status"] == "pass"
assert repository.state().generation == 0
authorization = tmp_path / "activation-authorization.txt"
authorization.write_bytes(b"exact operations authorization")
denied = public_policy_cli_main(
[
"--database-url", "sqlite+pysqlite:///:memory:",
"promote",
"--apply",
"--confirmation", "NOT_THEExact_CONFIRMATION",
"--authorization-file", str(authorization),
"--expected-generation", "0",
"--expected-current-set-id", "none",
"--idempotency-key", "policy-cli-v1",
"--trace-id", "policy-cli-trace-v1",
],
repository=repository,
settings=settings,
checked_at=datetime(2026, 8, 13, tzinfo=UTC),
)
denied_payload = json.loads(capsys.readouterr().out)
assert denied == 1
assert denied_payload["status"] == "error"
assert repository.state().generation == 0
promoted = public_policy_cli_main(
[
"--database-url", "sqlite+pysqlite:///:memory:",
"promote",
"--apply",
"--confirmation", PROMOTION_CONFIRMATION,
"--authorization-file", str(authorization),
"--expected-generation", "0",
"--expected-current-set-id", "none",
"--idempotency-key", "policy-cli-v1",
"--trace-id", "policy-cli-trace-v1",
],
repository=repository,
settings=settings,
checked_at=datetime(2026, 8, 13, tzinfo=UTC),
)
promotion_payload = json.loads(capsys.readouterr().out)
assert promoted == 0
assert promotion_payload["status"] == "promoted"
assert promotion_payload["state"]["generation"] == 1
assert repository.state().generation == 1
def test_public_policy_cli_can_roll_back_only_with_exact_expected_state(
tmp_path: Path,
monkeypatch,
capsys,
) -> None:
repository, _ = _repository(tmp_path)
policy_dir = tmp_path / "candidate"
policy_dir.mkdir()
effective_at = "2026-08-11T00:00:00+08:00"
approval_file, keyring_file, documents = _evidence(policy_dir, effective_at=effective_at)
_mock_verifier(monkeypatch, version="2026-08-12", effective_at=effective_at, documents=documents)
settings = _cli_settings(policy_dir, approval_file, keyring_file)
authorization = tmp_path / "rollback-authorization.txt"
authorization.write_bytes(b"exact rollback authorization")
promoted = public_policy_cli_main(
[
"--database-url", "sqlite+pysqlite:///:memory:",
"promote",
"--apply",
"--confirmation", PROMOTION_CONFIRMATION,
"--authorization-file", str(authorization),
"--expected-generation", "0",
"--expected-current-set-id", "none",
"--idempotency-key", "policy-cli-promote-v1",
"--trace-id", "policy-cli-promote-trace",
],
repository=repository,
settings=settings,
checked_at=datetime(2026, 8, 13, tzinfo=UTC),
)
promotion_payload = json.loads(capsys.readouterr().out)
assert promoted == 0
target_set_id = promotion_payload["state"]["current_set_id"]
denied = public_policy_cli_main(
[
"--database-url", "sqlite+pysqlite:///:memory:",
"rollback",
"--apply",
"--confirmation", PROMOTION_CONFIRMATION,
"--authorization-file", str(authorization),
"--target-set-id", target_set_id,
"--expected-generation", "99",
"--expected-current-set-id", target_set_id,
"--idempotency-key", "policy-cli-rollback-v1",
"--trace-id", "policy-cli-rollback-trace",
],
repository=repository,
settings=settings,
checked_at=datetime(2026, 8, 13, tzinfo=UTC),
)
assert denied == 1
assert repository.state().generation == 1
capsys.readouterr()
rolled_back = public_policy_cli_main(
[
"--database-url", "sqlite+pysqlite:///:memory:",
"rollback",
"--apply",
"--confirmation", ROLLBACK_CONFIRMATION,
"--authorization-file", str(authorization),
"--target-set-id", "missing-policy-set",
"--expected-generation", "1",
"--expected-current-set-id", target_set_id,
"--idempotency-key", "policy-cli-rollback-v1",
"--trace-id", "policy-cli-rollback-trace",
],
repository=repository,
settings=settings,
checked_at=datetime(2026, 8, 13, tzinfo=UTC),
)
rollback_payload = json.loads(capsys.readouterr().out)
assert rolled_back == 1
assert rollback_payload["status"] == "error"
assert repository.state().generation == 1
second_dir = tmp_path / "candidate-v2"
second_dir.mkdir()
second_approval, second_keyring, second_documents = _evidence(
second_dir,
version="2026-08-13",
effective_at=effective_at,
suffix="-v2",
)
_mock_verifier(
monkeypatch,
version="2026-08-13",
effective_at=effective_at,
documents=second_documents,
)
second_settings = _cli_settings(second_dir, second_approval, second_keyring)
second = public_policy_cli_main(
[
"--database-url", "sqlite+pysqlite:///:memory:",
"promote",
"--apply",
"--confirmation", PROMOTION_CONFIRMATION,
"--authorization-file", str(authorization),
"--expected-generation", "1",
"--expected-current-set-id", target_set_id,
"--idempotency-key", "policy-cli-promote-v2",
"--trace-id", "policy-cli-promote-trace-v2",
],
repository=repository,
settings=second_settings,
checked_at=datetime(2026, 8, 14, tzinfo=UTC),
)
second_payload = json.loads(capsys.readouterr().out)
assert second == 0
second_set_id = second_payload["state"]["current_set_id"]
restored = public_policy_cli_main(
[
"--database-url", "sqlite+pysqlite:///:memory:",
"rollback",
"--apply",
"--confirmation", ROLLBACK_CONFIRMATION,
"--authorization-file", str(authorization),
"--target-set-id", target_set_id,
"--expected-generation", "2",
"--expected-current-set-id", second_set_id,
"--idempotency-key", "policy-cli-rollback-v2",
"--trace-id", "policy-cli-rollback-trace-v2",
],
repository=repository,
settings=second_settings,
checked_at=datetime(2026, 8, 14, tzinfo=UTC),
)
restore_payload = json.loads(capsys.readouterr().out)
assert restored == 0
assert restore_payload["status"] == "rolled_back"
assert restore_payload["state"]["generation"] == 3
assert restore_payload["state"]["current_set_id"] == target_set_id
@@ -129,6 +129,14 @@ CONTRACT_API_ANTI_ABUSE_FILES_DAILY_LIMIT=50
CONTRACT_API_ANTI_ABUSE_BYTES_DAILY_LIMIT=524288000
CONTRACT_API_BATCH_DOWNLOAD_MAX_FILES=20
CONTRACT_API_BATCH_DOWNLOAD_MAX_BYTES=104857600
# Verified public-policy material is mounted read-only. The activation executor
# remains false; promotion is performed only through the named operations CLI.
PUBLIC_POLICY_CANDIDATE_DIR_HOST=/etc/yuqei/public-policy/2026-08-12
PUBLIC_POLICY_LEGAL_APPROVAL_FILE_HOST=/etc/yuqei/public-policy/2026-08-12/named-policy-legal-approval.json
PUBLIC_POLICY_TRUSTED_KEYRING_FILE_HOST=/etc/yuqei/public-policy/2026-08-12/policy-legal-keyring.trusted.json
CONTRACT_API_PUBLIC_POLICY_RELEASE_DIGEST=sha256:CHANGE_ME_64_HEX_DIGEST
CONTRACT_API_PUBLIC_POLICY_EXECUTOR_ENABLED=false
CONTRACT_REVIEW_WORKER_SEED_DEMO_DATA=false
CONTRACT_WEB_SESSION_SECRET=CHANGE_ME_USE_A_DIFFERENT_RANDOM_32_BYTE_SECRET
CONTRACT_WEB_SESSION_SECURE=true
@@ -24,8 +24,17 @@ services:
image: ${CONTRACT_API_IMAGE:?CONTRACT_API_IMAGE must reference an immutable production image}
pull_policy: always
# SMTP remains opt-in through docker-compose.production.smtp.yml so disabled mode has no secret-file dependency.
environment:
CONTRACT_API_PUBLIC_POLICY_CANDIDATE_DIR: /run/yuqei/public-policy/candidate
CONTRACT_API_PUBLIC_POLICY_LEGAL_APPROVAL_FILE: /run/yuqei/public-policy/named-policy-legal-approval.json
CONTRACT_API_PUBLIC_POLICY_TRUSTED_KEYRING_FILE: /run/yuqei/public-policy/policy-legal-keyring.trusted.json
CONTRACT_API_PUBLIC_POLICY_RELEASE_DIGEST: ${CONTRACT_API_PUBLIC_POLICY_RELEASE_DIGEST:?CONTRACT_API_PUBLIC_POLICY_RELEASE_DIGEST is required}
CONTRACT_API_PUBLIC_POLICY_EXECUTOR_ENABLED: ${CONTRACT_API_PUBLIC_POLICY_EXECUTOR_ENABLED:-false}
volumes:
- contract-files:/data/contract-files
- ${PUBLIC_POLICY_CANDIDATE_DIR_HOST:?PUBLIC_POLICY_CANDIDATE_DIR_HOST is required}:/run/yuqei/public-policy/candidate:ro
- ${PUBLIC_POLICY_LEGAL_APPROVAL_FILE_HOST:?PUBLIC_POLICY_LEGAL_APPROVAL_FILE_HOST is required}:/run/yuqei/public-policy/named-policy-legal-approval.json:ro
- ${PUBLIC_POLICY_TRUSTED_KEYRING_FILE_HOST:?PUBLIC_POLICY_TRUSTED_KEYRING_FILE_HOST is required}:/run/yuqei/public-policy/policy-legal-keyring.trusted.json:ro
contract-review-worker:
image: ${CONTRACT_API_IMAGE:?CONTRACT_API_IMAGE must reference an immutable production image}
+11
View File
@@ -91,6 +91,17 @@ not change a keyring, toggle the kill switch, import a receipt, or invoke any id
write/execute API. Make control changes through the independently approved operations
procedure, then use this workflow only to observe their propagation.
## Public policy promotion
The production Compose mounts the verified policy candidate directory, named legal approval, and trusted public keyring into Contract API read-only. Keep `CONTRACT_API_PUBLIC_POLICY_EXECUTOR_ENABLED=false`; there is no HTTP write route.
Before activation, run the read-only production preflight and follow:
```text
project-docs/V2-public-policy-production-promotion-runbook-2026-08-17.md
```
The named CLI performs `status`, `preflight`, dry-run, idempotent promotion, and historical rollback. A write requires `--apply`, the exact operation confirmation string, an owner-only authorization file, the approved expected generation/set, a trace ID, and an idempotency key. Policy private keys and the one-time authorization file must never be committed or added to Compose permanently.
## Alibaba Cloud frontend variables
The frontend deployment must use server-only values:
+201 -1
View File
@@ -28,6 +28,10 @@ PLACEHOLDER_MARKERS = ("change_me", "local-only", "example.com", "your-api-key",
LOOPBACK_VALUES = {"127.0.0.1", "::1", "localhost"}
PROVIDER_ACCEPTANCE_RECEIPT_MAX_BYTES = 1024 * 1024
PROVIDER_ACCEPTANCE_KEYRING_MAX_BYTES = 256 * 1024
PUBLIC_POLICY_CANDIDATE_MAX_BYTES = 2 * 1024 * 1024
PUBLIC_POLICY_APPROVAL_MAX_BYTES = 256 * 1024
PUBLIC_POLICY_KEYRING_MAX_BYTES = 256 * 1024
EXPECTED_PUBLIC_POLICY_IDS = {"user_terms", "privacy_policy", "service_agreement", "ai_use_notice"}
ALERT_WEBHOOK_MAX_BYTES = 16 * 1024
SMTP_CREDENTIAL_CONTAINER_PATH = "/run/secrets/contract_smtp_password"
LAUNCH_SCOPE_MANIFEST_PATH = (
@@ -146,6 +150,10 @@ def validate_environment(values: dict[str, str], *, template_mode: bool = False)
"AI_PLATFORM_PROVIDER_ACCEPTANCE_RECEIPT_FILE_HOST",
"AI_PLATFORM_PROVIDER_ACCEPTANCE_TRUSTED_KEYS_FILE_HOST",
"AI_PLATFORM_PROVIDER_ACCEPTANCE_EXPECTED_RECEIPT_SHA256",
"PUBLIC_POLICY_CANDIDATE_DIR_HOST",
"PUBLIC_POLICY_LEGAL_APPROVAL_FILE_HOST",
"PUBLIC_POLICY_TRUSTED_KEYRING_FILE_HOST",
"CONTRACT_API_PUBLIC_POLICY_RELEASE_DIGEST",
"PRODUCTION_ALERT_WEBHOOK_URL_FILE",
)
missing = [key for key in required if not values.get(key, "").strip()]
@@ -157,6 +165,7 @@ def validate_environment(values: dict[str, str], *, template_mode: bool = False)
"CONTRACT_API_SEED_DEMO_DATA": "false",
"CONTRACT_REVIEW_WORKER_SEED_DEMO_DATA": "false",
"CONTRACT_API_REQUIRE_WEB_PROXY_AUTH": "true",
"CONTRACT_API_PUBLIC_POLICY_EXECUTOR_ENABLED": "false",
"CONTRACT_WEB_SESSION_SECURE": "true",
"AI_PLATFORM_ENVIRONMENT": "production",
"AI_PLATFORM_DATABASE_AUTO_CREATE": "false",
@@ -250,6 +259,7 @@ def validate_environment(values: dict[str, str], *, template_mode: bool = False)
database_checks = _validate_database_urls(values)
checks.extend(database_checks)
checks.extend(_validate_provider_acceptance_deployment(values, template_mode=template_mode))
checks.extend(_validate_public_policy_deployment(values, template_mode=template_mode))
public_url = values.get("BACKEND_PUBLIC_BASE_URL", "")
parsed_public_url = urlparse(public_url)
@@ -280,7 +290,7 @@ def validate_files(values: dict[str, str], *, template_mode: bool = False) -> li
Check(
"protected_files",
"warning",
"Template mode skipped TLS, provider key, Provider acceptance receipt/keyring, optional SMTP secret, and backup mount checks.",
"Template mode skipped TLS, provider key, Provider acceptance receipt/keyring, public-policy evidence, optional SMTP secret, and backup mount checks.",
)
]
checks: list[Check] = []
@@ -338,6 +348,8 @@ def validate_files(values: dict[str, str], *, template_mode: bool = False) -> li
)
)
checks.extend(_validate_public_policy_files(values))
email_mode = values.get("CONTRACT_API_IDENTITY_EMAIL_DELIVERY_MODE", "").strip().lower()
smtp_username = values.get("CONTRACT_API_IDENTITY_SMTP_USERNAME", "").strip()
if email_mode == "smtp" and smtp_username:
@@ -412,6 +424,139 @@ def validate_files(values: dict[str, str], *, template_mode: bool = False) -> li
return checks
def _validate_public_policy_files(values: dict[str, str]) -> list[Check]:
checks: list[Check] = []
candidate_root = Path(values.get("PUBLIC_POLICY_CANDIDATE_DIR_HOST", ""))
candidate_ok = candidate_root.is_absolute() and candidate_root.is_dir() and not candidate_root.is_symlink()
total_size = 0
symlink_found = False
if candidate_ok:
try:
for path in candidate_root.rglob("*"):
if path.is_symlink():
symlink_found = True
elif path.is_file():
total_size += path.stat().st_size
except OSError:
candidate_ok = False
candidate_ok = candidate_ok and not symlink_found and 0 < total_size <= PUBLIC_POLICY_CANDIDATE_MAX_BYTES
checks.append(
_result(
"public_policy_candidate_dir",
candidate_ok,
"The public-policy candidate directory is absolute, bounded, and contains no symlinks."
if candidate_ok
else "The public-policy candidate directory must be absolute, non-empty, no larger than 2 MiB, and free of symlinks.",
)
)
manifest_ok = False
if candidate_ok:
manifest_path = candidate_root / "manifest.json"
try:
manifest = _read_strict_json_object(manifest_path)
documents = manifest.get("documents")
manifest_ok = (
manifest.get("schema_version") == "yuqei-public-policy-candidate-v1"
and manifest.get("existing_user_reconsent") == "required_on_next_login"
and isinstance(documents, list)
and len(documents) == 4
and {item.get("policy_id") for item in documents if isinstance(item, dict)}
== EXPECTED_PUBLIC_POLICY_IDS
)
if manifest_ok:
for item in documents:
source_file = item.get("source_file")
relative = Path(str(source_file))
resolved = (candidate_root / relative).resolve()
if (
not isinstance(source_file, str)
or not source_file.strip()
or relative.is_absolute()
or ".." in relative.parts
or not resolved.is_relative_to(candidate_root.resolve())
or not resolved.is_file()
or resolved.stat().st_size <= 0
):
manifest_ok = False
break
except (OSError, UnicodeError, ValueError, TypeError):
manifest_ok = False
checks.append(
_result(
"public_policy_manifest",
manifest_ok,
"The candidate manifest binds all four approved public-policy source files."
if manifest_ok
else "The candidate manifest must bind exactly user terms, privacy policy, service agreement, and AI notice with safe in-directory sources.",
)
)
approval_path = Path(values.get("PUBLIC_POLICY_LEGAL_APPROVAL_FILE_HOST", ""))
approval_ok = _is_bounded_absolute_file(approval_path, PUBLIC_POLICY_APPROVAL_MAX_BYTES)
if approval_ok:
try:
_read_strict_json_object(approval_path)
except (OSError, UnicodeError, ValueError):
approval_ok = False
checks.append(
_result(
"public_policy_legal_approval_file",
approval_ok,
"The named public-policy legal approval file is present and JSON-valid."
if approval_ok
else "The public-policy legal approval must be an absolute JSON file no larger than 256 KiB.",
)
)
keyring_path = Path(values.get("PUBLIC_POLICY_TRUSTED_KEYRING_FILE_HOST", ""))
keyring_ok = _is_bounded_absolute_file(keyring_path, PUBLIC_POLICY_KEYRING_MAX_BYTES)
if keyring_ok:
try:
keyring = _read_strict_json_object(keyring_path)
trust_basis = keyring.get("trust_basis")
keyring_ok = (
isinstance(trust_basis, dict)
and trust_basis.get("policy_activation_authorized") is False
)
except (OSError, UnicodeError, ValueError):
keyring_ok = False
checks.append(
_result(
"public_policy_trusted_keyring_file",
keyring_ok,
"The trusted public-policy keyring is JSON-valid and does not itself authorize activation."
if keyring_ok
else "The trusted public-policy keyring must be JSON-valid and keep policy_activation_authorized=false.",
)
)
if os.name != "nt":
for key, path, enabled in (
("public_policy_legal_approval_permissions", approval_path, approval_ok),
("public_policy_trusted_keyring_permissions", keyring_path, keyring_ok),
):
if enabled:
mode = stat.S_IMODE(path.stat().st_mode)
secure_mode = mode & 0o077 == 0
checks.append(
_result(
key,
secure_mode,
"The public-policy evidence file is owner-only."
if secure_mode
else f"Public-policy evidence file mode {oct(mode)} exposes group or other permissions.",
)
)
return checks
def _read_strict_json_object(path: Path) -> dict[str, object]:
value = json.loads(path.read_text(encoding="utf-8"))
if not isinstance(value, dict):
raise ValueError("JSON value is not an object")
return value
def _validate_identity_email_delivery(values: dict[str, str]) -> Check:
mode = values.get("CONTRACT_API_IDENTITY_EMAIL_DELIVERY_MODE", "").strip().lower()
self_service = values.get("CONTRACT_API_IDENTITY_SELF_SERVICE_ENABLED", "").strip().lower()
@@ -651,6 +796,61 @@ def _validate_database_urls(values: dict[str, str]) -> list[Check]:
return [_result("database_boundaries", not problems, "Database URLs target distinct production PostgreSQL databases on the internal service host." if not problems else f"Invalid database configuration keys: {', '.join(sorted(set(problems)))}.")]
def _validate_public_policy_deployment(
values: dict[str, str],
*,
template_mode: bool,
) -> list[Check]:
candidate_dir = values.get("PUBLIC_POLICY_CANDIDATE_DIR_HOST", "").strip()
approval_file = values.get("PUBLIC_POLICY_LEGAL_APPROVAL_FILE_HOST", "").strip()
trusted_keyring = values.get("PUBLIC_POLICY_TRUSTED_KEYRING_FILE_HOST", "").strip()
release_digest = values.get("CONTRACT_API_PUBLIC_POLICY_RELEASE_DIGEST", "").strip()
contract_image = values.get("CONTRACT_API_IMAGE", "").strip()
deployed_contract_digest = contract_image.rsplit("@", 1)[1] if "@" in contract_image else ""
paths = (candidate_dir, approval_file, trusted_keyring)
paths_ok = (
all(paths)
and all(_is_absolute_host_path(path) for path in paths)
and len(set(paths)) == len(paths)
)
executor_ok = values.get("CONTRACT_API_PUBLIC_POLICY_EXECUTOR_ENABLED", "").strip().lower() == "false"
digest_ok = (
bool(IMAGE_DIGEST_PATTERN.fullmatch(release_digest))
and release_digest == deployed_contract_digest.lower()
)
deployment_ok = paths_ok and executor_ok and digest_ok
forbidden = sorted(
key
for key, value in values.items()
if value.strip() and _is_forbidden_public_policy_key(key)
)
return [
_result(
"public_policy_deployment",
template_mode or deployment_ok,
"Public policy evidence uses distinct absolute paths, the executor is disabled, and the release digest matches the Contract API image."
if deployment_ok
else "Public policy requires distinct absolute evidence paths, executor=false, and a release digest matching the immutable Contract API image.",
warning=template_mode and not deployment_ok,
),
_result(
"public_policy_credential_isolation",
not forbidden,
"Production contains no public-policy private key, password, or staging credential."
if not forbidden
else f"Forbidden public-policy credentials are configured: {', '.join(forbidden)}.",
),
]
def _is_forbidden_public_policy_key(key: str) -> bool:
normalized = key.upper()
if "PUBLIC_POLICY" not in normalized:
return False
return any(token in normalized for token in ("PRIVATE_KEY", "PRIVATE_KEY_FILE", "PASSWORD", "SECRET", "STAGING"))
def _validate_provider_acceptance_deployment(
values: dict[str, str],
*,
@@ -60,6 +60,7 @@ def production_values() -> dict[str, str]:
"CONTRACT_API_IMAGE": f"registry.internal/contract-api@sha256:{'1' * 64}",
"AI_PLATFORM_API_IMAGE": f"registry.internal/ai-platform-api@sha256:{'2' * 64}",
"AI_PLATFORM_RELEASE_IMAGE_DIGEST": f"sha256:{'2' * 64}",
"CONTRACT_API_PUBLIC_POLICY_RELEASE_DIGEST": f"sha256:{'1' * 64}",
"AI_WORKER_IMAGE": f"registry.internal/ai-worker@sha256:{'3' * 64}",
"POSTGRES_IMAGE": f"registry.internal/pgvector@sha256:{'4' * 64}",
"POSTGRES_CLIENT_IMAGE": f"registry.internal/postgres@sha256:{'5' * 64}",
@@ -353,11 +354,119 @@ def test_production_compose_mounts_only_promotion_verification_material_read_onl
assert "AI_PLATFORM_PROVIDER_ACCEPTANCE_EXPECTED_RECEIPT_SHA256: ${" in override
assert ":/run/yuqei/provider-acceptance/receipt.json:ro" in override
assert ":/run/yuqei/provider-acceptance/trusted-keys.json:ro" in override
assert "CONTRACT_API_PUBLIC_POLICY_CANDIDATE_DIR: /run/yuqei/public-policy/candidate" in override
assert "${PUBLIC_POLICY_CANDIDATE_DIR_HOST:?PUBLIC_POLICY_CANDIDATE_DIR_HOST is required}:/run/yuqei/public-policy/candidate:ro" in override
assert "${PUBLIC_POLICY_LEGAL_APPROVAL_FILE_HOST:?PUBLIC_POLICY_LEGAL_APPROVAL_FILE_HOST is required}:/run/yuqei/public-policy/named-policy-legal-approval.json:ro" in override
assert "${PUBLIC_POLICY_TRUSTED_KEYRING_FILE_HOST:?PUBLIC_POLICY_TRUSTED_KEYRING_FILE_HOST is required}:/run/yuqei/public-policy/policy-legal-keyring.trusted.json:ro" in override
assert "CONTRACT_API_PUBLIC_POLICY_EXECUTOR_ENABLED: ${CONTRACT_API_PUBLIC_POLICY_EXECUTOR_ENABLED:-false}" in override
assert 'AI_PLATFORM_PROVIDER_LEGAL_ATTESTATION_SECRET: ""' in override
assert "SIGNING_PRIVATE_KEY" not in override
assert "STAGING_DATABASE_URL" not in override
def _write_public_policy_evidence(tmp_path: Path) -> dict[str, str]:
candidate = tmp_path / "public-policy"
candidate.mkdir()
documents = []
for policy_id, source_file in (
("user_terms", "user-terms.md"),
("privacy_policy", "privacy-policy.md"),
("service_agreement", "service-agreement.md"),
("ai_use_notice", "ai-use-notice.md"),
):
(candidate / source_file).write_text(f"# {policy_id}\n", encoding="utf-8")
documents.append({"policy_id": policy_id, "source_file": source_file})
(candidate / "manifest.json").write_text(
json.dumps(
{
"schema_version": "yuqei-public-policy-candidate-v1",
"existing_user_reconsent": "required_on_next_login",
"documents": documents,
},
ensure_ascii=False,
),
encoding="utf-8",
)
approval = tmp_path / "policy-approval.json"
approval.write_text('{"approval_id":"policy-approval"}', encoding="utf-8")
keyring = tmp_path / "policy-keyring.json"
keyring.write_text(
json.dumps({"trust_basis": {"policy_activation_authorized": False}}),
encoding="utf-8",
)
return {
"PUBLIC_POLICY_CANDIDATE_DIR_HOST": str(candidate.resolve()),
"PUBLIC_POLICY_LEGAL_APPROVAL_FILE_HOST": str(approval.resolve()),
"PUBLIC_POLICY_TRUSTED_KEYRING_FILE_HOST": str(keyring.resolve()),
}
def test_production_preflight_rejects_invalid_public_policy_deployment() -> None:
values = production_values()
values["CONTRACT_API_PUBLIC_POLICY_RELEASE_DIGEST"] = f"sha256:{'f' * 64}"
checks = {check.key: check for check in preflight.validate_environment(values)}
assert checks["public_policy_deployment"].status == "fail"
@pytest.mark.parametrize(
("key", "value"),
[
("PUBLIC_POLICY_CANDIDATE_DIR_HOST", "relative/public-policy"),
("PUBLIC_POLICY_LEGAL_APPROVAL_FILE_HOST", "/etc/yuqei/public-policy/2026-08-12/policy-legal-keyring.trusted.json"),
("CONTRACT_API_PUBLIC_POLICY_EXECUTOR_ENABLED", "true"),
],
)
def test_production_preflight_rejects_unsafe_public_policy_paths_or_executor(
key: str,
value: str,
) -> None:
values = production_values()
values[key] = value
checks = {check.key: check for check in preflight.validate_environment(values)}
assert checks["public_policy_deployment"].status == "fail"
def test_production_preflight_rejects_public_policy_private_material() -> None:
values = production_values()
values["CONTRACT_API_PUBLIC_POLICY_PRIVATE_KEY_FILE"] = "/etc/yuqei/private.pem"
checks = {check.key: check for check in preflight.validate_environment(values)}
assert checks["public_policy_credential_isolation"].status == "fail"
def test_public_policy_evidence_files_are_present_and_safe(tmp_path: Path) -> None:
values = production_values()
values.update(_write_public_policy_evidence(tmp_path))
checks = {check.key: check for check in preflight.validate_files(values)}
assert checks["public_policy_candidate_dir"].status == "pass"
assert checks["public_policy_manifest"].status == "pass"
assert checks["public_policy_legal_approval_file"].status == "pass"
assert checks["public_policy_trusted_keyring_file"].status == "pass"
def test_public_policy_preflight_fails_on_manifest_or_keyring_drift(tmp_path: Path) -> None:
values = production_values()
values.update(_write_public_policy_evidence(tmp_path))
candidate = Path(values["PUBLIC_POLICY_CANDIDATE_DIR_HOST"])
(candidate / "manifest.json").write_text('{"schema_version":"wrong"}', encoding="utf-8")
keyring = Path(values["PUBLIC_POLICY_TRUSTED_KEYRING_FILE_HOST"])
keyring.write_text(
json.dumps({"trust_basis": {"policy_activation_authorized": True}}),
encoding="utf-8",
)
checks = {check.key: check for check in preflight.validate_files(values)}
assert checks["public_policy_manifest"].status == "fail"
assert checks["public_policy_trusted_keyring_file"].status == "fail"
def test_shared_compose_wires_contract_anti_abuse_controls() -> None:
compose = (REPO_ROOT / "yuqei-infra/compose/docker-compose.yml").read_text(
encoding="utf-8"