1251 lines
39 KiB
Python
1251 lines
39 KiB
Python
from __future__ import annotations
|
|
|
|
import dataclasses
|
|
import json
|
|
import multiprocessing
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from cryptography.hazmat.primitives import serialization
|
|
from cryptography.hazmat.primitives.asymmetric.ed25519 import (
|
|
Ed25519PrivateKey,
|
|
)
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT / "active-runtime" / "app"))
|
|
|
|
from silent_release_hardening import ( # noqa: E402
|
|
CapabilityAuthorizationError,
|
|
BaselineController,
|
|
CapabilityController,
|
|
CapabilityLedger,
|
|
ClaimProbeResult,
|
|
ApprovalKeyRing,
|
|
ApprovalRevocationStore,
|
|
DeploymentPlan,
|
|
FinalizationPlan,
|
|
GovernedFinalizationCoordinator,
|
|
HardenedTagController,
|
|
JournalIntegrityError,
|
|
PreSwitchValidator,
|
|
ReleaseOrchestrator,
|
|
ReleaseCandidate,
|
|
ReleasePlan,
|
|
ReleasePlanStore,
|
|
ReleaseRecoveryWorker,
|
|
RegistryController,
|
|
ReleaseSagaStore,
|
|
RuntimeSwitchController,
|
|
SagaState,
|
|
SignedApprovalStore,
|
|
SignedApprovalVerifier,
|
|
SignedReleasePlanApproval,
|
|
SilentReleaseActionService,
|
|
TagPlan,
|
|
TagPublicationConflict,
|
|
TrustedCandidateBundle,
|
|
TrustedReleasePolicy,
|
|
StaleApprovalError,
|
|
)
|
|
from silent_release_controller import ( # noqa: E402
|
|
ReleaseCandidateStore,
|
|
ReleaseState,
|
|
)
|
|
|
|
|
|
COMMIT = "1" * 40
|
|
TREE = "2" * 40
|
|
ARTIFACT = "3" * 64
|
|
MANIFEST = "4" * 64
|
|
PLAN = "5" * 64
|
|
POLICY = "6" * 64
|
|
EVIDENCE = "7" * 40
|
|
OWNER_RECORD = "8" * 40
|
|
TAG_OID = "9" * 40
|
|
BASELINE = "a" * 64
|
|
REGISTRY = "b" * 64
|
|
TAG_REFSET = "c" * 64
|
|
PACKAGE = "d" * 64
|
|
SYSTEMD = "e" * 64
|
|
ABSENT = "f" * 64
|
|
|
|
|
|
def approval() -> SignedReleasePlanApproval:
|
|
return SignedReleasePlanApproval(
|
|
release_plan_id="release-plan-contract-001",
|
|
candidate_id="rc-contract-001",
|
|
source_commit=COMMIT,
|
|
tree_oid=TREE,
|
|
artifact_sha256=ARTIFACT,
|
|
manifest_sha256=MANIFEST,
|
|
deployment_plan_sha256=PLAN,
|
|
policy_sha256=POLICY,
|
|
approval_plan_id="owner-plan-contract-001",
|
|
evidence_closure_oid=EVIDENCE,
|
|
owner_approval_record_oid=OWNER_RECORD,
|
|
capabilities=(
|
|
"runtime_switch",
|
|
"tag_publish",
|
|
"baseline_registry_finalize",
|
|
),
|
|
expires_at="2099-01-01T00:00:00Z",
|
|
nonce="nonce-contract-000000001",
|
|
key_id="owner-key-contract-1",
|
|
signature="test-signature",
|
|
)
|
|
|
|
|
|
def candidate() -> ReleaseCandidate:
|
|
return ReleaseCandidate(
|
|
candidate_id="rc-contract-001",
|
|
source_commit=COMMIT,
|
|
tree_oid=TREE,
|
|
artifact_sha256=ARTIFACT,
|
|
manifest_sha256=MANIFEST,
|
|
)
|
|
|
|
|
|
def deployment_plan() -> DeploymentPlan:
|
|
return DeploymentPlan(
|
|
service="aios-gitops-model-api",
|
|
release_path="/data/aios-gitops/runtime/releases/" + COMMIT,
|
|
release_root="/data/aios-gitops/runtime/releases",
|
|
cutover_link=(
|
|
"/etc/systemd/system/aios-gitops-model-api.service.d/"
|
|
"production.conf"
|
|
),
|
|
dropin_path=(
|
|
"/data/aios-gitops/runtime/releases/"
|
|
+ COMMIT
|
|
+ "/deployment/production.conf"
|
|
),
|
|
dropin_sha256=SYSTEMD,
|
|
systemd_unit_sha256=SYSTEMD,
|
|
rollback_path="/usr/libexec/aios-gitops-runtime-anchor",
|
|
rollback_sha256="0" * 64,
|
|
source_commit=COMMIT,
|
|
tree_oid=TREE,
|
|
artifact_sha256=ARTIFACT,
|
|
manifest_sha256=MANIFEST,
|
|
package_closure_sha256=PACKAGE,
|
|
authoritative_git_dir="/data/aios-gitops/repo.git",
|
|
repository_identity="aios/aios",
|
|
writer_expected=False,
|
|
baseline_before_sha256=BASELINE,
|
|
registry_before_sha256=REGISTRY,
|
|
production_tag_refset_before_sha256=TAG_REFSET,
|
|
health_contract_version="silent-release-health/v1",
|
|
timeout_policy={"health_seconds": 30},
|
|
retry_policy={"health_attempts": 3},
|
|
policy_sha256=POLICY,
|
|
health_contract={
|
|
"runtime_commit": COMMIT,
|
|
"writer": False,
|
|
"repository_identity": "aios/aios",
|
|
},
|
|
)
|
|
|
|
|
|
def release_plan() -> ReleasePlan:
|
|
return ReleasePlan(
|
|
deployment=deployment_plan(),
|
|
tag=TagPlan(
|
|
version="V2R1C3",
|
|
tag_type="annotated",
|
|
expected_tag_oid=TAG_OID,
|
|
expected_target_oid=COMMIT,
|
|
evidence_closure_oid=EVIDENCE,
|
|
approval_plan_id="owner-plan-contract-001",
|
|
repository_identity="aios/aios",
|
|
gitea_owner="aios",
|
|
gitea_repository="aios",
|
|
current_remote_tag_oid=None,
|
|
remote_absent_evidence_sha256=ABSENT,
|
|
verification_contract="git+api+listing/v1",
|
|
force_push_allowed=False,
|
|
tag_rewrite_allowed=False,
|
|
),
|
|
finalization=FinalizationPlan(
|
|
version="V2R1C3",
|
|
expected_tag_oid=TAG_OID,
|
|
expected_target_oid=COMMIT,
|
|
expected_previous_baseline_sha256=BASELINE,
|
|
registry_expected_record={
|
|
"version": "V2R1C3",
|
|
"target_oid": COMMIT,
|
|
},
|
|
baseline_expected_record={
|
|
"version": "V2R1C3",
|
|
"target_oid": COMMIT,
|
|
},
|
|
final_consistency_contract="registry=baseline=tag=v1",
|
|
),
|
|
policy_sha256=POLICY,
|
|
)
|
|
|
|
|
|
def signing_material():
|
|
private = Ed25519PrivateKey.generate()
|
|
public = private.public_key().public_bytes(
|
|
serialization.Encoding.Raw,
|
|
serialization.PublicFormat.Raw,
|
|
)
|
|
return private, public
|
|
|
|
|
|
def strict_approval(
|
|
plan: ReleasePlan | None = None,
|
|
*,
|
|
issued_at: str = "2098-12-31T23:00:00Z",
|
|
not_before: str = "2098-12-31T23:00:00Z",
|
|
expires_at: str = "2099-01-01T00:00:00Z",
|
|
):
|
|
plan = plan or release_plan()
|
|
private, public = signing_material()
|
|
value = dataclasses.replace(
|
|
approval(),
|
|
deployment_plan_sha256=plan.deployment.sha256,
|
|
owner_identity="owner:imruyi",
|
|
trust_domain="aios.production",
|
|
issued_at=issued_at,
|
|
not_before=not_before,
|
|
expires_at=expires_at,
|
|
release_plan_sha256=plan.sha256,
|
|
signature="",
|
|
)
|
|
signature = private.sign(value.canonical_bytes())
|
|
value = dataclasses.replace(
|
|
value,
|
|
signature=__import__("base64").b64encode(signature).decode(),
|
|
)
|
|
return value, public
|
|
|
|
|
|
def _claim_worker(root: str, start, queue) -> None:
|
|
marker = Path(root) / "adapter-calls.log"
|
|
|
|
def adapter():
|
|
with marker.open("a", encoding="utf-8") as output:
|
|
output.write(f"{os.getpid()}\n")
|
|
output.flush()
|
|
os.fsync(output.fileno())
|
|
time.sleep(0.05)
|
|
return {"verified": True}
|
|
|
|
controller = CapabilityController(
|
|
capability="runtime_switch",
|
|
ledger=CapabilityLedger(Path(root) / "ledger"),
|
|
adapter=adapter,
|
|
executor_id=f"worker-{os.getpid()}",
|
|
lease_seconds=10,
|
|
)
|
|
start.wait()
|
|
result = controller.execute(
|
|
approval(), operation="runtime-switch-process-race"
|
|
)
|
|
queue.put(result["status"])
|
|
|
|
|
|
def test_journal_is_hash_chained_and_snapshot_tracks_head(tmp_path):
|
|
store = ReleaseSagaStore(tmp_path / "sagas")
|
|
saga = store.create("rc-contract-001", "release-plan-contract-001")
|
|
saga = store.compare_and_set(
|
|
saga.candidate_id,
|
|
expected_revision=saga.revision,
|
|
expected_states={SagaState.PREPARING},
|
|
target=SagaState.VERIFYING,
|
|
event="verification_started",
|
|
)
|
|
records = store.verify_journal(saga.candidate_id)
|
|
assert [record["sequence"] for record in records] == [1, 2]
|
|
assert records[0]["previous_hash"] == "0" * 64
|
|
assert records[1]["previous_hash"] == records[0]["event_hash"]
|
|
assert saga.journal_sequence == 2
|
|
assert saga.journal_head_hash == records[-1]["event_hash"]
|
|
|
|
|
|
def test_journal_single_line_tamper_enters_manual_review(tmp_path):
|
|
store = ReleaseSagaStore(tmp_path / "sagas")
|
|
saga = store.create("rc-contract-001", "release-plan-contract-001")
|
|
journal = tmp_path / "sagas" / "rc-contract-001.events.jsonl"
|
|
record = json.loads(journal.read_text(encoding="utf-8").splitlines()[0])
|
|
record["payload"]["state"] = SagaState.COMPLETED.value
|
|
journal.write_text(json.dumps(record) + "\n", encoding="utf-8")
|
|
recovered = store.load(saga.candidate_id)
|
|
assert recovered.state == SagaState.MANUAL_OWNER_REVIEW_REQUIRED
|
|
assert recovered.journal_integrity is False
|
|
|
|
|
|
def test_journal_truncation_is_not_silently_accepted(tmp_path):
|
|
store = ReleaseSagaStore(tmp_path / "sagas")
|
|
saga = store.create("rc-contract-001", "release-plan-contract-001")
|
|
store.compare_and_set(
|
|
saga.candidate_id,
|
|
expected_revision=0,
|
|
expected_states={SagaState.PREPARING},
|
|
target=SagaState.VERIFYING,
|
|
event="verification_started",
|
|
)
|
|
journal = tmp_path / "sagas" / "rc-contract-001.events.jsonl"
|
|
journal.write_text(
|
|
journal.read_text(encoding="utf-8").splitlines()[0] + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
recovered = store.load(saga.candidate_id)
|
|
assert recovered.state == SagaState.MANUAL_OWNER_REVIEW_REQUIRED
|
|
assert recovered.journal_integrity is False
|
|
|
|
|
|
def test_journal_reordering_enters_manual_review(tmp_path):
|
|
store = ReleaseSagaStore(tmp_path / "sagas")
|
|
saga = store.create("rc-contract-001", "release-plan-contract-001")
|
|
saga = store.compare_and_set(
|
|
saga.candidate_id,
|
|
expected_revision=0,
|
|
expected_states={SagaState.PREPARING},
|
|
target=SagaState.VERIFYING,
|
|
event="verification_started",
|
|
)
|
|
store.compare_and_set(
|
|
saga.candidate_id,
|
|
expected_revision=1,
|
|
expected_states={SagaState.VERIFYING},
|
|
target=SagaState.READY_FOR_APPROVAL,
|
|
event="verification_passed",
|
|
)
|
|
journal = tmp_path / "sagas" / "rc-contract-001.events.jsonl"
|
|
lines = journal.read_text(encoding="utf-8").splitlines()
|
|
journal.write_text(
|
|
"\n".join([lines[0], lines[2], lines[1]]) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
assert (
|
|
store.load(saga.candidate_id).state
|
|
== SagaState.MANUAL_OWNER_REVIEW_REQUIRED
|
|
)
|
|
|
|
|
|
def test_journal_missing_middle_line_enters_manual_review(tmp_path):
|
|
store = ReleaseSagaStore(tmp_path / "sagas")
|
|
saga = store.create("rc-contract-001", "release-plan-contract-001")
|
|
saga = store.compare_and_set(
|
|
saga.candidate_id,
|
|
expected_revision=0,
|
|
expected_states={SagaState.PREPARING},
|
|
target=SagaState.VERIFYING,
|
|
event="verification_started",
|
|
)
|
|
store.compare_and_set(
|
|
saga.candidate_id,
|
|
expected_revision=1,
|
|
expected_states={SagaState.VERIFYING},
|
|
target=SagaState.READY_FOR_APPROVAL,
|
|
event="verification_passed",
|
|
)
|
|
journal = tmp_path / "sagas" / "rc-contract-001.events.jsonl"
|
|
lines = journal.read_text(encoding="utf-8").splitlines()
|
|
journal.write_text(
|
|
"\n".join([lines[0], lines[2]]) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
assert (
|
|
store.load(saga.candidate_id).state
|
|
== SagaState.MANUAL_OWNER_REVIEW_REQUIRED
|
|
)
|
|
|
|
|
|
def test_crash_after_journal_before_snapshot_recovers_from_journal(tmp_path):
|
|
crashed = {"done": False}
|
|
|
|
def hook(event):
|
|
if event["payload"]["event"] == "verification_started":
|
|
crashed["done"] = True
|
|
raise RuntimeError("simulated crash after journal fsync")
|
|
|
|
root = tmp_path / "sagas"
|
|
store = ReleaseSagaStore(root, after_journal_hook=hook)
|
|
saga = store.create("rc-contract-001", "release-plan-contract-001")
|
|
with pytest.raises(RuntimeError, match="simulated crash"):
|
|
store.compare_and_set(
|
|
saga.candidate_id,
|
|
expected_revision=0,
|
|
expected_states={SagaState.PREPARING},
|
|
target=SagaState.VERIFYING,
|
|
event="verification_started",
|
|
)
|
|
assert crashed["done"] is True
|
|
recovered = ReleaseSagaStore(root).load(saga.candidate_id)
|
|
assert recovered.state == SagaState.VERIFYING
|
|
assert recovered.revision == 1
|
|
assert recovered.journal_sequence == 2
|
|
|
|
|
|
def test_capability_claim_is_exclusive_and_busy(tmp_path):
|
|
ledger = CapabilityLedger(tmp_path / "ledger")
|
|
first = ledger.claim(
|
|
approval(),
|
|
"runtime_switch",
|
|
operation="runtime-switch-001",
|
|
executor_id="worker-a",
|
|
lease_seconds=60,
|
|
)
|
|
second = ledger.claim(
|
|
approval(),
|
|
"runtime_switch",
|
|
operation="runtime-switch-001",
|
|
executor_id="worker-b",
|
|
lease_seconds=60,
|
|
)
|
|
assert first["status"] == "CLAIMED"
|
|
assert second["status"] == "BUSY"
|
|
assert second["claim_id"] == first["claim_id"]
|
|
|
|
|
|
def test_capability_complete_requires_matching_claim_id(tmp_path):
|
|
ledger = CapabilityLedger(tmp_path / "ledger")
|
|
claim = ledger.claim(
|
|
approval(),
|
|
"runtime_switch",
|
|
operation="runtime-switch-001",
|
|
executor_id="worker-a",
|
|
lease_seconds=60,
|
|
)
|
|
with pytest.raises(CapabilityAuthorizationError):
|
|
ledger.complete(
|
|
approval(),
|
|
"runtime_switch",
|
|
operation="runtime-switch-001",
|
|
claim_id="wrong-claim",
|
|
evidence={"verified": True},
|
|
)
|
|
result = ledger.complete(
|
|
approval(),
|
|
"runtime_switch",
|
|
operation="runtime-switch-001",
|
|
claim_id=claim["claim_id"],
|
|
evidence={"verified": True},
|
|
)
|
|
assert result["status"] == "COMPLETED"
|
|
|
|
|
|
def test_expired_claim_requires_side_effect_probe_before_takeover(
|
|
tmp_path, monkeypatch
|
|
):
|
|
ledger = CapabilityLedger(tmp_path / "ledger")
|
|
first = ledger.claim(
|
|
approval(),
|
|
"tag_publish",
|
|
operation="tag-V2R1C3",
|
|
executor_id="worker-a",
|
|
lease_seconds=1,
|
|
now=100.0,
|
|
)
|
|
blocked = ledger.claim(
|
|
approval(),
|
|
"tag_publish",
|
|
operation="tag-V2R1C3",
|
|
executor_id="worker-b",
|
|
lease_seconds=10,
|
|
now=102.0,
|
|
)
|
|
assert blocked["status"] == "PROBE_REQUIRED"
|
|
takeover = ledger.claim(
|
|
approval(),
|
|
"tag_publish",
|
|
operation="tag-V2R1C3",
|
|
executor_id="worker-b",
|
|
lease_seconds=10,
|
|
now=102.0,
|
|
side_effect_probe=lambda: ClaimProbeResult.NOT_APPLIED,
|
|
)
|
|
assert takeover["status"] == "CLAIMED"
|
|
assert takeover["claim_id"] != first["claim_id"]
|
|
assert takeover["attempt"] == 2
|
|
|
|
|
|
def test_two_processes_execute_adapter_only_once(tmp_path):
|
|
context = multiprocessing.get_context("fork")
|
|
start = context.Event()
|
|
queue = context.Queue()
|
|
workers = [
|
|
context.Process(
|
|
target=_claim_worker,
|
|
args=(str(tmp_path), start, queue),
|
|
)
|
|
for _ in range(2)
|
|
]
|
|
for worker in workers:
|
|
worker.start()
|
|
start.set()
|
|
for worker in workers:
|
|
worker.join(timeout=10)
|
|
assert worker.exitcode == 0
|
|
statuses = sorted(queue.get(timeout=2) for _ in workers)
|
|
assert statuses in (
|
|
["BUSY", "EXECUTED"],
|
|
["EXECUTED", "IDEMPOTENT"],
|
|
)
|
|
calls = (tmp_path / "adapter-calls.log").read_text(
|
|
encoding="utf-8"
|
|
).splitlines()
|
|
assert len(calls) == 1
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"probe,expected",
|
|
[
|
|
(ClaimProbeResult.APPLIED_VERIFIED, "IDEMPOTENT"),
|
|
(
|
|
ClaimProbeResult.INDETERMINATE,
|
|
"RECONCILIATION_REQUIRED",
|
|
),
|
|
(
|
|
ClaimProbeResult.CONFLICT,
|
|
"MANUAL_OWNER_REVIEW_REQUIRED",
|
|
),
|
|
],
|
|
)
|
|
def test_expired_claim_probe_is_fail_closed(tmp_path, probe, expected):
|
|
ledger = CapabilityLedger(tmp_path / "ledger")
|
|
ledger.claim(
|
|
approval(),
|
|
"tag_publish",
|
|
operation="tag-V2R1C3",
|
|
executor_id="worker-a",
|
|
lease_seconds=1,
|
|
now=100.0,
|
|
)
|
|
result = ledger.claim(
|
|
approval(),
|
|
"tag_publish",
|
|
operation="tag-V2R1C3",
|
|
executor_id="worker-b",
|
|
lease_seconds=10,
|
|
now=102.0,
|
|
side_effect_probe=lambda: probe,
|
|
)
|
|
assert result["status"] == expected
|
|
|
|
|
|
def test_signed_approval_covers_complete_release_plan(tmp_path):
|
|
plan = release_plan()
|
|
signed, public = strict_approval(plan)
|
|
verifier = SignedApprovalVerifier(
|
|
ApprovalKeyRing({"owner-key-contract-1": public}),
|
|
ApprovalRevocationStore(tmp_path / "revocations"),
|
|
allowed_trust_domains={"aios.production"},
|
|
maximum_approval_lifetime_seconds=3600,
|
|
)
|
|
verified = verifier.verify_release_plan(
|
|
signed,
|
|
candidate(),
|
|
plan,
|
|
POLICY,
|
|
now="2098-12-31T23:30:00Z",
|
|
)
|
|
assert verified.release_plan_sha256 == plan.sha256
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"mutator",
|
|
[
|
|
lambda plan: dataclasses.replace(
|
|
plan, tag=dataclasses.replace(plan.tag, version="V2R1C4")
|
|
),
|
|
lambda plan: dataclasses.replace(
|
|
plan,
|
|
tag=dataclasses.replace(
|
|
plan.tag, expected_tag_oid="1" * 40
|
|
),
|
|
),
|
|
lambda plan: dataclasses.replace(
|
|
plan,
|
|
tag=dataclasses.replace(
|
|
plan.tag, expected_target_oid="2" * 40
|
|
),
|
|
),
|
|
lambda plan: dataclasses.replace(
|
|
plan,
|
|
tag=dataclasses.replace(
|
|
plan.tag, remote_absent_evidence_sha256="1" * 64
|
|
),
|
|
),
|
|
lambda plan: dataclasses.replace(
|
|
plan,
|
|
deployment=dataclasses.replace(
|
|
plan.deployment, service="substituted-service"
|
|
),
|
|
),
|
|
lambda plan: dataclasses.replace(
|
|
plan,
|
|
deployment=dataclasses.replace(
|
|
plan.deployment, cutover_link="/tmp/substituted.conf"
|
|
),
|
|
),
|
|
lambda plan: dataclasses.replace(
|
|
plan,
|
|
deployment=dataclasses.replace(
|
|
plan.deployment, release_root="/tmp/releases"
|
|
),
|
|
),
|
|
lambda plan: dataclasses.replace(
|
|
plan,
|
|
deployment=dataclasses.replace(
|
|
plan.deployment, baseline_before_sha256="1" * 64
|
|
),
|
|
),
|
|
lambda plan: dataclasses.replace(
|
|
plan,
|
|
deployment=dataclasses.replace(
|
|
plan.deployment, registry_before_sha256="2" * 64
|
|
),
|
|
),
|
|
],
|
|
)
|
|
def test_release_plan_mutation_invalidates_owner_signature(
|
|
tmp_path, mutator
|
|
):
|
|
original = release_plan()
|
|
signed, public = strict_approval(original)
|
|
verifier = SignedApprovalVerifier(
|
|
ApprovalKeyRing({"owner-key-contract-1": public}),
|
|
ApprovalRevocationStore(tmp_path / "revocations"),
|
|
allowed_trust_domains={"aios.production"},
|
|
maximum_approval_lifetime_seconds=3600,
|
|
)
|
|
with pytest.raises(CapabilityAuthorizationError):
|
|
verifier.verify_release_plan(
|
|
signed,
|
|
candidate(),
|
|
mutator(original),
|
|
POLICY,
|
|
now="2098-12-31T23:30:00Z",
|
|
)
|
|
|
|
|
|
def test_approval_not_before_and_maximum_ttl_are_enforced(tmp_path):
|
|
plan = release_plan()
|
|
future, public = strict_approval(
|
|
plan,
|
|
issued_at="2098-12-31T23:00:00Z",
|
|
not_before="2098-12-31T23:45:00Z",
|
|
expires_at="2099-01-01T00:00:00Z",
|
|
)
|
|
verifier = SignedApprovalVerifier(
|
|
ApprovalKeyRing({"owner-key-contract-1": public}),
|
|
ApprovalRevocationStore(tmp_path / "revocations"),
|
|
allowed_trust_domains={"aios.production"},
|
|
maximum_approval_lifetime_seconds=1800,
|
|
)
|
|
with pytest.raises(CapabilityAuthorizationError):
|
|
verifier.verify_release_plan(
|
|
future,
|
|
candidate(),
|
|
plan,
|
|
POLICY,
|
|
now="2098-12-31T23:30:00Z",
|
|
)
|
|
|
|
|
|
def test_trusted_policy_rejects_request_defined_allowlists():
|
|
policy = TrustedReleasePolicy(
|
|
services=frozenset({"aios-gitops-model-api"}),
|
|
release_roots=frozenset(
|
|
{"/data/aios-gitops/runtime/releases"}
|
|
),
|
|
cutover_links=frozenset(
|
|
{
|
|
"/etc/systemd/system/"
|
|
"aios-gitops-model-api.service.d/production.conf"
|
|
}
|
|
),
|
|
dropin_paths=frozenset({deployment_plan().dropin_path}),
|
|
authoritative_git_dirs=frozenset(
|
|
{"/data/aios-gitops/repo.git"}
|
|
),
|
|
repository_identities=frozenset({"aios/aios"}),
|
|
)
|
|
policy.validate_deployment(deployment_plan())
|
|
with pytest.raises(ValueError):
|
|
policy.validate_deployment(
|
|
dataclasses.replace(
|
|
deployment_plan(), service="request-chosen-service"
|
|
)
|
|
)
|
|
|
|
|
|
def trusted_policy() -> TrustedReleasePolicy:
|
|
plan = deployment_plan()
|
|
return TrustedReleasePolicy(
|
|
services=frozenset({plan.service}),
|
|
release_roots=frozenset({plan.release_root}),
|
|
cutover_links=frozenset({plan.cutover_link}),
|
|
dropin_paths=frozenset({plan.dropin_path}),
|
|
authoritative_git_dirs=frozenset(
|
|
{plan.authoritative_git_dir}
|
|
),
|
|
repository_identities=frozenset(
|
|
{plan.repository_identity}
|
|
),
|
|
)
|
|
|
|
|
|
def ready_candidate() -> ReleaseCandidate:
|
|
value = candidate()
|
|
value.tests = {"passed": True}
|
|
value.canary = {"passed": True}
|
|
value.rollback_ready = True
|
|
value.git_consistency = {"consistent": True}
|
|
value.gitea_consistency = {"consistent": True}
|
|
value.state = ReleaseState.READY_FOR_APPROVAL
|
|
return value
|
|
|
|
|
|
class PreSwitchProof:
|
|
def collect(self, candidate, plan):
|
|
return {
|
|
"source_commit": candidate.source_commit,
|
|
"tree_oid": candidate.tree_oid,
|
|
"artifact_sha256": candidate.artifact_sha256,
|
|
"manifest_sha256": candidate.manifest_sha256,
|
|
"deployment_plan_sha256": plan.sha256,
|
|
"policy_sha256": plan.policy_sha256,
|
|
"rollback_ready": True,
|
|
"git_refs_consistent": True,
|
|
"repository_identity": plan.repository_identity,
|
|
"writer_disabled": True,
|
|
}
|
|
|
|
|
|
def runtime_proof(candidate, plan):
|
|
return {
|
|
"verified": True,
|
|
"source_commit": candidate.source_commit,
|
|
"tree_oid": candidate.tree_oid,
|
|
"artifact_sha256": candidate.artifact_sha256,
|
|
"manifest_sha256": candidate.manifest_sha256,
|
|
"repository_identity": plan.deployment.repository_identity,
|
|
"authoritative_git_dir": (
|
|
plan.deployment.authoritative_git_dir
|
|
),
|
|
"writer_disabled": True,
|
|
}
|
|
|
|
|
|
def tag_proof(plan):
|
|
return {
|
|
"evidence_closure_oid": EVIDENCE,
|
|
"approval_plan_id": "owner-plan-contract-001",
|
|
"repository_identity": "aios/aios",
|
|
"remote_tag_oid": TAG_OID,
|
|
"remote_peeled_target_oid": COMMIT,
|
|
"gitea_api_tag_oid": TAG_OID,
|
|
"gitea_api_peeled_target_oid": COMMIT,
|
|
"gitea_listing_visible": True,
|
|
}
|
|
|
|
|
|
def finalization_proof(plan):
|
|
return {
|
|
"verified": True,
|
|
"tag_verified": True,
|
|
"remote_tag_oid": plan.tag.expected_tag_oid,
|
|
"remote_peeled_target_oid": plan.tag.expected_target_oid,
|
|
"gitea_api_peeled_target_oid": plan.tag.expected_target_oid,
|
|
"gitea_listing_visible": True,
|
|
"evidence_closure_oid": plan.tag.evidence_closure_oid,
|
|
"approval_plan_id": plan.tag.approval_plan_id,
|
|
"registry_record": (
|
|
plan.finalization.registry_expected_record
|
|
),
|
|
"baseline_record": (
|
|
plan.finalization.baseline_expected_record
|
|
),
|
|
"baseline_written_last": True,
|
|
}
|
|
|
|
|
|
def orchestrator_fixture(
|
|
tmp_path,
|
|
*,
|
|
runtime_verify=runtime_proof,
|
|
runtime_rollback=None,
|
|
runtime_adapter=None,
|
|
tag_adapter=None,
|
|
finalizer_adapter=None,
|
|
):
|
|
root = tmp_path / "contract"
|
|
candidates = ReleaseCandidateStore(root / "candidates")
|
|
plans = ReleasePlanStore(root / "plans")
|
|
approvals = SignedApprovalStore(root / "approvals")
|
|
sagas = ReleaseSagaStore(root / "sagas")
|
|
value = ready_candidate()
|
|
plan = release_plan()
|
|
signed, public = strict_approval(
|
|
plan,
|
|
issued_at="2026-01-01T00:00:00Z",
|
|
not_before="2026-01-01T00:00:00Z",
|
|
expires_at="2099-01-01T00:00:00Z",
|
|
)
|
|
candidates.save(value)
|
|
plans.save(value.candidate_id, plan)
|
|
approvals.save(value.candidate_id, signed)
|
|
saga = sagas.create(value.candidate_id, signed.release_plan_id)
|
|
saga = sagas.compare_and_set(
|
|
value.candidate_id,
|
|
expected_revision=saga.revision,
|
|
expected_states={SagaState.PREPARING},
|
|
target=SagaState.READY_FOR_APPROVAL,
|
|
event="candidate_ready",
|
|
)
|
|
sagas.compare_and_set(
|
|
value.candidate_id,
|
|
expected_revision=saga.revision,
|
|
expected_states={SagaState.READY_FOR_APPROVAL},
|
|
target=SagaState.APPROVED,
|
|
event="approval_accepted",
|
|
)
|
|
calls = []
|
|
|
|
def runtime_switch(**kwargs):
|
|
calls.append("runtime_switch")
|
|
if runtime_adapter:
|
|
return runtime_adapter(**kwargs)
|
|
return {"verified": True}
|
|
|
|
def runtime_check(**kwargs):
|
|
calls.append("runtime_verify")
|
|
return runtime_verify(**kwargs)
|
|
|
|
def rollback(**kwargs):
|
|
calls.append("rollback")
|
|
if runtime_rollback:
|
|
return runtime_rollback(**kwargs)
|
|
return {"verified": True}
|
|
|
|
def publish(**kwargs):
|
|
calls.append("tag_publish")
|
|
if tag_adapter:
|
|
return tag_adapter(**kwargs)
|
|
return tag_proof(plan)
|
|
|
|
def registry_finalize(**kwargs):
|
|
calls.append("registry_finalize")
|
|
return {
|
|
**finalization_proof(plan),
|
|
"registry_target_oid": COMMIT,
|
|
}
|
|
|
|
def baseline_finalize(**kwargs):
|
|
calls.append("baseline_finalize")
|
|
if finalizer_adapter:
|
|
return finalizer_adapter(**kwargs)
|
|
return {
|
|
**finalization_proof(plan),
|
|
"baseline_target_oid": COMMIT,
|
|
}
|
|
|
|
ledger = CapabilityLedger(root / "claims")
|
|
verifier = SignedApprovalVerifier(
|
|
ApprovalKeyRing({"owner-key-contract-1": public}),
|
|
ApprovalRevocationStore(root / "revocations"),
|
|
allowed_trust_domains={"aios.production"},
|
|
maximum_approval_lifetime_seconds=10**10,
|
|
)
|
|
orchestrator = ReleaseOrchestrator(
|
|
candidates=candidates,
|
|
plans=plans,
|
|
approvals=approvals,
|
|
sagas=sagas,
|
|
approval_verifier=verifier,
|
|
trusted_policy=trusted_policy(),
|
|
actual_plan_factory=lambda candidate: plan,
|
|
pre_switch_validator=PreSwitchValidator(PreSwitchProof()),
|
|
runtime_switch=RuntimeSwitchController(
|
|
ledger, runtime_switch
|
|
),
|
|
runtime_verify=runtime_check,
|
|
runtime_rollback=rollback,
|
|
tag_publish=HardenedTagController(
|
|
ledger,
|
|
publish,
|
|
expected_repository="aios/aios",
|
|
),
|
|
finalizer=GovernedFinalizationCoordinator(
|
|
RegistryController(ledger, registry_finalize),
|
|
BaselineController(ledger, baseline_finalize),
|
|
),
|
|
)
|
|
return orchestrator, calls, signed, plan, sagas
|
|
|
|
|
|
def test_orchestrator_executes_fixed_order_and_baseline_last(tmp_path):
|
|
orchestrator, calls, _, _, sagas = orchestrator_fixture(tmp_path)
|
|
result = orchestrator.execute("rc-contract-001")
|
|
assert result["state"] == SagaState.COMPLETED.value
|
|
assert calls == [
|
|
"runtime_switch",
|
|
"runtime_verify",
|
|
"tag_publish",
|
|
"registry_finalize",
|
|
"baseline_finalize",
|
|
]
|
|
assert sagas.load("rc-contract-001").state == SagaState.COMPLETED
|
|
|
|
|
|
def test_runtime_verification_failure_rolls_back_before_tag(tmp_path):
|
|
def invalid_runtime(candidate, plan):
|
|
return {"verified": False}
|
|
|
|
orchestrator, calls, _, _, _ = orchestrator_fixture(
|
|
tmp_path, runtime_verify=invalid_runtime
|
|
)
|
|
result = orchestrator.execute("rc-contract-001")
|
|
assert result["state"] == SagaState.ROLLED_BACK.value
|
|
assert calls == ["runtime_switch", "runtime_verify", "rollback"]
|
|
assert "tag_publish" not in calls
|
|
assert "registry_finalize" not in calls
|
|
assert "baseline_finalize" not in calls
|
|
|
|
|
|
def test_tag_conflict_stops_in_manual_owner_review(tmp_path):
|
|
def conflict(**kwargs):
|
|
raise TagPublicationConflict("same name different OID")
|
|
|
|
orchestrator, calls, _, _, _ = orchestrator_fixture(
|
|
tmp_path, tag_adapter=conflict
|
|
)
|
|
result = orchestrator.execute("rc-contract-001")
|
|
assert result["state"] == SagaState.MANUAL_OWNER_REVIEW_REQUIRED
|
|
assert calls[-1] == "tag_publish"
|
|
assert "registry_finalize" not in calls
|
|
assert "baseline_finalize" not in calls
|
|
|
|
|
|
def test_failure_after_tag_enters_reconciliation_without_rollback(
|
|
tmp_path,
|
|
):
|
|
failures = {"count": 0}
|
|
|
|
def fail_once(**kwargs):
|
|
failures["count"] += 1
|
|
if failures["count"] == 1:
|
|
raise RuntimeError("registry temporarily unavailable")
|
|
return {
|
|
**finalization_proof(release_plan()),
|
|
"baseline_target_oid": COMMIT,
|
|
}
|
|
|
|
orchestrator, calls, _, _, _ = orchestrator_fixture(
|
|
tmp_path, finalizer_adapter=fail_once
|
|
)
|
|
first = orchestrator.execute("rc-contract-001")
|
|
assert first["state"] == SagaState.RECONCILIATION_REQUIRED
|
|
assert "rollback" not in calls
|
|
second = orchestrator.execute("rc-contract-001")
|
|
assert second["state"] == SagaState.RECONCILIATION_REQUIRED
|
|
assert calls.count("tag_publish") == 1
|
|
assert calls.count("registry_finalize") == 1
|
|
assert calls.count("baseline_finalize") == 1
|
|
assert failures["count"] == 1
|
|
|
|
|
|
def test_actual_plan_change_blocks_before_any_adapter(tmp_path):
|
|
orchestrator, calls, _, _, _, = orchestrator_fixture(tmp_path)
|
|
orchestrator.actual_plan_factory = lambda candidate: dataclasses.replace(
|
|
release_plan(),
|
|
deployment=dataclasses.replace(
|
|
deployment_plan(), systemd_unit_sha256="1" * 64
|
|
),
|
|
)
|
|
with pytest.raises(StaleApprovalError):
|
|
orchestrator.execute("rc-contract-001")
|
|
assert calls == []
|
|
|
|
|
|
class TrustedBuilder:
|
|
def __init__(self, bundle):
|
|
self.bundle = bundle
|
|
self.calls = []
|
|
|
|
def build(self, intent):
|
|
self.calls.append(dict(intent))
|
|
return self.bundle
|
|
|
|
|
|
def test_action_rejects_caller_supplied_acceptance_evidence(tmp_path):
|
|
builder = TrustedBuilder(
|
|
TrustedCandidateBundle(
|
|
ready_candidate(),
|
|
release_plan(),
|
|
{name: True for name in TrustedCandidateBundle.REQUIRED_CHECKS},
|
|
)
|
|
)
|
|
service = SilentReleaseActionService(
|
|
tmp_path,
|
|
candidate_builder=builder,
|
|
trusted_policy=trusted_policy(),
|
|
policy_sha256=POLICY,
|
|
)
|
|
with pytest.raises(CapabilityAuthorizationError):
|
|
service.create_release_candidate(
|
|
{"source_commit": COMMIT, "tests": {"passed": True}}
|
|
)
|
|
assert builder.calls == []
|
|
|
|
|
|
def test_action_uses_trusted_builder_and_marks_ready(tmp_path):
|
|
bundle = TrustedCandidateBundle(
|
|
ready_candidate(),
|
|
release_plan(),
|
|
{name: True for name in TrustedCandidateBundle.REQUIRED_CHECKS},
|
|
)
|
|
builder = TrustedBuilder(bundle)
|
|
service = SilentReleaseActionService(
|
|
tmp_path,
|
|
candidate_builder=builder,
|
|
trusted_policy=trusted_policy(),
|
|
policy_sha256=POLICY,
|
|
)
|
|
result = service.create_release_candidate(
|
|
{"source_commit": COMMIT, "version": "V2R1C3"}
|
|
)
|
|
assert result["saga_state"] == SagaState.READY_FOR_APPROVAL
|
|
assert builder.calls == [
|
|
{"source_commit": COMMIT, "version": "V2R1C3"}
|
|
]
|
|
|
|
|
|
def test_execution_enabled_requires_every_production_component(tmp_path):
|
|
with pytest.raises(
|
|
CapabilityAuthorizationError,
|
|
match="production execution components are missing",
|
|
):
|
|
SilentReleaseActionService(
|
|
tmp_path, execution_enabled=True
|
|
)
|
|
|
|
|
|
def test_start_deployment_ignores_caller_approval_object(tmp_path):
|
|
class Orchestrator:
|
|
def __init__(self):
|
|
self.calls = []
|
|
|
|
def execute(self, candidate_id):
|
|
self.calls.append(candidate_id)
|
|
return {"status": "DURABLE_APPROVAL_LOADED"}
|
|
|
|
orchestrator = Orchestrator()
|
|
builder = TrustedBuilder(
|
|
TrustedCandidateBundle(
|
|
ready_candidate(),
|
|
release_plan(),
|
|
{name: True for name in TrustedCandidateBundle.REQUIRED_CHECKS},
|
|
)
|
|
)
|
|
service = SilentReleaseActionService(
|
|
tmp_path,
|
|
execution_enabled=True,
|
|
approval_verifier=object(),
|
|
candidate_builder=builder,
|
|
builder_attestation_verifier=object(),
|
|
trusted_policy=trusted_policy(),
|
|
pre_switch_validator=object(),
|
|
orchestrator=orchestrator,
|
|
policy_sha256=POLICY,
|
|
)
|
|
service.candidates.save(ready_candidate())
|
|
result = service.start_deployment(
|
|
"rc-contract-001",
|
|
approval=dataclasses.replace(
|
|
approval(), approval_plan_id="attacker-plan-001"
|
|
),
|
|
)
|
|
assert result["status"] == "DURABLE_APPROVAL_LOADED"
|
|
assert orchestrator.calls == ["rc-contract-001"]
|
|
|
|
|
|
def test_recovery_worker_shadow_reports_without_execution(tmp_path):
|
|
orchestrator, _, _, _, sagas = orchestrator_fixture(tmp_path)
|
|
|
|
class Bomb:
|
|
def execute(self, candidate_id):
|
|
raise AssertionError("shadow worker executed")
|
|
|
|
result = ReleaseRecoveryWorker(
|
|
sagas, Bomb(), shadow_mode=True
|
|
).run_once()
|
|
assert result["reports"][0]["status"] == "SHADOW_REPORTED"
|
|
|
|
|
|
def test_recovery_worker_resumes_nonterminal_saga(tmp_path):
|
|
orchestrator, calls, _, _, sagas = orchestrator_fixture(tmp_path)
|
|
result = ReleaseRecoveryWorker(
|
|
sagas, orchestrator, shadow_mode=False
|
|
).run_once()
|
|
assert result["reports"][0]["state"] == SagaState.COMPLETED
|
|
assert calls[-1] == "baseline_finalize"
|
|
|
|
|
|
def test_finalization_controller_rejects_unverified_tag(tmp_path):
|
|
controller = __import__(
|
|
"silent_release_hardening"
|
|
).BaselineRegistryFinalizeController(
|
|
CapabilityLedger(tmp_path / "claims"),
|
|
lambda: {
|
|
**finalization_proof(release_plan()),
|
|
"tag_verified": False,
|
|
},
|
|
)
|
|
with pytest.raises(CapabilityAuthorizationError):
|
|
controller.execute(
|
|
approval(),
|
|
operation="finalize-without-tag-verification",
|
|
)
|
|
|
|
|
|
def test_submit_owner_approval_durably_transitions_to_approved(tmp_path):
|
|
plan = release_plan()
|
|
bundle = TrustedCandidateBundle(
|
|
ready_candidate(),
|
|
plan,
|
|
{name: True for name in TrustedCandidateBundle.REQUIRED_CHECKS},
|
|
)
|
|
builder = TrustedBuilder(bundle)
|
|
signed, public = strict_approval(
|
|
plan,
|
|
issued_at="2026-01-01T00:00:00Z",
|
|
not_before="2026-01-01T00:00:00Z",
|
|
expires_at="2099-01-01T00:00:00Z",
|
|
)
|
|
verifier = SignedApprovalVerifier(
|
|
ApprovalKeyRing({"owner-key-contract-1": public}),
|
|
ApprovalRevocationStore(tmp_path / "revocations"),
|
|
allowed_trust_domains={"aios.production"},
|
|
maximum_approval_lifetime_seconds=10**10,
|
|
)
|
|
service = SilentReleaseActionService(
|
|
tmp_path / "service",
|
|
candidate_builder=builder,
|
|
trusted_policy=trusted_policy(),
|
|
approval_verifier=verifier,
|
|
policy_sha256=POLICY,
|
|
)
|
|
service.create_release_candidate(
|
|
{"source_commit": COMMIT, "version": "V2R1C3"}
|
|
)
|
|
result = service.submit_owner_approval(signed.to_dict())
|
|
assert result["saga_state"] == SagaState.APPROVED
|
|
assert (
|
|
service.approvals.load("rc-contract-001").fingerprint
|
|
== signed.fingerprint
|
|
)
|
|
|
|
|
|
def test_action_execution_true_runs_complete_orchestrator(tmp_path):
|
|
orchestrator, calls, _, _, _ = orchestrator_fixture(tmp_path)
|
|
service = SilentReleaseActionService(
|
|
tmp_path / "contract",
|
|
execution_enabled=True,
|
|
approval_verifier=orchestrator.approval_verifier,
|
|
candidate_builder=object(),
|
|
builder_attestation_verifier=object(),
|
|
trusted_policy=orchestrator.trusted_policy,
|
|
pre_switch_validator=orchestrator.pre_switch_validator,
|
|
orchestrator=orchestrator,
|
|
policy_sha256=POLICY,
|
|
)
|
|
result = service.start_deployment("rc-contract-001")
|
|
assert result["state"] == SagaState.COMPLETED
|
|
assert calls == [
|
|
"runtime_switch",
|
|
"runtime_verify",
|
|
"tag_publish",
|
|
"registry_finalize",
|
|
"baseline_finalize",
|
|
]
|
|
|
|
|
|
def test_recovery_worker_skips_manual_owner_review(tmp_path):
|
|
orchestrator, _, _, _, sagas = orchestrator_fixture(tmp_path)
|
|
saga = sagas.load("rc-contract-001")
|
|
sagas.compare_and_set(
|
|
saga.candidate_id,
|
|
expected_revision=saga.revision,
|
|
expected_states={SagaState.APPROVED},
|
|
target=SagaState.MANUAL_OWNER_REVIEW_REQUIRED,
|
|
event="manual_gate_test",
|
|
)
|
|
|
|
class Bomb:
|
|
def execute(self, candidate_id):
|
|
raise AssertionError("manual Saga was resumed")
|
|
|
|
result = ReleaseRecoveryWorker(
|
|
sagas, Bomb(), shadow_mode=False
|
|
).run_once()
|
|
assert result["reports"][0]["status"] == "SKIPPED_MANUAL_REVIEW"
|
|
|
|
|
|
def test_action_api_start_request_cannot_carry_approval():
|
|
source = (
|
|
ROOT / "active-runtime" / "app" / "aios_chatgpt_actions_api.py"
|
|
).read_text(encoding="utf-8")
|
|
schema = (
|
|
ROOT / "docs" / "gitops" / "openapi-chatgpt-actions.yaml"
|
|
).read_text(encoding="utf-8")
|
|
assert "data['approval']" not in source
|
|
start_schema = schema.split("StartDeploymentRequest:", 1)[1].split(
|
|
"paths:", 1
|
|
)[0]
|
|
assert "approval:" not in start_schema
|
|
assert "additionalProperties: false" in start_schema
|
|
|
|
|
|
def test_default_action_rejects_legacy_caller_evidence(tmp_path):
|
|
service = SilentReleaseActionService(tmp_path)
|
|
with pytest.raises(
|
|
CapabilityAuthorizationError,
|
|
match="trusted candidate builder is not configured",
|
|
):
|
|
service.create_release_candidate(candidate().to_dict())
|
|
|
|
|
|
def test_finalization_coordinator_writes_baseline_last(tmp_path):
|
|
calls = []
|
|
proof = finalization_proof(release_plan())
|
|
ledger = CapabilityLedger(tmp_path / "claims")
|
|
coordinator = GovernedFinalizationCoordinator(
|
|
RegistryController(
|
|
ledger,
|
|
lambda **kwargs: (
|
|
calls.append("registry")
|
|
or {
|
|
**proof,
|
|
"registry_target_oid": COMMIT,
|
|
}
|
|
),
|
|
),
|
|
BaselineController(
|
|
ledger,
|
|
lambda **kwargs: (
|
|
calls.append("baseline")
|
|
or {
|
|
**proof,
|
|
"baseline_target_oid": COMMIT,
|
|
}
|
|
),
|
|
),
|
|
)
|
|
result = coordinator.execute(
|
|
approval(),
|
|
operation="finalize-contract",
|
|
finalization_plan=release_plan().finalization,
|
|
)
|
|
assert result["status"] == "EXECUTED"
|
|
assert calls == ["registry", "baseline"]
|