tegwick
7b38f05eb2
Apply the user-directed Emergent Security guidance to close the redundant role gate. Add closed request/receipt validation bound to run, cluster, implementation, mode and authority reference; preserve platform enforcement and operation admission gates. Decision-Class: task-scoped-code-and-tests Delegation: explicit-user-task; Emergent-Security-A6 Evidence: 653-tests-passed; 1-skipped; focused-43-passed; ruff-clean Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a100de-2e1c-7e23-9ae9-2da0e0499411
129 lines
4.8 KiB
Python
129 lines
4.8 KiB
Python
"""Offline owner-handoff contract checks; no platform calls or credentials."""
|
|
from datetime import UTC, datetime, timedelta
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
from activity_core.recovery_contract import parse_recovery_request, validate_recovery_receipt
|
|
|
|
NOW = datetime(2026, 10, 3, 10, tzinfo=UTC)
|
|
|
|
|
|
def request(mode="apply"):
|
|
return parse_recovery_request({
|
|
"run_id": str(uuid4()), "mode": mode,
|
|
"cluster_uid": str(uuid4()), "implementation_sha256": "b" * 64,
|
|
"authority_ref": str(uuid4()) if mode == "apply" else None,
|
|
})
|
|
|
|
|
|
def receipt(req):
|
|
return {
|
|
"schema_id": "activity.platform-recovery-receipt.v1",
|
|
"run_id": str(req.run_id), "operation": req.operation, "target": req.target,
|
|
"cluster_uid": str(req.cluster_uid),
|
|
"mode": req.mode, "authority_ref": str(req.authority_ref) if req.authority_ref else None,
|
|
"outcome": "succeeded" if req.mode == "apply" else "planned",
|
|
"started_at": (NOW - timedelta(minutes=5)).isoformat(),
|
|
"finished_at": (NOW - timedelta(minutes=1)).isoformat(),
|
|
"validation_succeeded": req.mode == "apply",
|
|
"cleanup": "completed" if req.mode == "apply" else "not_created",
|
|
"evidence_sha256": "a" * 64, "implementation_sha256": "b" * 64,
|
|
}
|
|
|
|
|
|
def check(req, value):
|
|
return validate_recovery_receipt(req, value, observed_at=NOW, max_age=timedelta(hours=1))
|
|
|
|
|
|
@pytest.mark.parametrize("mode", ["apply", "dry_run"])
|
|
def test_exact_owner_receipt_preserves_outcome(mode):
|
|
req = request(mode)
|
|
result = check(req, receipt(req))
|
|
assert result.outcome == ("succeeded" if mode == "apply" else "planned")
|
|
|
|
|
|
@pytest.mark.parametrize("change", [
|
|
{"operation": "platform_root"}, {"target": "forgejo-db"},
|
|
{"command": "arbitrary"}, {"credential": "SECRET_CANARY"},
|
|
{"authority_ref": None}, {"mode": "unknown"},
|
|
])
|
|
def test_request_rejects_widening_and_missing_apply_reference_without_echo(change):
|
|
value = request().model_dump(mode="json")
|
|
value.update(change)
|
|
with pytest.raises(ValueError) as exc:
|
|
parse_recovery_request(value)
|
|
assert str(exc.value) == "invalid recovery request envelope"
|
|
assert exc.value.__suppress_context__
|
|
|
|
|
|
@pytest.mark.parametrize("change", [
|
|
{"run_id": str(uuid4())}, {"authority_ref": str(uuid4())},
|
|
{"cluster_uid": str(uuid4())}, {"implementation_sha256": "c" * 64},
|
|
{"target": "production"}, {"mode": "dry_run"},
|
|
{"validation_succeeded": False}, {"validation_succeeded": "true"},
|
|
{"cleanup": "failed"}, {"cleanup": "unknown"},
|
|
{"finished_at": (NOW + timedelta(seconds=1)).isoformat()},
|
|
{"started_at": NOW.isoformat()},
|
|
{"finished_at": "2026-10-03T09:59:00"},
|
|
{"evidence_sha256": "invalid"},
|
|
{"stderr": "SECRET_CANARY"}, {"evidence_sha256": "SECRET_CANARY"},
|
|
])
|
|
def test_receipt_rejects_mismatch_invalid_proof_and_sensitive_fields(change):
|
|
req = request()
|
|
value = receipt(req)
|
|
value.update(change)
|
|
with pytest.raises(ValueError) as exc:
|
|
check(req, value)
|
|
assert "SECRET_CANARY" not in str(exc.value)
|
|
|
|
|
|
def test_stale_receipt_is_not_freshened_by_scheduler():
|
|
req = request()
|
|
value = receipt(req)
|
|
value.update(started_at=(NOW - timedelta(hours=3)).isoformat(),
|
|
finished_at=(NOW - timedelta(hours=2)).isoformat())
|
|
with pytest.raises(ValueError, match="stale"):
|
|
check(req, value)
|
|
|
|
|
|
@pytest.mark.parametrize("outcome", ["failed", "rejected", "unknown"])
|
|
def test_failure_and_ambiguity_remain_visible(outcome):
|
|
req = request()
|
|
value = receipt(req)
|
|
value.update(outcome=outcome, validation_succeeded=False, cleanup="unknown")
|
|
assert check(req, value).outcome == outcome
|
|
|
|
|
|
def test_dry_run_cannot_claim_a_restore():
|
|
req = request("dry_run")
|
|
value = receipt(req)
|
|
value.update(outcome="succeeded", validation_succeeded=True, cleanup="completed")
|
|
with pytest.raises(ValueError):
|
|
check(req, value)
|
|
|
|
|
|
@pytest.mark.parametrize("observed_at,max_age", [
|
|
(NOW.replace(tzinfo=None), timedelta(hours=1)),
|
|
(NOW, timedelta(0)), (NOW, timedelta(seconds=-1)),
|
|
])
|
|
def test_freshness_policy_must_be_explicit_and_valid(observed_at, max_age):
|
|
req = request()
|
|
with pytest.raises(ValueError):
|
|
validate_recovery_receipt(req, receipt(req), observed_at=observed_at, max_age=max_age)
|
|
|
|
|
|
def test_receipt_requires_an_explicit_supported_schema():
|
|
req = request()
|
|
for schema in (None, "activity.platform-recovery-receipt.v2"):
|
|
value = receipt(req)
|
|
if schema is None:
|
|
del value["schema_id"]
|
|
else:
|
|
value["schema_id"] = schema
|
|
with pytest.raises(ValueError, match="invalid recovery receipt envelope"):
|
|
check(req, value)
|
|
|
|
|
|
def test_offline_contract_does_not_register_an_operation():
|
|
from activity_core.bounded_operations import BOUNDED_OPERATION_REGISTRY
|
|
assert "apps_pg_restore_evidence" not in BOUNDED_OPERATION_REGISTRY
|