tegwick
fd80975d37
Some checks failed
Use-case regression gate / usecase-regressions (push) Failing after 3m35s
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a17f16-d99c-7ea3-95f0-8212897fba6b
301 lines
10 KiB
Python
301 lines
10 KiB
Python
"""Exact independent generations survive primary age expiry and foreign objects."""
|
|
|
|
import copy
|
|
import importlib.util
|
|
from datetime import timedelta
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from tests.test_backup_health import NOW
|
|
from tests.test_backup_health import health_state as health_state
|
|
from tests.test_backup_health import monitor as monitor
|
|
|
|
|
|
@pytest.fixture
|
|
def reviewer():
|
|
path = Path(__file__).resolve().parents[1] / "scripts/review-backup-retention.py"
|
|
spec = importlib.util.spec_from_file_location("retention_review", path)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
@pytest.fixture
|
|
def evidence(reviewer, health_state):
|
|
_, receipt, mirrored = health_state
|
|
recovery = dict(
|
|
receipt,
|
|
status="application_restored_verified",
|
|
application_restore_verified=True,
|
|
application_data_images_and_recovery_tools_from_provider_kit=True,
|
|
)
|
|
recovery["protected_kit"] = {
|
|
"project_id": reviewer.health.PROJECT,
|
|
"bucket": reviewer.KIT,
|
|
"object_key": "kits/" + "a" * 64 + "/recovery-kit.tar",
|
|
"version_id": "kit-version",
|
|
"sha256": "a" * 64,
|
|
"bytes": 100,
|
|
"automatic_expiry": False,
|
|
}
|
|
recovery["protected_recovery_proof"] = {
|
|
"project_id": reviewer.health.PROJECT,
|
|
"bucket": reviewer.KIT,
|
|
"object_key": "kits/receipts/" + "b" * 64 + "/recovery-proof.json",
|
|
"version_id": "proof-version",
|
|
"sha256": "b" * 64,
|
|
"bytes": 80,
|
|
"automatic_expiry": False,
|
|
"independent_readback_verified": True,
|
|
}
|
|
roles = [
|
|
(
|
|
reviewer.PRIMARY,
|
|
receipt["object_key"],
|
|
receipt["version_id"],
|
|
receipt["ciphertext_sha256"],
|
|
receipt["ciphertext_bytes"],
|
|
),
|
|
(
|
|
reviewer.health.MIRROR_BUCKET,
|
|
mirrored["recovery_mirror"]["object_key"],
|
|
mirrored["recovery_mirror"]["version_id"],
|
|
receipt["ciphertext_sha256"],
|
|
receipt["ciphertext_bytes"],
|
|
),
|
|
]
|
|
for name in ["protected_kit", "protected_recovery_proof"]:
|
|
ref = recovery[name]
|
|
roles.append(
|
|
(ref["bucket"], ref["object_key"], ref["version_id"], ref["sha256"], ref["bytes"])
|
|
)
|
|
buckets = []
|
|
for bucket, prefix in [
|
|
(reviewer.PRIMARY, reviewer.health.PREFIX),
|
|
(reviewer.health.MIRROR_BUCKET, "archives/"),
|
|
(reviewer.KIT, "kits/"),
|
|
]:
|
|
rules = (
|
|
[
|
|
{
|
|
"ID": "existing-global-30-day",
|
|
"Status": "Enabled",
|
|
"Filter": {"Prefix": ""},
|
|
"Expiration": {"Days": 30},
|
|
}
|
|
]
|
|
if bucket == reviewer.PRIMARY
|
|
else []
|
|
)
|
|
buckets.append(
|
|
{
|
|
"bucket": bucket,
|
|
"prefix": prefix,
|
|
"inventory_complete": True,
|
|
"versioning": {"Status": "Enabled"},
|
|
"lifecycle": {"status": "observed", "Rules": rules},
|
|
"versions": [
|
|
{"Key": k, "VersionId": v, "Size": n, "IsLatest": True}
|
|
for b, k, v, h, n in roles
|
|
if b == bucket
|
|
],
|
|
}
|
|
)
|
|
observed = {
|
|
"schema": "coulomb.provider-retention-observation.v1",
|
|
"provider_mutations": False,
|
|
"observed_at": NOW.isoformat(),
|
|
"head_observed_at": NOW.isoformat(),
|
|
"observations": buckets,
|
|
"version_heads": [
|
|
{"bucket": b, "object_key": k, "version_id": v, "sha256": h, "bytes": n}
|
|
for b, k, v, h, n in roles
|
|
],
|
|
}
|
|
point = {
|
|
k: receipt[k] for k in ["object_key", "version_id", "ciphertext_sha256", "ciphertext_bytes"]
|
|
}
|
|
selection = {
|
|
"application": "coulomb-social",
|
|
"status": "reviewed",
|
|
"daily_generations": 14,
|
|
"weekly_generations": 4,
|
|
"protected": point,
|
|
"keep": [point],
|
|
"expiration_candidates": [],
|
|
}
|
|
return [mirrored], recovery, observed, selection
|
|
|
|
|
|
def test_review_maps_logical_point_to_independent_version_without_expiry(reviewer, evidence):
|
|
before = copy.deepcopy(evidence)
|
|
plan = reviewer.review(*evidence, NOW)
|
|
assert plan["generation_selection_valid"] and not plan["blockers"]
|
|
assert len(plan["keep"]) == 1 and not plan["expiration_candidates"]
|
|
assert plan["keep"][0]["bucket"] == reviewer.health.MIRROR_BUCKET
|
|
assert plan["protected_kit_and_proof_versions_verified"]
|
|
assert not plan["ready_to_execute"] and not plan["deletion_capability"]
|
|
assert evidence == before
|
|
|
|
|
|
def test_primary_version_expiry_does_not_discard_held_mirror_generation(reviewer, evidence):
|
|
for bucket in evidence[2]["observations"]:
|
|
if bucket["bucket"] == reviewer.PRIMARY:
|
|
bucket["versions"] = []
|
|
evidence[2]["version_heads"] = [
|
|
h for h in evidence[2]["version_heads"] if h["bucket"] != reviewer.PRIMARY
|
|
]
|
|
plan = reviewer.review(*evidence, NOW)
|
|
assert plan["generation_selection_valid"] and plan["primary_versions_already_absent"] == 1
|
|
assert len(plan["keep"]) == 1
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"bucket_role,reason",
|
|
[
|
|
("mirror", "mirror_lifecycle_cannot_guarantee_generation_selection"),
|
|
("kit", "protected_bucket_has_lifecycle_rules"),
|
|
],
|
|
)
|
|
def test_fixed_age_expiry_cannot_certify_generation_or_protected_custody(
|
|
reviewer,
|
|
evidence,
|
|
bucket_role,
|
|
reason,
|
|
):
|
|
target = reviewer.KIT if bucket_role == "kit" else reviewer.health.MIRROR_BUCKET
|
|
for bucket in evidence[2]["observations"]:
|
|
if bucket["bucket"] == target:
|
|
bucket["lifecycle"]["Rules"] = [{"Status": "Enabled", "Expiration": {"Days": 30}}]
|
|
plan = reviewer.review(*evidence, NOW)
|
|
assert not plan["generation_selection_valid"] and reason in plan["blockers"]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"failure",
|
|
[
|
|
"partial",
|
|
"stale",
|
|
"hash",
|
|
"kit_version",
|
|
"protected_selection",
|
|
"recovery_binding",
|
|
"independent_proof",
|
|
],
|
|
)
|
|
def test_weak_or_changed_provider_recovery_evidence_refuses(reviewer, evidence, failure):
|
|
if failure == "partial":
|
|
evidence[2]["observations"][0]["inventory_complete"] = False
|
|
elif failure == "stale":
|
|
evidence[2]["head_observed_at"] = (NOW - timedelta(hours=27)).isoformat()
|
|
elif failure == "hash":
|
|
evidence[2]["version_heads"][1]["sha256"] = "c" * 64
|
|
elif failure == "kit_version":
|
|
evidence[1]["protected_kit"]["version_id"] = "unreviewed-version"
|
|
elif failure == "recovery_binding":
|
|
evidence[1]["database_sha256"] = "f" * 64
|
|
elif failure == "independent_proof":
|
|
evidence[1]["protected_recovery_proof"]["independent_readback_verified"] = False
|
|
else:
|
|
evidence[3]["keep"] = []
|
|
with pytest.raises(ValueError):
|
|
reviewer.review(*evidence, NOW)
|
|
|
|
|
|
def test_unknown_objects_are_counted_and_preserved(reviewer, evidence):
|
|
mirror = next(
|
|
b for b in evidence[2]["observations"] if b["bucket"] == reviewer.health.MIRROR_BUCKET
|
|
)
|
|
mirror["versions"].append(
|
|
{
|
|
"Key": "archives/unassigned-admission-fixture",
|
|
"VersionId": "fixture",
|
|
"Size": 123,
|
|
"IsLatest": True,
|
|
}
|
|
)
|
|
plan = reviewer.review(*evidence, NOW)
|
|
assert plan["unknown_objects_preserved"] and not plan["expiration_candidates"]
|
|
assert plan["bucket_version_counts"][reviewer.health.MIRROR_BUCKET] == 2
|
|
|
|
|
|
def test_owner_source_tampering_refuses_before_import(reviewer, tmp_path):
|
|
for name in reviewer.OWNER_FILES:
|
|
(tmp_path / name).write_text('raise AssertionError("must not execute")\n')
|
|
with pytest.raises(ValueError, match="Owner planner source differs"):
|
|
reviewer.load_owner(tmp_path)
|
|
|
|
|
|
@pytest.mark.parametrize("sidecar_present", [True, False])
|
|
def test_expiry_mapping_requires_exact_independent_receipt_evidence(
|
|
reviewer,
|
|
evidence,
|
|
sidecar_present,
|
|
):
|
|
# Exercise mapping of a preselected candidate; selection itself is supplied
|
|
# by the hash-pinned owner planner in the CLI, never by an expiry caller.
|
|
older = copy.deepcopy(evidence[0][0])
|
|
stamp = NOW - timedelta(days=40)
|
|
name = stamp.strftime("%Y%m%dT%H%M%SZ-") + "f" * 24 + ".tar.age"
|
|
key = reviewer.health.PREFIX + stamp.strftime("%Y/%m/%d/") + name
|
|
older.update(
|
|
captured_at=stamp.isoformat(),
|
|
local_archive=name,
|
|
object_key=key,
|
|
destination="s3://" + reviewer.PRIMARY + "/" + key,
|
|
version_id="older-primary",
|
|
)
|
|
ref = older["recovery_mirror"]
|
|
ref.update(
|
|
object_key="archives/" + key,
|
|
version_id="older-mirror",
|
|
receipt_object_key="archives/" + key + "." + ref["receipt_sha256"] + ".receipt.json",
|
|
receipt_version_id="older-sidecar",
|
|
)
|
|
evidence[0].append(older)
|
|
evidence[3]["expiration_candidates"].append(
|
|
{k: older[k] for k in ["object_key", "version_id", "ciphertext_sha256", "ciphertext_bytes"]}
|
|
)
|
|
mirror = next(
|
|
b for b in evidence[2]["observations"] if b["bucket"] == reviewer.health.MIRROR_BUCKET
|
|
)
|
|
triples = [
|
|
(
|
|
ref["object_key"],
|
|
ref["version_id"],
|
|
older["ciphertext_sha256"],
|
|
older["ciphertext_bytes"],
|
|
)
|
|
]
|
|
if sidecar_present:
|
|
triples.append(
|
|
(
|
|
ref["receipt_object_key"],
|
|
ref["receipt_version_id"],
|
|
ref["receipt_sha256"],
|
|
ref["receipt_bytes"],
|
|
)
|
|
)
|
|
for object_key, version, digest, size in triples:
|
|
mirror["versions"].append({"Key": object_key, "VersionId": version, "Size": size})
|
|
evidence[2]["version_heads"].append(
|
|
{
|
|
"bucket": reviewer.health.MIRROR_BUCKET,
|
|
"object_key": object_key,
|
|
"version_id": version,
|
|
"sha256": digest,
|
|
"bytes": size,
|
|
}
|
|
)
|
|
if not sidecar_present:
|
|
with pytest.raises(ValueError, match="Exact retained version"):
|
|
reviewer.review(*evidence, NOW)
|
|
else:
|
|
plan = reviewer.review(*evidence, NOW)
|
|
assert len(plan["expiration_candidates"]) == 1
|
|
target = plan["expiration_candidates"][0]
|
|
assert target["bucket"] == reviewer.health.MIRROR_BUCKET
|
|
assert target["version_id"] == "older-mirror"
|
|
assert target["associated_receipt"]["receipt_version_id"] == "older-sidecar"
|
|
assert not plan["ready_to_execute"]
|