1336 lines
51 KiB
Python
1336 lines
51 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
import copy
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import stat
|
|
import sys
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from cryptography.hazmat.primitives import serialization
|
|
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
|
|
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[2]
|
|
NOW = datetime(2026, 9, 21, 6, 0, tzinfo=UTC)
|
|
|
|
|
|
def _load_script():
|
|
path = REPO_ROOT / "yuqei-infra/deploy/plan-rollback-rehearsal.py"
|
|
spec = importlib.util.spec_from_file_location("rollback_rehearsal_planner", path)
|
|
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
|
|
|
|
|
|
planner = _load_script()
|
|
|
|
|
|
def _write_json(path: Path, payload: dict[str, Any]) -> str:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
raw = planner.canonical_json_bytes(payload) + b"\n"
|
|
path.write_bytes(raw)
|
|
return hashlib.sha256(raw).hexdigest()
|
|
|
|
|
|
def _bootstrap_package_fixture(
|
|
tmp_path: Path,
|
|
*,
|
|
schema: str,
|
|
omitted_tool_input: str | None = None,
|
|
omitted_artifact: str | None = None,
|
|
extra_artifact: str | None = None,
|
|
repo_root: Path = REPO_ROOT,
|
|
) -> tuple[Path, str, dict[str, bytes]]:
|
|
source_items = []
|
|
expected_raw: dict[str, bytes] = {}
|
|
for relative in planner.REHEARSAL_TOOL_INPUTS:
|
|
source = repo_root / relative
|
|
raw = source.read_bytes()
|
|
expected_raw[relative] = raw
|
|
if relative != omitted_tool_input:
|
|
source_items.append(
|
|
{
|
|
"path": relative,
|
|
"sha256": hashlib.sha256(raw).hexdigest(),
|
|
}
|
|
)
|
|
source_manifest = {
|
|
"schema_version": "yuqei.release-source-manifest.v1",
|
|
"files": {"items": source_items},
|
|
}
|
|
release_dir = tmp_path / "release"
|
|
source_path = release_dir / "release-source-manifest.json"
|
|
source_sha = _write_json(source_path, source_manifest)
|
|
required = (
|
|
planner.V2_BOOTSTRAP_ARTIFACTS
|
|
if schema == planner.PACKAGE_SCHEMA_V2
|
|
else planner.V1_BOOTSTRAP_ARTIFACTS
|
|
)
|
|
artifacts = {
|
|
name: {"path": f"artifacts/{name}.json", "sha256": "0" * 64}
|
|
for name in required
|
|
if name != omitted_artifact
|
|
}
|
|
artifacts["release_manifest"] = {
|
|
"path": "release-source-manifest.json",
|
|
"sha256": source_sha,
|
|
}
|
|
if extra_artifact is not None:
|
|
artifacts[extra_artifact] = {
|
|
"path": f"artifacts/{extra_artifact}.json",
|
|
"sha256": "1" * 64,
|
|
}
|
|
package_path = release_dir / "release-package.json"
|
|
package_sha = _write_json(
|
|
package_path,
|
|
{"schema_version": schema, "artifacts": artifacts},
|
|
)
|
|
return package_path, package_sha, expected_raw
|
|
|
|
|
|
def _with_hash(payload: dict[str, Any], field: str) -> dict[str, Any]:
|
|
return {**payload, field: planner.canonical_sha256(payload)}
|
|
|
|
|
|
def _raw_public_key(private: Ed25519PrivateKey) -> bytes:
|
|
return private.public_key().public_bytes(
|
|
encoding=serialization.Encoding.Raw,
|
|
format=serialization.PublicFormat.Raw,
|
|
)
|
|
|
|
|
|
def _image_map(prefix: str) -> dict[str, dict[str, str]]:
|
|
keys = (
|
|
"CONTRACT_API_IMAGE",
|
|
"CONTRACT_WEB_IMAGE",
|
|
"AI_PLATFORM_API_IMAGE",
|
|
"AI_WORKER_IMAGE",
|
|
)
|
|
return {
|
|
key: {
|
|
"reference": f"registry.test/{prefix}/{key.lower()}@sha256:{str(index) * 64}",
|
|
"digest": "sha256:" + str(index) * 64,
|
|
}
|
|
for index, key in enumerate(keys, start=1)
|
|
}
|
|
|
|
|
|
def _packages():
|
|
predecessor_images = _image_map("release-1")
|
|
predecessor_import = tuple(
|
|
{
|
|
"role": key.lower(),
|
|
"environment_key": key,
|
|
"archive": {
|
|
"path": f"images/{key.lower()}.tar",
|
|
"sha256": str(index + 4) * 64,
|
|
},
|
|
"expected_loaded_image_id": value["digest"],
|
|
}
|
|
for index, (key, value) in enumerate(predecessor_images.items())
|
|
)
|
|
predecessor = SimpleNamespace(
|
|
package={"release_id": "release-1", "compose_project_name": "yuqei-prod"},
|
|
package_sha256="b" * 64,
|
|
application_images=predecessor_images,
|
|
installable_manifest_sha256="c" * 64,
|
|
trusted_build_receipt_sha256="d" * 64,
|
|
docker_import_plan=predecessor_import,
|
|
)
|
|
current = SimpleNamespace(
|
|
package={"release_id": "release-2", "compose_project_name": "yuqei-prod"},
|
|
package_sha256="a" * 64,
|
|
rollback_release_id="release-1",
|
|
rollback_images=predecessor_images,
|
|
)
|
|
return current, predecessor
|
|
|
|
|
|
def _root_sign_keyring(
|
|
keyring: dict[str, Any], root_private: Ed25519PrivateKey
|
|
) -> None:
|
|
unsigned = {
|
|
key: value
|
|
for key, value in keyring.items()
|
|
if key not in {"keyring_payload_sha256", "root_signature_base64"}
|
|
}
|
|
keyring["keyring_payload_sha256"] = planner.canonical_sha256(unsigned)
|
|
keyring["root_signature_base64"] = base64.b64encode(
|
|
root_private.sign(planner.canonical_json_bytes(unsigned))
|
|
).decode("ascii")
|
|
|
|
|
|
def _resign_evidence(
|
|
evidence: dict[str, Any],
|
|
approver_keys: list[Ed25519PrivateKey],
|
|
*,
|
|
expires_at: str = "2026-09-21T12:00:00Z",
|
|
signed_at: str = "2026-09-21T05:45:00Z",
|
|
approval_id: str = "approval-1",
|
|
) -> None:
|
|
payload = {
|
|
key: evidence[key]
|
|
for key in (
|
|
"schema_version",
|
|
"transition",
|
|
"runtime",
|
|
"target",
|
|
"backup",
|
|
"compatibility",
|
|
)
|
|
}
|
|
payload_hash = planner.canonical_sha256(payload)
|
|
assert len(approver_keys) == 1
|
|
unsigned = {
|
|
"schema_version": planner.APPROVAL_SCHEMA,
|
|
"approval_id": approval_id,
|
|
"approver_id": planner.SOLE_REVIEWER_ID,
|
|
"role": "release_owner",
|
|
"key_id": "release-owner-key-1",
|
|
"signed_at": signed_at,
|
|
"expires_at": expires_at,
|
|
"evidence_payload_sha256": payload_hash,
|
|
}
|
|
approvals = [
|
|
{
|
|
**unsigned,
|
|
"signature_base64": base64.b64encode(
|
|
approver_keys[0].sign(planner.canonical_json_bytes(unsigned))
|
|
).decode("ascii"),
|
|
}
|
|
]
|
|
evidence["evidence_payload_sha256"] = payload_hash
|
|
evidence["approvals"] = approvals
|
|
|
|
|
|
def _context(tmp_path: Path, monkeypatch) -> dict[str, Any]:
|
|
current, predecessor = _packages()
|
|
root_private = Ed25519PrivateKey.generate()
|
|
root_raw = _raw_public_key(root_private)
|
|
root_file = tmp_path / "organization-root.ed25519.pub"
|
|
root_file.parent.mkdir(parents=True, exist_ok=True)
|
|
root_file.write_bytes(root_raw)
|
|
monkeypatch.setattr(
|
|
planner,
|
|
"TRUSTED_ROOT_PUBLIC_KEY_SHA256",
|
|
hashlib.sha256(root_raw).hexdigest(),
|
|
)
|
|
approver_keys = [Ed25519PrivateKey.generate()]
|
|
identities = (("release-owner-key-1", planner.SOLE_REVIEWER_ID, "release_owner"),)
|
|
key_items = []
|
|
for private, (key_id, approver_id, role) in zip(
|
|
approver_keys, identities, strict=True
|
|
):
|
|
key_items.append(
|
|
{
|
|
"key_id": key_id,
|
|
"approver_id": approver_id,
|
|
"role": role,
|
|
"algorithm": "ed25519",
|
|
"public_key_base64": base64.b64encode(_raw_public_key(private)).decode(
|
|
"ascii"
|
|
),
|
|
"not_before": "2026-09-20T00:00:00Z",
|
|
"not_after": "2026-09-22T00:00:00Z",
|
|
"revoked": False,
|
|
}
|
|
)
|
|
keyring = {
|
|
"schema_version": planner.KEYRING_SCHEMA,
|
|
"root_key_id": planner.TRUSTED_ROOT_KEY_ID,
|
|
"epoch": 1,
|
|
"issued_at": "2026-09-20T00:00:00Z",
|
|
"expires_at": "2026-09-23T00:00:00Z",
|
|
"keys": key_items,
|
|
}
|
|
_root_sign_keyring(keyring, root_private)
|
|
monkeypatch.setattr(
|
|
planner,
|
|
"TRUSTED_CURRENT_KEYRING_PAYLOAD_SHA256",
|
|
keyring["keyring_payload_sha256"],
|
|
)
|
|
monkeypatch.setattr(planner, "MIN_KEYRING_EPOCH", 1)
|
|
|
|
target = {
|
|
"environment_id": "uat-cn-1",
|
|
"environment_class": "isolated_uat",
|
|
"host_id": "uat-host-1",
|
|
"host_identity_sha256": "4" * 64,
|
|
"docker_daemon_identity_sha256": "5" * 64,
|
|
"database_cluster_id": "uat-pg-cluster-1",
|
|
"source_database_id": "contracts-current",
|
|
"restore_database_id": "contracts-restore-drill",
|
|
"database_identity_sha256": planner.canonical_sha256({
|
|
"environment_id": "uat-cn-1",
|
|
"database_cluster_id": "uat-pg-cluster-1",
|
|
"source_database_id": "contracts-current",
|
|
"restore_database_id": "contracts-restore-drill",
|
|
}),
|
|
"compose_project_name": "yuqei-prod",
|
|
}
|
|
backup_artifact_path = tmp_path / "database-backup.dump"
|
|
backup_artifact_path.write_bytes(b"verified database backup fixture\n")
|
|
backup_artifact_sha = hashlib.sha256(backup_artifact_path.read_bytes()).hexdigest()
|
|
backup_artifact_size = backup_artifact_path.stat().st_size
|
|
|
|
backup_manifest = _with_hash(
|
|
{
|
|
"schema_version": planner.BACKUP_MANIFEST_SCHEMA,
|
|
"backup_id": "backup-release-2-001",
|
|
"source_release_id": "release-2",
|
|
"source_schema_revision": "schema-2026-09-21-001",
|
|
"environment_id": target["environment_id"],
|
|
"database_cluster_id": target["database_cluster_id"],
|
|
"source_database_id": target["source_database_id"],
|
|
"backup_artifact": {
|
|
"path": str(backup_artifact_path.resolve()),
|
|
"sha256": backup_artifact_sha,
|
|
"size_bytes": backup_artifact_size,
|
|
},
|
|
"completed_at": "2026-09-21T05:00:00Z",
|
|
"status": "complete",
|
|
},
|
|
"manifest_payload_sha256",
|
|
)
|
|
backup_path = tmp_path / "backup-manifest.json"
|
|
backup_sha = _write_json(backup_path, backup_manifest)
|
|
restore = _with_hash(
|
|
{
|
|
"schema_version": planner.RESTORE_EVIDENCE_SCHEMA,
|
|
"backup_id": "backup-release-2-001",
|
|
"backup_manifest_sha256": backup_sha,
|
|
"source_release_id": "release-2",
|
|
"environment_id": target["environment_id"],
|
|
"database_cluster_id": target["database_cluster_id"],
|
|
"source_database_id": target["source_database_id"],
|
|
"restore_database_id": target["restore_database_id"],
|
|
"backup_artifact_sha256": backup_artifact_sha,
|
|
"backup_artifact_size_bytes": backup_artifact_size,
|
|
"artifact_verified": True,
|
|
"restored_schema_revision": "schema-2026-09-21-001",
|
|
"tested_at": "2026-09-21T05:30:00Z",
|
|
"status": "passed",
|
|
},
|
|
"evidence_payload_sha256",
|
|
)
|
|
restore_path = tmp_path / "restore-evidence.json"
|
|
restore_sha = _write_json(restore_path, restore)
|
|
compatibility_artifact = _with_hash(
|
|
{
|
|
"schema_version": planner.COMPATIBILITY_EVIDENCE_SCHEMA,
|
|
"assessment_id": "compat-release-2-to-1-001",
|
|
"from_release_id": "release-2",
|
|
"to_release_id": "release-1",
|
|
"current_package_sha256": current.package_sha256,
|
|
"predecessor_package_sha256": predecessor.package_sha256,
|
|
"current_schema_revision": "schema-2026-09-21-001",
|
|
"database_downgrade_required": False,
|
|
"current_schema_is_backward_compatible": True,
|
|
"application_compatibility_status": "passed",
|
|
"approved_application_digests": planner._image_digests(predecessor),
|
|
"assessed_at": "2026-09-21T05:40:00Z",
|
|
},
|
|
"evidence_payload_sha256",
|
|
)
|
|
compatibility_path = tmp_path / "compatibility-evidence.json"
|
|
compatibility_sha = _write_json(compatibility_path, compatibility_artifact)
|
|
command = [
|
|
"/usr/bin/docker",
|
|
"--host",
|
|
"unix:///var/run/docker.sock",
|
|
"compose",
|
|
"up",
|
|
"-d",
|
|
"--pull",
|
|
"never",
|
|
"--no-build",
|
|
"--no-deps",
|
|
"contract-api",
|
|
]
|
|
runtime = planner._build_runtime_binding(
|
|
current_package=current,
|
|
predecessor_package=predecessor,
|
|
env_file_sha256="8" * 64,
|
|
compose_project_name="yuqei-prod",
|
|
command_argv=command,
|
|
compose_input_sha256={
|
|
"yuqei-infra/compose/docker-compose.yml": "1" * 64,
|
|
"yuqei-infra/compose/docker-compose.production.yml": "2" * 64,
|
|
"yuqei-infra/compose/docker-compose.production.offline.yml": "3" * 64,
|
|
},
|
|
tool_input_sha256={
|
|
path: f"{index + 10:064x}"
|
|
for index, path in enumerate(planner.REHEARSAL_TOOL_INPUTS)
|
|
},
|
|
target_identity_sha256=planner.canonical_sha256(target),
|
|
production_preflight_sha256="6" * 64,
|
|
compose_gate_evidence_sha256="7" * 64,
|
|
)
|
|
transition = {
|
|
"from_release_id": "release-2",
|
|
"from_package_sha256": current.package_sha256,
|
|
"to_release_id": "release-1",
|
|
"to_package_sha256": predecessor.package_sha256,
|
|
"predecessor_installable_manifest_sha256": predecessor.installable_manifest_sha256,
|
|
"predecessor_trusted_build_receipt_sha256": predecessor.trusted_build_receipt_sha256,
|
|
"predecessor_import_plan_sha256": planner.canonical_sha256(
|
|
[dict(item) for item in predecessor.docker_import_plan]
|
|
),
|
|
"rollback_application_digests": planner._image_digests(predecessor),
|
|
}
|
|
evidence = {
|
|
"schema_version": planner.EVIDENCE_SCHEMA,
|
|
"transition": transition,
|
|
"runtime": dict(runtime),
|
|
"target": target,
|
|
"backup": {
|
|
"backup_id": "backup-release-2-001",
|
|
"source_release_id": "release-2",
|
|
"source_schema_revision": "schema-2026-09-21-001",
|
|
"environment_id": target["environment_id"],
|
|
"database_cluster_id": target["database_cluster_id"],
|
|
"source_database_id": target["source_database_id"],
|
|
"backup_manifest": {
|
|
"path": str(backup_path.resolve()),
|
|
"sha256": backup_sha,
|
|
},
|
|
"restore_evidence": {
|
|
"path": str(restore_path.resolve()),
|
|
"sha256": restore_sha,
|
|
},
|
|
},
|
|
"compatibility": {
|
|
"assessment_id": "compat-release-2-to-1-001",
|
|
"current_schema_revision": "schema-2026-09-21-001",
|
|
"target_application_release_id": "release-1",
|
|
"current_package_sha256": current.package_sha256,
|
|
"predecessor_package_sha256": predecessor.package_sha256,
|
|
"evidence_artifact": {
|
|
"path": str(compatibility_path.resolve()),
|
|
"sha256": compatibility_sha,
|
|
},
|
|
},
|
|
}
|
|
_resign_evidence(evidence, approver_keys)
|
|
return {
|
|
"current": current,
|
|
"predecessor": predecessor,
|
|
"root_private": root_private,
|
|
"root_file": root_file,
|
|
"approver_keys": approver_keys,
|
|
"keyring": keyring,
|
|
"evidence": evidence,
|
|
"runtime": runtime,
|
|
"target": target,
|
|
"command": command,
|
|
"backup_path": backup_path,
|
|
"backup_artifact_path": backup_artifact_path,
|
|
"restore_path": restore_path,
|
|
"compatibility_path": compatibility_path,
|
|
}
|
|
|
|
|
|
def _validate(context: dict[str, Any]) -> dict[str, Any]:
|
|
root, _sha = planner._load_trusted_root_public_key(context["root_file"].resolve())
|
|
return planner.validate_rehearsal_evidence(
|
|
context["evidence"],
|
|
keyring=context["keyring"],
|
|
current_package=context["current"],
|
|
predecessor_package=context["predecessor"],
|
|
trusted_root_public_key=root,
|
|
runtime_binding=context["runtime"],
|
|
now=NOW,
|
|
)
|
|
|
|
|
|
def _rewrite_artifact(context: dict[str, Any], kind: str, mutate) -> None:
|
|
path = context[f"{kind}_path"]
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
hash_field = (
|
|
"manifest_payload_sha256" if kind == "backup" else "evidence_payload_sha256"
|
|
)
|
|
payload.pop(hash_field)
|
|
mutate(payload)
|
|
digest = _write_json(path, _with_hash(payload, hash_field))
|
|
if kind == "backup":
|
|
context["evidence"]["backup"]["backup_manifest"]["sha256"] = digest
|
|
restore = json.loads(context["restore_path"].read_text(encoding="utf-8"))
|
|
restore.pop("evidence_payload_sha256")
|
|
restore["backup_manifest_sha256"] = digest
|
|
restore_digest = _write_json(
|
|
context["restore_path"],
|
|
_with_hash(restore, "evidence_payload_sha256"),
|
|
)
|
|
context["evidence"]["backup"]["restore_evidence"]["sha256"] = restore_digest
|
|
elif kind == "restore":
|
|
context["evidence"]["backup"]["restore_evidence"]["sha256"] = digest
|
|
else:
|
|
context["evidence"]["compatibility"]["evidence_artifact"]["sha256"] = digest
|
|
_resign_evidence(context["evidence"], context["approver_keys"])
|
|
|
|
|
|
def test_root_signed_single_reviewer_evidence_unlocks_plan_generation_only(
|
|
tmp_path: Path, monkeypatch
|
|
) -> None:
|
|
context = _context(tmp_path, monkeypatch)
|
|
validated = _validate(context)
|
|
plan = planner.build_rehearsal_plan(
|
|
current_package=context["current"],
|
|
predecessor_package=context["predecessor"],
|
|
validated_evidence=validated,
|
|
evidence_file_sha256="5" * 64,
|
|
keyring_file_sha256="6" * 64,
|
|
trusted_root_key_sha256=planner.TRUSTED_ROOT_PUBLIC_KEY_SHA256,
|
|
command_argv=context["command"],
|
|
deployment_scope="backend_only",
|
|
deployed_service_set=["contract-api"],
|
|
generated_at=NOW,
|
|
)
|
|
assert plan["status"] == "approved_for_plan_only_rehearsal"
|
|
assert plan["execution_entrypoint_available"] is False
|
|
assert plan["apply_allowed"] is False
|
|
assert plan["docker_invoked"] is False
|
|
assert plan["remote_connection_attempted"] is False
|
|
assert plan["env_file_sha256"] == "8" * 64
|
|
assert plan["keyring_epoch"] == 1
|
|
assert plan["keyring_payload_sha256"] == context["keyring"][
|
|
"keyring_payload_sha256"
|
|
]
|
|
assert len({item["approver_id"] for item in validated["approvals"]}) == 1
|
|
assert {item["role"] for item in validated["approvals"]} == {"release_owner"}
|
|
assert len({item["key_id"] for item in validated["approvals"]}) == 1
|
|
assert plan["reviewer"] == {
|
|
"approver_id": "chen-yuqi",
|
|
"name": "陈钰麒",
|
|
"title": "总经理",
|
|
}
|
|
assert plan["compose_input_sha256"] == context["runtime"][
|
|
"compose_input_sha256"
|
|
]
|
|
assert plan["command_argv_sha256"] == context["runtime"]["command_argv_sha256"]
|
|
assert plan["backup"]["backup_artifact_sha256"] == hashlib.sha256(
|
|
context["backup_artifact_path"].read_bytes()
|
|
).hexdigest()
|
|
assert plan["backup"]["backup_artifact_size_bytes"] == context[
|
|
"backup_artifact_path"
|
|
].stat().st_size
|
|
assert len(plan["predecessor_import_plan"]) == 4
|
|
assert planner.canonical_sha256(
|
|
{key: value for key, value in plan.items() if key != "plan_sha256"}
|
|
) == plan["plan_sha256"]
|
|
|
|
|
|
def test_unprovisioned_or_substituted_root_key_fails_closed(
|
|
tmp_path: Path, monkeypatch
|
|
) -> None:
|
|
context = _context(tmp_path, monkeypatch)
|
|
monkeypatch.setattr(planner, "TRUSTED_ROOT_PUBLIC_KEY_SHA256", "")
|
|
with pytest.raises(planner.RehearsalProofError, match="not provisioned"):
|
|
planner._load_trusted_root_public_key(context["root_file"].resolve())
|
|
other = tmp_path / "other-root.pub"
|
|
other.write_bytes(_raw_public_key(Ed25519PrivateKey.generate()))
|
|
monkeypatch.setattr(
|
|
planner,
|
|
"TRUSTED_ROOT_PUBLIC_KEY_SHA256",
|
|
hashlib.sha256(context["root_file"].read_bytes()).hexdigest(),
|
|
)
|
|
with pytest.raises(planner.RehearsalProofError, match="frozen trust anchor"):
|
|
planner._load_trusted_root_public_key(other.resolve())
|
|
|
|
|
|
def test_unprovisioned_current_keyring_state_fails_closed(
|
|
tmp_path: Path, monkeypatch
|
|
) -> None:
|
|
context = _context(tmp_path, monkeypatch)
|
|
root, _sha = planner._load_trusted_root_public_key(context["root_file"].resolve())
|
|
monkeypatch.setattr(planner, "TRUSTED_CURRENT_KEYRING_PAYLOAD_SHA256", "")
|
|
with pytest.raises(planner.RehearsalProofError, match="keyring pin is not provisioned"):
|
|
planner._validate_keyring(context["keyring"], now=NOW, trusted_root_public_key=root)
|
|
monkeypatch.setattr(
|
|
planner,
|
|
"TRUSTED_CURRENT_KEYRING_PAYLOAD_SHA256",
|
|
context["keyring"]["keyring_payload_sha256"],
|
|
)
|
|
monkeypatch.setattr(planner, "MIN_KEYRING_EPOCH", None)
|
|
with pytest.raises(planner.RehearsalProofError, match="epoch is not provisioned"):
|
|
planner._validate_keyring(context["keyring"], now=NOW, trusted_root_public_key=root)
|
|
|
|
|
|
def test_old_unexpired_keyring_is_rejected_after_new_revocation_state(
|
|
tmp_path: Path, monkeypatch
|
|
) -> None:
|
|
context = _context(tmp_path, monkeypatch)
|
|
old_keyring = copy.deepcopy(context["keyring"])
|
|
new_keyring = copy.deepcopy(old_keyring)
|
|
new_keyring["epoch"] = 2
|
|
new_keyring["keys"][0]["revoked"] = True
|
|
rotated_key = Ed25519PrivateKey.generate()
|
|
new_keyring["keys"].append(
|
|
{
|
|
"key_id": "release-owner-key-2",
|
|
"approver_id": planner.SOLE_REVIEWER_ID,
|
|
"role": "release_owner",
|
|
"algorithm": "ed25519",
|
|
"public_key_base64": base64.b64encode(
|
|
_raw_public_key(rotated_key)
|
|
).decode("ascii"),
|
|
"not_before": "2026-09-20T00:00:00Z",
|
|
"not_after": "2026-09-22T00:00:00Z",
|
|
"revoked": False,
|
|
}
|
|
)
|
|
_root_sign_keyring(new_keyring, context["root_private"])
|
|
monkeypatch.setattr(
|
|
planner,
|
|
"TRUSTED_CURRENT_KEYRING_PAYLOAD_SHA256",
|
|
new_keyring["keyring_payload_sha256"],
|
|
)
|
|
monkeypatch.setattr(planner, "MIN_KEYRING_EPOCH", 2)
|
|
root, _sha = planner._load_trusted_root_public_key(context["root_file"].resolve())
|
|
trusted = planner._validate_keyring(
|
|
new_keyring,
|
|
now=NOW,
|
|
trusted_root_public_key=root,
|
|
)
|
|
assert "release-owner-key-1" not in trusted
|
|
assert "release-owner-key-2" in trusted
|
|
with pytest.raises(planner.RehearsalProofError, match="older than the trusted minimum"):
|
|
planner._validate_keyring(old_keyring, now=NOW, trusted_root_public_key=root)
|
|
|
|
|
|
def test_keyring_forged_by_untrusted_root_is_rejected(tmp_path: Path, monkeypatch) -> None:
|
|
context = _context(tmp_path, monkeypatch)
|
|
_root_sign_keyring(context["keyring"], Ed25519PrivateKey.generate())
|
|
with pytest.raises(planner.RehearsalProofError, match="organization-root signature"):
|
|
_validate(context)
|
|
|
|
|
|
def test_keyring_rejects_approval_key_reused_as_offline_root(
|
|
tmp_path: Path, monkeypatch
|
|
) -> None:
|
|
context = _context(tmp_path, monkeypatch)
|
|
context["keyring"]["keys"][0]["public_key_base64"] = base64.b64encode(
|
|
context["root_file"].read_bytes()
|
|
).decode("ascii")
|
|
_root_sign_keyring(context["keyring"], context["root_private"])
|
|
monkeypatch.setattr(
|
|
planner,
|
|
"TRUSTED_CURRENT_KEYRING_PAYLOAD_SHA256",
|
|
context["keyring"]["keyring_payload_sha256"],
|
|
)
|
|
with pytest.raises(planner.RehearsalProofError, match="must differ from the offline root"):
|
|
_validate(context)
|
|
|
|
|
|
def test_keyring_rejects_another_reviewer_identity(
|
|
tmp_path: Path, monkeypatch
|
|
) -> None:
|
|
context = _context(tmp_path, monkeypatch)
|
|
context["keyring"]["keys"][0]["approver_id"] = "another-reviewer"
|
|
_root_sign_keyring(context["keyring"], context["root_private"])
|
|
monkeypatch.setattr(
|
|
planner,
|
|
"TRUSTED_CURRENT_KEYRING_PAYLOAD_SHA256",
|
|
context["keyring"]["keyring_payload_sha256"],
|
|
)
|
|
with pytest.raises(planner.RehearsalProofError, match="unauthorized reviewer"):
|
|
_validate(context)
|
|
|
|
|
|
def test_rollback_approval_rejects_more_than_one_signature(
|
|
tmp_path: Path, monkeypatch
|
|
) -> None:
|
|
context = _context(tmp_path, monkeypatch)
|
|
context["evidence"]["approvals"].append(
|
|
copy.deepcopy(context["evidence"]["approvals"][0])
|
|
)
|
|
with pytest.raises(planner.RehearsalProofError, match="Exactly one"):
|
|
_validate(context)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("case", "message"),
|
|
(
|
|
("missing", "missing or unknown fields"),
|
|
("expired", "expired or not yet valid"),
|
|
("wrong_release", "does not match the immutable"),
|
|
("wrong_image", "does not match the immutable"),
|
|
("wrong_schema", "Unsupported rollback rehearsal evidence schema"),
|
|
("wrong_signature", "signature verification failed"),
|
|
("extra_approval", "Exactly one"),
|
|
("runtime", "not bound to this exact env"),
|
|
),
|
|
)
|
|
def test_untrusted_or_incomplete_signed_evidence_is_rejected(
|
|
tmp_path: Path, monkeypatch, case: str, message: str
|
|
) -> None:
|
|
context = _context(tmp_path, monkeypatch)
|
|
evidence = context["evidence"]
|
|
if case == "missing":
|
|
evidence["backup"].pop("backup_manifest")
|
|
_resign_evidence(evidence, context["approver_keys"])
|
|
elif case == "expired":
|
|
_resign_evidence(
|
|
evidence,
|
|
context["approver_keys"],
|
|
expires_at="2026-09-21T05:59:59Z",
|
|
)
|
|
elif case == "wrong_release":
|
|
evidence["transition"]["to_release_id"] = "release-0"
|
|
_resign_evidence(evidence, context["approver_keys"])
|
|
elif case == "wrong_image":
|
|
evidence["transition"]["rollback_application_digests"]["CONTRACT_API_IMAGE"] = (
|
|
"sha256:" + "0" * 64
|
|
)
|
|
_resign_evidence(evidence, context["approver_keys"])
|
|
elif case == "wrong_schema":
|
|
evidence["schema_version"] = "yuqei.rollback-rehearsal-evidence.v0"
|
|
_resign_evidence(evidence, context["approver_keys"])
|
|
elif case == "wrong_signature":
|
|
evidence["approvals"][0]["signature_base64"] = base64.b64encode(b"0" * 64).decode()
|
|
elif case == "extra_approval":
|
|
evidence["approvals"].append(copy.deepcopy(evidence["approvals"][0]))
|
|
else:
|
|
evidence["runtime"]["env_file_sha256"] = "0" * 64
|
|
_resign_evidence(evidence, context["approver_keys"])
|
|
with pytest.raises(planner.RehearsalProofError, match=message):
|
|
_validate(context)
|
|
|
|
|
|
def test_current_and_predecessor_release_must_differ(tmp_path: Path, monkeypatch) -> None:
|
|
context = _context(tmp_path, monkeypatch)
|
|
context["current"].package["release_id"] = "release-1"
|
|
context["current"].rollback_release_id = "release-1"
|
|
with pytest.raises(planner.RehearsalProofError, match="must be different"):
|
|
_validate(context)
|
|
|
|
|
|
def test_missing_artifact_file_is_rejected(tmp_path: Path, monkeypatch) -> None:
|
|
context = _context(tmp_path, monkeypatch)
|
|
context["restore_path"].unlink()
|
|
with pytest.raises(planner.RehearsalProofError, match="unavailable"):
|
|
_validate(context)
|
|
|
|
|
|
def test_artifact_hash_mismatch_is_rejected(tmp_path: Path, monkeypatch) -> None:
|
|
context = _context(tmp_path, monkeypatch)
|
|
context["compatibility_path"].write_bytes(b'{"tampered":true}\n')
|
|
with pytest.raises(planner.RehearsalProofError, match="externally supplied SHA-256"):
|
|
_validate(context)
|
|
|
|
|
|
def test_backup_artifact_content_and_size_are_verified(
|
|
tmp_path: Path, monkeypatch
|
|
) -> None:
|
|
context = _context(tmp_path, monkeypatch)
|
|
context["backup_artifact_path"].write_bytes(b"tampered database backup fixture\n")
|
|
with pytest.raises(planner.RehearsalProofError, match="identity or size|content"):
|
|
_validate(context)
|
|
|
|
|
|
def test_target_and_database_identity_cannot_be_replayed_across_environments(
|
|
tmp_path: Path, monkeypatch
|
|
) -> None:
|
|
context = _context(tmp_path, monkeypatch)
|
|
context["evidence"]["backup"]["environment_id"] = "uat-cn-2"
|
|
_resign_evidence(context["evidence"], context["approver_keys"])
|
|
with pytest.raises(planner.RehearsalProofError, match="different target database"):
|
|
_validate(context)
|
|
|
|
context = _context(tmp_path / "target-drift", monkeypatch)
|
|
context["evidence"]["target"]["host_id"] = "other-host"
|
|
_resign_evidence(context["evidence"], context["approver_keys"])
|
|
with pytest.raises(planner.RehearsalProofError, match="not bound to the signed target"):
|
|
_validate(context)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("kind", "mutation", "message"),
|
|
(
|
|
("backup", lambda value: value.update(status="failed"), "does not match the signed proof"),
|
|
("restore", lambda value: value.update(status="failed"), "does not match the signed backup"),
|
|
(
|
|
"compatibility",
|
|
lambda value: value.update(application_compatibility_status="failed"),
|
|
"compatibility has not passed",
|
|
),
|
|
(
|
|
"compatibility",
|
|
lambda value: value.update(database_downgrade_required=True),
|
|
"Database downgrade is forbidden",
|
|
),
|
|
),
|
|
)
|
|
def test_artifact_semantics_are_verified(
|
|
tmp_path: Path, monkeypatch, kind: str, mutation, message: str
|
|
) -> None:
|
|
context = _context(tmp_path, monkeypatch)
|
|
_rewrite_artifact(context, kind, mutation)
|
|
with pytest.raises(planner.RehearsalProofError, match=message):
|
|
_validate(context)
|
|
|
|
|
|
def test_stale_backup_artifact_is_rejected(tmp_path: Path, monkeypatch) -> None:
|
|
context = _context(tmp_path, monkeypatch)
|
|
_rewrite_artifact(
|
|
context,
|
|
"backup",
|
|
lambda value: value.update(completed_at="2026-09-20T05:59:59Z"),
|
|
)
|
|
_rewrite_artifact(
|
|
context,
|
|
"restore",
|
|
lambda value: value.update(tested_at="2026-09-20T06:30:00Z"),
|
|
)
|
|
with pytest.raises(planner.RehearsalProofError, match="older than the 24-hour"):
|
|
_validate(context)
|
|
|
|
|
|
def test_approval_must_not_predate_latest_bound_evidence(
|
|
tmp_path: Path, monkeypatch
|
|
) -> None:
|
|
context = _context(tmp_path, monkeypatch)
|
|
_resign_evidence(
|
|
context["evidence"],
|
|
context["approver_keys"],
|
|
signed_at="2026-09-21T05:39:59Z",
|
|
)
|
|
with pytest.raises(planner.RehearsalProofError, match="predates backup, restore, or compatibility"):
|
|
_validate(context)
|
|
|
|
|
|
def test_signed_compose_hash_substitution_is_rejected_against_live_inputs(
|
|
tmp_path: Path, monkeypatch
|
|
) -> None:
|
|
context = _context(tmp_path, monkeypatch)
|
|
context["evidence"]["runtime"]["compose_input_sha256"][
|
|
"yuqei-infra/compose/docker-compose.production.yml"
|
|
] = "9" * 64
|
|
unsigned_runtime = {
|
|
key: value
|
|
for key, value in context["evidence"]["runtime"].items()
|
|
if key != "offline_plan_inputs_sha256"
|
|
}
|
|
context["evidence"]["runtime"]["offline_plan_inputs_sha256"] = (
|
|
planner.canonical_sha256(unsigned_runtime)
|
|
)
|
|
_resign_evidence(context["evidence"], context["approver_keys"])
|
|
with pytest.raises(planner.RehearsalProofError, match="not bound to this exact env"):
|
|
_validate(context)
|
|
|
|
|
|
def test_compose_inputs_are_directly_hashed_and_bound_to_source_manifest(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
relative_paths = (
|
|
"yuqei-infra/compose/docker-compose.yml",
|
|
"yuqei-infra/compose/docker-compose.production.yml",
|
|
"yuqei-infra/compose/docker-compose.production.offline.yml",
|
|
)
|
|
items = []
|
|
expected = {}
|
|
for index, relative in enumerate(relative_paths, start=1):
|
|
target = tmp_path / relative
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
target.write_bytes(f"compose-{index}\n".encode())
|
|
digest = hashlib.sha256(target.read_bytes()).hexdigest()
|
|
items.append({"path": relative, "sha256": digest})
|
|
expected[relative] = digest
|
|
release_package = SimpleNamespace(
|
|
BASE_COMPOSE_RELATIVE=relative_paths[0],
|
|
PRODUCTION_COMPOSE_RELATIVE=relative_paths[1],
|
|
OFFLINE_COMPOSE_RELATIVE=relative_paths[2],
|
|
)
|
|
current = SimpleNamespace(release_manifest={"files": {"items": items}})
|
|
assert planner._validated_compose_input_hashes(
|
|
release_package=release_package,
|
|
current_package=current,
|
|
repo_root=tmp_path.resolve(),
|
|
) == expected
|
|
(tmp_path / relative_paths[1]).write_bytes(b"tampered\n")
|
|
with pytest.raises(planner.RehearsalProofError, match="differs from package identity"):
|
|
planner._validated_compose_input_hashes(
|
|
release_package=release_package,
|
|
current_package=current,
|
|
repo_root=tmp_path.resolve(),
|
|
)
|
|
|
|
|
|
def test_rollback_tool_sources_are_bound_to_current_package(tmp_path: Path) -> None:
|
|
items = []
|
|
expected = {}
|
|
for index, relative in enumerate(planner.REHEARSAL_TOOL_INPUTS, start=1):
|
|
target = tmp_path / relative
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
target.write_bytes(f"tool-{index}\n".encode())
|
|
digest = hashlib.sha256(target.read_bytes()).hexdigest()
|
|
items.append({"path": relative, "sha256": digest})
|
|
expected[relative] = digest
|
|
current = SimpleNamespace(release_manifest={"files": {"items": items}})
|
|
assert planner._validated_tool_input_hashes(
|
|
current_package=current,
|
|
repo_root=tmp_path.resolve(),
|
|
) == expected
|
|
(tmp_path / planner.REHEARSAL_TOOL_INPUTS[0]).write_bytes(b"tampered\n")
|
|
with pytest.raises(planner.RehearsalProofError, match="differs from package identity"):
|
|
planner._validated_tool_input_hashes(
|
|
current_package=current,
|
|
repo_root=tmp_path.resolve(),
|
|
)
|
|
|
|
|
|
def test_rollback_tool_sources_are_bootstrapped_before_module_execution(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
package_path, package_sha, expected_raw = _bootstrap_package_fixture(
|
|
tmp_path,
|
|
schema=planner.PACKAGE_SCHEMA_V2,
|
|
)
|
|
|
|
verified_sources, verified_hashes = planner._bootstrap_verified_tool_sources(
|
|
current_package_dir=package_path.parent.resolve(),
|
|
current_package_sha256=package_sha,
|
|
repo_root=REPO_ROOT,
|
|
)
|
|
assert verified_sources == expected_raw
|
|
assert verified_hashes == {
|
|
path: hashlib.sha256(raw).hexdigest()
|
|
for path, raw in expected_raw.items()
|
|
}
|
|
loaded = planner._load_release_package_module(verified_sources)
|
|
assert loaded._INSTALLABLE_VALIDATOR_MODULE is not None
|
|
assert loaded._PREFLIGHT_MODULE is not None
|
|
assert loaded._COMPOSE_GATE_MODULE is not None
|
|
assert loaded._INFRASTRUCTURE_BUNDLE_MODULE is not None
|
|
assert loaded._THIRD_PARTY_MATERIALS_MODULE is not None
|
|
assert loaded._SNAPSHOT_RESTORE_PLANNER_MODULE is not None
|
|
assert (
|
|
loaded._SNAPSHOT_RESTORE_PLANNER_MODULE._load_release_package_validator(
|
|
REPO_ROOT
|
|
)
|
|
is loaded.validate_release_package
|
|
)
|
|
|
|
(package_path.parent / "release-source-manifest.json").write_bytes(b"tampered\n")
|
|
with pytest.raises(planner.RehearsalProofError, match="externally supplied SHA-256"):
|
|
planner._bootstrap_verified_tool_sources(
|
|
current_package_dir=package_path.parent.resolve(),
|
|
current_package_sha256=package_sha,
|
|
repo_root=REPO_ROOT,
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"relative",
|
|
(
|
|
"yuqei-infra/scripts/build-infrastructure-image-bundle.py",
|
|
"yuqei-infra/scripts/build-third-party-materials.py",
|
|
"yuqei-infra/deploy/plan-first-release-snapshot-restore.py",
|
|
"yuqei-infra/release-trust/third-party-materials-baseline.json",
|
|
),
|
|
)
|
|
def test_v2_bootstrap_rejects_missing_new_source_closure(
|
|
tmp_path: Path,
|
|
relative: str,
|
|
) -> None:
|
|
package_path, package_sha, _expected_raw = _bootstrap_package_fixture(
|
|
tmp_path,
|
|
schema=planner.PACKAGE_SCHEMA_V2,
|
|
omitted_tool_input=relative,
|
|
)
|
|
with pytest.raises(planner.RehearsalProofError, match="does not bind"):
|
|
planner._bootstrap_verified_tool_sources(
|
|
current_package_dir=package_path.parent.resolve(),
|
|
current_package_sha256=package_sha,
|
|
repo_root=REPO_ROOT,
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"relative",
|
|
(
|
|
"yuqei-infra/scripts/build-infrastructure-image-bundle.py",
|
|
"yuqei-infra/scripts/build-third-party-materials.py",
|
|
"yuqei-infra/deploy/plan-first-release-snapshot-restore.py",
|
|
"yuqei-infra/release-trust/third-party-materials-baseline.json",
|
|
),
|
|
)
|
|
def test_v2_bootstrap_rejects_new_source_drift(
|
|
tmp_path: Path,
|
|
relative: str,
|
|
) -> None:
|
|
candidate_root = tmp_path / "candidate"
|
|
for tool_input in planner.REHEARSAL_TOOL_INPUTS:
|
|
target = candidate_root / tool_input
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
target.write_bytes((REPO_ROOT / tool_input).read_bytes())
|
|
package_path, package_sha, _expected_raw = _bootstrap_package_fixture(
|
|
tmp_path / "package",
|
|
schema=planner.PACKAGE_SCHEMA_V2,
|
|
repo_root=candidate_root,
|
|
)
|
|
(candidate_root / relative).write_bytes(b"candidate drift\n")
|
|
with pytest.raises(planner.RehearsalProofError, match="differs from package identity"):
|
|
planner._bootstrap_verified_tool_sources(
|
|
current_package_dir=package_path.parent.resolve(),
|
|
current_package_sha256=package_sha,
|
|
repo_root=candidate_root.resolve(),
|
|
)
|
|
|
|
|
|
def test_verified_v2_dependency_modules_are_injected_from_bootstrapped_bytes(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
_package_path, _package_sha, verified_sources = _bootstrap_package_fixture(
|
|
tmp_path,
|
|
schema=planner.PACKAGE_SCHEMA_V2,
|
|
)
|
|
injected = dict(verified_sources)
|
|
module_paths = (
|
|
"yuqei-infra/scripts/build-infrastructure-image-bundle.py",
|
|
"yuqei-infra/scripts/build-third-party-materials.py",
|
|
"yuqei-infra/deploy/plan-first-release-snapshot-restore.py",
|
|
)
|
|
for index, relative in enumerate(module_paths, start=1):
|
|
injected[relative] += f"\nVERIFIED_BYTE_SENTINEL = {index}\n".encode()
|
|
loaded = planner._load_release_package_module(injected)
|
|
assert loaded._INFRASTRUCTURE_BUNDLE_MODULE.VERIFIED_BYTE_SENTINEL == 1
|
|
assert loaded._THIRD_PARTY_MATERIALS_MODULE.VERIFIED_BYTE_SENTINEL == 2
|
|
assert loaded._SNAPSHOT_RESTORE_PLANNER_MODULE.VERIFIED_BYTE_SENTINEL == 3
|
|
|
|
|
|
def test_bootstrap_accepts_complete_legacy_v1_for_read_only_validation(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
package_path, package_sha, expected_raw = _bootstrap_package_fixture(
|
|
tmp_path,
|
|
schema=planner.PACKAGE_SCHEMA_V1,
|
|
)
|
|
verified_sources, _verified_hashes = planner._bootstrap_verified_tool_sources(
|
|
current_package_dir=package_path.parent.resolve(),
|
|
current_package_sha256=package_sha,
|
|
repo_root=REPO_ROOT,
|
|
)
|
|
assert verified_sources == expected_raw
|
|
|
|
|
|
def test_bootstrap_rejects_incomplete_v2_and_v2_artifact_downgrade(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
package_path, package_sha, _expected_raw = _bootstrap_package_fixture(
|
|
tmp_path / "incomplete-v2",
|
|
schema=planner.PACKAGE_SCHEMA_V2,
|
|
omitted_artifact="third_party_materials_verification_receipt",
|
|
)
|
|
with pytest.raises(planner.RehearsalProofError, match="v2 bootstrap artifact closure"):
|
|
planner._bootstrap_verified_tool_sources(
|
|
current_package_dir=package_path.parent.resolve(),
|
|
current_package_sha256=package_sha,
|
|
repo_root=REPO_ROOT,
|
|
)
|
|
|
|
package_path, package_sha, _expected_raw = _bootstrap_package_fixture(
|
|
tmp_path / "downgraded-v1",
|
|
schema=planner.PACKAGE_SCHEMA_V1,
|
|
extra_artifact="infrastructure_image_bundle",
|
|
)
|
|
with pytest.raises(planner.RehearsalProofError, match="v1 bootstrap must remain read-only"):
|
|
planner._bootstrap_verified_tool_sources(
|
|
current_package_dir=package_path.parent.resolve(),
|
|
current_package_sha256=package_sha,
|
|
repo_root=REPO_ROOT,
|
|
)
|
|
|
|
|
|
def test_strict_env_preflight_and_compose_gate_evidence_fail_closed(
|
|
tmp_path: Path, monkeypatch
|
|
) -> None:
|
|
preflight = planner._load_preflight_module()
|
|
passed = preflight.Check(key="fixture", status="pass", message="passed")
|
|
monkeypatch.setattr(preflight, "validate_environment", lambda *_args, **_kwargs: [passed])
|
|
monkeypatch.setattr(preflight, "validate_files", lambda *_args, **_kwargs: [])
|
|
values, digest = planner._parse_and_validate_production_env(
|
|
b"COMPOSE_PROJECT_NAME=yuqei-prod\n"
|
|
)
|
|
assert values["COMPOSE_PROJECT_NAME"] == "yuqei-prod"
|
|
assert len(digest) == 64
|
|
with pytest.raises(planner.RehearsalProofError, match="Invalid strict"):
|
|
planner._parse_and_validate_production_env(
|
|
b"COMPOSE_PROJECT_NAME=yuqei-prod\nBROKEN LINE\n"
|
|
)
|
|
|
|
compose_hashes = {
|
|
"yuqei-infra/compose/docker-compose.yml": "1" * 64,
|
|
"yuqei-infra/compose/docker-compose.production.yml": "2" * 64,
|
|
"yuqei-infra/compose/docker-compose.production.offline.yml": "3" * 64,
|
|
}
|
|
profiles = {
|
|
"default": [],
|
|
"backend-hosted-web": ["backend-hosted-web"],
|
|
"ops": ["ops"],
|
|
"backend-hosted-web+ops": ["backend-hosted-web", "ops"],
|
|
}
|
|
gate = {
|
|
"schema_version": planner.COMPOSE_GATE_SCHEMA,
|
|
"status": "passed",
|
|
"env_file_sha256": "8" * 64,
|
|
"compose_inputs": [
|
|
{"path": path, "sha256": sha}
|
|
for path, sha in compose_hashes.items()
|
|
],
|
|
"application_service_count": 9,
|
|
"production_service_count": 17,
|
|
"cases": [
|
|
{
|
|
"mode": mode,
|
|
"profile_case": profile,
|
|
"profiles": profile_values,
|
|
"service_count": planner.EXPECTED_PRODUCTION_SERVICE_COUNTS[profile],
|
|
"validated_application_services": [
|
|
service
|
|
for service in planner.EXPECTED_APPLICATION_SERVICES
|
|
if service != "contract-web" or "backend-hosted-web" in profile_values
|
|
],
|
|
"config_sha256": str(index + 1) * 64,
|
|
}
|
|
for index, (mode, profile, profile_values) in enumerate(
|
|
(mode, profile, values)
|
|
for mode in ("online", "offline")
|
|
for profile, values in profiles.items()
|
|
)
|
|
],
|
|
}
|
|
assert planner._validate_compose_gate_evidence(
|
|
gate,
|
|
raw_sha256="7" * 64,
|
|
env_file_sha256="8" * 64,
|
|
compose_input_sha256=compose_hashes,
|
|
) == "7" * 64
|
|
gate["cases"].pop()
|
|
with pytest.raises(planner.RehearsalProofError, match="all eight"):
|
|
planner._validate_compose_gate_evidence(
|
|
gate,
|
|
raw_sha256="7" * 64,
|
|
env_file_sha256="8" * 64,
|
|
compose_input_sha256=compose_hashes,
|
|
)
|
|
|
|
gate["cases"] = [
|
|
{
|
|
"mode": mode,
|
|
"profile_case": profile,
|
|
"profiles": profile_values,
|
|
"service_count": planner.EXPECTED_PRODUCTION_SERVICE_COUNTS[profile],
|
|
"validated_application_services": [
|
|
service
|
|
for service in planner.EXPECTED_APPLICATION_SERVICES
|
|
if service != "contract-web" or "backend-hosted-web" in profile_values
|
|
],
|
|
"config_sha256": str(index + 1) * 64,
|
|
}
|
|
for index, (mode, profile, profile_values) in enumerate(
|
|
(mode, profile, values)
|
|
for mode in ("online", "offline")
|
|
for profile, values in profiles.items()
|
|
)
|
|
]
|
|
gate["cases"][0]["validated_application_services"].pop()
|
|
with pytest.raises(planner.RehearsalProofError, match="case inventory"):
|
|
planner._validate_compose_gate_evidence(
|
|
gate,
|
|
raw_sha256="7" * 64,
|
|
env_file_sha256="8" * 64,
|
|
compose_input_sha256=compose_hashes,
|
|
)
|
|
|
|
|
|
def _make_symlink_or_simulate(link: Path, target: Path, monkeypatch) -> None:
|
|
try:
|
|
os.symlink(target, link)
|
|
except (OSError, NotImplementedError):
|
|
link.write_bytes(target.read_bytes())
|
|
original = os.lstat
|
|
|
|
def simulated_lstat(path):
|
|
result = original(path)
|
|
if Path(path).absolute() == link.absolute():
|
|
values = list(result)
|
|
values[0] = stat.S_IFLNK | 0o777
|
|
return os.stat_result(values)
|
|
return result
|
|
|
|
monkeypatch.setattr(os, "lstat", simulated_lstat)
|
|
|
|
|
|
def test_artifact_symlink_is_rejected(tmp_path: Path, monkeypatch) -> None:
|
|
context = _context(tmp_path, monkeypatch)
|
|
target = context["restore_path"]
|
|
link = tmp_path / "restore-link.json"
|
|
_make_symlink_or_simulate(link, target, monkeypatch)
|
|
context["evidence"]["backup"]["restore_evidence"]["path"] = str(link.absolute())
|
|
_resign_evidence(context["evidence"], context["approver_keys"])
|
|
with pytest.raises(planner.RehearsalProofError, match="symbolic links"):
|
|
_validate(context)
|
|
|
|
|
|
def test_pinned_json_requires_absolute_path_and_exact_content(tmp_path: Path) -> None:
|
|
path = tmp_path / "evidence.json"
|
|
path.write_bytes(b'{"status":"test"}\n')
|
|
digest = hashlib.sha256(path.read_bytes()).hexdigest()
|
|
with pytest.raises(planner.RehearsalProofError, match="must be absolute"):
|
|
planner._read_pinned_json(Path("evidence.json"), digest, label="test evidence")
|
|
with pytest.raises(planner.RehearsalProofError, match="externally supplied"):
|
|
planner._read_pinned_json(path.resolve(), "0" * 64, label="test evidence")
|
|
|
|
|
|
def test_output_is_atomically_exclusive_and_never_overwrites(tmp_path: Path) -> None:
|
|
output = (tmp_path / "plan.json").resolve()
|
|
planner._write_exclusive_output(output, b'{"plan":"one"}\n')
|
|
assert output.read_bytes() == b'{"plan":"one"}\n'
|
|
with pytest.raises(planner.RehearsalProofError, match="must not already exist"):
|
|
planner._write_exclusive_output(output, b'{"plan":"two"}\n')
|
|
assert output.read_bytes() == b'{"plan":"one"}\n'
|
|
assert list(tmp_path.glob(".*.tmp")) == []
|
|
|
|
|
|
def test_output_rejects_symlink_parent(tmp_path: Path, monkeypatch) -> None:
|
|
real = tmp_path / "real"
|
|
real.mkdir()
|
|
link = tmp_path / "linked"
|
|
try:
|
|
os.symlink(real, link, target_is_directory=True)
|
|
except (OSError, NotImplementedError):
|
|
link.mkdir()
|
|
original = os.lstat
|
|
|
|
def simulated_lstat(path):
|
|
result = original(path)
|
|
if Path(path).absolute() == link.absolute():
|
|
values = list(result)
|
|
values[0] = stat.S_IFLNK | 0o777
|
|
return os.stat_result(values)
|
|
return result
|
|
|
|
monkeypatch.setattr(os, "lstat", simulated_lstat)
|
|
with pytest.raises(planner.RehearsalProofError, match="symbolic links"):
|
|
planner._write_exclusive_output((link / "plan.json").absolute(), b"{}\n")
|
|
|
|
|
|
def test_cli_has_no_apply_and_planner_does_not_call_release_build_plan() -> None:
|
|
option_strings = {
|
|
option
|
|
for action in planner.build_parser()._actions
|
|
for option in action.option_strings
|
|
}
|
|
assert "--apply" not in option_strings
|
|
assert "--confirmation" not in option_strings
|
|
assert "--idempotency-key" not in option_strings
|
|
source = (REPO_ROOT / "yuqei-infra/deploy/plan-rollback-rehearsal.py").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
assert "release_package.build_plan" not in source
|
|
|
|
|
|
def test_plan_entrypoint_rejects_relative_release_paths_before_validation(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
with pytest.raises(planner.RehearsalProofError, match="current package directory"):
|
|
planner.plan_from_files(
|
|
current_package_dir=Path("current-package"),
|
|
current_package_sha256="a" * 64,
|
|
predecessor_package_dir=(tmp_path / "predecessor-package").resolve(),
|
|
predecessor_package_sha256="b" * 64,
|
|
predecessor_repo_root=(tmp_path / "predecessor-repo").resolve(),
|
|
evidence_file=(tmp_path / "evidence.json").resolve(),
|
|
evidence_sha256="c" * 64,
|
|
keyring_file=(tmp_path / "keyring.json").resolve(),
|
|
keyring_sha256="d" * 64,
|
|
compose_gate_evidence_file=(tmp_path / "compose-gate.json").resolve(),
|
|
compose_gate_evidence_sha256="e" * 64,
|
|
trusted_root_public_key_file=(tmp_path / "root.pub").resolve(),
|
|
env_file=(tmp_path / "production.env").resolve(),
|
|
)
|
|
|
|
|
|
def test_plan_entrypoint_uses_each_package_own_verified_validator(
|
|
tmp_path: Path,
|
|
monkeypatch,
|
|
) -> None:
|
|
current_package = (tmp_path / "current-package").resolve()
|
|
predecessor_package = (tmp_path / "predecessor-package").resolve()
|
|
current_repo = (tmp_path / "current-repo").resolve()
|
|
predecessor_repo = (tmp_path / "predecessor-repo").resolve()
|
|
bootstrap_calls: list[tuple[Path, str, Path]] = []
|
|
load_calls: list[tuple[object, Path]] = []
|
|
validation_calls: list[tuple[Path, Path, str]] = []
|
|
|
|
def bootstrap(*, current_package_dir, current_package_sha256, repo_root):
|
|
bootstrap_calls.append(
|
|
(current_package_dir, current_package_sha256, repo_root)
|
|
)
|
|
marker = repo_root.name
|
|
return {"marker": marker.encode()}, {"marker": marker}
|
|
|
|
def load(verified_sources, *, verified_repo_root=planner.REPO_ROOT):
|
|
marker = verified_sources["marker"].decode()
|
|
load_calls.append((verified_sources, verified_repo_root))
|
|
|
|
def validate(package_dir, *, expected_package_sha256, repo_root):
|
|
validation_calls.append(
|
|
(package_dir, repo_root, expected_package_sha256)
|
|
)
|
|
return SimpleNamespace(marker=marker)
|
|
|
|
return SimpleNamespace(validate_release_package=validate)
|
|
|
|
class StopAfterPackageValidation(RuntimeError):
|
|
pass
|
|
|
|
monkeypatch.setattr(planner, "_bootstrap_verified_tool_sources", bootstrap)
|
|
monkeypatch.setattr(planner, "_load_release_package_module", load)
|
|
monkeypatch.setattr(
|
|
planner,
|
|
"_read_pinned_json",
|
|
lambda *_args, **_kwargs: (_ for _ in ()).throw(
|
|
StopAfterPackageValidation()
|
|
),
|
|
)
|
|
|
|
with pytest.raises(StopAfterPackageValidation):
|
|
planner.plan_from_files(
|
|
current_package_dir=current_package,
|
|
current_package_sha256="a" * 64,
|
|
predecessor_package_dir=predecessor_package,
|
|
predecessor_package_sha256="b" * 64,
|
|
predecessor_repo_root=predecessor_repo,
|
|
evidence_file=(tmp_path / "evidence.json").resolve(),
|
|
evidence_sha256="c" * 64,
|
|
keyring_file=(tmp_path / "keyring.json").resolve(),
|
|
keyring_sha256="d" * 64,
|
|
compose_gate_evidence_file=(tmp_path / "compose-gate.json").resolve(),
|
|
compose_gate_evidence_sha256="e" * 64,
|
|
trusted_root_public_key_file=(tmp_path / "root.pub").resolve(),
|
|
env_file=(tmp_path / "production.env").resolve(),
|
|
repo_root=current_repo,
|
|
)
|
|
|
|
assert bootstrap_calls == [
|
|
(current_package, "a" * 64, current_repo),
|
|
(predecessor_package, "b" * 64, predecessor_repo),
|
|
]
|
|
assert [item[1] for item in load_calls] == [current_repo, predecessor_repo]
|
|
assert validation_calls == [
|
|
(current_package, current_repo, "a" * 64),
|
|
(predecessor_package, predecessor_repo, "b" * 64),
|
|
]
|