diff --git a/project-docs/RC06-single-reviewer-one-click-signing-runbook-2026-09-27.md b/project-docs/RC06-single-reviewer-one-click-signing-runbook-2026-09-27.md index 25beb8fa..fa12fac2 100644 --- a/project-docs/RC06-single-reviewer-one-click-signing-runbook-2026-09-27.md +++ b/project-docs/RC06-single-reviewer-one-click-signing-runbook-2026-09-27.md @@ -6,7 +6,7 @@ 本工具只处理 RC-06 的一个精确发布批次：60 个合同资产、14 个 Prompt 模块和 31 项处置结论。唯一人工审核人为陈钰麒（总经理）。离线根密钥和 apply 服务密钥由系统按用途隔离管理，它们不是人工审核人。 -当前没有生成真实审核私钥、正式签名、候选 tag、发布收据或生产授权。内部试点、UAT、生产及公众开放继续 `NO_GO`。一次 RC-06 签名不等于资产已经发布，也不等于授权公众生产切流。 +陈钰麒已在受控本机完成审核密钥初始化；该事实只表示审核私钥已加密保存并完成公钥注册准备。RC-06/recovery/apply 技术密钥、源码 pin、正式签名、候选 tag、发布收据和生产授权仍未形成。内部试点、UAT、生产及公众开放继续 `NO_GO`。一次 RC-06 签名不等于资产已经发布，也不等于授权公众生产切流。 ## 2. 审核人只需要执行的两个入口 @@ -54,6 +54,30 @@ - 后续通过业务 API 应用 74 项内容，并由服务密钥签署 74 项机器结果和在线 hash； - 继续验证 1 个起草＋8 个审查运行时 Prompt 快照以及 RC-06 UAT。 +系统侧使用 `yuqei-infra/scripts/Launch-ReleaseAuthority.ps1` 调用 +`rc06-release-trust-bootstrap.py`。该入口与审核人入口分离，默认非交互： + +- `initialize` 一次性生成互不相同的 RC-06 approval root、recovery root 和 + apply-service 三把 Ed25519 技术密钥；私钥只以 Windows CurrentUser DPAPI + 密文保存在源码树外，并继承仅当前账户与 SYSTEM 可访问的 DACL； +- `build-recovery-keyrings` 使用 recovery root 分别签发 first-release 与 + rollback 两种 schema/key-id 的 keyring；两者可共享 recovery root，但不能 + 混用 schema、root key id 或签名 payload； +- `pin-source` 默认只预览，显式 `--apply` 才会以严格匹配和事务回退方式固定 + 三类公钥 SHA、两个 recovery keyring payload SHA 和精确 epoch；不写入私钥； +- `sign-rc06-keyring` 只在冻结 manifest 形成后签发绑定精确 manifest 原始 SHA + 的候选专属 keyring； +- `sign-apply-payloads` 只签署已经包含 74 项在线 hash、且可由正式 RC-06 + verifier 完整验证的 canonical apply payload；所有 receipt 先在内存生成并 + 逐项自验，再作为禁止覆盖的完整 bundle 原子写出，不调用业务 API； +- `provision-apply-private-key` 在安全 secret-store 交付适配器完成前保持 + fail-closed，不会导出明文或 DPAPI 私钥。 + +DPAPI 与 DACL 不能抵御已经控制当前 Windows 账户的进程。正式初始化仍只能在 +发布负责人确认的干净受控主机上执行；临时 UAT 主机、源码仓库、聊天、邮件、 +共享盘和构建 artifact 均不得成为技术私钥的保存位置。当前新增工具代码本身不 +表示真实技术密钥、pin、keyring、apply receipt 或发布授权已经形成。 + ## 4. 不得执行的操作 - 不得手工编辑 manifest、request、keyring、approval、SHA 或签名 JSON； diff --git a/yuqei-infra/scripts/Launch-ReleaseAuthority.ps1 b/yuqei-infra/scripts/Launch-ReleaseAuthority.ps1 new file mode 100644 index 00000000..a389c304 --- /dev/null +++ b/yuqei-infra/scripts/Launch-ReleaseAuthority.ps1 @@ -0,0 +1,102 @@ +[CmdletBinding()] +param( + [Parameter(Mandatory = $true, Position = 0)] + [ValidateSet( + "runtime-check", + "initialize", + "status", + "build-recovery-keyrings", + "pin-source", + "sign-rc06-keyring", + "sign-apply-payloads", + "provision-apply-private-key" + )] + [string]$Command, + + [Parameter(ValueFromRemainingArguments = $true)] + [string[]]$CommandArguments +) + +$ErrorActionPreference = "Stop" +$env:PSModulePath = @( + (Join-Path $env:SystemRoot "System32\WindowsPowerShell\v1.0\Modules") + (Join-Path $env:ProgramFiles "WindowsPowerShell\Modules") +) -join ";" + +function Get-TrustedPythonCandidates { + $candidatePaths = [System.Collections.Generic.List[string]]::new() + foreach ($pattern in @( + (Join-Path $env:LOCALAPPDATA "Python\pythoncore-*\python.exe") + (Join-Path $env:LOCALAPPDATA "Programs\Python\Python*\python.exe") + (Join-Path $env:ProgramFiles "Python*\python.exe") + )) { + Get-ChildItem -Path $pattern -File -ErrorAction SilentlyContinue | + ForEach-Object { $candidatePaths.Add($_.FullName) } + } + foreach ($registryRoot in @( + "Registry::HKEY_CURRENT_USER\Software\Python\PythonCore" + "Registry::HKEY_LOCAL_MACHINE\Software\Python\PythonCore" + )) { + Get-ChildItem -LiteralPath $registryRoot -ErrorAction SilentlyContinue | + ForEach-Object { + $properties = Get-ItemProperty -LiteralPath (Join-Path $_.PSPath "InstallPath") -ErrorAction SilentlyContinue + if ($properties.ExecutablePath) { + $candidatePaths.Add([string]$properties.ExecutablePath) + } elseif ($properties.'(default)') { + $candidatePaths.Add((Join-Path ([string]$properties.'(default)') "python.exe")) + } + } + } + + $trusted = foreach ($candidatePath in $candidatePaths | Sort-Object -Unique) { + if (-not (Test-Path -LiteralPath $candidatePath -PathType Leaf)) { continue } + $item = Get-Item -LiteralPath $candidatePath -Force + if (($item.Attributes -band [System.IO.FileAttributes]::ReparsePoint) -ne 0) { continue } + if ($item.FullName -like "*\Microsoft\WindowsApps\*") { continue } + $version = [version]::new( + [Math]::Max(0, $item.VersionInfo.FileMajorPart), + [Math]::Max(0, $item.VersionInfo.FileMinorPart), + [Math]::Max(0, $item.VersionInfo.FileBuildPart), + [Math]::Max(0, $item.VersionInfo.FilePrivatePart) + ) + if ($version -lt [version]"3.11") { continue } + $signature = Get-AuthenticodeSignature -LiteralPath $item.FullName + if ( + $signature.Status -ne [System.Management.Automation.SignatureStatus]::Valid -or + $null -eq $signature.SignerCertificate -or + $signature.SignerCertificate.Subject -notlike "*CN=Python Software Foundation*" + ) { continue } + [pscustomobject]@{ Path = $item.FullName; Version = $version } + } + return $trusted | Sort-Object Version -Descending +} + +try { + $authorityScript = Join-Path $PSScriptRoot "rc06-release-trust-bootstrap.py" + if (-not (Test-Path -LiteralPath $authorityScript -PathType Leaf)) { + throw "The release authority program is missing" + } + Remove-Item Env:PYTHONPATH -ErrorAction SilentlyContinue + Remove-Item Env:PYTHONHOME -ErrorAction SilentlyContinue + Remove-Item Env:PYTHONSTARTUP -ErrorAction SilentlyContinue + Remove-Item Env:PYTHONUSERBASE -ErrorAction SilentlyContinue + $runtime = $null + $ErrorActionPreference = "Continue" + foreach ($candidate in @(Get-TrustedPythonCandidates)) { + $preflightOutput = @(& $candidate.Path -I -S -B $authorityScript runtime-check 2>&1) + if ($LASTEXITCODE -eq 0) { + $runtime = $candidate + break + } + } + $ErrorActionPreference = "Stop" + if ($null -eq $runtime) { + throw "No trusted offline Python runtime has DPAPI and the required cryptography dependency" + } + & $runtime.Path -I -S -B $authorityScript $Command @CommandArguments + exit $LASTEXITCODE +} catch { + Write-Host ("Authority startup failed: " + $_.Exception.Message) -ForegroundColor Red + Write-Host "No authority key, source pin, keyring, signature, or private-key export was created." + exit 2 +} diff --git a/yuqei-infra/scripts/build-rc06-publication-batch.py b/yuqei-infra/scripts/build-rc06-publication-batch.py index 7c5377ba..1150884b 100644 --- a/yuqei-infra/scripts/build-rc06-publication-batch.py +++ b/yuqei-infra/scripts/build-rc06-publication-batch.py @@ -134,6 +134,7 @@ GOLDEN_INVENTORY_PATH = ( FROZEN_IMPLEMENTATION_PATHS = ( "yuqei-infra/scripts/build-rc06-publication-batch.py", "yuqei-infra/scripts/rc06-single-reviewer-signing.py", + "yuqei-infra/scripts/rc06-release-trust-bootstrap.py", "yuqei-contract-platform/services/contract-api/src/yuqei_contract_api/contract_templates.py", "yuqei-contract-platform/services/contract-api/src/yuqei_contract_api/contract_scene_assets.py", "yuqei-contract-platform/services/contract-api/src/yuqei_contract_api/template_inputs.py", diff --git a/yuqei-infra/scripts/rc06-release-trust-bootstrap.py b/yuqei-infra/scripts/rc06-release-trust-bootstrap.py new file mode 100644 index 00000000..6668b514 --- /dev/null +++ b/yuqei-infra/scripts/rc06-release-trust-bootstrap.py @@ -0,0 +1,1756 @@ +#!/usr/bin/env python3 +"""Bootstrap and operate the technical trust keys used by RC-06 and recovery. + +The authority keys are deliberately separate from the human reviewer's key: + +* ``rc06-approval-root`` signs candidate-specific RC-06 approval keyrings; +* ``recovery-root`` signs the first-release and rollback recovery keyrings; +* ``apply-service`` signs system-generated publication apply receipts. + +Private key material is never written in plaintext. The supported bootstrap +runtime is Windows and uses CurrentUser DPAPI plus an owner-only local directory +ACL. A process already running as the same Windows user is inside this trust +boundary. Cross-platform apply-key delivery is intentionally not implemented; +the explicit provisioning command fails closed instead of exporting a key. +""" + +from __future__ import annotations + +import argparse +import base64 +import ctypes +import importlib.util +import json +import os +import re +import stat +import sys +import sysconfig +import tempfile +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta +from pathlib import Path +from typing import Any, Callable, Mapping + + +def _is_relative_to(path: Path, parent: Path) -> bool: + try: + path.relative_to(parent) + return True + except ValueError: + return False + + +def _bootstrap_isolated_site_packages() -> tuple[Path, ...]: + if not (sys.flags.isolated and sys.flags.no_site): + return () + base = Path(sys.base_prefix).resolve(strict=True) + allowed: list[Path] = [] + for name in ("purelib", "platlib"): + configured = Path(sysconfig.get_path(name)).resolve(strict=True) + if not _is_relative_to(configured, base): + raise RuntimeError("isolated Python package path escapes the selected runtime") + if configured not in allowed: + sys.path.append(os.fspath(configured)) + allowed.append(configured) + return tuple(allowed) + + +ISOLATED_SITE_PACKAGES = _bootstrap_isolated_site_packages() + +import cryptography +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric.ed25519 import ( + Ed25519PrivateKey, + Ed25519PublicKey, +) + + +REPO_ROOT = Path(__file__).resolve().parents[2] +SCRIPTS_ROOT = Path(__file__).resolve().parent + + +def _load_script_module(name: str, path: Path) -> Any: + spec = importlib.util.spec_from_file_location(name, path) + if spec is None or spec.loader is None: # pragma: no cover - import failure + raise RuntimeError(f"cannot load required release module: {path.name}") + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +signer = _load_script_module( + "yuqei_rc06_single_reviewer_signer", + SCRIPTS_ROOT / "rc06-single-reviewer-signing.py", +) +batch = signer.batch + +AUTHORITY_BUNDLE_SCHEMA = "yuqei.release-authority-key-bundle.v1" +AUTHORITY_REGISTRATION_SCHEMA = "yuqei.release-authority-key-registration.v1" +PROTECTED_KEY_SCHEMA = "yuqei.release-authority-dpapi-private-key.v1" +RECOVERY_BUNDLE_SCHEMA = "yuqei.release-recovery-keyring-bundle.v1" +RC06_KEYRING_BUNDLE_SCHEMA = "yuqei.rc06-candidate-keyring-bundle.v1" +APPLY_RECEIPT_BUNDLE_SCHEMA = "yuqei.rc06-apply-receipt-bundle.v1" +PIN_PLAN_SCHEMA = "yuqei.release-authority-pin-plan.v1" +PROTECTION_NAME = "windows-dpapi-current-user" +DPAPI_DESCRIPTION = "Yuqei release authority private key" +DPAPI_ENTROPY_PREFIX = b"YUQEI_RELEASE_AUTHORITY_V1\0" +KEYS_DIRECTORY_NAME = "keys" +AUTHORITY_MANIFEST_FILENAME = "authority-keys.json" +MAX_PRIVATE_BUNDLE_BYTES = 256 * 1024 +MAX_REGISTRATION_BYTES = 64 * 1024 +DEFAULT_RECOVERY_VALIDITY = timedelta(days=365) +DEFAULT_RC06_VALIDITY = timedelta(days=7) +HEX64_RE = re.compile(r"^[0-9a-f]{64}$") + +ROLE_RC06_ROOT = "rc06-approval-root" +ROLE_RECOVERY_ROOT = "recovery-root" +ROLE_APPLY_SERVICE = "apply-service" +AUTHORITY_ROLES = (ROLE_RC06_ROOT, ROLE_RECOVERY_ROOT, ROLE_APPLY_SERVICE) + +ROLE_KEY_IDS = { + ROLE_RC06_ROOT: batch.TRUSTED_APPROVAL_ROOT_KEY_ID, + ROLE_RECOVERY_ROOT: "yuqei-recovery-authority-root-v1", + ROLE_APPLY_SERVICE: "yuqei-rc06-apply-service-v1", +} + +FIRST_RELEASE_KEYRING_SCHEMA = "yuqei.first-release-recovery-keyring.v2" +FIRST_RELEASE_ROOT_KEY_ID = "yuqei-first-release-recovery-root-v1" +ROLLBACK_KEYRING_SCHEMA = "yuqei.rollback-rehearsal-keyring.v3" +ROLLBACK_ROOT_KEY_ID = "yuqei-rollback-rehearsal-root-v1" +REVIEWER_REGISTRATION_SCHEMA = "yuqei.rc06-reviewer-key-registration.v1" + + +class AuthorityError(RuntimeError): + """Raised when authority material cannot be created or verified safely.""" + + +ProtectFunction = Callable[[bytes, str], bytes] +UnprotectFunction = Callable[[bytes, str], bytes] + + +@dataclass(frozen=True) +class AuthorityKey: + role: str + key_id: str + private_key: Ed25519PrivateKey + public_key: Ed25519PublicKey + public_raw: bytes + public_sha256: str + registration: dict[str, Any] + + +@dataclass(frozen=True) +class PinFileChange: + path: Path + original: bytes + proposed: bytes + + @property + def original_sha256(self) -> str: + return batch._sha256_bytes(self.original) + + @property + def proposed_sha256(self) -> str: + return batch._sha256_bytes(self.proposed) + + +@dataclass(frozen=True) +class PinPlan: + epoch: int + changes: tuple[PinFileChange, ...] + values: Mapping[str, str | int] + + +class _DataBlob(ctypes.Structure): + _fields_ = [ + ("cbData", ctypes.c_uint32), + ("pbData", ctypes.POINTER(ctypes.c_ubyte)), + ] + + +def _blob(data: bytes) -> tuple[_DataBlob, Any]: + buffer = ctypes.create_string_buffer(data) + return ( + _DataBlob(len(data), ctypes.cast(buffer, ctypes.POINTER(ctypes.c_ubyte))), + buffer, + ) + + +def _dpapi_crypt(raw: bytes, role: str, *, decrypt: bool) -> bytes: + if os.name != "nt": + raise AuthorityError( + "release authority private keys require Windows CurrentUser DPAPI; " + "cross-platform private-key bootstrap is not supported" + ) + crypt32 = ctypes.WinDLL("crypt32", use_last_error=True) + kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) + input_blob, input_buffer = _blob(raw) + entropy_blob, entropy_buffer = _blob(DPAPI_ENTROPY_PREFIX + role.encode("ascii")) + output_blob = _DataBlob() + flags = 0x1 # CRYPTPROTECT_UI_FORBIDDEN + if decrypt: + function = crypt32.CryptUnprotectData + function.argtypes = [ + ctypes.POINTER(_DataBlob), + ctypes.c_void_p, + ctypes.POINTER(_DataBlob), + ctypes.c_void_p, + ctypes.c_void_p, + ctypes.c_uint32, + ctypes.POINTER(_DataBlob), + ] + args = ( + ctypes.byref(input_blob), + None, + ctypes.byref(entropy_blob), + None, + None, + flags, + ctypes.byref(output_blob), + ) + else: + function = crypt32.CryptProtectData + function.argtypes = [ + ctypes.POINTER(_DataBlob), + ctypes.c_wchar_p, + ctypes.POINTER(_DataBlob), + ctypes.c_void_p, + ctypes.c_void_p, + ctypes.c_uint32, + ctypes.POINTER(_DataBlob), + ] + args = ( + ctypes.byref(input_blob), + DPAPI_DESCRIPTION, + ctypes.byref(entropy_blob), + None, + None, + flags, + ctypes.byref(output_blob), + ) + function.restype = ctypes.c_int + # Keep the ctypes buffers alive until the API returns. + _ = (input_buffer, entropy_buffer) + if not function(*args): + error = ctypes.get_last_error() + raise AuthorityError(f"Windows DPAPI operation failed with error {error}") + try: + return ctypes.string_at(output_blob.pbData, output_blob.cbData) + finally: + kernel32.LocalFree.argtypes = [ctypes.c_void_p] + kernel32.LocalFree(output_blob.pbData) + + +def _protect_current_user(raw: bytes, role: str) -> bytes: + return _dpapi_crypt(raw, role, decrypt=False) + + +def _unprotect_current_user(raw: bytes, role: str) -> bytes: + return _dpapi_crypt(raw, role, decrypt=True) + + +def _now_utc() -> datetime: + return datetime.now(UTC).replace(microsecond=0) + + +def _format_time(value: datetime) -> str: + normalized = value.astimezone(UTC).replace(microsecond=0) + return normalized.isoformat().replace("+00:00", "Z") + + +def _parse_time(value: Any, label: str) -> datetime: + if not isinstance(value, str): + raise AuthorityError(f"{label} must be an RFC3339 string") + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError as exc: + raise AuthorityError(f"{label} is invalid") from exc + if parsed.tzinfo is None: + raise AuthorityError(f"{label} must include a timezone") + return parsed.astimezone(UTC) + + +def _default_state_directory() -> Path: + if os.name != "nt": + raise AuthorityError("the default release authority directory is Windows-only") + configured = os.environ.get("LOCALAPPDATA") + if not configured: + raise AuthorityError("LOCALAPPDATA is unavailable") + return Path(configured) / "Yuqei" / "ReleaseAuthority" + + +def _private_filename(role: str) -> str: + return f"{role}.private.dpapi.json" + + +def _public_filename(role: str) -> str: + return f"{role}.public.raw" + + +def _registration_filename(role: str) -> str: + return f"{role}.registration.json" + + +def _expected_authority_names() -> set[str]: + names = {AUTHORITY_MANIFEST_FILENAME} + for role in AUTHORITY_ROLES: + names.update( + { + _private_filename(role), + _public_filename(role), + _registration_filename(role), + } + ) + return names + + +def _canonical(value: Any) -> bytes: + try: + return batch._canonical(value) + except batch.PublicationBatchError as exc: + raise AuthorityError(str(exc)) from exc + + +def _public_raw(public_key: Ed25519PublicKey) -> bytes: + return public_key.public_bytes( + serialization.Encoding.Raw, + serialization.PublicFormat.Raw, + ) + + +def _private_der(private_key: Ed25519PrivateKey) -> bytes: + return private_key.private_bytes( + serialization.Encoding.DER, + serialization.PrivateFormat.PKCS8, + serialization.NoEncryption(), + ) + + +def _ensure_epoch(epoch: Any) -> int: + if not isinstance(epoch, int) or isinstance(epoch, bool) or epoch < 1: + raise AuthorityError("keyring epoch must be a positive integer") + return epoch + + +def _read_json(path: Path, label: str, *, maximum_bytes: int = MAX_REGISTRATION_BYTES) -> dict[str, Any]: + try: + payload, _ = batch.secure_read_json(path, label) + except batch.PublicationBatchError as exc: + raise AuthorityError(str(exc)) from exc + if not isinstance(payload, dict): + raise AuthorityError(f"{label} must be a JSON object") + try: + size = path.stat().st_size + except OSError as exc: + raise AuthorityError(f"cannot inspect {label}: {exc}") from exc + if size > maximum_bytes: + raise AuthorityError(f"{label} is too large") + return payload + + +def _authority_files( + *, + protect: ProtectFunction, + created_at: datetime, +) -> tuple[dict[str, bytes], dict[str, str]]: + files: dict[str, bytes] = {} + fingerprints: dict[str, str] = {} + registrations: list[dict[str, Any]] = [] + public_values: set[bytes] = set() + for role in AUTHORITY_ROLES: + private_key = Ed25519PrivateKey.generate() + public_raw = _public_raw(private_key.public_key()) + if public_raw in public_values: + raise AuthorityError("authority key generation unexpectedly reused a public key") + public_values.add(public_raw) + fingerprint = batch._sha256_bytes(public_raw) + key_id = ROLE_KEY_IDS[role] + protected = protect(_private_der(private_key), role) + if not protected: + raise AuthorityError(f"{role} protection returned an empty ciphertext") + envelope = { + "schema_version": PROTECTED_KEY_SCHEMA, + "role": role, + "key_id": key_id, + "algorithm": "Ed25519", + "protection": PROTECTION_NAME, + "public_key_raw_sha256": fingerprint, + "ciphertext_base64": base64.b64encode(protected).decode("ascii"), + } + registration = { + "schema_version": AUTHORITY_REGISTRATION_SCHEMA, + "role": role, + "key_id": key_id, + "algorithm": "Ed25519", + "created_at": _format_time(created_at), + "public_key_raw_sha256": fingerprint, + "public_key_filename": _public_filename(role), + } + files[_private_filename(role)] = _canonical(envelope) + files[_public_filename(role)] = public_raw + files[_registration_filename(role)] = _canonical(registration) + fingerprints[role] = fingerprint + registrations.append(registration) + manifest = { + "schema_version": AUTHORITY_BUNDLE_SCHEMA, + "created_at": _format_time(created_at), + "protection": PROTECTION_NAME, + "roles": registrations, + "private_key_export_supported": False, + "same_account_compromise_outside_boundary": True, + } + files[AUTHORITY_MANIFEST_FILENAME] = _canonical(manifest) + return files, fingerprints + + +def initialize_authority( + state_directory: Path, + *, + protect: ProtectFunction = _protect_current_user, + unprotect: UnprotectFunction = _unprotect_current_user, + created_at: datetime | None = None, +) -> dict[str, Any]: + state_directory = signer._assert_local_safe_path( + state_directory, + "release authority state directory", + allow_existing=True, + ) + with signer._state_operation_lock(state_directory): + keys_directory = state_directory / KEYS_DIRECTORY_NAME + if keys_directory.exists() or keys_directory.is_symlink(): + keys = load_authority(state_directory, unprotect=unprotect) + return { + "status": "already_initialized_and_verified", + "roles": {role: keys[role].public_sha256 for role in AUTHORITY_ROLES}, + "state_directory": os.fspath(state_directory), + "private_key_export_supported": False, + } + files, expected_fingerprints = _authority_files( + protect=protect, + created_at=(created_at or _now_utc()), + ) + try: + signer._write_bundle_atomically(keys_directory, files, private_directory=True) + for role in AUTHORITY_ROLES: + signer._harden_private_file(keys_directory / _private_filename(role)) + keys = load_authority(state_directory, unprotect=unprotect) + except (signer.SignerError, batch.PublicationBatchError) as exc: + raise AuthorityError(str(exc)) from exc + actual = {role: keys[role].public_sha256 for role in AUTHORITY_ROLES} + if actual != expected_fingerprints: + raise AuthorityError("authority key self-verification failed") + return { + "status": "initialized_not_pinned_not_signed", + "roles": actual, + "state_directory": os.fspath(state_directory), + "private_key_export_supported": False, + } + + +def load_authority( + state_directory: Path, + *, + unprotect: UnprotectFunction = _unprotect_current_user, +) -> dict[str, AuthorityKey]: + state_directory = signer._assert_local_safe_path( + state_directory, + "release authority state directory", + allow_existing=True, + ) + keys_directory = state_directory / KEYS_DIRECTORY_NAME + try: + signer._verify_complete_bundle( + keys_directory, + _expected_authority_names(), + "release authority key bundle", + ) + signer._harden_private_directory(keys_directory) + except (signer.SignerError, batch.PublicationBatchError) as exc: + raise AuthorityError(str(exc)) from exc + manifest = _read_json( + keys_directory / AUTHORITY_MANIFEST_FILENAME, + "release authority manifest", + ) + if set(manifest) != { + "schema_version", + "created_at", + "protection", + "roles", + "private_key_export_supported", + "same_account_compromise_outside_boundary", + } or manifest["schema_version"] != AUTHORITY_BUNDLE_SCHEMA: + raise AuthorityError("release authority manifest has an unsupported shape") + if ( + manifest["protection"] != PROTECTION_NAME + or manifest["private_key_export_supported"] is not False + or manifest["same_account_compromise_outside_boundary"] is not True + ): + raise AuthorityError("release authority manifest weakens its trust boundary") + rows = manifest["roles"] + if not isinstance(rows, list) or [row.get("role") for row in rows if isinstance(row, dict)] != list( + AUTHORITY_ROLES + ): + raise AuthorityError("release authority manifest role inventory drift") + result: dict[str, AuthorityKey] = {} + seen_public: set[bytes] = set() + for role in AUTHORITY_ROLES: + registration = _read_json( + keys_directory / _registration_filename(role), + f"{role} registration", + ) + expected_registration_fields = { + "schema_version", + "role", + "key_id", + "algorithm", + "created_at", + "public_key_raw_sha256", + "public_key_filename", + } + if set(registration) != expected_registration_fields: + raise AuthorityError(f"{role} registration has an invalid shape") + if ( + registration["schema_version"] != AUTHORITY_REGISTRATION_SCHEMA + or registration["role"] != role + or registration["key_id"] != ROLE_KEY_IDS[role] + or registration["algorithm"] != "Ed25519" + or registration["public_key_filename"] != _public_filename(role) + ): + raise AuthorityError(f"{role} registration identity drift") + _parse_time(registration["created_at"], f"{role} created_at") + public_raw = batch.secure_read_bytes( + keys_directory / _public_filename(role), + f"{role} public key", + maximum_bytes=32, + ) + if len(public_raw) != 32: + raise AuthorityError(f"{role} public key must contain exactly 32 bytes") + fingerprint = batch._sha256_bytes(public_raw) + if registration["public_key_raw_sha256"] != fingerprint or not HEX64_RE.fullmatch(fingerprint): + raise AuthorityError(f"{role} public key fingerprint drift") + envelope = _read_json( + keys_directory / _private_filename(role), + f"{role} protected private key", + maximum_bytes=MAX_PRIVATE_BUNDLE_BYTES, + ) + if set(envelope) != { + "schema_version", + "role", + "key_id", + "algorithm", + "protection", + "public_key_raw_sha256", + "ciphertext_base64", + }: + raise AuthorityError(f"{role} protected private key has an invalid shape") + if ( + envelope["schema_version"] != PROTECTED_KEY_SCHEMA + or envelope["role"] != role + or envelope["key_id"] != ROLE_KEY_IDS[role] + or envelope["algorithm"] != "Ed25519" + or envelope["protection"] != PROTECTION_NAME + or envelope["public_key_raw_sha256"] != fingerprint + ): + raise AuthorityError(f"{role} protected private key identity drift") + try: + protected = base64.b64decode(envelope["ciphertext_base64"], validate=True) + private_der = unprotect(protected, role) + private_key = serialization.load_der_private_key(private_der, password=None) + except (TypeError, ValueError) as exc: + raise AuthorityError(f"{role} protected private key cannot be opened") from exc + if not isinstance(private_key, Ed25519PrivateKey): + raise AuthorityError(f"{role} private key must be Ed25519") + public_key = private_key.public_key() + if _public_raw(public_key) != public_raw: + raise AuthorityError(f"{role} private/public key mismatch") + if public_raw in seen_public: + raise AuthorityError("authority roles reuse an Ed25519 key") + seen_public.add(public_raw) + signer._harden_private_file(keys_directory / _private_filename(role)) + result[role] = AuthorityKey( + role=role, + key_id=registration["key_id"], + private_key=private_key, + public_key=public_key, + public_raw=public_raw, + public_sha256=fingerprint, + registration=registration, + ) + return result + + +def load_reviewer_registration(path: Path) -> tuple[dict[str, Any], Ed25519PublicKey, bytes]: + registration = _read_json(path, "reviewer public registration") + expected = { + "schema_version", + "key_id", + "algorithm", + "reviewer", + "created_at", + "public_key_pem", + "public_key_raw_sha256", + "public_key_pem_sha256", + } + if set(registration) != expected: + raise AuthorityError("reviewer public registration has an invalid shape") + if ( + registration["schema_version"] != REVIEWER_REGISTRATION_SCHEMA + or registration["algorithm"] != "Ed25519" + or registration["reviewer"] != batch.SOLE_RELEASE_REVIEWER + ): + raise AuthorityError("reviewer public registration identity drift") + _parse_time(registration["created_at"], "reviewer registration created_at") + try: + public_pem = registration["public_key_pem"].encode("ascii") + public_key = serialization.load_pem_public_key(public_pem) + except (AttributeError, UnicodeError, ValueError) as exc: + raise AuthorityError("reviewer registration public key is invalid") from exc + if not isinstance(public_key, Ed25519PublicKey): + raise AuthorityError("reviewer registration public key must be Ed25519") + public_raw = _public_raw(public_key) + if ( + registration["public_key_raw_sha256"] != batch._sha256_bytes(public_raw) + or registration["public_key_pem_sha256"] != batch._sha256_bytes(public_pem) + ): + raise AuthorityError("reviewer registration public key fingerprint drift") + return registration, public_key, public_raw + + +def _assert_reviewer_separate(keys: Mapping[str, AuthorityKey], reviewer_raw: bytes) -> None: + if reviewer_raw in {item.public_raw for item in keys.values()}: + raise AuthorityError("reviewer key must differ from every technical authority key") + + +def _sign_recovery_keyring( + *, + schema_version: str, + root_key_id: str, + epoch: int, + issued_at: datetime, + expires_at: datetime, + reviewer_registration: Mapping[str, Any], + reviewer_raw: bytes, + recovery_root: AuthorityKey, + include_algorithm: bool, +) -> dict[str, Any]: + key = { + "key_id": reviewer_registration["key_id"], + "approver_id": batch.SOLE_RELEASE_REVIEWER["approver_id"], + "role": "release_owner", + "public_key_base64": base64.b64encode(reviewer_raw).decode("ascii"), + "not_before": _format_time(issued_at), + "not_after": _format_time(expires_at), + "revoked": False, + } + if include_algorithm: + key = {**key, "algorithm": "ed25519"} + unsigned = { + "schema_version": schema_version, + "root_key_id": root_key_id, + "epoch": epoch, + "issued_at": _format_time(issued_at), + "expires_at": _format_time(expires_at), + "keys": [key], + } + signature = recovery_root.private_key.sign(_canonical(unsigned)) + return { + **unsigned, + "keyring_payload_sha256": batch._sha256_bytes(_canonical(unsigned)), + "root_signature_base64": base64.b64encode(signature).decode("ascii"), + } + + +def _validate_recovery_keyrings( + first_release: Mapping[str, Any], + rollback: Mapping[str, Any], + recovery_root: AuthorityKey, + *, + now: datetime, +) -> None: + first_module = _load_script_module( + "yuqei_first_release_snapshot_restore_validation", + REPO_ROOT / "yuqei-infra" / "deploy" / "plan-first-release-snapshot-restore.py", + ) + rollback_module = _load_script_module( + "yuqei_rollback_rehearsal_validation", + REPO_ROOT / "yuqei-infra" / "deploy" / "plan-rollback-rehearsal.py", + ) + for module, keyring in ((first_module, first_release), (rollback_module, rollback)): + module.TRUSTED_ROOT_PUBLIC_KEY_SHA256 = recovery_root.public_sha256 + module.TRUSTED_CURRENT_KEYRING_PAYLOAD_SHA256 = keyring["keyring_payload_sha256"] + module.MIN_KEYRING_EPOCH = keyring["epoch"] + try: + first_module._validate_keyring( + first_release, + root_public_key=recovery_root.public_key, + now=now, + ) + rollback_module._validate_keyring( + rollback, + now=now, + trusted_root_public_key=recovery_root.public_key, + ) + except Exception as exc: + raise AuthorityError(f"recovery keyring self-verification failed: {exc}") from exc + + +def build_recovery_keyrings( + state_directory: Path, + reviewer_registration_file: Path, + *, + epoch: int, + unprotect: UnprotectFunction = _unprotect_current_user, + now: datetime | None = None, + validity: timedelta = DEFAULT_RECOVERY_VALIDITY, +) -> dict[str, Any]: + epoch = _ensure_epoch(epoch) + current = (now or _now_utc()).astimezone(UTC).replace(microsecond=0) + if validity <= timedelta(0): + raise AuthorityError("recovery keyring validity must be positive") + state_directory = signer._assert_local_safe_path( + state_directory, + "release authority state directory", + allow_existing=True, + ) + with signer._state_operation_lock(state_directory): + keys = load_authority(state_directory, unprotect=unprotect) + reviewer_registration, _reviewer_public, reviewer_raw = load_reviewer_registration( + reviewer_registration_file + ) + _assert_reviewer_separate(keys, reviewer_raw) + recovery_root = keys[ROLE_RECOVERY_ROOT] + output_directory = state_directory / "keyrings" / f"recovery-epoch-{epoch}" + if output_directory.exists() or output_directory.is_symlink(): + return _load_existing_recovery_bundle( + output_directory, + keys=keys, + reviewer_registration=reviewer_registration, + reviewer_raw=reviewer_raw, + now=current, + ) + expires = current + validity + first_release = _sign_recovery_keyring( + schema_version=FIRST_RELEASE_KEYRING_SCHEMA, + root_key_id=FIRST_RELEASE_ROOT_KEY_ID, + epoch=epoch, + issued_at=current, + expires_at=expires, + reviewer_registration=reviewer_registration, + reviewer_raw=reviewer_raw, + recovery_root=recovery_root, + include_algorithm=False, + ) + rollback = _sign_recovery_keyring( + schema_version=ROLLBACK_KEYRING_SCHEMA, + root_key_id=ROLLBACK_ROOT_KEY_ID, + epoch=epoch, + issued_at=current, + expires_at=expires, + reviewer_registration=reviewer_registration, + reviewer_raw=reviewer_raw, + recovery_root=recovery_root, + include_algorithm=True, + ) + _validate_recovery_keyrings( + first_release, + rollback, + recovery_root, + now=current, + ) + receipt = { + "schema_version": RECOVERY_BUNDLE_SCHEMA, + "epoch": epoch, + "issued_at": _format_time(current), + "expires_at": _format_time(expires), + "recovery_root_public_key_sha256": recovery_root.public_sha256, + "reviewer_key_id": reviewer_registration["key_id"], + "reviewer_public_key_sha256": reviewer_registration["public_key_raw_sha256"], + "first_release_keyring_payload_sha256": first_release["keyring_payload_sha256"], + "rollback_keyring_payload_sha256": rollback["keyring_payload_sha256"], + "private_keys_in_bundle": False, + } + files = { + "first-release-recovery-keyring.json": _canonical(first_release), + "rollback-rehearsal-keyring.json": _canonical(rollback), + "recovery-root-public.raw": recovery_root.public_raw, + "recovery-keyring-bundle.json": _canonical(receipt), + } + try: + signer._write_bundle_atomically(output_directory, files, private_directory=True) + except signer.SignerError as exc: + raise AuthorityError(str(exc)) from exc + return { + "status": "recovery_keyrings_created_not_pinned", + "directory": os.fspath(output_directory), + **receipt, + } + + +def _load_existing_recovery_bundle( + directory: Path, + *, + keys: Mapping[str, AuthorityKey], + reviewer_registration: Mapping[str, Any], + reviewer_raw: bytes, + now: datetime, +) -> dict[str, Any]: + expected_names = { + "first-release-recovery-keyring.json", + "rollback-rehearsal-keyring.json", + "recovery-root-public.raw", + "recovery-keyring-bundle.json", + } + try: + signer._verify_complete_bundle(directory, expected_names, "recovery keyring bundle") + except signer.SignerError as exc: + raise AuthorityError(str(exc)) from exc + first_release = _read_json( + directory / "first-release-recovery-keyring.json", + "first-release recovery keyring", + ) + rollback = _read_json( + directory / "rollback-rehearsal-keyring.json", + "rollback rehearsal keyring", + ) + receipt = _read_json(directory / "recovery-keyring-bundle.json", "recovery keyring receipt") + recovery_root = keys[ROLE_RECOVERY_ROOT] + public_raw = batch.secure_read_bytes( + directory / "recovery-root-public.raw", + "recovery root public key", + maximum_bytes=32, + ) + if public_raw != recovery_root.public_raw: + raise AuthorityError("existing recovery bundle uses a different root") + _assert_reviewer_separate(keys, reviewer_raw) + _validate_recovery_keyrings(first_release, rollback, recovery_root, now=now) + expected_receipt = { + "schema_version": RECOVERY_BUNDLE_SCHEMA, + "epoch": first_release["epoch"], + "issued_at": first_release["issued_at"], + "expires_at": first_release["expires_at"], + "recovery_root_public_key_sha256": recovery_root.public_sha256, + "reviewer_key_id": reviewer_registration["key_id"], + "reviewer_public_key_sha256": reviewer_registration["public_key_raw_sha256"], + "first_release_keyring_payload_sha256": first_release["keyring_payload_sha256"], + "rollback_keyring_payload_sha256": rollback["keyring_payload_sha256"], + "private_keys_in_bundle": False, + } + if receipt != expected_receipt: + raise AuthorityError("existing recovery keyring receipt drift") + return { + "status": "recovery_keyrings_already_created_and_verified_not_pinned", + "directory": os.fspath(directory), + **receipt, + } + + +def build_rc06_candidate_keyring( + state_directory: Path, + reviewer_registration_file: Path, + manifest_file: Path, + manifest_sha256: str, + *, + unprotect: UnprotectFunction = _unprotect_current_user, + now: datetime | None = None, + validity: timedelta = DEFAULT_RC06_VALIDITY, +) -> dict[str, Any]: + current = (now or _now_utc()).astimezone(UTC).replace(microsecond=0) + if validity <= timedelta(0) or validity > DEFAULT_RC06_VALIDITY: + raise AuthorityError("RC-06 keyring validity must be positive and no longer than seven days") + if not HEX64_RE.fullmatch(manifest_sha256): + raise AuthorityError("publication manifest SHA-256 is invalid") + state_directory = signer._assert_local_safe_path( + state_directory, + "release authority state directory", + allow_existing=True, + ) + with signer._state_operation_lock(state_directory): + keys = load_authority(state_directory, unprotect=unprotect) + registration, _reviewer_public, reviewer_raw = load_reviewer_registration( + reviewer_registration_file + ) + _assert_reviewer_separate(keys, reviewer_raw) + rc06_root = keys[ROLE_RC06_ROOT] + if batch.TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256 != rc06_root.public_sha256: + raise AuthorityError("frozen RC-06 root pin does not match the initialized authority") + epoch = _ensure_epoch(batch.FROZEN_APPROVAL_KEYRING_EPOCH) + try: + manifest_raw = batch.secure_read_bytes(manifest_file, "RC-06 publication manifest") + except batch.PublicationBatchError as exc: + raise AuthorityError(str(exc)) from exc + if batch._sha256_bytes(manifest_raw) != manifest_sha256: + raise AuthorityError("publication manifest raw SHA-256 mismatch") + try: + manifest = json.loads( + manifest_raw, + object_pairs_hook=batch._reject_duplicate_json_pairs, + parse_constant=batch._reject_json_constant, + ) + batch.validate_manifest(manifest) + except (UnicodeDecodeError, json.JSONDecodeError, batch.PublicationBatchError) as exc: + raise AuthorityError(f"publication manifest is invalid: {exc}") from exc + output_directory = state_directory / "keyrings" / f"rc06-{manifest_sha256[:24]}" + if output_directory.exists() or output_directory.is_symlink(): + return _load_existing_rc06_keyring_bundle( + output_directory, + keys=keys, + registration=registration, + manifest_sha256=manifest_sha256, + now=current, + ) + expires = current + validity + unsigned = { + "schema_version": batch.APPROVAL_KEYRING_SCHEMA_VERSION, + "root_key_id": batch.TRUSTED_APPROVAL_ROOT_KEY_ID, + "epoch": epoch, + "issued_at": _format_time(current), + "expires_at": _format_time(expires), + "authorized_batch_manifest_raw_sha256": manifest_sha256, + "keys": [ + { + "key_id": registration["key_id"], + "algorithm": "Ed25519", + "public_key_pem": registration["public_key_pem"], + **batch.SOLE_RELEASE_REVIEWER, + "not_before": _format_time(current), + "not_after": _format_time(expires), + "revoked": False, + } + ], + } + keyring = { + **unsigned, + "keyring_payload_sha256": batch._sha256_json(unsigned), + "root_signature_base64": base64.b64encode( + rc06_root.private_key.sign( + batch.APPROVAL_KEYRING_SIGNATURE_DOMAIN + _canonical(unsigned) + ) + ).decode("ascii"), + } + try: + batch._validate_approval_keyring( + keyring, + root_public_key=rc06_root.public_key, + approved_at=current, + ) + except batch.PublicationBatchError as exc: + raise AuthorityError(f"RC-06 keyring self-verification failed: {exc}") from exc + receipt = { + "schema_version": RC06_KEYRING_BUNDLE_SCHEMA, + "epoch": epoch, + "manifest_raw_sha256": manifest_sha256, + "keyring_payload_sha256": keyring["keyring_payload_sha256"], + "root_public_key_sha256": rc06_root.public_sha256, + "reviewer_key_id": registration["key_id"], + "reviewer_public_key_sha256": registration["public_key_raw_sha256"], + "issued_at": keyring["issued_at"], + "expires_at": keyring["expires_at"], + "private_keys_in_bundle": False, + } + files = { + "rc06-approval-keyring.json": _canonical(keyring), + "rc06-approval-root-public-key.raw": rc06_root.public_raw, + "rc06-keyring-bundle.json": _canonical(receipt), + } + try: + signer._write_bundle_atomically(output_directory, files, private_directory=True) + except signer.SignerError as exc: + raise AuthorityError(str(exc)) from exc + return { + "status": "rc06_candidate_keyring_created_not_approved", + "directory": os.fspath(output_directory), + **receipt, + } + + +def _load_existing_rc06_keyring_bundle( + directory: Path, + *, + keys: Mapping[str, AuthorityKey], + registration: Mapping[str, Any], + manifest_sha256: str, + now: datetime, +) -> dict[str, Any]: + expected_names = { + "rc06-approval-keyring.json", + "rc06-approval-root-public-key.raw", + "rc06-keyring-bundle.json", + } + try: + signer._verify_complete_bundle(directory, expected_names, "RC-06 keyring bundle") + except signer.SignerError as exc: + raise AuthorityError(str(exc)) from exc + keyring = _read_json(directory / "rc06-approval-keyring.json", "RC-06 approval keyring") + receipt = _read_json(directory / "rc06-keyring-bundle.json", "RC-06 keyring receipt") + root = keys[ROLE_RC06_ROOT] + root_raw = batch.secure_read_bytes( + directory / "rc06-approval-root-public-key.raw", + "RC-06 approval root public key", + maximum_bytes=32, + ) + if root_raw != root.public_raw or keyring.get("authorized_batch_manifest_raw_sha256") != manifest_sha256: + raise AuthorityError("existing RC-06 keyring bundle identity drift") + try: + batch._validate_approval_keyring(keyring, root_public_key=root.public_key, approved_at=now) + except batch.PublicationBatchError as exc: + raise AuthorityError(f"existing RC-06 keyring cannot be verified: {exc}") from exc + expected = { + "schema_version": RC06_KEYRING_BUNDLE_SCHEMA, + "epoch": keyring["epoch"], + "manifest_raw_sha256": manifest_sha256, + "keyring_payload_sha256": keyring["keyring_payload_sha256"], + "root_public_key_sha256": root.public_sha256, + "reviewer_key_id": registration["key_id"], + "reviewer_public_key_sha256": registration["public_key_raw_sha256"], + "issued_at": keyring["issued_at"], + "expires_at": keyring["expires_at"], + "private_keys_in_bundle": False, + } + if receipt != expected: + raise AuthorityError("existing RC-06 keyring receipt drift") + return { + "status": "rc06_candidate_keyring_already_created_and_verified_not_approved", + "directory": os.fspath(directory), + **receipt, + } + + +def _read_expected_json( + path: Path, + label: str, + expected_sha256: str, +) -> tuple[Any, bytes]: + if not HEX64_RE.fullmatch(expected_sha256): + raise AuthorityError(f"{label} expected SHA-256 is invalid") + try: + raw = batch.secure_read_bytes(path, label, maximum_bytes=16 * 1024 * 1024) + except batch.PublicationBatchError as exc: + raise AuthorityError(str(exc)) from exc + if batch._sha256_bytes(raw) != expected_sha256: + raise AuthorityError(f"{label} raw SHA-256 mismatch") + try: + payload = json.loads( + raw, + object_pairs_hook=batch._reject_duplicate_json_pairs, + parse_constant=batch._reject_json_constant, + ) + except (UnicodeDecodeError, json.JSONDecodeError, batch.PublicationBatchError) as exc: + raise AuthorityError(f"{label} is invalid JSON: {exc}") from exc + return payload, raw + + +def _normalize_apply_payloads(payload: Any, raw: bytes) -> list[dict[str, Any]]: + values = payload if isinstance(payload, list) else [payload] + if not values or len(values) > 64 or any(not isinstance(value, dict) for value in values): + raise AuthorityError("apply payload input must contain one to 64 JSON objects") + if _canonical(payload) != raw: + raise AuthorityError("apply payload input must use exact canonical JSON bytes") + identities: set[str] = set() + normalized: list[dict[str, Any]] = [] + for index, value in enumerate(values): + identity = batch._sha256_json(value) + if identity in identities: + raise AuthorityError("apply payload input repeats a canonical payload") + identities.add(identity) + if value.get("schema_version") != batch.APPLY_PAYLOAD_SCHEMA_VERSION: + raise AuthorityError(f"apply payload {index} uses an unsupported schema") + normalized.append(value) + return normalized + + +def _build_apply_receipt( + payload: Mapping[str, Any], + apply_service: AuthorityKey, +) -> dict[str, Any]: + signature = apply_service.private_key.sign( + batch.APPLY_SIGNATURE_DOMAIN + _canonical(payload) + ) + return { + "schema_version": batch.APPLY_SCHEMA_VERSION, + "receipt_id": ( + "rc06-apply-receipt-" + + batch._sha256_bytes(_canonical(payload) + signature)[:24] + ), + "payload": dict(payload), + "signature": base64.b64encode(signature).decode("ascii"), + } + + +def sign_apply_payloads( + state_directory: Path, + *, + manifest_file: Path, + manifest_sha256: str, + approval_file: Path, + approval_sha256: str, + approval_keyring_file: Path, + approval_keyring_sha256: str, + approval_root_public_key_file: Path, + payloads_file: Path, + payloads_sha256: str, + output_directory: Path, + unprotect: UnprotectFunction = _unprotect_current_user, +) -> dict[str, Any]: + """Sign already-applied, verifier-ready payloads without calling an API. + + The caller supplies online results, but cannot select a trust root or an + apply key. Every payload is checked by the production verifier before any + output directory is published. All receipts are built in memory and then + written as one no-replace bundle, so a failure cannot publish a subset. + """ + + state_directory = signer._assert_local_safe_path( + state_directory, + "release authority state directory", + allow_existing=True, + ) + output_directory = signer._assert_local_safe_path( + output_directory, + "apply receipt output directory", + allow_existing=True, + ) + with signer._state_operation_lock(state_directory): + keys = load_authority(state_directory, unprotect=unprotect) + rc06_root = keys[ROLE_RC06_ROOT] + apply_service = keys[ROLE_APPLY_SERVICE] + if batch.TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256 != rc06_root.public_sha256: + raise AuthorityError("frozen RC-06 root pin does not match the initialized authority") + if batch.TRUSTED_APPLY_SERVICE_PUBLIC_KEY_SHA256 != apply_service.public_sha256: + raise AuthorityError("frozen apply-service pin does not match the initialized authority") + + manifest, manifest_raw = _read_expected_json( + manifest_file, + "RC-06 publication manifest", + manifest_sha256, + ) + approval, approval_raw = _read_expected_json( + approval_file, + "RC-06 approval receipt", + approval_sha256, + ) + keyring, _keyring_raw = _read_expected_json( + approval_keyring_file, + "RC-06 approval keyring", + approval_keyring_sha256, + ) + payload_input, payload_input_raw = _read_expected_json( + payloads_file, + "RC-06 apply payload input", + payloads_sha256, + ) + payloads = _normalize_apply_payloads(payload_input, payload_input_raw) + try: + validated_manifest = batch.validate_manifest(manifest) + root_raw = batch.secure_read_bytes( + approval_root_public_key_file, + "RC-06 approval root public key", + maximum_bytes=32, + ) + if root_raw != rc06_root.public_raw: + raise AuthorityError("approval root public key does not match the initialized authority") + root_public_key = Ed25519PublicKey.from_public_bytes(root_raw) + batch.validate_approval_receipt( + validated_manifest, + manifest_sha256, + approval, + keyring, + root_public_key, + ) + except (ValueError, batch.PublicationBatchError) as exc: + if isinstance(exc, AuthorityError): + raise + raise AuthorityError(f"RC-06 approval chain is invalid: {exc}") from exc + + receipt_files: dict[str, bytes] = {} + receipt_index: list[dict[str, Any]] = [] + for index, payload in enumerate(payloads, start=1): + receipt = _build_apply_receipt(payload, apply_service) + try: + batch.validate_apply_receipt( + validated_manifest, + manifest_sha256, + approval, + keyring, + root_public_key, + apply_service.public_key, + approval_sha256, + receipt, + ) + except batch.PublicationBatchError as exc: + raise AuthorityError(f"apply payload {index} failed formal verification: {exc}") from exc + filename = f"apply-receipt-{index:03d}.json" + receipt_raw = _canonical(receipt) + receipt_files[filename] = receipt_raw + receipt_index.append( + { + "filename": filename, + "receipt_id": receipt["receipt_id"], + "payload_sha256": batch._sha256_json(payload), + "raw_sha256": batch._sha256_bytes(receipt_raw), + } + ) + + bundle = { + "schema_version": APPLY_RECEIPT_BUNDLE_SCHEMA, + "candidate_commit": validated_manifest["candidate"]["commit"], + "candidate_tag": validated_manifest["candidate"]["tag"], + "manifest_raw_sha256": manifest_sha256, + "manifest_payload_sha256": validated_manifest["manifest_payload_sha256"], + "approval_raw_sha256": approval_sha256, + "approval_keyring_payload_sha256": keyring["keyring_payload_sha256"], + "payload_input_raw_sha256": payloads_sha256, + "apply_service_public_key_sha256": apply_service.public_sha256, + "receipt_count": len(receipt_index), + "receipts": receipt_index, + "private_keys_in_bundle": False, + "business_api_calls": 0, + } + expected_files = { + **receipt_files, + "apply-receipt-bundle.json": _canonical(bundle), + } + if output_directory.exists() or output_directory.is_symlink(): + try: + signer._assert_bundle_matches_files( + output_directory, + expected_files, + "existing apply receipt bundle", + ) + except signer.SignerError as exc: + raise AuthorityError(str(exc)) from exc + return { + "status": "apply_receipts_already_signed_and_verified_not_published", + "directory": os.fspath(output_directory), + **bundle, + } + try: + signer._write_bundle_atomically( + output_directory, + expected_files, + private_directory=True, + ) + signer._assert_bundle_matches_files( + output_directory, + expected_files, + "completed apply receipt bundle", + ) + except signer.SignerError as exc: + raise AuthorityError(str(exc)) from exc + return { + "status": "apply_receipts_signed_and_verified_not_published", + "directory": os.fspath(output_directory), + **bundle, + } + + +def _replace_assignment( + text: str, + *, + name: str, + rendered_value: str, + allowed_existing: Callable[[str], bool], +) -> tuple[str, str]: + pattern = re.compile( + rf"(?m)^(?P{re.escape(name)}(?:\s*:\s*[^=]+)?\s*=\s*)" + rf"(?P[^\r\n]*)(?=\r?$)" + ) + matches = list(pattern.finditer(text)) + if len(matches) != 1: + raise AuthorityError(f"source assignment {name} must appear exactly once") + current = matches[0].group("value").strip() + if not allowed_existing(current): + raise AuthorityError(f"source assignment {name} contains an unexpected existing value") + if current == rendered_value: + return text, current + match = matches[0] + replacement = match.group("prefix") + rendered_value + return text[: match.start()] + replacement + text[match.end() :], current + + +def _read_source_file(path: Path, repo: Path, label: str) -> bytes: + resolved_repo = repo.resolve(strict=True) + resolved = path.resolve(strict=True) + if not _is_relative_to(resolved, resolved_repo): + raise AuthorityError(f"{label} escapes the repository") + try: + batch._assert_no_symlink_components(path, label) + raw = batch.secure_read_bytes(path, label, maximum_bytes=4 * 1024 * 1024) + except batch.PublicationBatchError as exc: + raise AuthorityError(str(exc)) from exc + if b"\x00" in raw: + raise AuthorityError(f"{label} is not a text source file") + return raw + + +def _prepare_source_change( + path: Path, + repo: Path, + replacements: list[tuple[str, str, Callable[[str], bool]]], +) -> PinFileChange: + original = _read_source_file(path, repo, path.name) + try: + text = original.decode("utf-8") + except UnicodeDecodeError as exc: + raise AuthorityError(f"{path.name} is not UTF-8") from exc + proposed = text + for name, rendered, predicate in replacements: + proposed, _ = _replace_assignment( + proposed, + name=name, + rendered_value=rendered, + allowed_existing=predicate, + ) + return PinFileChange(path=path, original=original, proposed=proposed.encode("utf-8")) + + +def _quoted_pin(value: str) -> str: + if not HEX64_RE.fullmatch(value): + raise AuthorityError("public-key or payload SHA-256 is invalid") + return json.dumps(value) + + +def _existing_pin(value: str) -> bool: + if value == '""': + return True + try: + parsed = json.loads(value) + except json.JSONDecodeError: + return False + return isinstance(parsed, str) and bool(HEX64_RE.fullmatch(parsed)) + + +def _existing_epoch(value: str) -> bool: + return value in {"0", "None"} or (value.isdigit() and int(value) >= 1) + + +def plan_pin_updates( + repo: Path, + state_directory: Path, + recovery_keyring_directory: Path, + *, + epoch: int, + unprotect: UnprotectFunction = _unprotect_current_user, + expected_current_epoch: int | None = None, +) -> PinPlan: + epoch = _ensure_epoch(epoch) + if expected_current_epoch is not None: + _ensure_epoch(expected_current_epoch) + if epoch <= expected_current_epoch: + raise AuthorityError("rotated keyring epoch must increase") + repo = repo.resolve(strict=True) + keys = load_authority(state_directory, unprotect=unprotect) + receipt = _read_json( + recovery_keyring_directory / "recovery-keyring-bundle.json", + "recovery keyring receipt", + ) + if receipt.get("epoch") != epoch: + raise AuthorityError("recovery keyring epoch does not match the requested frozen epoch") + if receipt.get("recovery_root_public_key_sha256") != keys[ROLE_RECOVERY_ROOT].public_sha256: + raise AuthorityError("recovery keyring root does not match the initialized authority") + first_sha = receipt.get("first_release_keyring_payload_sha256") + rollback_sha = receipt.get("rollback_keyring_payload_sha256") + if not isinstance(first_sha, str) or not isinstance(rollback_sha, str): + raise AuthorityError("recovery keyring receipt is incomplete") + first_keyring = _read_json( + recovery_keyring_directory / "first-release-recovery-keyring.json", + "first-release recovery keyring", + ) + rollback_keyring = _read_json( + recovery_keyring_directory / "rollback-rehearsal-keyring.json", + "rollback rehearsal keyring", + ) + _validate_recovery_keyrings( + first_keyring, + rollback_keyring, + keys[ROLE_RECOVERY_ROOT], + now=_now_utc(), + ) + try: + first_reviewer_raw = base64.b64decode( + first_keyring["keys"][0]["public_key_base64"], validate=True + ) + rollback_reviewer_raw = base64.b64decode( + rollback_keyring["keys"][0]["public_key_base64"], validate=True + ) + except (KeyError, IndexError, TypeError, ValueError) as exc: + raise AuthorityError("recovery keyring reviewer binding is invalid") from exc + if ( + first_keyring["keyring_payload_sha256"] != first_sha + or rollback_keyring["keyring_payload_sha256"] != rollback_sha + or first_keyring["epoch"] != epoch + or rollback_keyring["epoch"] != epoch + or first_keyring["keys"][0]["key_id"] != rollback_keyring["keys"][0]["key_id"] + or first_reviewer_raw != rollback_reviewer_raw + or batch._sha256_bytes(first_reviewer_raw) != receipt.get("reviewer_public_key_sha256") + ): + raise AuthorityError("recovery keyring receipt or reviewer binding drift") + + def allowed_pin(desired: str) -> Callable[[str], bool]: + if expected_current_epoch is None: + return lambda current: current in {'""', desired} + return _existing_pin + + def allowed_epoch(desired: str) -> Callable[[str], bool]: + if expected_current_epoch is None: + return lambda current: current in {"0", "None", desired} + return lambda current: current in {str(expected_current_epoch), desired} + + rc06_root_pin = _quoted_pin(keys[ROLE_RC06_ROOT].public_sha256) + apply_pin = _quoted_pin(keys[ROLE_APPLY_SERVICE].public_sha256) + recovery_root_pin = _quoted_pin(keys[ROLE_RECOVERY_ROOT].public_sha256) + first_payload_pin = _quoted_pin(first_sha) + rollback_payload_pin = _quoted_pin(rollback_sha) + rendered_epoch = str(epoch) + + rc06_path = repo / "yuqei-infra" / "scripts" / "build-rc06-publication-batch.py" + first_path = repo / "yuqei-infra" / "deploy" / "plan-first-release-snapshot-restore.py" + rollback_path = repo / "yuqei-infra" / "deploy" / "plan-rollback-rehearsal.py" + changes = ( + _prepare_source_change( + rc06_path, + repo, + [ + ( + "TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256", + rc06_root_pin, + allowed_pin(rc06_root_pin), + ), + ( + "FROZEN_APPROVAL_KEYRING_EPOCH", + rendered_epoch, + allowed_epoch(rendered_epoch), + ), + ( + "TRUSTED_APPLY_SERVICE_PUBLIC_KEY_SHA256", + apply_pin, + allowed_pin(apply_pin), + ), + ], + ), + _prepare_source_change( + first_path, + repo, + [ + ( + "TRUSTED_ROOT_PUBLIC_KEY_SHA256", + recovery_root_pin, + allowed_pin(recovery_root_pin), + ), + ( + "TRUSTED_CURRENT_KEYRING_PAYLOAD_SHA256", + first_payload_pin, + allowed_pin(first_payload_pin), + ), + ("MIN_KEYRING_EPOCH", rendered_epoch, allowed_epoch(rendered_epoch)), + ], + ), + _prepare_source_change( + rollback_path, + repo, + [ + ( + "TRUSTED_ROOT_PUBLIC_KEY_SHA256", + recovery_root_pin, + allowed_pin(recovery_root_pin), + ), + ( + "TRUSTED_CURRENT_KEYRING_PAYLOAD_SHA256", + rollback_payload_pin, + allowed_pin(rollback_payload_pin), + ), + ("MIN_KEYRING_EPOCH", rendered_epoch, allowed_epoch(rendered_epoch)), + ], + ), + ) + if expected_current_epoch is None: + # First provisioning may start from placeholders or may be an exact + # idempotent rerun. A different already-provisioned epoch is refused. + for change in changes: + source = change.original.decode("utf-8") + for found in re.findall( + r"(?m)^(?:FROZEN_APPROVAL_KEYRING_EPOCH|MIN_KEYRING_EPOCH(?:\s*:\s*[^=]+)?)\s*=\s*(\d+|None)$", + source, + ): + if found not in {"0", "None", str(epoch)}: + raise AuthorityError( + "source already contains a different epoch; explicit rotation expectation is required" + ) + else: + for change in changes: + source = change.original.decode("utf-8") + epochs = re.findall( + r"(?m)^(?:FROZEN_APPROVAL_KEYRING_EPOCH|MIN_KEYRING_EPOCH(?:\s*:\s*[^=]+)?)\s*=\s*(\d+|None)$", + source, + ) + if epochs and any(value != str(expected_current_epoch) for value in epochs): + raise AuthorityError("source epoch does not match the explicit rotation expectation") + return PinPlan( + epoch=epoch, + changes=changes, + values={ + "rc06_root_public_key_sha256": keys[ROLE_RC06_ROOT].public_sha256, + "apply_service_public_key_sha256": keys[ROLE_APPLY_SERVICE].public_sha256, + "recovery_root_public_key_sha256": keys[ROLE_RECOVERY_ROOT].public_sha256, + "first_release_keyring_payload_sha256": first_sha, + "rollback_keyring_payload_sha256": rollback_sha, + "epoch": epoch, + }, + ) + + +def _write_atomic_replacement(path: Path, raw: bytes) -> None: + parent = path.parent + temporary: Path | None = None + descriptor: int | None = None + try: + descriptor, temporary_name = tempfile.mkstemp(prefix=f".{path.name}.", dir=parent) + temporary = Path(temporary_name) + offset = 0 + while offset < len(raw): + written = os.write(descriptor, raw[offset:]) + if written <= 0: + raise AuthorityError(f"cannot write complete replacement for {path.name}") + offset += written + os.fsync(descriptor) + os.close(descriptor) + descriptor = None + if temporary.is_symlink() or not stat.S_ISREG(temporary.stat().st_mode): + raise AuthorityError(f"replacement for {path.name} is not a regular file") + os.replace(temporary, path) + temporary = None + finally: + if descriptor is not None: + os.close(descriptor) + if temporary is not None: + try: + if temporary.is_file() and not temporary.is_symlink(): + temporary.unlink() + except OSError: + pass + + +def apply_pin_plan(plan: PinPlan) -> dict[str, Any]: + changed = [change for change in plan.changes if change.original != change.proposed] + for change in plan.changes: + actual = batch.secure_read_bytes(change.path, change.path.name, maximum_bytes=4 * 1024 * 1024) + if actual != change.original: + raise AuthorityError(f"source changed after preview: {change.path.name}") + completed: list[PinFileChange] = [] + try: + for change in changed: + _write_atomic_replacement(change.path, change.proposed) + completed.append(change) + for change in plan.changes: + actual = batch.secure_read_bytes(change.path, change.path.name, maximum_bytes=4 * 1024 * 1024) + if actual != change.proposed: + raise AuthorityError(f"source pin verification failed: {change.path.name}") + except Exception as exc: + restoration_errors: list[str] = [] + for change in reversed(completed): + try: + _write_atomic_replacement(change.path, change.original) + except Exception as restore_exc: # pragma: no cover - catastrophic filesystem failure + restoration_errors.append(f"{change.path}: {restore_exc}") + if restoration_errors: + raise AuthorityError( + "source pin transaction failed and rollback was incomplete: " + + "; ".join(restoration_errors) + ) from exc + if isinstance(exc, AuthorityError): + raise + raise AuthorityError(f"source pin transaction failed: {exc}") from exc + return { + "status": "pins_applied_not_frozen_not_signed", + "epoch": plan.epoch, + "changed_file_count": len(changed), + "files": [ + { + "path": os.fspath(change.path), + "original_sha256": change.original_sha256, + "proposed_sha256": change.proposed_sha256, + "changed": change.original != change.proposed, + } + for change in plan.changes + ], + } + + +def pin_plan_summary(plan: PinPlan) -> dict[str, Any]: + return { + "schema_version": PIN_PLAN_SCHEMA, + "status": "preview_only_not_applied", + "epoch": plan.epoch, + "values": dict(plan.values), + "files": [ + { + "path": os.fspath(change.path), + "original_sha256": change.original_sha256, + "proposed_sha256": change.proposed_sha256, + "changed": change.original != change.proposed, + } + for change in plan.changes + ], + } + + +def _assert_runtime() -> None: + try: + signer._assert_trusted_runtime() + except signer.SignerError as exc: + raise AuthorityError(str(exc)) from exc + + +def _parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description=__doc__) + commands = parser.add_subparsers(dest="command", required=True) + commands.add_parser("runtime-check", help="verify the isolated authority runtime") + initialize = commands.add_parser("initialize", help="initialize three protected technical keys") + initialize.add_argument("--state-directory", type=Path) + status = commands.add_parser("status", help="verify the existing protected authority") + status.add_argument("--state-directory", type=Path) + recovery = commands.add_parser( + "build-recovery-keyrings", + help="root-sign the first-release and rollback recovery keyrings", + ) + recovery.add_argument("--state-directory", type=Path) + recovery.add_argument("--reviewer-registration", type=Path, required=True) + recovery.add_argument("--epoch", type=int, required=True) + pin = commands.add_parser( + "pin-source", + help="preview or apply strict public-pin and epoch source updates", + ) + pin.add_argument("--state-directory", type=Path) + pin.add_argument("--repo", type=Path, required=True) + pin.add_argument("--recovery-keyrings", type=Path, required=True) + pin.add_argument("--epoch", type=int, required=True) + pin.add_argument("--expected-current-epoch", type=int) + pin.add_argument("--apply", action="store_true") + rc06 = commands.add_parser( + "sign-rc06-keyring", + help="root-sign one candidate-specific RC-06 keyring after manifest formation", + ) + rc06.add_argument("--state-directory", type=Path) + rc06.add_argument("--reviewer-registration", type=Path, required=True) + rc06.add_argument("--manifest", type=Path, required=True) + rc06.add_argument("--manifest-sha256", required=True) + apply_sign = commands.add_parser( + "sign-apply-payloads", + help="sign verifier-ready apply payloads and immediately self-verify every receipt", + ) + apply_sign.add_argument("--state-directory", type=Path) + apply_sign.add_argument("--manifest", type=Path, required=True) + apply_sign.add_argument("--manifest-sha256", required=True) + apply_sign.add_argument("--approval", type=Path, required=True) + apply_sign.add_argument("--approval-sha256", required=True) + apply_sign.add_argument("--approval-keyring", type=Path, required=True) + apply_sign.add_argument("--approval-keyring-sha256", required=True) + apply_sign.add_argument("--approval-root-public-key", type=Path, required=True) + apply_sign.add_argument("--payloads", type=Path, required=True) + apply_sign.add_argument("--payloads-sha256", required=True) + apply_sign.add_argument("--output", type=Path, required=True) + provision = commands.add_parser( + "provision-apply-private-key", + help="fail closed until an approved secret-store delivery adapter exists", + ) + provision.add_argument("--state-directory", type=Path) + return parser + + +def main(argv: list[str] | None = None) -> int: + if hasattr(sys.stdout, "reconfigure"): + sys.stdout.reconfigure(encoding="utf-8", errors="replace") + args = _parser().parse_args(argv) + try: + _assert_runtime() + if args.command == "runtime-check": + if os.name != "nt": + raise AuthorityError("Windows CurrentUser DPAPI is required") + probe = b"yuqei-release-authority-runtime-probe" + protected = _protect_current_user(probe, "runtime-check") + if _unprotect_current_user(protected, "runtime-check") != probe: + raise AuthorityError("Windows CurrentUser DPAPI round-trip failed") + result = { + "status": "isolated_runtime_and_dpapi_verified", + "python_version": ".".join(str(part) for part in sys.version_info[:3]), + "cryptography_version": cryptography.__version__, + } + else: + state_directory = args.state_directory or _default_state_directory() + if args.command == "initialize": + result = initialize_authority(state_directory) + elif args.command == "status": + keys = load_authority(state_directory) + result = { + "status": "authority_verified", + "roles": {role: keys[role].public_sha256 for role in AUTHORITY_ROLES}, + "private_key_export_supported": False, + } + elif args.command == "build-recovery-keyrings": + result = build_recovery_keyrings( + state_directory, + args.reviewer_registration, + epoch=args.epoch, + ) + elif args.command == "pin-source": + plan = plan_pin_updates( + args.repo, + state_directory, + args.recovery_keyrings, + epoch=args.epoch, + expected_current_epoch=args.expected_current_epoch, + ) + result = apply_pin_plan(plan) if args.apply else pin_plan_summary(plan) + elif args.command == "sign-rc06-keyring": + result = build_rc06_candidate_keyring( + state_directory, + args.reviewer_registration, + args.manifest, + args.manifest_sha256, + ) + elif args.command == "sign-apply-payloads": + result = sign_apply_payloads( + state_directory, + manifest_file=args.manifest, + manifest_sha256=args.manifest_sha256, + approval_file=args.approval, + approval_sha256=args.approval_sha256, + approval_keyring_file=args.approval_keyring, + approval_keyring_sha256=args.approval_keyring_sha256, + approval_root_public_key_file=args.approval_root_public_key, + payloads_file=args.payloads, + payloads_sha256=args.payloads_sha256, + output_directory=args.output, + ) + else: + raise AuthorityError( + "apply-service private-key export/provisioning is not implemented; " + "no private material was exported" + ) + except ( + AuthorityError, + signer.SignerError, + batch.PublicationBatchError, + OSError, + KeyboardInterrupt, + ) as exc: + message = "operation cancelled" if isinstance(exc, KeyboardInterrupt) else str(exc) + print( + json.dumps( + { + "status": "failed_closed", + "error": message, + "private_material_exported": False, + }, + ensure_ascii=False, + separators=(",", ":"), + ), + file=sys.stderr, + ) + return 2 + print(json.dumps(result, ensure_ascii=False, sort_keys=True, separators=(",", ":"))) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/yuqei-infra/tests/test_rc06_release_trust_bootstrap.py b/yuqei-infra/tests/test_rc06_release_trust_bootstrap.py new file mode 100644 index 00000000..17d543c6 --- /dev/null +++ b/yuqei-infra/tests/test_rc06_release_trust_bootstrap.py @@ -0,0 +1,713 @@ +from __future__ import annotations + +import base64 +import importlib.util +import json +import os +import re +import sys +from datetime import UTC, datetime, timedelta +from pathlib import Path + +import pytest +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + + +REPO_ROOT = Path(__file__).resolve().parents[2] +SCRIPT = REPO_ROOT / "yuqei-infra" / "scripts" / "rc06-release-trust-bootstrap.py" +EXISTING_BATCH_TEST = REPO_ROOT / "yuqei-infra" / "tests" / "test_rc06_publication_batch.py" +AUTHORITY_LAUNCHER = REPO_ROOT / "yuqei-infra" / "scripts" / "Launch-ReleaseAuthority.ps1" + + +def _load_script(): + spec = importlib.util.spec_from_file_location("rc06_release_trust_bootstrap", SCRIPT) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +authority = _load_script() + + +def _load_existing_batch_test_module(): + spec = importlib.util.spec_from_file_location( + "rc06_publication_batch_fixture_helpers", + EXISTING_BATCH_TEST, + ) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +def _protect(raw: bytes, role: str) -> bytes: + return b"fixture-protected\0" + role.encode("ascii") + b"\0" + raw[::-1] + + +def _unprotect(raw: bytes, role: str) -> bytes: + prefix = b"fixture-protected\0" + role.encode("ascii") + b"\0" + if not raw.startswith(prefix): + raise ValueError("wrong fixture protection domain") + return raw[len(prefix) :][::-1] + + +def _reviewer_registration(path: Path, *, private_key: Ed25519PrivateKey | None = None) -> dict: + private_key = private_key or Ed25519PrivateKey.generate() + public_key = private_key.public_key() + public_raw = public_key.public_bytes( + serialization.Encoding.Raw, + serialization.PublicFormat.Raw, + ) + public_pem = public_key.public_bytes( + serialization.Encoding.PEM, + serialization.PublicFormat.SubjectPublicKeyInfo, + ) + fingerprint = authority.batch._sha256_bytes(public_raw) + registration = { + "schema_version": authority.REVIEWER_REGISTRATION_SCHEMA, + "key_id": f"chen-yuqi-rc06-{fingerprint[:16]}", + "algorithm": "Ed25519", + "reviewer": dict(authority.batch.SOLE_RELEASE_REVIEWER), + "created_at": "2026-09-27T00:00:00Z", + "public_key_pem": public_pem.decode("ascii"), + "public_key_raw_sha256": fingerprint, + "public_key_pem_sha256": authority.batch._sha256_bytes(public_pem), + } + path.write_bytes(authority._canonical(registration)) + return registration + + +def _initialize(tmp_path: Path): + state = tmp_path / "authority-state" + result = authority.initialize_authority( + state, + protect=_protect, + unprotect=_unprotect, + created_at=datetime(2026, 9, 27, 1, 0, tzinfo=UTC), + ) + return state, result, authority.load_authority(state, unprotect=_unprotect) + + +def _copy_pin_sources(tmp_path: Path) -> Path: + repo = tmp_path / "candidate" + paths = ( + "yuqei-infra/scripts/build-rc06-publication-batch.py", + "yuqei-infra/deploy/plan-first-release-snapshot-restore.py", + "yuqei-infra/deploy/plan-rollback-rehearsal.py", + ) + for relative in paths: + source = REPO_ROOT / relative + target = repo / relative + target.parent.mkdir(parents=True, exist_ok=True) + target.write_bytes(source.read_bytes()) + return repo + + +def _formal_apply_materials(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> dict: + state, _result, keys = _initialize(tmp_path) + reviewer_private = Ed25519PrivateKey.generate() + reviewer_file = tmp_path / "reviewer-registration.json" + registration = _reviewer_registration(reviewer_file, private_key=reviewer_private) + helpers = _load_existing_batch_test_module() + inventory = helpers.batch.load_runtime_inventory(REPO_ROOT) + manifest = helpers._manifest(inventory, resolved=True) + manifest_raw = authority._canonical(manifest) + manifest_file = tmp_path / "rc06-publication-manifest.json" + manifest_file.write_bytes(manifest_raw) + manifest_sha = authority.batch._sha256_bytes(manifest_raw) + monkeypatch.setattr( + authority.batch, + "TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256", + keys[authority.ROLE_RC06_ROOT].public_sha256, + ) + monkeypatch.setattr( + authority.batch, + "TRUSTED_APPLY_SERVICE_PUBLIC_KEY_SHA256", + keys[authority.ROLE_APPLY_SERVICE].public_sha256, + ) + monkeypatch.setattr(authority.batch, "FROZEN_APPROVAL_KEYRING_EPOCH", 1) + keyring_result = authority.build_rc06_candidate_keyring( + state, + reviewer_file, + manifest_file, + manifest_sha, + unprotect=_unprotect, + now=datetime(2026, 9, 27, 3, 0, tzinfo=UTC), + ) + keyring_directory = Path(keyring_result["directory"]) + keyring_file = keyring_directory / "rc06-approval-keyring.json" + keyring_raw = keyring_file.read_bytes() + keyring = json.loads(keyring_raw) + approved_at = datetime(2026, 9, 27, 3, 10, tzinfo=UTC) + expires_at = datetime(2026, 9, 27, 5, 0, tzinfo=UTC) + approval_payload = authority.batch.build_approval_payload( + manifest, + manifest_raw_sha256=manifest_sha, + key_id=registration["key_id"], + approval_keyring_epoch=1, + approval_keyring_payload_sha256=keyring["keyring_payload_sha256"], + approved_at=approved_at, + expires_at=expires_at, + ) + approval_signature = reviewer_private.sign( + authority.batch.APPROVAL_SIGNATURE_DOMAIN + authority._canonical(approval_payload) + ) + approval = { + "schema_version": authority.batch.APPROVAL_SCHEMA_VERSION, + "approval_id": ( + "rc06-batch-approval-" + + authority.batch._sha256_bytes( + authority._canonical(approval_payload) + approval_signature + )[:24] + ), + "payload": approval_payload, + "signature": base64.b64encode(approval_signature).decode("ascii"), + } + approval_file = tmp_path / "rc06-batch-approval.json" + approval_raw = authority._canonical(approval) + approval_file.write_bytes(approval_raw) + item_results = [ + { + "kind": "asset", + "item_id": item["template_id"], + "expected_content_sha256": item["content_sha256"], + "online_content_sha256": item["content_sha256"], + "status": "applied_and_verified", + } + for item in manifest["asset_batch"]["items"] + ] + [ + { + "kind": "prompt_module", + "item_id": item["key"], + "expected_content_sha256": item["content_sha256"], + "online_content_sha256": item["content_sha256"], + "status": "applied_and_verified", + } + for item in manifest["prompt_module_batch"]["items"] + ] + item_results.sort(key=lambda item: (item["kind"], item["item_id"])) + apply_payload = { + "schema_version": authority.batch.APPLY_PAYLOAD_SCHEMA_VERSION, + "batch_manifest_payload_sha256": manifest["manifest_payload_sha256"], + "approval_receipt_sha256": authority.batch._sha256_bytes(approval_raw), + "environment_id": "isolated-uat-fixture", + "applied_at": "2026-09-27T04:00:00Z", + "generated_by": {"type": "system", "component": "rc06-publication-service"}, + "item_results": item_results, + "write_count": 74, + "status": "applied_and_verified", + } + payloads_file = tmp_path / "apply-payloads.json" + payloads_raw = authority._canonical(apply_payload) + payloads_file.write_bytes(payloads_raw) + return { + "state": state, + "keys": keys, + "reviewer_private": reviewer_private, + "registration": registration, + "manifest": manifest, + "manifest_file": manifest_file, + "manifest_sha": manifest_sha, + "keyring": keyring, + "keyring_file": keyring_file, + "keyring_sha": authority.batch._sha256_bytes(keyring_raw), + "root_file": keyring_directory / "rc06-approval-root-public-key.raw", + "approval": approval, + "approval_file": approval_file, + "approval_sha": authority.batch._sha256_bytes(approval_raw), + "apply_payload": apply_payload, + "payloads_file": payloads_file, + "payloads_sha": authority.batch._sha256_bytes(payloads_raw), + } + + +def test_authority_initialization_creates_three_distinct_protected_keys_and_is_idempotent( + tmp_path: Path, +) -> None: + state, result, keys = _initialize(tmp_path) + + assert result["status"] == "initialized_not_pinned_not_signed" + assert result["private_key_export_supported"] is False + assert set(keys) == set(authority.AUTHORITY_ROLES) + assert len({item.public_raw for item in keys.values()}) == 3 + key_dir = state / authority.KEYS_DIRECTORY_NAME + assert set(child.name for child in key_dir.iterdir()) == ( + authority._expected_authority_names() | {authority.signer.BUNDLE_COMPLETE_FILENAME} + ) + for role, item in keys.items(): + protected_raw = (key_dir / authority._private_filename(role)).read_bytes() + private_der = authority._private_der(item.private_key) + assert private_der not in protected_raw + assert b"BEGIN PRIVATE KEY" not in protected_raw + envelope = json.loads(protected_raw) + assert envelope["protection"] == authority.PROTECTION_NAME + assert base64.b64decode(envelope["ciphertext_base64"], validate=True) != private_der + + repeated = authority.initialize_authority( + state, + protect=lambda _raw, _role: pytest.fail("idempotent initialization generated a new key"), + unprotect=_unprotect, + ) + assert repeated["status"] == "already_initialized_and_verified" + assert repeated["roles"] == result["roles"] + + +def test_authority_rejects_partial_or_tampered_bundle(tmp_path: Path) -> None: + state, _result, _keys = _initialize(tmp_path) + key_dir = state / authority.KEYS_DIRECTORY_NAME + (key_dir / authority._public_filename(authority.ROLE_APPLY_SERVICE)).unlink() + + with pytest.raises(authority.AuthorityError, match="incomplete"): + authority.initialize_authority(state, protect=_protect, unprotect=_unprotect) + + +def test_recovery_keyrings_share_only_the_recovery_root_and_verify_both_schemas( + tmp_path: Path, +) -> None: + state, _result, keys = _initialize(tmp_path) + reviewer_file = tmp_path / "reviewer-registration.json" + reviewer = _reviewer_registration(reviewer_file) + now = datetime(2026, 9, 27, 2, 0, tzinfo=UTC) + + result = authority.build_recovery_keyrings( + state, + reviewer_file, + epoch=1, + unprotect=_unprotect, + now=now, + validity=timedelta(days=30), + ) + + assert result["status"] == "recovery_keyrings_created_not_pinned" + assert result["recovery_root_public_key_sha256"] == keys[ + authority.ROLE_RECOVERY_ROOT + ].public_sha256 + assert result["first_release_keyring_payload_sha256"] != result[ + "rollback_keyring_payload_sha256" + ] + bundle = Path(result["directory"]) + first_release = json.loads((bundle / "first-release-recovery-keyring.json").read_bytes()) + rollback = json.loads((bundle / "rollback-rehearsal-keyring.json").read_bytes()) + assert first_release["root_key_id"] == authority.FIRST_RELEASE_ROOT_KEY_ID + assert rollback["root_key_id"] == authority.ROLLBACK_ROOT_KEY_ID + assert "algorithm" not in first_release["keys"][0] + assert rollback["keys"][0]["algorithm"] == "ed25519" + assert first_release["keys"][0]["public_key_base64"] == rollback["keys"][0][ + "public_key_base64" + ] + assert reviewer["public_key_raw_sha256"] not in { + item.public_sha256 for item in keys.values() + } + + repeated = authority.build_recovery_keyrings( + state, + reviewer_file, + epoch=1, + unprotect=_unprotect, + now=now + timedelta(hours=1), + ) + assert repeated["status"] == "recovery_keyrings_already_created_and_verified_not_pinned" + assert repeated["first_release_keyring_payload_sha256"] == result[ + "first_release_keyring_payload_sha256" + ] + + +def test_recovery_keyrings_reject_reviewer_reuse_of_any_technical_key(tmp_path: Path) -> None: + state, _result, keys = _initialize(tmp_path) + technical_private = keys[authority.ROLE_APPLY_SERVICE].private_key + reviewer_file = tmp_path / "reviewer-registration.json" + _reviewer_registration(reviewer_file, private_key=technical_private) + + with pytest.raises(authority.AuthorityError, match="must differ"): + authority.build_recovery_keyrings( + state, + reviewer_file, + epoch=1, + unprotect=_unprotect, + now=datetime(2026, 9, 27, 2, 0, tzinfo=UTC), + ) + + +def test_pin_preview_and_apply_update_only_public_hashes_and_epoch_in_temp_sources( + tmp_path: Path, +) -> None: + state, _result, keys = _initialize(tmp_path) + reviewer_file = tmp_path / "reviewer-registration.json" + _reviewer_registration(reviewer_file) + recovery = authority.build_recovery_keyrings( + state, + reviewer_file, + epoch=1, + unprotect=_unprotect, + now=datetime(2026, 9, 27, 2, 0, tzinfo=UTC), + ) + repo = _copy_pin_sources(tmp_path) + before = { + path.relative_to(repo): path.read_bytes() + for path in repo.rglob("*.py") + } + + plan = authority.plan_pin_updates( + repo, + state, + Path(recovery["directory"]), + epoch=1, + unprotect=_unprotect, + ) + preview = authority.pin_plan_summary(plan) + assert preview["status"] == "preview_only_not_applied" + assert all(path.read_bytes() == before[path.relative_to(repo)] for path in repo.rglob("*.py")) + assert all(len(str(value)) in {1, 64} for value in preview["values"].values()) + + applied = authority.apply_pin_plan(plan) + assert applied["status"] == "pins_applied_not_frozen_not_signed" + assert applied["changed_file_count"] == 3 + rc06 = (repo / "yuqei-infra/scripts/build-rc06-publication-batch.py").read_text( + encoding="utf-8" + ) + first = (repo / "yuqei-infra/deploy/plan-first-release-snapshot-restore.py").read_text( + encoding="utf-8" + ) + rollback = (repo / "yuqei-infra/deploy/plan-rollback-rehearsal.py").read_text( + encoding="utf-8" + ) + assert f'TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256 = "{keys[authority.ROLE_RC06_ROOT].public_sha256}"' in rc06 + assert f'TRUSTED_APPLY_SERVICE_PUBLIC_KEY_SHA256 = "{keys[authority.ROLE_APPLY_SERVICE].public_sha256}"' in rc06 + assert "FROZEN_APPROVAL_KEYRING_EPOCH = 1" in rc06 + for text in (first, rollback): + assert f'TRUSTED_ROOT_PUBLIC_KEY_SHA256 = "{keys[authority.ROLE_RECOVERY_ROOT].public_sha256}"' in text + assert re.search(r"(?m)^MIN_KEYRING_EPOCH: int \| None = 1$", text) + assert "PRIVATE KEY" not in text + + repeated = authority.plan_pin_updates( + repo, + state, + Path(recovery["directory"]), + epoch=1, + unprotect=_unprotect, + ) + assert all(change.original == change.proposed for change in repeated.changes) + assert authority.apply_pin_plan(repeated)["changed_file_count"] == 0 + + +def test_pin_planner_refuses_unknown_existing_pin_without_overwriting(tmp_path: Path) -> None: + state, _result, _keys = _initialize(tmp_path) + reviewer_file = tmp_path / "reviewer-registration.json" + _reviewer_registration(reviewer_file) + recovery = authority.build_recovery_keyrings( + state, + reviewer_file, + epoch=1, + unprotect=_unprotect, + now=datetime(2026, 9, 27, 2, 0, tzinfo=UTC), + ) + repo = _copy_pin_sources(tmp_path) + rc06 = repo / "yuqei-infra/scripts/build-rc06-publication-batch.py" + original = rc06.read_text(encoding="utf-8") + rc06.write_text( + original.replace( + 'TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256 = ""', + 'TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256 = "' + "f" * 64 + '"', + ), + encoding="utf-8", + ) + + # A different already-provisioned pin cannot be replaced by a first-time + # provisioning run. + with pytest.raises(authority.AuthorityError, match="different epoch|unexpected existing value"): + authority.plan_pin_updates( + repo, + state, + Path(recovery["directory"]), + epoch=1, + unprotect=_unprotect, + ) + + +def test_pin_apply_rolls_back_completed_files_when_later_replace_fails( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + state, _result, _keys = _initialize(tmp_path) + reviewer_file = tmp_path / "reviewer-registration.json" + _reviewer_registration(reviewer_file) + recovery = authority.build_recovery_keyrings( + state, + reviewer_file, + epoch=1, + unprotect=_unprotect, + now=datetime(2026, 9, 27, 2, 0, tzinfo=UTC), + ) + repo = _copy_pin_sources(tmp_path) + plan = authority.plan_pin_updates( + repo, + state, + Path(recovery["directory"]), + epoch=1, + unprotect=_unprotect, + ) + real_replace = authority._write_atomic_replacement + calls = 0 + + def fail_second(path: Path, raw: bytes) -> None: + nonlocal calls + calls += 1 + if calls == 2: + raise OSError("fixture replacement failure") + real_replace(path, raw) + + monkeypatch.setattr(authority, "_write_atomic_replacement", fail_second) + with pytest.raises(authority.AuthorityError, match="transaction failed"): + authority.apply_pin_plan(plan) + assert all(change.path.read_bytes() == change.original for change in plan.changes) + + +def test_rc06_candidate_keyring_requires_frozen_pin_and_binds_exact_manifest( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + state, _result, keys = _initialize(tmp_path) + reviewer_file = tmp_path / "reviewer-registration.json" + registration = _reviewer_registration(reviewer_file) + manifest_file = tmp_path / "manifest.json" + manifest_raw = authority._canonical({"fixture": "validated elsewhere in this unit test"}) + manifest_file.write_bytes(manifest_raw) + manifest_sha = authority.batch._sha256_bytes(manifest_raw) + monkeypatch.setattr(authority.batch, "validate_manifest", lambda value: value) + + with pytest.raises(authority.AuthorityError, match="frozen RC-06 root pin"): + authority.build_rc06_candidate_keyring( + state, + reviewer_file, + manifest_file, + manifest_sha, + unprotect=_unprotect, + now=datetime(2026, 9, 27, 3, 0, tzinfo=UTC), + ) + + monkeypatch.setattr( + authority.batch, + "TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256", + keys[authority.ROLE_RC06_ROOT].public_sha256, + ) + monkeypatch.setattr(authority.batch, "FROZEN_APPROVAL_KEYRING_EPOCH", 1) + result = authority.build_rc06_candidate_keyring( + state, + reviewer_file, + manifest_file, + manifest_sha, + unprotect=_unprotect, + now=datetime(2026, 9, 27, 3, 0, tzinfo=UTC), + ) + assert result["status"] == "rc06_candidate_keyring_created_not_approved" + bundle = Path(result["directory"]) + keyring = json.loads((bundle / "rc06-approval-keyring.json").read_bytes()) + assert keyring["authorized_batch_manifest_raw_sha256"] == manifest_sha + assert keyring["epoch"] == 1 + assert keyring["keys"][0]["key_id"] == registration["key_id"] + assert keyring["root_key_id"] == authority.batch.TRUSTED_APPROVAL_ROOT_KEY_ID + + repeated = authority.build_rc06_candidate_keyring( + state, + reviewer_file, + manifest_file, + manifest_sha, + unprotect=_unprotect, + now=datetime(2026, 9, 27, 4, 0, tzinfo=UTC), + ) + assert repeated["status"] == "rc06_candidate_keyring_already_created_and_verified_not_approved" + with pytest.raises(authority.AuthorityError, match="SHA-256 mismatch"): + authority.build_rc06_candidate_keyring( + state, + reviewer_file, + manifest_file, + "f" * 64, + unprotect=_unprotect, + now=datetime(2026, 9, 27, 4, 0, tzinfo=UTC), + ) + + +def test_apply_private_key_provision_command_fails_closed_without_export( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + monkeypatch.setattr(authority, "_assert_runtime", lambda: None) + result = authority.main( + [ + "provision-apply-private-key", + "--state-directory", + os.fspath(tmp_path / "authority-state"), + ] + ) + captured = capsys.readouterr() + assert result == 2 + assert "not implemented" in captured.err + assert '"private_material_exported":false' in captured.err + assert list(tmp_path.iterdir()) == [] + + +def test_apply_payload_signing_is_bound_verified_atomic_and_idempotent( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + materials = _formal_apply_materials(tmp_path, monkeypatch) + output = tmp_path / "signed-apply-receipts" + result = authority.sign_apply_payloads( + materials["state"], + manifest_file=materials["manifest_file"], + manifest_sha256=materials["manifest_sha"], + approval_file=materials["approval_file"], + approval_sha256=materials["approval_sha"], + approval_keyring_file=materials["keyring_file"], + approval_keyring_sha256=materials["keyring_sha"], + approval_root_public_key_file=materials["root_file"], + payloads_file=materials["payloads_file"], + payloads_sha256=materials["payloads_sha"], + output_directory=output, + unprotect=_unprotect, + ) + assert result["status"] == "apply_receipts_signed_and_verified_not_published" + assert result["receipt_count"] == 1 + assert result["business_api_calls"] == 0 + assert result["private_keys_in_bundle"] is False + receipt = json.loads((output / "apply-receipt-001.json").read_bytes()) + assert authority.batch.validate_apply_receipt( + materials["manifest"], + materials["manifest_sha"], + materials["approval"], + materials["keyring"], + materials["keys"][authority.ROLE_RC06_ROOT].public_key, + materials["keys"][authority.ROLE_APPLY_SERVICE].public_key, + materials["approval_sha"], + receipt, + ) == receipt + repeated = authority.sign_apply_payloads( + materials["state"], + manifest_file=materials["manifest_file"], + manifest_sha256=materials["manifest_sha"], + approval_file=materials["approval_file"], + approval_sha256=materials["approval_sha"], + approval_keyring_file=materials["keyring_file"], + approval_keyring_sha256=materials["keyring_sha"], + approval_root_public_key_file=materials["root_file"], + payloads_file=materials["payloads_file"], + payloads_sha256=materials["payloads_sha"], + output_directory=output, + unprotect=_unprotect, + ) + assert repeated["status"] == "apply_receipts_already_signed_and_verified_not_published" + assert repeated["receipts"] == result["receipts"] + + +def test_apply_payload_signing_rejects_tamper_cross_approval_and_partial_bundle( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + materials = _formal_apply_materials(tmp_path, monkeypatch) + + tampered = dict(materials["apply_payload"]) + tampered["batch_manifest_payload_sha256"] = "f" * 64 + tampered_file = tmp_path / "tampered-payload.json" + tampered_raw = authority._canonical(tampered) + tampered_file.write_bytes(tampered_raw) + tampered_output = tmp_path / "tampered-output" + with pytest.raises(authority.AuthorityError, match="not bound to this batch"): + authority.sign_apply_payloads( + materials["state"], + manifest_file=materials["manifest_file"], + manifest_sha256=materials["manifest_sha"], + approval_file=materials["approval_file"], + approval_sha256=materials["approval_sha"], + approval_keyring_file=materials["keyring_file"], + approval_keyring_sha256=materials["keyring_sha"], + approval_root_public_key_file=materials["root_file"], + payloads_file=tampered_file, + payloads_sha256=authority.batch._sha256_bytes(tampered_raw), + output_directory=tampered_output, + unprotect=_unprotect, + ) + assert not tampered_output.exists() + + second_payload = dict(materials["approval"]["payload"]) + second_payload["approved_at"] = "2026-09-27T03:15:00Z" + second_signature = materials["reviewer_private"].sign( + authority.batch.APPROVAL_SIGNATURE_DOMAIN + authority._canonical(second_payload) + ) + second_approval = { + "schema_version": authority.batch.APPROVAL_SCHEMA_VERSION, + "approval_id": ( + "rc06-batch-approval-" + + authority.batch._sha256_bytes( + authority._canonical(second_payload) + second_signature + )[:24] + ), + "payload": second_payload, + "signature": base64.b64encode(second_signature).decode("ascii"), + } + second_approval_file = tmp_path / "second-approval.json" + second_approval_raw = authority._canonical(second_approval) + second_approval_file.write_bytes(second_approval_raw) + with pytest.raises(authority.AuthorityError, match="not bound to the approval receipt"): + authority.sign_apply_payloads( + materials["state"], + manifest_file=materials["manifest_file"], + manifest_sha256=materials["manifest_sha"], + approval_file=second_approval_file, + approval_sha256=authority.batch._sha256_bytes(second_approval_raw), + approval_keyring_file=materials["keyring_file"], + approval_keyring_sha256=materials["keyring_sha"], + approval_root_public_key_file=materials["root_file"], + payloads_file=materials["payloads_file"], + payloads_sha256=materials["payloads_sha"], + output_directory=tmp_path / "cross-approval-output", + unprotect=_unprotect, + ) + + partial = dict(materials["apply_payload"]) + partial["item_results"] = partial["item_results"][:-1] + payload_list = [materials["apply_payload"], partial] + payload_list_file = tmp_path / "payload-list.json" + payload_list_raw = authority._canonical(payload_list) + payload_list_file.write_bytes(payload_list_raw) + partial_output = tmp_path / "partial-output" + with pytest.raises(authority.AuthorityError, match="74-item online evidence drift"): + authority.sign_apply_payloads( + materials["state"], + manifest_file=materials["manifest_file"], + manifest_sha256=materials["manifest_sha"], + approval_file=materials["approval_file"], + approval_sha256=materials["approval_sha"], + approval_keyring_file=materials["keyring_file"], + approval_keyring_sha256=materials["keyring_sha"], + approval_root_public_key_file=materials["root_file"], + payloads_file=payload_list_file, + payloads_sha256=authority.batch._sha256_bytes(payload_list_raw), + output_directory=partial_output, + unprotect=_unprotect, + ) + assert not partial_output.exists() + + +def test_authority_launcher_uses_authority_script_and_forwards_non_secret_arguments() -> None: + text = AUTHORITY_LAUNCHER.read_text(encoding="utf-8-sig") + assert '"rc06-release-trust-bootstrap.py"' in text + assert "ValueFromRemainingArguments = $true" in text + assert "$authorityScript $Command @CommandArguments" in text + assert "rc06-single-reviewer-signing.py" not in text + assert "password" not in text.casefold() + assert "passphrase" not in text.casefold() + + +@pytest.mark.skipif(os.name != "nt", reason="Windows CurrentUser DPAPI is platform-specific") +def test_windows_current_user_dpapi_round_trip_has_role_entropy() -> None: + raw = b"not-a-real-private-key-fixture" + protected = authority._protect_current_user(raw, authority.ROLE_RC06_ROOT) + assert protected != raw + assert authority._unprotect_current_user(protected, authority.ROLE_RC06_ROOT) == raw + with pytest.raises(authority.AuthorityError): + authority._unprotect_current_user(protected, authority.ROLE_APPLY_SERVICE)