diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 04acef3..e04dffe 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -57,6 +57,7 @@ jobs: tests/test_decision_engine_replay.py \ tests/test_evaluate_agent_routing.py \ tests/test_gcp_artifact_workflow.py \ + tests/test_gcp_infra_readiness_checker.py \ tests/test_gcp_runtime_baseline_apply.py \ tests/test_gcp_service_communications.py \ tests/test_gcp_cloudsql_restore_drill.py \ diff --git a/ops/check_gcp_infra_readiness.py b/ops/check_gcp_infra_readiness.py index 08aa65b..4c94d07 100755 --- a/ops/check_gcp_infra_readiness.py +++ b/ops/check_gcp_infra_readiness.py @@ -413,13 +413,29 @@ def check_backup_buckets() -> Check: problems = [] details = [] for bucket in BACKUP_BUCKETS: - versioning = run_text(["gsutil", "versioning", "get", f"gs://{bucket}"]) - ubla = run_text(["gsutil", "uniformbucketlevelaccess", "get", f"gs://{bucket}"]) - versioning_ok = "Enabled" in versioning - ubla_ok = "Enabled: True" in ubla - if not versioning_ok or not ubla_ok: + metadata = run_json( + [ + "gcloud", + "storage", + "buckets", + "describe", + f"gs://{bucket}", + "--format=json", + ] + ) + versioning_ok = metadata.get("versioning", {}).get("enabled") is True + ubla_ok = ( + metadata.get("iamConfiguration", {}) + .get("uniformBucketLevelAccess", {}) + .get("enabled") + is True + ) + public_access_prevention_ok = metadata.get("iamConfiguration", {}).get("publicAccessPrevention") == "enforced" + if not versioning_ok or not ubla_ok or not public_access_prevention_ok: problems.append(bucket) - details.append(f"{bucket}:versioning={versioning_ok}:ubla={ubla_ok}") + details.append( + f"{bucket}:versioning={versioning_ok}:ubla={ubla_ok}:publicAccessPrevention={public_access_prevention_ok}" + ) if problems: return Check("backup_buckets", "fail", " ".join(details)) return Check("backup_buckets", "pass", " ".join(details)) @@ -570,8 +586,25 @@ def check_local_sqlite_postgres_restore_canary() -> Check: current_tier="T1_foundation", ) - latest = candidates[-1] - proof = json.loads(latest.read_text()) + valid_proofs = [] + invalid_proofs = [] + for candidate in candidates: + try: + proof = json.loads(candidate.read_text()) + except json.JSONDecodeError: + invalid_proofs.append(candidate) + continue + valid_proofs.append((str(proof.get("generated_at_utc") or ""), candidate, proof)) + if not valid_proofs: + return Check( + "local_sqlite_postgres_restore_canary", + "fail", + f"no valid sqlite-postgres restore canary proof found under {PROOF_ROOT}; invalid={len(invalid_proofs)}", + required_tier="T2_local_restore_canary", + current_tier="T1_foundation", + ) + + _, latest, proof = max(valid_proofs, key=lambda item: (item[0], item[1].stat().st_mtime)) mismatches = proof.get("mismatched_tables") or [] source_tables = proof.get("source_table_count") target_tables = proof.get("target_table_count") diff --git a/tests/test_gcp_infra_readiness_checker.py b/tests/test_gcp_infra_readiness_checker.py new file mode 100644 index 0000000..f5ab25a --- /dev/null +++ b/tests/test_gcp_infra_readiness_checker.py @@ -0,0 +1,119 @@ +from __future__ import annotations + +import json +from pathlib import Path + +import ops.check_gcp_infra_readiness as checker + + +def write_restore_proof(path: Path, *, generated_at: str, rows: int, status: str = "pass") -> None: + path.write_text( + json.dumps( + { + "generated_at_utc": generated_at, + "mismatched_tables": [], + "postgres_image": "postgres:16-alpine", + "source_integrity_check": "ok", + "source_table_count": 2, + "source_total_rows": rows, + "status": status, + "target_table_count": 2, + "target_total_rows": rows, + } + ) + ) + + +def test_backup_bucket_check_uses_gcloud_storage_json(monkeypatch) -> None: + calls: list[list[str]] = [] + + def fake_run_json(args: list[str]) -> dict: + calls.append(args) + return { + "versioning": {"enabled": True}, + "iamConfiguration": { + "uniformBucketLevelAccess": {"enabled": True}, + "publicAccessPrevention": "enforced", + }, + } + + monkeypatch.setattr(checker, "BACKUP_BUCKETS", ("teleo-test-backups",)) + monkeypatch.setattr(checker, "run_json", fake_run_json) + + result = checker.check_backup_buckets() + + assert result.status == "pass" + assert calls == [ + [ + "gcloud", + "storage", + "buckets", + "describe", + "gs://teleo-test-backups", + "--format=json", + ] + ] + assert "versioning=True" in result.detail + assert "ubla=True" in result.detail + assert "publicAccessPrevention=True" in result.detail + + +def test_backup_bucket_check_rejects_weak_bucket_settings(monkeypatch) -> None: + def fake_run_json(args: list[str]) -> dict: + assert args[:4] == ["gcloud", "storage", "buckets", "describe"] + return { + "versioning": {"enabled": False}, + "iamConfiguration": { + "uniformBucketLevelAccess": {"enabled": True}, + "publicAccessPrevention": "inherited", + }, + } + + monkeypatch.setattr(checker, "BACKUP_BUCKETS", ("teleo-test-backups",)) + monkeypatch.setattr(checker, "run_json", fake_run_json) + + result = checker.check_backup_buckets() + + assert result.status == "fail" + assert "versioning=False" in result.detail + assert "ubla=True" in result.detail + assert "publicAccessPrevention=False" in result.detail + + +def test_readiness_checker_does_not_use_gsutil() -> None: + assert "gsutil" not in Path("ops/check_gcp_infra_readiness.py").read_text() + + +def test_restore_canary_check_uses_newest_generated_at_not_filename(tmp_path, monkeypatch) -> None: + write_restore_proof( + tmp_path / "sqlite-postgres-restore-canary-zzzz.json", + generated_at="2026-07-07T10:00:00+00:00", + rows=10, + ) + write_restore_proof( + tmp_path / "sqlite-postgres-restore-canary-aaaa.json", + generated_at="2026-07-07T12:00:00+00:00", + rows=20, + ) + monkeypatch.setattr(checker, "PROOF_ROOT", tmp_path) + + result = checker.check_local_sqlite_postgres_restore_canary() + + assert result.status == "pass" + assert "sqlite-postgres-restore-canary-aaaa.json" in result.detail + assert "rows=20" in result.detail + + +def test_restore_canary_check_ignores_invalid_json_when_valid_proof_exists(tmp_path, monkeypatch) -> None: + (tmp_path / "sqlite-postgres-restore-canary-newer-name.json").write_text("{") + write_restore_proof( + tmp_path / "sqlite-postgres-restore-canary-valid.json", + generated_at="2026-07-07T12:00:00+00:00", + rows=20, + ) + monkeypatch.setattr(checker, "PROOF_ROOT", tmp_path) + + result = checker.check_local_sqlite_postgres_restore_canary() + + assert result.status == "pass" + assert "sqlite-postgres-restore-canary-valid.json" in result.detail