test(gitops): close silent release production contract

This commit is contained in:
chatgpt-gitops-bot
2026-07-24 01:12:22 +08:00
parent e0b48dae24
commit 9cc04bf146
4 changed files with 619 additions and 5 deletions
@@ -2030,11 +2030,27 @@ class BaselineRegistryFinalizeController(CapabilityController):
adapter=adapter,
)
def validate_result(self, approval, result):
required = {
"verified": True,
"tag_verified": True,
"remote_peeled_target_oid": approval.source_commit,
"gitea_api_peeled_target_oid": approval.source_commit,
"gitea_listing_visible": True,
"evidence_closure_oid": approval.evidence_closure_oid,
"approval_plan_id": approval.approval_plan_id,
}
if any(result.get(key) != value for key, value in required.items()):
raise CapabilityAuthorizationError(
"finalization is blocked until exact tag proof is verified"
)
class BaselineController(BaselineRegistryFinalizeController):
"""Narrow Baseline half of the governed finalization capability."""
def validate_result(self, approval, result):
super().validate_result(approval, result)
required = {
"verified": True,
"baseline_target_oid": approval.source_commit,
@@ -2051,6 +2067,7 @@ class RegistryController(BaselineRegistryFinalizeController):
"""Narrow Registry half of the governed finalization capability."""
def validate_result(self, approval, result):
super().validate_result(approval, result)
required = {
"verified": True,
"registry_target_oid": approval.source_commit,
@@ -0,0 +1,74 @@
# Silent Release Controller Production Contract
Status: review-only implementation. Production execution remains disabled.
## Trust boundary
The caller submits only release intent. A configured trusted builder owns ref
validation, build, tests, artifact and manifest hashing, package closure,
canary, rollback readiness, Git/Gitea consistency, and secret scanning. Caller
supplied success evidence is rejected.
The resulting immutable `ReleasePlan` contains deployment, tag, and
finalization plans. One Ed25519 owner signature binds its SHA-256 digest,
candidate evidence, policy digest, owner identity, trust domain, validity
window, evidence lineage, and three narrow capabilities.
## Durable execution
The Saga journal is the source of truth. Every record is sequence-numbered and
SHA-256 hash-chained from a fixed genesis hash. Snapshots are rebuildable
caches. Journal corruption, truncation, reordering, missing records, or hash
mismatch moves the release to `MANUAL_OWNER_REVIEW_REQUIRED`.
Capability execution uses an exclusive, leased claim. A second worker receives
`BUSY`; an expired claim is not reassigned until a capability-specific
side-effect probe reports that the operation was not applied. Runtime, tag,
and finalization remain separately claimed and recoverable.
## Execution order
```text
APPROVED
-> PRE_SWITCH_REVALIDATING
-> SWITCH_PENDING
-> SWITCHED
-> RUNTIME_VERIFYING
-> RUNTIME_VERIFIED
-> TAG_PENDING
-> TAG_VERIFIED
-> BASELINE_REGISTRY_PENDING
-> COMPLETED
```
The actual adapter configuration is regenerated before execution and must
match the signed plan. Trusted allowlists come from controlled policy, not the
request.
A failure before verified tag publication rolls the runtime back. A failure
after the exact tag is verified enters `RECONCILIATION_REQUIRED`; the tag is
never deleted or rewritten. A same-name/different-OID conflict enters
`MANUAL_OWNER_REVIEW_REQUIRED`. Registry/Baseline finalization requires exact
remote and Gitea tag proof, and Baseline is recorded last.
## Action and recovery contract
`submitOwnerApproval` verifies and durably stores the signed plan, then moves
the Saga to `APPROVED`. `startDeployment` loads that durable approval and
ignores caller-supplied approval content. Enabling execution without the
trusted builder, verifier, validator, policy, and orchestrator fails during
service construction.
The recovery worker scans incomplete journals and resumes through the same
claims and verification gates. It never automatically resumes a manual-review
Saga. In shadow mode it only reports state and performs no adapter side
effects.
## Governance
- `force_push`: forbidden
- `tag_rewrite`: forbidden
- `direct_main_push`: forbidden
- production execution default: disabled
- production tag operations in this change: not executed
- Baseline/Registry mutations in this change: not executed
+7 -5
View File
@@ -460,6 +460,10 @@ def test_finalize_controllers_require_governance_lineage(
target_field: verified.source_commit,
"evidence_closure_oid": verified.evidence_closure_oid,
"approval_plan_id": verified.approval_plan_id,
"tag_verified": True,
"remote_peeled_target_oid": verified.source_commit,
"gitea_api_peeled_target_oid": verified.source_commit,
"gitea_listing_visible": True,
}
controller = controller_type(
CapabilityLedger(Path(td) / "capabilities"),
@@ -728,12 +732,10 @@ def test_action_service_candidate_and_saga_are_durable():
def test_action_service_rejects_execution_without_signed_approval():
with tempfile.TemporaryDirectory() as td:
service = SilentReleaseActionService(
Path(td), execution_enabled=True
)
service.create_release_candidate(candidate().to_dict())
with pytest.raises(CapabilityAuthorizationError):
service.start_deployment("rc-hardening-001")
SilentReleaseActionService(
Path(td), execution_enabled=True
)
def test_deployment_plan_rejects_release_root_escape():
@@ -27,15 +27,30 @@ from silent_release_hardening import ( # noqa: E402
ApprovalRevocationStore,
DeploymentPlan,
FinalizationPlan,
HardenedTagController,
JournalIntegrityError,
PreSwitchValidator,
ReleaseOrchestrator,
ReleaseCandidate,
ReleasePlan,
ReleasePlanStore,
ReleaseRecoveryWorker,
ReleaseSagaStore,
RuntimeSwitchController,
SagaState,
SignedApprovalStore,
SignedApprovalVerifier,
SignedReleasePlanApproval,
SilentReleaseActionService,
TagPlan,
TagPublicationConflict,
TrustedCandidateBundle,
TrustedReleasePolicy,
StaleApprovalError,
)
from silent_release_controller import ( # noqa: E402
ReleaseCandidateStore,
ReleaseState,
)
@@ -658,3 +673,509 @@ def test_trusted_policy_rejects_request_defined_allowlists():
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 finalize(**kwargs):
calls.append("finalize")
if finalizer_adapter:
return finalizer_adapter(**kwargs)
return finalization_proof(plan)
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=__import__(
"silent_release_hardening"
).BaselineRegistryFinalizeController(ledger, 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",
"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 "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 "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())
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.COMPLETED
assert calls.count("tag_publish") == 1
assert calls.count("finalize") == 2
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,
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] == "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(),
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",
"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