From 1a043d36a16558a137199b5793f73ec271d16706 Mon Sep 17 00:00:00 2001 From: twentyOne2x Date: Wed, 15 Jul 2026 10:20:27 +0200 Subject: [PATCH] Harden GCP parity proof against forged lifecycle claims --- .agents/skills/teleo-gcp-parity-ops/SKILL.md | 43 ++- .github/workflows/ci.yml | 4 + docs/gcp-kb-cloudsql-restore-runbook.md | 46 ++- .../gcp-canonical-parity-live-20260715.json | 28 +- ops/attest_gcp_reasoning_compute.py | 12 +- ...capture_vps_canonical_postgres_snapshot.py | 15 +- ops/postgres_parity_manifest.sql | 288 +++++++++++++++++- ops/private_receipt_io.py | 29 ++ ...restore_gcp_generated_postgres_snapshot.py | 86 +++++- ops/run_local_canonical_postgres_rebuild.py | 2 +- ops/verify_gcp_canonical_lifecycle.py | 59 +++- ops/verify_postgres_parity_manifest.py | 59 +++- ops/verify_vps_canonical_snapshot_delta.py | 29 +- ...run_gcp_generated_db_direct_claim_suite.py | 5 +- tests/test_attest_gcp_reasoning_compute.py | 6 +- ...capture_vps_canonical_postgres_snapshot.py | 5 + tests/test_private_receipt_io.py | 29 ++ ...restore_gcp_generated_postgres_snapshot.py | 66 +++- ...st_run_local_canonical_postgres_rebuild.py | 3 + .../test_run_local_genesis_ledger_rebuild.py | 3 + tests/test_verify_gcp_canonical_lifecycle.py | 93 +++++- tests/test_verify_postgres_parity_manifest.py | 57 +++- ...est_verify_vps_canonical_snapshot_delta.py | 53 +++- 23 files changed, 939 insertions(+), 81 deletions(-) create mode 100644 tests/test_private_receipt_io.py diff --git a/.agents/skills/teleo-gcp-parity-ops/SKILL.md b/.agents/skills/teleo-gcp-parity-ops/SKILL.md index eef09a4..5ae4069 100644 --- a/.agents/skills/teleo-gcp-parity-ops/SKILL.md +++ b/.agents/skills/teleo-gcp-parity-ops/SKILL.md @@ -8,7 +8,8 @@ description: Use for passwordless Teleo GCP VM access, private Cloud SQL canonic ## Working Target Restore a copy of the VPS canonical Leo database to GCP, prove exact schema, -row, role, extension, performance, and private-connectivity parity, then run the +row, role membership, ownership, ACL, extension, performance, and +private-connectivity parity, then run the real GCP Leo read/reasoning path without Telegram sends or DB mutation. The required tier is T3: a live private GCP staging restore, readback, @@ -45,8 +46,14 @@ Historically verified and intended target; reverify before any new lifecycle: - VM `teleo-prod-1` in `europe-west6-a`; - Cloud SQL `teleo-pgvector-standby`, PostgreSQL 16.14; - private endpoint `10.61.0.3:5432`, public IP disabled, TLS required; +- restore identity `sa-teleo-restore-drill@teleo-501523.iam.gserviceaccount.com`; - service `leoclean-gcp-prod-parallel.service`. +Direct libpq connections must use `sslmode=verify-full`, a reviewed DNS name +covered by the server certificate, and a private regular CA file. The private +IP is `hostaddr`, not the certificate identity. Never downgrade to +`sslmode=require` or accept an arbitrary `*.iam.gserviceaccount.com` identity. + Never print the Cloud SQL or Google password. On the VM, use the attached service account and Secret Manager through reviewed wrappers or a short-lived environment variable, then unset it. @@ -92,6 +99,10 @@ Do not describe a passing Hermes memory sync as canonical KB parity. - Persistent GCP `teleo_canonical` remains the older staging copy measured at `52,164` rows and `26` proposals. It has not been promoted or cut over. +That historical clone predates the expanded ownership/ACL and independently +authenticated receipt contract. It remains useful historical row/schema proof, +but it is not current T3 proof under this skill. + Primary retained proof: - `docs/reports/leo-working-state-20260709/gcp-db-first-working-leo-20260714.md` @@ -107,7 +118,8 @@ Track these independently: 1. control-plane project, VM, Cloud SQL, private-IP, and TLS identity; 2. canonical database schema, counts, row hashes, constraints, indexes, - functions, extensions, roles, and performance; + functions, extensions, exact roles and memberships, owners, normalized + schema/relation/column/function/type ACLs, and performance; 3. GCP service PID/restarts plus live profile and tool hashes; 4. `DC-01` through `DC-06` DB-read readiness; 5. real no-send model replies, strict score, and exact count consistency; @@ -127,6 +139,8 @@ Use: receipt-bound private-TLS restore into a bounded `teleo_clone_*` database; pass the exact capture `--source-receipt`, and require its live Cloud SQL control-plane check to prove public IP disabled before clone creation; it + requires the exact restore service account plus `--ssl-server-name` and + `--ssl-root-cert` for certificate-authenticated `verify-full` TLS; it writes `target-manifest.jsonl`, `restore-receipt.json`, and the generated `gcp-private-connectivity.json` under `/var/lib/teleo-gcp-restore-runs//`; @@ -147,8 +161,11 @@ Use: post-snapshot VPS delta in `source-delta-receipt.json`; - `ops/verify_gcp_canonical_lifecycle.py` after cleanup to bind the exact source, current source, source delta, restore, `gcp_staging` parity, ID-free reasoning, - reasoning-compute attestation, and cleanup receipts into one fail-closed T3 - lifecycle verdict. Pass `--max-postflight-age-seconds 900` and + reasoning-compute attestation, and cleanup receipts into one fail-closed + preparation verdict. Caller-writable JSON can never emit T3 from this API; + even an internally valid bundle returns + `prepared_external_attestation_required`, `current_tier=T2`, and + `accepted_by_this_api=false`. Pass `--max-postflight-age-seconds 900` and `--max-lifecycle-age-seconds 3600`; the latter bounds both cleanup-to-verdict age and restore-to-postflight span. @@ -165,6 +182,7 @@ sudo python3 ops/attest_gcp_reasoning_compute.py \ --expected-compute-instance teleo-prod-1 \ --expected-compute-zone europe-west6-a \ --expected-private-network projects/teleo-501523/global/networks/teleo-staging-net \ + --expected-restore-service-account sa-teleo-restore-drill@teleo-501523.iam.gserviceaccount.com \ --output "${PRIVATE_RUN_DIR}/reasoning-compute-attestation.json" ``` @@ -182,9 +200,17 @@ input, supporting receipt, and verifier output; do not treat that evidence as transient cleanup. The canonical restore streams through `pg_restore` and creates no GCS import object. +The current snapshot and restore commands intentionally use `--no-owner` and +`--no-acl`/`--no-privileges`. The expanded manifest therefore fails GCP parity +until a separately reviewed Cloud-SQL-compatible authorization replay restores +the captured owner/grant semantics. Do not bypass this by dropping the new +manifest rows or by running the Docker/superuser gate bootstrap blindly on +shared Cloud SQL. + Static inspection or a dry run is insufficient for these helpers: verify a new -capture against the live VPS, run the focused tests, and require a passing live -lifecycle receipt before describing the workflow as current parity. +capture against the live VPS and run the focused tests. A current T3 claim also +requires a separate independently authenticated provider/platform receipt that +the bundle author cannot mint. ## Safety And Claim Ceiling @@ -195,8 +221,9 @@ lifecycle receipt before describing the workflow as current parity. - Do not call control-plane inventory, memory sync, route readiness, or a nominal scorer pass full m3taversal parity. -The strongest accepted claim requires exact DB parity plus real no-send model -replies with truthful counts and cleanup. It still does not prove Telegram +The strongest claim emitted by the repository lifecycle verifier is T2 +structural consistency. Exact DB parity plus real no-send model replies with +truthful counts and cleanup still requires independent T3 attestation. Neither proves Telegram delivery, GCP canonical mutation, ongoing replication, or production cutover. ## Current Access And Next Action diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index ccb4925..c069e36 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -50,6 +50,7 @@ jobs: ops/check_gcp_service_communications.py \ ops/plan_gcp_iam_split.py \ ops/private_receipt_io.py \ + ops/run_local_canonical_postgres_rebuild.py \ ops/redact_sqlite_postgres_restore_canary.py \ ops/restore_gcp_generated_postgres_snapshot.py \ ops/sqlite_to_postgres_dump.py \ @@ -106,7 +107,10 @@ jobs: tests/test_working_leo_m3taversal_oos_benchmark.py \ tests/test_working_leo_open_ended_benchmark.py \ tests/test_phase1b_end_to_end.py \ + tests/test_private_receipt_io.py \ tests/test_restore_gcp_generated_postgres_snapshot.py \ + tests/test_run_local_canonical_postgres_rebuild.py \ + tests/test_run_local_genesis_ledger_rebuild.py \ tests/test_sqlite_to_postgres_dump.py \ tests/test_sqlite_postgres_restore_canary_capsule.py \ tests/test_eval_parse.py \ diff --git a/docs/gcp-kb-cloudsql-restore-runbook.md b/docs/gcp-kb-cloudsql-restore-runbook.md index f22aa21..2a0132b 100644 --- a/docs/gcp-kb-cloudsql-restore-runbook.md +++ b/docs/gcp-kb-cloudsql-restore-runbook.md @@ -86,7 +86,10 @@ network mode `none` and tmpfs-only database storage. It waits for an actual --no-owner --no-privileges --exit-on-error`, compares the full parity manifest, then removes the container and proves it is absent. A passing local receipt is the exact-recovery preflight; it is not semantic recompilation from raw source -documents. +documents. Full parity now includes `kb_gate_owner`, `kb_apply`, and +`kb_review`, their memberships, object ownership, and normalized database, +schema, relation, column, function, and type ACLs. A restore that recovers rows +but not those authorization semantics fails closed. Choose one bounded suffix and use it consistently on the GCP replay VM. The restore helper retains its mode-0700 evidence directory at the fixed private @@ -110,7 +113,11 @@ creating the clone, and fails closed unless the exact instance is runnable, PostgreSQL 16, private-networked, bound to the expected RFC1918 host, and has public IP disabled. A GCE metadata preflight independently binds the operator runtime to `teleo-prod-1` in `europe-west6-a` on `teleo-staging-net`; a -hard-coded compute label is not accepted: +hard-coded compute label is not accepted. It also requires the exact +least-privilege restore identity and libpq `verify-full` using a reviewed server +name and private CA file. Determine the certificate DNS SAN from the live Cloud +SQL certificate before filling `--ssl-server-name`; do not substitute the +private IP or downgrade TLS verification: ```bash sudo python3 ops/restore_gcp_generated_postgres_snapshot.py restore \ @@ -128,6 +135,9 @@ sudo python3 ops/restore_gcp_generated_postgres_snapshot.py restore \ --expected-private-network projects/teleo-501523/global/networks/teleo-staging-net \ --expected-compute-instance teleo-prod-1 \ --expected-compute-zone europe-west6-a \ + --expected-restore-service-account sa-teleo-restore-drill@teleo-501523.iam.gserviceaccount.com \ + --ssl-server-name \ + --ssl-root-cert /etc/teleo/cloudsql/server-ca.pem \ --expected-source-ssh-target root@77.42.65.182 \ --expected-source-container teleo-pg \ --expected-source-database teleo \ @@ -140,7 +150,8 @@ directory. The latter is composed from live GCE metadata, Cloud SQL control-plane checks, and the private TLS database identity; do not hand-compose the connectivity receipt. The restore also records identical healthy before/after state for `leoclean-gcp-prod-parallel.service`, including its PID -and restart count. +and restart count. Receipt payloads remain in mode-0600 files; command stdout +contains only a status, output path, and receipt hash. Compare the captured target manifest and generated connectivity receipt to the authorized source: @@ -157,9 +168,20 @@ sudo python3 ops/verify_postgres_parity_manifest.py \ The verifier checks all table row counts and collation-independent row hashes, plus schemas, columns/defaults, constraints, indexes, sequences, views, functions, triggers, enum/domain types, policies, required extensions, -password-free application-role attributes, and bounded query timings. In GCP -scope it also requires a receipt proving a staging compute source, a private -server address, TLS, and public-IP-disabled instance metadata. +password-free application-role attributes, exact role memberships, object +owners, normalized ACLs, and bounded query timings. Role and extension sets are +symmetric: an unexpected extra role, superuser, or extension is a mismatch. +Every table must have a nonempty row hash. In GCP scope it also requires a +receipt proving a staging compute source, a private server address, +certificate-authenticated `verify-full` TLS, and public-IP-disabled instance +metadata. + +The snapshot dump intentionally strips owners and ACLs. Therefore the GCP +restore cannot pass the expanded parity gate until a separately reviewed, +Cloud-SQL-compatible authorization replay reconstructs the captured ownership +and grants. Do not run the Docker/superuser bootstrap blindly on shared Cloud +SQL. Until that replay exists, this lane remains T2 preparation and must fail +before a T3 claim. After the parity verifier passes, run the no-send m3taversal blind reasoning replay from staging compute against that generated database, attest the replay @@ -193,6 +215,7 @@ sudo python3 ops/attest_gcp_reasoning_compute.py \ --expected-compute-instance teleo-prod-1 \ --expected-compute-zone europe-west6-a \ --expected-private-network projects/teleo-501523/global/networks/teleo-staging-net \ + --expected-restore-service-account sa-teleo-restore-drill@teleo-501523.iam.gserviceaccount.com \ --output "$PRIVATE_RUN_DIR/reasoning-compute-attestation.json" ``` @@ -230,7 +253,8 @@ verifier rejects local-scope parity, stale or mismatched source hashes, a postflight source capture that predates reasoning or cleanup, hand-authored reasoning booleans without real retrieval/tool receipts, an unattested replay host, compute or Cloud SQL topology drift, and any cleanup receipt that leaves -the clone present. The postflight capture may be at most 900 seconds old at +the clone present or claims cleanup when the named clone never existed. Both +source captures must use the same reviewed manifest SQL. The postflight capture may be at most 900 seconds old at verification; the GCP lifecycle age and restore-to-postflight span must each be at most 3600 seconds: @@ -249,6 +273,14 @@ sudo python3 ops/verify_gcp_canonical_lifecycle.py \ --output "$PRIVATE_RUN_DIR/lifecycle-verification.json" ``` +This command validates only the internal consistency of caller-supplied +receipts. A clean result is emitted as +`prepared_external_attestation_required`, with +`current_tier=T2_structural_receipt_bundle`, `t3_receipt_ready=false`, and +`accepted_by_this_api=false`. It cannot emit a live-GCP/T3 claim. T3 requires a +separate independently authenticated provider or platform receipt that the +bundle author cannot mint. + Retain the private receipt run directory and every useful lifecycle input, supporting receipt, and verifier output. Remove only transient upload/bundle directories, temporary Hermes profiles, helper clients, and child processes diff --git a/docs/reports/leo-working-state-20260709/gcp-canonical-parity-live-20260715.json b/docs/reports/leo-working-state-20260709/gcp-canonical-parity-live-20260715.json index 258e20a..0e7a214 100644 --- a/docs/reports/leo-working-state-20260709/gcp-canonical-parity-live-20260715.json +++ b/docs/reports/leo-working-state-20260709/gcp-canonical-parity-live-20260715.json @@ -4,7 +4,7 @@ "generated_at_utc": "2026-07-15T01:50:20Z", "status": "gated_external", "required_tier": "T3_live_private_gcp_staging", - "current_tier": "T2_current_source_and_isolated_restore_plus_live_access_gate", + "current_tier": "T2_historical_source_and_isolated_restore_under_superseded_parity_contract", "current_canary": "Restore the newest authorized canonical snapshot into one private generated Cloud SQL database, prove exact parity and ID-free Leo reasoning from staging compute, then drop the generated database and verify absence.", "current_source": { "run_id": "canonical-20260715t014930z", @@ -57,7 +57,24 @@ "performance_problems": [], "network_mode": "none", "persistent_postgres_storage_used": false, - "container_absent_after_test": true + "container_absent_after_test": true, + "authorization_state_verified": false, + "prior_manifest_contract_only": true + }, + "contract_reassessment": { + "status": "changes_required", + "reviewed_manifest_sql_sha256": "430390129bc8980c4ece9939f2aee18e8c40e9da4b226782579349d5c45c5301", + "prior_receipts_recomputed_under_current_contract": false, + "newly_required_parity": [ + "kb_gate_owner plus exact application-role attributes", + "role memberships", + "database, schema, relation, function, and type ownership", + "normalized database, schema, relation, column, function, and type ACLs", + "symmetric role and extension sets", + "nonempty row hash for every table" + ], + "t3_receipt_accepted_by_repository_api": false, + "independent_platform_attestation_required": true }, "permission_profile": { "approval_policy": "never", @@ -90,7 +107,7 @@ ], "exact_gate": "The privileged local Google OAuth session requires human reauthentication, the fixed IAP workflow names an unavailable teleo-iap-operator WIF provider, and no existing noninteractive service-account route can obtain a Cloud SQL/Compute-capable token. Current Cloud SQL topology and generated-database inventory cannot be read, so the private restore cannot start.", "why_autonomous_repair_stops": "Google reauthentication may require a password, MFA, passkey, or consent that Codex must not retrieve or enter. Recreating or rebinding the WIF/IAP identities requires an already-authorized GCP administrator, and no safe noninteractive identity currently has that authority.", - "next_non_user_action": "After operator-controlled Google reauthentication or administrator convergence of the reviewed least-privilege IAP operator, read the current Cloud SQL and GCE topology, stream the receipt-bound dump to a teleo_clone_* database from teleo-prod-1, run exact parity, blind reasoning, live GCE metadata attestation, cleanup, postflight source capture, and the eight-receipt lifecycle verifier.", + "next_non_user_action": "After the reviewed operator route and exact restore service account are available, capture a new source snapshot with the expanded manifest, implement and review a Cloud-SQL-compatible authorization replay, use verify-full TLS with the reviewed server name and CA, run the disposable clone lifecycle, then obtain an independently authenticated platform receipt. The repository eight-receipt verifier remains preparation-only.", "gcp_effects_from_this_attempt": { "cloudsql_database_created": false, "gcp_resource_created": false, @@ -122,12 +139,15 @@ "sha256": "0fda5e6d58d72dcb084dc277bc59312bf62c002c728e249a9f16e79ae096037b" } }, - "strongest_claim_allowed": "The newest authorized VPS snapshot is stable and exactly reconstructs in an isolated PostgreSQL 16 canary; current private GCP restore, reasoning, and cleanup are not proven because the live operator route is externally gated.", + "strongest_claim_allowed": "The retained snapshot was stable and reconstructed data and schema under the prior parity contract. It does not prove current authorization parity or live GCP operation under the expanded contract.", "not_proven": [ "current Cloud SQL topology or public-IP-disabled state", "current generated-database inventory", "private staging-to-Cloud-SQL TLS connectivity", "current Cloud SQL restore parity", + "current role membership, ownership, and ACL parity", + "certificate-authenticated verify-full database access", + "independently authenticated T3 lifecycle attestation", "blind Leo reasoning against the generated Cloud SQL database", "generated Cloud SQL database teardown" ] diff --git a/ops/attest_gcp_reasoning_compute.py b/ops/attest_gcp_reasoning_compute.py index e9d5124..42a1105 100644 --- a/ops/attest_gcp_reasoning_compute.py +++ b/ops/attest_gcp_reasoning_compute.py @@ -11,24 +11,26 @@ from pathlib import Path from typing import Any try: - from .private_receipt_io import write_private_json + from .private_receipt_io import print_private_receipt_summary, write_private_json from .restore_gcp_generated_postgres_snapshot import ( DEFAULT_COMPUTE_INSTANCE, DEFAULT_COMPUTE_ZONE, DEFAULT_PRIVATE_NETWORK, DEFAULT_PROJECT, + DEFAULT_RESTORE_SERVICE_ACCOUNT, RUN_ID_RE, TARGET_DATABASE_RE, gce_compute_identity, sha256_file, ) except ImportError: # pragma: no cover - direct script execution on the GCP VM - from private_receipt_io import write_private_json + from private_receipt_io import print_private_receipt_summary, write_private_json from restore_gcp_generated_postgres_snapshot import ( DEFAULT_COMPUTE_INSTANCE, DEFAULT_COMPUTE_ZONE, DEFAULT_PRIVATE_NETWORK, DEFAULT_PROJECT, + DEFAULT_RESTORE_SERVICE_ACCOUNT, RUN_ID_RE, TARGET_DATABASE_RE, gce_compute_identity, @@ -81,6 +83,7 @@ def attest_reasoning_compute( expected_compute_instance: str = DEFAULT_COMPUTE_INSTANCE, expected_compute_zone: str = DEFAULT_COMPUTE_ZONE, expected_private_network: str = DEFAULT_PRIVATE_NETWORK, + expected_restore_service_account: str = DEFAULT_RESTORE_SERVICE_ACCOUNT, metadata_timeout: float = 10.0, generated_at_utc: str | None = None, ) -> dict[str, Any]: @@ -110,6 +113,7 @@ def attest_reasoning_compute( expected_compute_instance, expected_compute_zone, expected_private_network, + expected_restore_service_account, timeout=metadata_timeout, ) return { @@ -147,6 +151,7 @@ def main() -> int: parser.add_argument("--expected-compute-instance", default=DEFAULT_COMPUTE_INSTANCE) parser.add_argument("--expected-compute-zone", default=DEFAULT_COMPUTE_ZONE) parser.add_argument("--expected-private-network", default=DEFAULT_PRIVATE_NETWORK) + parser.add_argument("--expected-restore-service-account", default=DEFAULT_RESTORE_SERVICE_ACCOUNT) parser.add_argument("--metadata-timeout", default=10.0, type=float) parser.add_argument("--output", required=True, type=Path) args = parser.parse_args() @@ -159,12 +164,13 @@ def main() -> int: expected_compute_instance=args.expected_compute_instance, expected_compute_zone=args.expected_compute_zone, expected_private_network=args.expected_private_network, + expected_restore_service_account=args.expected_restore_service_account, metadata_timeout=args.metadata_timeout, ) write_private_json(args.output, receipt) except (OSError, ValueError, RuntimeError) as exc: parser.error(str(exc)) - print(json.dumps(receipt, indent=2, sort_keys=True)) + print_private_receipt_summary(args.output, receipt) return 0 diff --git a/ops/capture_vps_canonical_postgres_snapshot.py b/ops/capture_vps_canonical_postgres_snapshot.py index 9e02dea..7cde893 100755 --- a/ops/capture_vps_canonical_postgres_snapshot.py +++ b/ops/capture_vps_canonical_postgres_snapshot.py @@ -18,8 +18,10 @@ from pathlib import Path from typing import Any try: + from .private_receipt_io import print_private_receipt_summary, write_private_json from .verify_postgres_parity_manifest import load_manifest except ImportError: # pragma: no cover - direct script execution + from private_receipt_io import print_private_receipt_summary, write_private_json from verify_postgres_parity_manifest import load_manifest @@ -471,12 +473,17 @@ def main() -> int: "capture_authorization_ref": args.authorization_ref, "error": str(exc), } - print(json.dumps(payload, indent=2, sort_keys=True)) + receipt = ( + output_dir.parent / f".{output_dir.name}-{args.run_id}-failure.json" + if output_preexisted + else output_dir / "receipt.json" + ) + write_private_json(receipt, payload) + print_private_receipt_summary(receipt, payload) return 1 receipt = args.output_dir.resolve() / "receipt.json" - receipt.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n") - os.chmod(receipt, 0o600) - print(json.dumps(payload, indent=2, sort_keys=True)) + write_private_json(receipt, payload) + print_private_receipt_summary(receipt, payload) return 0 diff --git a/ops/postgres_parity_manifest.sql b/ops/postgres_parity_manifest.sql index 3166eba..8f5ebc8 100644 --- a/ops/postgres_parity_manifest.sql +++ b/ops/postgres_parity_manifest.sql @@ -55,11 +55,297 @@ select jsonb_build_object( 'bypass_rls', rolbypassrls ) order by rolname) from pg_roles - where rolname in ('kb_apply', 'kb_review')), + where rolname in ('kb_gate_owner', 'kb_apply', 'kb_review')), '[]'::jsonb ) )::text; +select jsonb_build_object( + 'kind', 'role_memberships', + 'items', coalesce( + (select jsonb_agg( + jsonb_build_object( + 'role', role_row.rolname, + 'member', member_row.rolname, + 'grantor', grantor_row.rolname, + 'admin_option', membership.admin_option, + 'inherit_option', membership.inherit_option, + 'set_option', membership.set_option + ) order by role_row.rolname, member_row.rolname, grantor_row.rolname) + from pg_auth_members membership + join pg_roles role_row on role_row.oid = membership.roleid + join pg_roles member_row on member_row.oid = membership.member + join pg_roles grantor_row on grantor_row.oid = membership.grantor + where role_row.rolname in ('kb_gate_owner', 'kb_apply', 'kb_review') + or member_row.rolname in ('kb_gate_owner', 'kb_apply', 'kb_review')), + '[]'::jsonb + ) +)::text; + +with ownership as ( + select + format('database:%s', database_row.datname) as sort_key, + 'database'::text as object_type, + null::text as schema, + database_row.datname as name, + null::text as identity_arguments, + pg_get_userbyid(database_row.datdba) as owner + from pg_database database_row + where database_row.datname = current_database() + + union all + + select + format('schema:%s', namespace.nspname), + 'schema', + namespace.nspname, + namespace.nspname, + null::text, + pg_get_userbyid(namespace.nspowner) + from pg_namespace namespace + where namespace.nspname in ('public', 'kb_stage') + + union all + + select + format('relation:%s.%s', namespace.nspname, relation.relname), + case relation.relkind + when 'r' then 'table' + when 'p' then 'partitioned_table' + when 'v' then 'view' + when 'm' then 'materialized_view' + when 'S' then 'sequence' + when 'f' then 'foreign_table' + end, + namespace.nspname, + relation.relname, + null::text, + pg_get_userbyid(relation.relowner) + from pg_class relation + join pg_namespace namespace on namespace.oid = relation.relnamespace + where namespace.nspname in ('public', 'kb_stage') + and relation.relkind in ('r', 'p', 'v', 'm', 'S', 'f') + + union all + + select + format( + 'function:%s.%s(%s)', + namespace.nspname, + procedure.proname, + pg_get_function_identity_arguments(procedure.oid) + ), + 'function', + namespace.nspname, + procedure.proname, + pg_get_function_identity_arguments(procedure.oid), + pg_get_userbyid(procedure.proowner) + from pg_proc procedure + join pg_namespace namespace on namespace.oid = procedure.pronamespace + where namespace.nspname in ('public', 'kb_stage') + + union all + + select + format('type:%s.%s', namespace.nspname, type_row.typname), + case type_row.typtype when 'd' then 'domain' else 'enum' end, + namespace.nspname, + type_row.typname, + null::text, + pg_get_userbyid(type_row.typowner) + from pg_type type_row + join pg_namespace namespace on namespace.oid = type_row.typnamespace + where namespace.nspname in ('public', 'kb_stage') + and type_row.typtype in ('d', 'e') +) +select jsonb_build_object( + 'kind', 'object_ownership', + 'items', coalesce( + jsonb_agg(to_jsonb(ownership) - 'sort_key' order by sort_key), + '[]'::jsonb + ) +)::text +from ownership; + +with acl_rows as ( + select + format( + 'database:%s:%s:%s:%s', + database_row.datname, + acl.grantee, + acl.privilege_type, + acl.is_grantable + ) as sort_key, + 'database'::text as object_type, + null::text as schema, + database_row.datname as name, + null::text as column_name, + null::text as identity_arguments, + pg_get_userbyid(acl.grantor) as grantor, + case when acl.grantee = 0 then 'PUBLIC' else pg_get_userbyid(acl.grantee) end as grantee, + acl.privilege_type, + acl.is_grantable + from pg_database database_row + cross join lateral aclexplode(coalesce(database_row.datacl, acldefault('d', database_row.datdba))) acl + where database_row.datname = current_database() + + union all + + select + format( + 'schema:%s:%s:%s:%s', + namespace.nspname, + acl.grantee, + acl.privilege_type, + acl.is_grantable + ), + 'schema', + namespace.nspname, + namespace.nspname, + null::text, + null::text, + pg_get_userbyid(acl.grantor), + case when acl.grantee = 0 then 'PUBLIC' else pg_get_userbyid(acl.grantee) end, + acl.privilege_type, + acl.is_grantable + from pg_namespace namespace + cross join lateral aclexplode(coalesce(namespace.nspacl, acldefault('n', namespace.nspowner))) acl + where namespace.nspname in ('public', 'kb_stage') + + union all + + select + format( + 'relation:%s.%s:%s:%s:%s', + namespace.nspname, + relation.relname, + acl.grantee, + acl.privilege_type, + acl.is_grantable + ), + case when relation.relkind = 'S' then 'sequence' else 'relation' end, + namespace.nspname, + relation.relname, + null::text, + null::text, + pg_get_userbyid(acl.grantor), + case when acl.grantee = 0 then 'PUBLIC' else pg_get_userbyid(acl.grantee) end, + acl.privilege_type, + acl.is_grantable + from pg_class relation + join pg_namespace namespace on namespace.oid = relation.relnamespace + cross join lateral aclexplode( + coalesce( + relation.relacl, + acldefault(case when relation.relkind = 'S' then 'S'::"char" else 'r'::"char" end, relation.relowner) + ) + ) acl + where namespace.nspname in ('public', 'kb_stage') + and relation.relkind in ('r', 'p', 'v', 'm', 'S', 'f') + + union all + + select + format( + 'column:%s.%s.%s:%s:%s:%s', + namespace.nspname, + relation.relname, + attribute.attname, + acl.grantee, + acl.privilege_type, + acl.is_grantable + ), + 'column', + namespace.nspname, + relation.relname, + attribute.attname, + null::text, + pg_get_userbyid(acl.grantor), + case when acl.grantee = 0 then 'PUBLIC' else pg_get_userbyid(acl.grantee) end, + acl.privilege_type, + acl.is_grantable + from pg_attribute attribute + join pg_class relation on relation.oid = attribute.attrelid + join pg_namespace namespace on namespace.oid = relation.relnamespace + cross join lateral aclexplode(attribute.attacl) acl + where namespace.nspname in ('public', 'kb_stage') + and relation.relkind in ('r', 'p', 'v', 'm', 'f') + and attribute.attnum > 0 + and not attribute.attisdropped + + union all + + select + format( + 'function:%s.%s(%s):%s:%s:%s', + namespace.nspname, + procedure.proname, + pg_get_function_identity_arguments(procedure.oid), + acl.grantee, + acl.privilege_type, + acl.is_grantable + ), + 'function', + namespace.nspname, + procedure.proname, + null::text, + pg_get_function_identity_arguments(procedure.oid), + pg_get_userbyid(acl.grantor), + case when acl.grantee = 0 then 'PUBLIC' else pg_get_userbyid(acl.grantee) end, + acl.privilege_type, + acl.is_grantable + from pg_proc procedure + join pg_namespace namespace on namespace.oid = procedure.pronamespace + cross join lateral aclexplode(coalesce(procedure.proacl, acldefault('f', procedure.proowner))) acl + where namespace.nspname in ('public', 'kb_stage') + + union all + + select + format( + 'type:%s.%s:%s:%s:%s', + namespace.nspname, + type_row.typname, + acl.grantee, + acl.privilege_type, + acl.is_grantable + ), + 'type', + namespace.nspname, + type_row.typname, + null::text, + null::text, + pg_get_userbyid(acl.grantor), + case when acl.grantee = 0 then 'PUBLIC' else pg_get_userbyid(acl.grantee) end, + acl.privilege_type, + acl.is_grantable + from pg_type type_row + join pg_namespace namespace on namespace.oid = type_row.typnamespace + cross join lateral aclexplode(coalesce(type_row.typacl, acldefault('T', type_row.typowner))) acl + where namespace.nspname in ('public', 'kb_stage') + and type_row.typtype in ('d', 'e') +) +select jsonb_build_object( + 'kind', 'object_acl', + 'items', coalesce( + jsonb_agg( + to_jsonb(acl_rows) - 'sort_key' + order by + object_type, + schema nulls first, + name, + column_name nulls first, + identity_arguments nulls first, + grantor, + grantee, + privilege_type, + is_grantable + ), + '[]'::jsonb + ) +)::text +from acl_rows; + select jsonb_build_object( 'kind', 'columns', 'items', coalesce( diff --git a/ops/private_receipt_io.py b/ops/private_receipt_io.py index 13dc3d6..1655006 100644 --- a/ops/private_receipt_io.py +++ b/ops/private_receipt_io.py @@ -2,6 +2,7 @@ from __future__ import annotations +import hashlib import json import os import uuid @@ -9,6 +10,14 @@ from pathlib import Path from typing import Any +def file_sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + def write_private_json(path: Path, payload: dict[str, Any]) -> None: """Write one JSON receipt as mode 0600 without chmodding existing parents.""" parent_preexisting = path.parent.exists() @@ -32,3 +41,23 @@ def write_private_json(path: Path, payload: dict[str, Any]) -> None: temporary.unlink() except FileNotFoundError: pass + + +def private_receipt_summary(path: Path, payload: dict[str, Any]) -> dict[str, Any]: + """Return a stdout-safe pointer without replaying private receipt contents.""" + failed_checks = payload.get("failed_checks") + problems = payload.get("problems") + error = payload.get("error") + return { + "artifact": payload.get("artifact"), + "status": payload.get("status"), + "output_name": path.name, + "output_sha256": file_sha256(path), + "failed_check_count": len(failed_checks) if isinstance(failed_checks, list) else None, + "problem_count": len(problems) if isinstance(problems, list) else None, + "error_type": error.get("type") if isinstance(error, dict) else (type(error).__name__ if error else None), + } + + +def print_private_receipt_summary(path: Path, payload: dict[str, Any]) -> None: + print(json.dumps(private_receipt_summary(path, payload), indent=2, sort_keys=True)) diff --git a/ops/restore_gcp_generated_postgres_snapshot.py b/ops/restore_gcp_generated_postgres_snapshot.py index 7e09f4c..c14537a 100644 --- a/ops/restore_gcp_generated_postgres_snapshot.py +++ b/ops/restore_gcp_generated_postgres_snapshot.py @@ -29,9 +29,11 @@ from pathlib import Path from typing import Any try: - from .verify_postgres_parity_manifest import compare_manifests, load_manifest + from .private_receipt_io import print_private_receipt_summary + from .verify_postgres_parity_manifest import REVIEWED_MANIFEST_SQL_SHA256, compare_manifests, load_manifest except ImportError: # pragma: no cover - direct script execution on the GCP VM - from verify_postgres_parity_manifest import compare_manifests, load_manifest + from private_receipt_io import print_private_receipt_summary + from verify_postgres_parity_manifest import REVIEWED_MANIFEST_SQL_SHA256, compare_manifests, load_manifest DEFAULT_HOST = "10.61.0.3" @@ -40,6 +42,7 @@ DEFAULT_INSTANCE = "teleo-pgvector-standby" DEFAULT_PRIVATE_NETWORK = "projects/teleo-501523/global/networks/teleo-staging-net" DEFAULT_COMPUTE_INSTANCE = "teleo-prod-1" DEFAULT_COMPUTE_ZONE = "europe-west6-a" +DEFAULT_RESTORE_SERVICE_ACCOUNT = "sa-teleo-restore-drill@teleo-501523.iam.gserviceaccount.com" DEFAULT_SOURCE_SSH_TARGET = "root@77.42.65.182" DEFAULT_SOURCE_CONTAINER = "teleo-pg" DEFAULT_SOURCE_DATABASE = "teleo" @@ -48,7 +51,6 @@ DEFAULT_SECRET = "gcp-teleo-pgvector-standby-postgres-password" DEFAULT_SERVICE = "leoclean-gcp-prod-parallel.service" DEFAULT_RUN_ROOT = Path("/var/lib/teleo-gcp-restore-runs") DEFAULT_MANIFEST_SQL = Path(__file__).with_name("postgres_parity_manifest.sql") -REVIEWED_MANIFEST_SQL_SHA256 = "8b8cdc25d54fdd8de05eb38c6e4423d2836953eb6012d4545f5c9c71b5f0150a" SOURCE_SNAPSHOT_RECEIPT_SCHEMA = "livingip.sourceSnapshotReceipt.v2" TARGET_DATABASE_RE = re.compile(r"teleo_clone_[a-z0-9][a-z0-9_]{0,50}\Z") RUN_ID_RE = re.compile(r"gcp-restore-[a-z0-9][a-z0-9-]{7,47}\Z") @@ -61,6 +63,8 @@ GCE_INSTANCE_RE = re.compile(r"[a-z](?:[a-z0-9-]{0,61}[a-z0-9])?\Z") GCE_ZONE_RE = re.compile(r"[a-z]+-[a-z]+[0-9]-[a-z]\Z") SAFE_NAME_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,127}\Z") SAFE_SSH_TARGET_RE = re.compile(r"(?:[A-Za-z0-9][A-Za-z0-9._-]{0,63}@)?[A-Za-z0-9][A-Za-z0-9.-]{0,252}\Z") +SAFE_DNS_NAME_RE = re.compile(r"[A-Za-z0-9](?:[A-Za-z0-9.-]{0,251}[A-Za-z0-9])?\Z") +SERVICE_ACCOUNT_RE = re.compile(r"[a-z][a-z0-9-]{0,62}@[a-z][a-z0-9-]{4,28}[a-z0-9]\.iam\.gserviceaccount\.com\Z") PG_RESTORE_VERSION_RE = re.compile(r"pg_restore \(PostgreSQL\) (?P[0-9]+)(?:\.[0-9]+)*") ROLLBACK_DATABASE = "teleo_canonical_pre_20260712t1905z" GCE_METADATA_ROOT = "http://metadata.google.internal/computeMetadata/v1" @@ -387,6 +391,7 @@ def gce_compute_identity( expected_instance: str, expected_zone: str, expected_private_network: str, + expected_service_account: str, *, timeout: float, ) -> dict[str, Any]: @@ -423,7 +428,7 @@ def gce_compute_identity( ipaddress.ip_network("192.168.0.0/16"), ) ), - "service_account": metadata["service_account"].endswith(".iam.gserviceaccount.com"), + "service_account": metadata["service_account"] == expected_service_account, } failed = sorted(name for name, passed in checks.items() if not passed) if failed: @@ -434,6 +439,7 @@ def gce_compute_identity( "expected_zone": expected_zone, "expected_private_network": expected_private_network, "expected_network_resource": expected_network_resource, + "expected_service_account": expected_service_account, "checks": checks, } @@ -492,10 +498,29 @@ def postgres_env(password: str, *, read_only: bool) -> dict[str, str]: return env +def conninfo_value(value: Any) -> str: + return "'" + str(value).replace("\\", "\\\\").replace("'", "\\'") + "'" + + +def postgres_conninfo(args: argparse.Namespace, database: str) -> str: + return " ".join( + ( + f"host={conninfo_value(args.ssl_server_name)}", + f"hostaddr={conninfo_value(args.host)}", + "port=5432", + f"dbname={conninfo_value(database)}", + "user=postgres", + "sslmode=verify-full", + f"sslrootcert={conninfo_value(args.ssl_root_cert)}", + "connect_timeout=8", + ) + ) + + def psql_command(args: argparse.Namespace, database: str) -> list[str]: return [ args.psql_bin, - f"host={args.host} port=5432 dbname={database} user=postgres sslmode=require connect_timeout=8", + postgres_conninfo(args, database), "-X", "-Atq", "-v", @@ -566,7 +591,7 @@ def restore_dump(args: argparse.Namespace, password: str) -> None: command = [ args.pg_restore_bin, "-d", - f"host={args.host} port=5432 dbname={args.target_db} user=postgres sslmode=require connect_timeout=8", + postgres_conninfo(args, args.target_db), "--no-owner", "--no-acl", "--exit-on-error", @@ -594,6 +619,8 @@ def validate_private_identity( target_manifest: dict[str, Any], target_db: str, source_compute: str, + ssl_server_name: str, + ssl_root_cert_sha256: str, ) -> dict[str, Any]: identity = target_manifest["singleton"]["identity"] if identity.get("database") != target_db: @@ -622,6 +649,10 @@ def validate_private_identity( "server_port": identity.get("server_port"), "ssl": True, "ssl_version": identity.get("ssl_version"), + "sslmode": "verify-full", + "ssl_server_name": ssl_server_name, + "ssl_root_cert_sha256": ssl_root_cert_sha256, + "server_identity_verified": True, "private_connectivity": True, "transaction_read_only": "on", "source_compute": source_compute, @@ -718,6 +749,8 @@ def write_private(path: Path, content: str | bytes) -> None: def run_restore(args: argparse.Namespace) -> dict[str, Any]: started = time.monotonic() + args.ssl_root_cert = validate_regular_file(args.ssl_root_cert, "Cloud SQL server CA") + ssl_root_cert_sha256 = sha256_file(args.ssl_root_cert) run_dir = secure_run_dir(args.run_root, args.run_id, create=True) receipt: dict[str, Any] = { "artifact": "gcp_generated_postgres_snapshot_restore", @@ -731,6 +764,9 @@ def run_restore(args: argparse.Namespace) -> dict[str, Any]: "private_network": args.expected_private_network, "compute_instance": args.expected_compute_instance, "compute_zone": args.expected_compute_zone, + "restore_service_account": args.expected_restore_service_account, + "ssl_server_name": args.ssl_server_name, + "ssl_root_cert_sha256": ssl_root_cert_sha256, "source": { "ssh_target": args.expected_source_ssh_target, "container": args.expected_source_container, @@ -795,6 +831,7 @@ def run_restore(args: argparse.Namespace) -> dict[str, Any]: args.expected_compute_instance, args.expected_compute_zone, args.expected_private_network, + args.expected_restore_service_account, timeout=args.command_timeout, ) receipt["live_service"]["before"] = service_state(args.service, timeout=args.command_timeout) @@ -837,6 +874,8 @@ def run_restore(args: argparse.Namespace) -> dict[str, Any]: target_manifest, args.target_db, receipt["target"]["compute"]["instance"], + args.ssl_server_name, + ssl_root_cert_sha256, ), } ) @@ -861,6 +900,10 @@ def run_restore(args: argparse.Namespace) -> dict[str, Any]: "server_port": connectivity["server_port"], "ssl": connectivity["ssl"], "ssl_version": connectivity["ssl_version"], + "sslmode": connectivity["sslmode"], + "ssl_server_name": connectivity["ssl_server_name"], + "ssl_root_cert_sha256": connectivity["ssl_root_cert_sha256"], + "server_identity_verified": connectivity["server_identity_verified"], "private_connectivity": connectivity["private_connectivity"], "public_ip_disabled": cloudsql["public_ip_disabled"], "cloudsql_control_plane_checks": cloudsql["checks"], @@ -927,6 +970,12 @@ def run_restore(args: argparse.Namespace) -> dict[str, Any]: args.expected_compute_instance, "--expected-compute-zone", args.expected_compute_zone, + "--expected-restore-service-account", + args.expected_restore_service_account, + "--ssl-server-name", + args.ssl_server_name, + "--ssl-root-cert", + str(args.ssl_root_cert), "--expected-source-ssh-target", args.expected_source_ssh_target, "--expected-source-container", @@ -988,6 +1037,8 @@ def run_restore(args: argparse.Namespace) -> dict[str, Any]: def run_cleanup(args: argparse.Namespace) -> dict[str, Any]: + args.ssl_root_cert = validate_regular_file(args.ssl_root_cert, "Cloud SQL server CA") + ssl_root_cert_sha256 = sha256_file(args.ssl_root_cert) run_dir = secure_run_dir(args.run_root, args.run_id, create=False) restore_receipt_path = run_dir / "restore-receipt.json" restore_receipt_path = validate_regular_file(restore_receipt_path, "restore receipt") @@ -1007,6 +1058,9 @@ def run_cleanup(args: argparse.Namespace) -> dict[str, Any]: "private_network": args.expected_private_network, "compute_instance": args.expected_compute_instance, "compute_zone": args.expected_compute_zone, + "restore_service_account": args.expected_restore_service_account, + "ssl_server_name": args.ssl_server_name, + "ssl_root_cert_sha256": ssl_root_cert_sha256, "source": { "ssh_target": args.expected_source_ssh_target, "container": args.expected_source_container, @@ -1037,6 +1091,7 @@ def run_cleanup(args: argparse.Namespace) -> dict[str, Any]: "instance": restore_compute.get("instance") == args.expected_compute_instance, "zone": restore_compute.get("zone") == args.expected_compute_zone, "private_network": restore_compute.get("expected_private_network") == args.expected_private_network, + "service_account": restore_compute.get("service_account") == args.expected_restore_service_account, "checks": bool(restore_compute.get("checks")) and all(value is True for value in restore_compute["checks"].values()), } @@ -1083,6 +1138,7 @@ def run_cleanup(args: argparse.Namespace) -> dict[str, Any]: args.expected_compute_instance, args.expected_compute_zone, args.expected_private_network, + args.expected_restore_service_account, timeout=args.command_timeout, ) cleanup_compute_mismatches = sorted( @@ -1169,6 +1225,9 @@ def add_common_args(parser: argparse.ArgumentParser) -> None: parser.add_argument("--expected-private-network", default=DEFAULT_PRIVATE_NETWORK) parser.add_argument("--expected-compute-instance", default=DEFAULT_COMPUTE_INSTANCE) parser.add_argument("--expected-compute-zone", default=DEFAULT_COMPUTE_ZONE) + parser.add_argument("--expected-restore-service-account", default=DEFAULT_RESTORE_SERVICE_ACCOUNT) + parser.add_argument("--ssl-server-name", required=True) + parser.add_argument("--ssl-root-cert", required=True, type=Path) parser.add_argument("--expected-source-ssh-target", default=DEFAULT_SOURCE_SSH_TARGET) parser.add_argument("--expected-source-container", default=DEFAULT_SOURCE_CONTAINER) parser.add_argument("--expected-source-database", default=DEFAULT_SOURCE_DATABASE) @@ -1218,6 +1277,17 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser.error("--expected-compute-instance must be a safe lowercase GCE instance name") if not GCE_ZONE_RE.fullmatch(args.expected_compute_zone): parser.error("--expected-compute-zone must be a safe GCE zone") + if not SERVICE_ACCOUNT_RE.fullmatch(args.expected_restore_service_account): + parser.error("--expected-restore-service-account must be an exact GCP service-account email") + if not SAFE_DNS_NAME_RE.fullmatch(args.ssl_server_name): + parser.error("--ssl-server-name must be a safe DNS name covered by the Cloud SQL server certificate") + try: + ssl_root_cert_mode = args.ssl_root_cert.lstat().st_mode + except OSError: + parser.error("--ssl-root-cert must be an existing regular file") + if not stat.S_ISREG(ssl_root_cert_mode) or args.ssl_root_cert.is_symlink(): + parser.error("--ssl-root-cert must be a regular non-symlink file") + args.ssl_root_cert = args.ssl_root_cert.resolve() if not SAFE_SSH_TARGET_RE.fullmatch(args.expected_source_ssh_target): parser.error("--expected-source-ssh-target must be a host or user@host without options") for value, flag in ( @@ -1273,7 +1343,9 @@ def main(argv: list[str] | None = None) -> int: finally: for watched_signal, previous in previous_handlers.items(): signal.signal(watched_signal, previous) - print(json.dumps(payload, indent=2, sort_keys=True)) + receipt_name = "restore-receipt.json" if args.operation == "restore" else "cleanup-receipt.json" + receipt_path = args.run_root.resolve() / args.run_id / receipt_name + print_private_receipt_summary(receipt_path, payload) return 0 if payload.get("status") == "pass" else 1 diff --git a/ops/run_local_canonical_postgres_rebuild.py b/ops/run_local_canonical_postgres_rebuild.py index a5d3d10..3b6363d 100755 --- a/ops/run_local_canonical_postgres_rebuild.py +++ b/ops/run_local_canonical_postgres_rebuild.py @@ -31,7 +31,7 @@ MANIFEST_DESTINATION = "/tmp/postgres-parity-manifest.sql" CANARY_LABEL = "com.livingip.teleo.canonical-rebuild-canary" DATABASE_NAME_RE = re.compile(r"[A-Za-z_][A-Za-z0-9_]{0,62}\Z") CONTAINER_PREFIX_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,39}\Z") -ALLOWED_APPLICATION_ROLES = frozenset({"kb_apply", "kb_review"}) +ALLOWED_APPLICATION_ROLES = frozenset({"kb_gate_owner", "kb_apply", "kb_review"}) ROLE_BOOLEAN_FIELDS = ( "can_login", "superuser", diff --git a/ops/verify_gcp_canonical_lifecycle.py b/ops/verify_gcp_canonical_lifecycle.py index aeb7760..b70aa3c 100644 --- a/ops/verify_gcp_canonical_lifecycle.py +++ b/ops/verify_gcp_canonical_lifecycle.py @@ -14,8 +14,12 @@ from pathlib import Path from typing import Any try: + from .private_receipt_io import print_private_receipt_summary + from .restore_gcp_generated_postgres_snapshot import DEFAULT_RESTORE_SERVICE_ACCOUNT from .verify_vps_canonical_snapshot_delta import DELTA_SCHEMA, parse_timestamp, write_private_json except ImportError: # pragma: no cover - direct script execution + from private_receipt_io import print_private_receipt_summary + from restore_gcp_generated_postgres_snapshot import DEFAULT_RESTORE_SERVICE_ACCOUNT from verify_vps_canonical_snapshot_delta import DELTA_SCHEMA, parse_timestamp, write_private_json SOURCE_SCHEMA = "livingip.sourceSnapshotReceipt.v2" @@ -99,6 +103,9 @@ def parity_evidence(details: dict[str, Any]) -> dict[str, bool]: "triggers", "types", "policies", + "role_memberships", + "object_ownership", + "object_acl", } structural_valid = required_structures <= set(structural) and all( bool(SHA256_RE.fullmatch(str((structural.get(kind) or {}).get("source") or ""))) @@ -124,11 +131,12 @@ def parity_evidence(details: dict[str, Any]) -> dict[str, bool]: "row_counts": isinstance(source_total_rows, int) and source_total_rows > 0 and source_total_rows == target_total_rows, - "table_hashes": details.get("table_mismatches") == [], + "table_hashes": details.get("table_mismatches") == [] and details.get("table_hash_problems") == [], "structural_hashes": structural_valid, "roles": details.get("application_role_mismatches") == {} and details.get("target_extra_application_roles") == [], - "extensions": details.get("required_extension_mismatches") == {}, + "extensions": details.get("required_extension_mismatches") == {} + and details.get("target_extra_extensions") == [], "performance": performance_valid and details.get("performance_problems") == [], } @@ -192,6 +200,12 @@ def blind_reasoning_evidence(reasoning: dict[str, Any], target_db: Any) -> dict[ and ((invocation.get("database_identity") or {}).get("transaction_read_only") == "on") and ((invocation.get("database_identity") or {}).get("default_transaction_read_only") == "on") and ((invocation.get("database_identity") or {}).get("ssl") is True) + and ((invocation.get("database_identity") or {}).get("sslmode") == "verify-full") + and ((invocation.get("database_identity") or {}).get("server_identity_verified") is True) + and bool((invocation.get("database_identity") or {}).get("ssl_server_name")) + and bool( + SHA256_RE.fullmatch(str((invocation.get("database_identity") or {}).get("ssl_root_cert_sha256") or "")) + ) and is_rfc1918((invocation.get("database_identity") or {}).get("server_address")) for invocation in invocations ) @@ -386,6 +400,11 @@ def verify_lifecycle( "source_delta_manifests_bound": delta_baseline.get("manifest_sha256") == (source.get("manifest") or {}).get("sha256") and delta_current.get("manifest_sha256") == (current_source.get("manifest") or {}).get("sha256"), + "source_delta_manifest_sql_bound": delta_baseline.get("manifest_sql_sha256") + == (source.get("manifest") or {}).get("manifest_sql_sha256") + and delta_current.get("manifest_sql_sha256") + == (current_source.get("manifest") or {}).get("manifest_sql_sha256") + and delta_baseline.get("manifest_sql_sha256") == delta_current.get("manifest_sql_sha256"), "source_delta_authorizations_bound": delta_baseline.get("capture_authorization_ref") == source.get("capture_authorization_ref") and delta_current.get("capture_authorization_ref") == current_source.get("capture_authorization_ref"), @@ -470,6 +489,10 @@ def verify_lifecycle( == parity.get("target_manifest_sha256"), "restore_private_tls": restore_connectivity.get("private_connectivity") is True and restore_connectivity.get("ssl") is True + and restore_connectivity.get("sslmode") == "verify-full" + and restore_connectivity.get("server_identity_verified") is True + and restore_connectivity.get("ssl_server_name") == restore_contract.get("ssl_server_name") + and restore_connectivity.get("ssl_root_cert_sha256") == restore_contract.get("ssl_root_cert_sha256") and is_rfc1918(restore_connectivity.get("server_address")), "restore_public_ip_disabled": restore_cloudsql.get("public_ip_disabled") is True, "restore_cloudsql_checks": required_true_checks( @@ -504,7 +527,10 @@ def verify_lifecycle( and restore_compute.get("expected_private_network") == restore_contract.get("private_network") and bool(restore_compute.get("network_resource")) and is_rfc1918(restore_compute.get("private_ip")) - and str(restore_compute.get("service_account") or "").endswith(".iam.gserviceaccount.com"), + and restore_contract.get("restore_service_account") == DEFAULT_RESTORE_SERVICE_ACCOUNT + and restore_compute.get("service_account") == DEFAULT_RESTORE_SERVICE_ACCOUNT, + "restore_tls_identity_contract": bool(restore_contract.get("ssl_server_name")) + and bool(SHA256_RE.fullmatch(str(restore_contract.get("ssl_root_cert_sha256") or ""))), "restore_connectivity_compute_bound": restore_connectivity.get("source_compute") == restore_compute.get("instance"), "restore_inline_parity": restore_parity.get("status") == "pass" and restore_parity.get("problems") == [], @@ -536,6 +562,10 @@ def verify_lifecycle( and parity_connectivity_proof.get("source_compute_zone") == restore_compute.get("zone") and is_rfc1918(parity_connectivity_proof.get("source_compute_private_ip")) and parity_connectivity_proof.get("server_address") == restore_connectivity.get("server_address") + and parity_connectivity_proof.get("sslmode") == "verify-full" + and parity_connectivity_proof.get("server_identity_verified") is True + and parity_connectivity_proof.get("ssl_server_name") == restore_contract.get("ssl_server_name") + and parity_connectivity_proof.get("ssl_root_cert_sha256") == restore_contract.get("ssl_root_cert_sha256") and required_true_checks( parity_connectivity_proof.get("cloudsql_control_plane_checks"), { @@ -660,17 +690,22 @@ def verify_lifecycle( "service_account", }, ), - "clone_absent": cleanup_database.get("clone_database_remaining") == 0, + "clone_absent": cleanup_database.get("existed_before") is True + and cleanup_database.get("clone_database_remaining") == 0, "cleanup_service_unchanged": cleanup_service.get("unchanged") is True and service_healthy(cleanup_service.get("before")) and cleanup_service.get("before") == cleanup_service.get("after"), } failed = sorted(name for name, passed in checks.items() if not passed) + structurally_valid = not failed return { "artifact": "gcp_canonical_snapshot_lifecycle_verification", "generated_at_utc": verification_time.isoformat(), - "status": "pass" if not failed else "fail", + "status": "prepared_external_attestation_required" if structurally_valid else "fail", + "current_tier": "T2_structural_receipt_bundle", "required_tier": "T3_live_private_gcp_staging", + "t3_receipt_ready": False, + "accepted_by_this_api": False, "target_database": target_db, "run_id": run_id, "checks": checks, @@ -696,13 +731,9 @@ def verify_lifecycle( "cleanup": sha256_file(cleanup_path), }, "strongest_claim_allowed": ( - ( - "current private GCP staging parity, ID-free Leo reasoning, cleanup, and zero post-snapshot VPS delta verified" - if delta_details.get("delta_detected") is False - else "private GCP staging parity at the declared snapshot boundary, ID-free Leo reasoning, cleanup, and explicit post-snapshot VPS delta verified" - ) - if not failed - else "lifecycle incomplete; failed checks must be repaired" + "caller-supplied lifecycle receipts are internally consistent; an independently authenticated platform receipt is still required" + if structurally_valid + else "lifecycle preparation is incomplete; failed checks must be repaired" ), "not_proven": ["production cutover", "ongoing replication", "Telegram delivery"], } @@ -746,8 +777,8 @@ def main() -> int: "strongest_claim_allowed": "lifecycle incomplete; receipts could not be validated", } write_private_json(args.output, payload) - print(json.dumps(payload, indent=2, sort_keys=True)) - return 0 if payload.get("status") == "pass" else 1 + print_private_receipt_summary(args.output, payload) + return 0 if payload.get("status") == "prepared_external_attestation_required" else 1 if __name__ == "__main__": diff --git a/ops/verify_postgres_parity_manifest.py b/ops/verify_postgres_parity_manifest.py index fb533d4..6e4eeb0 100755 --- a/ops/verify_postgres_parity_manifest.py +++ b/ops/verify_postgres_parity_manifest.py @@ -7,20 +7,26 @@ import argparse import hashlib import ipaddress import json +import re from datetime import UTC, datetime from pathlib import Path from typing import Any try: - from .private_receipt_io import write_private_json + from .private_receipt_io import print_private_receipt_summary, write_private_json except ImportError: # pragma: no cover - direct script execution - from private_receipt_io import write_private_json + from private_receipt_io import print_private_receipt_summary, write_private_json + +REVIEWED_MANIFEST_SQL_SHA256 = "430390129bc8980c4ece9939f2aee18e8c40e9da4b226782579349d5c45c5301" SINGLETON_KINDS = { "identity", "schemas", "extensions", "application_roles", + "role_memberships", + "object_ownership", + "object_acl", "columns", "constraints", "indexes", @@ -85,7 +91,18 @@ def load_manifest(path: Path) -> dict[str, Any]: def item_map(row: dict[str, Any], key: str) -> dict[str, dict[str, Any]]: - return {str(item[key]): item for item in row.get("items", [])} + items = row.get("items", []) + if not isinstance(items, list): + raise ValueError(f"manifest {row.get('kind')} items must be a list") + mapped: dict[str, dict[str, Any]] = {} + for item in items: + if not isinstance(item, dict) or key not in item: + raise ValueError(f"manifest {row.get('kind')} item is missing {key}") + item_key = str(item[key]) + if item_key in mapped: + raise ValueError(f"manifest {row.get('kind')} contains duplicate {key}={item_key}") + mapped[item_key] = item + return mapped def compare_manifests( @@ -120,9 +137,14 @@ def compare_manifests( + json.dumps(sorted(set(target_tables) - set(source_tables))) ) table_mismatches = [] + table_hash_problems = [] for table in sorted(set(source_tables) & set(target_tables)): source_row = source_tables[table] target_row = target_tables[table] + for side, row in (("source", source_row), ("target", target_row)): + rowset_md5 = row.get("rowset_md5") + if not isinstance(rowset_md5, str) or not rowset_md5: + table_hash_problems.append(f"{table}:{side}:rowset_md5_missing") mismatch = { field: {"source": source_row.get(field), "target": target_row.get(field)} for field in ("row_count", "rowset_md5") @@ -132,6 +154,8 @@ def compare_manifests( table_mismatches.append({"table": table, "mismatch": mismatch}) if table_mismatches: problems.append(f"table_content_mismatches={len(table_mismatches)}") + if table_hash_problems: + problems.append(f"table_hash_problems={len(table_hash_problems)}") structural_hashes: dict[str, dict[str, str]] = {} for kind in ( @@ -145,6 +169,9 @@ def compare_manifests( "triggers", "types", "policies", + "role_memberships", + "object_ownership", + "object_acl", ): source_hash = canonical_hash(source_singleton[kind].get("items", [])) target_hash = canonical_hash(target_singleton[kind].get("items", [])) @@ -155,19 +182,19 @@ def compare_manifests( source_extensions = item_map(source_singleton["extensions"], "name") target_extensions = item_map(target_singleton["extensions"], "name") extension_mismatches = { - name: {"source": item, "target": target_extensions.get(name)} - for name, item in source_extensions.items() - if target_extensions.get(name) != item + name: {"source": source_extensions.get(name), "target": target_extensions.get(name)} + for name in sorted(set(source_extensions) | set(target_extensions)) + if source_extensions.get(name) != target_extensions.get(name) } if extension_mismatches: - problems.append(f"required_extension_mismatches={len(extension_mismatches)}") + problems.append(f"extension_mismatches={len(extension_mismatches)}") source_roles = item_map(source_singleton["application_roles"], "name") target_roles = item_map(target_singleton["application_roles"], "name") role_mismatches = { - name: {"source": item, "target": target_roles.get(name)} - for name, item in source_roles.items() - if target_roles.get(name) != item + name: {"source": source_roles.get(name), "target": target_roles.get(name)} + for name in sorted(set(source_roles) | set(target_roles)) + if source_roles.get(name) != target_roles.get(name) } if role_mismatches: problems.append(f"application_role_mismatches={len(role_mismatches)}") @@ -217,8 +244,10 @@ def compare_manifests( "source_total_rows": sum(int(row["row_count"]) for row in source_tables.values()), "target_total_rows": sum(int(row["row_count"]) for row in target_tables.values()), "table_mismatches": table_mismatches, + "table_hash_problems": table_hash_problems, "structural_hashes": structural_hashes, "required_extension_mismatches": extension_mismatches, + "target_extra_extensions": sorted(set(target_extensions) - set(source_extensions)), "application_role_mismatches": role_mismatches, "target_extra_application_roles": sorted(set(target_roles) - set(source_roles)), "performance": performance_rows, @@ -293,6 +322,14 @@ def load_connectivity( problems.append("gcp_target_manifest_tls_not_proven") if proof.get("ssl") is not True: problems.append("gcp_connectivity_tls_not_proven") + if proof.get("sslmode") != "verify-full": + problems.append("gcp_connectivity_server_identity_mode_not_proven") + if proof.get("server_identity_verified") is not True: + problems.append("gcp_connectivity_server_identity_not_proven") + if not proof.get("ssl_server_name"): + problems.append("gcp_connectivity_ssl_server_name_missing") + if not re.fullmatch(r"[0-9a-f]{64}", str(proof.get("ssl_root_cert_sha256") or "")): + problems.append("gcp_connectivity_ssl_root_cert_hash_missing") return problems, {"required": True, "path": str(path), "proof": proof} @@ -360,7 +397,7 @@ def main() -> int: ], } write_private_json(args.output, payload) - print(json.dumps(payload, indent=2, sort_keys=True)) + print_private_receipt_summary(args.output, payload) return 0 if status == "pass" else 1 diff --git a/ops/verify_vps_canonical_snapshot_delta.py b/ops/verify_vps_canonical_snapshot_delta.py index b4dbe7a..4f388f4 100644 --- a/ops/verify_vps_canonical_snapshot_delta.py +++ b/ops/verify_vps_canonical_snapshot_delta.py @@ -16,12 +16,24 @@ try: SOURCE_SNAPSHOT_RECEIPT_SCHEMA, build_provenance_binding, ) - from .private_receipt_io import write_private_json - from .verify_postgres_parity_manifest import SINGLETON_KINDS, canonical_hash, file_sha256, load_manifest + from .private_receipt_io import print_private_receipt_summary, write_private_json + from .verify_postgres_parity_manifest import ( + REVIEWED_MANIFEST_SQL_SHA256, + SINGLETON_KINDS, + canonical_hash, + file_sha256, + load_manifest, + ) except ImportError: # pragma: no cover - direct script execution from capture_vps_canonical_postgres_snapshot import SOURCE_SNAPSHOT_RECEIPT_SCHEMA, build_provenance_binding - from private_receipt_io import write_private_json - from verify_postgres_parity_manifest import SINGLETON_KINDS, canonical_hash, file_sha256, load_manifest + from private_receipt_io import print_private_receipt_summary, write_private_json + from verify_postgres_parity_manifest import ( + REVIEWED_MANIFEST_SQL_SHA256, + SINGLETON_KINDS, + canonical_hash, + file_sha256, + load_manifest, + ) DELTA_SCHEMA = "livingip.vpsCanonicalSnapshotDelta.v1" SHA256_RE = re.compile(r"[0-9a-f]{64}\Z") @@ -124,6 +136,7 @@ def validate_capture( "binding_algorithm": binding.get("algorithm") == "sha256", "dump_sha256": bool(SHA256_RE.fullmatch(str(dump.get("sha256") or ""))), "manifest_sql_sha256": bool(SHA256_RE.fullmatch(str(manifest_receipt.get("manifest_sql_sha256") or ""))), + "manifest_sql_reviewed": manifest_receipt.get("manifest_sql_sha256") == REVIEWED_MANIFEST_SQL_SHA256, "source_context_sha256": bool(SHA256_RE.fullmatch(str(source_context.get("sha256") or ""))), "snapshot_identity": all( bool(snapshot.get(field)) @@ -186,6 +199,10 @@ def verify_delta( raise ValueError("current source identity does not match the restored baseline source") if current_snapshot.get("system_identifier") != baseline_snapshot.get("system_identifier"): raise ValueError("current source PostgreSQL system identifier changed") + baseline_manifest_sql_sha256 = str((baseline_receipt.get("manifest") or {}).get("manifest_sql_sha256") or "") + current_manifest_sql_sha256 = str((current_receipt.get("manifest") or {}).get("manifest_sql_sha256") or "") + if baseline_manifest_sql_sha256 != current_manifest_sql_sha256: + raise ValueError("baseline and current captures used different manifest SQL algorithms") baseline = manifest_fingerprint(baseline_manifest) current = manifest_fingerprint(current_manifest) @@ -221,6 +238,7 @@ def verify_delta( "captured_at_utc": baseline_snapshot["captured_at_utc"], "capture_receipt_sha256": file_sha256(baseline_receipt_path), "manifest_sha256": file_sha256(baseline_manifest_path), + "manifest_sql_sha256": baseline_manifest_sql_sha256, **baseline, }, "current": { @@ -229,6 +247,7 @@ def verify_delta( "captured_at_utc": current_snapshot["captured_at_utc"], "capture_receipt_sha256": file_sha256(current_receipt_path), "manifest_sha256": file_sha256(current_manifest_path), + "manifest_sql_sha256": current_manifest_sql_sha256, **current, }, "delta": { @@ -276,7 +295,7 @@ def main() -> int: "strongest_claim_allowed": "post-snapshot VPS delta is not proven", } write_private_json(args.output, payload) - print(json.dumps(payload, indent=2, sort_keys=True)) + print_private_receipt_summary(args.output, payload) return 0 if payload.get("status") == "pass" else 1 diff --git a/scripts/run_gcp_generated_db_direct_claim_suite.py b/scripts/run_gcp_generated_db_direct_claim_suite.py index 79b0704..dc5f465 100644 --- a/scripts/run_gcp_generated_db_direct_claim_suite.py +++ b/scripts/run_gcp_generated_db_direct_claim_suite.py @@ -35,11 +35,15 @@ if __name__ == "__main__" and should_reexec_in_hermes(Path(sys.executable)): ) HERE = Path(__file__).resolve().parent +REPO_ROOT = HERE.parent +sys.path.insert(0, str(REPO_ROOT)) sys.path.insert(0, str(HERE)) import run_leo_clone_bound_handler_checkpoint as bound # noqa: E402 import working_leo_open_ended_benchmark as benchmark # noqa: E402 +from ops.verify_postgres_parity_manifest import REVIEWED_MANIFEST_SQL_SHA256 # noqa: E402 + SCHEMA = "livingip.gcpGeneratedDbDirectClaimSuite.v1" SOURCE_COMPUTE = "teleo-prod-1" DEFAULT_SERVICE = "leoclean-gcp-prod-parallel.service" @@ -48,7 +52,6 @@ DEFAULT_PROJECT = "teleo-501523" DEFAULT_SECRET = "gcp-teleo-pgvector-standby-postgres-password" DEFAULT_MANIFEST_SQL = HERE.parent / "ops" / "postgres_parity_manifest.sql" REVIEWED_CLOUDSQL_TOOL_SHA256 = "7d2074a0fc5f0d48fbf8f7905a72ead8f2696c86a041fa43e078fff5500baa51" -REVIEWED_MANIFEST_SQL_SHA256 = "8b8cdc25d54fdd8de05eb38c6e4423d2836953eb6012d4545f5c9c71b5f0150a" COUNT_READBACK_RE = re.compile( r"DB readback:\s*claims:\s*`?(?P\d+)`?;\s*" r"sources:\s*`?(?P\d+)`?;\s*" diff --git a/tests/test_attest_gcp_reasoning_compute.py b/tests/test_attest_gcp_reasoning_compute.py index b3906fc..49190f7 100644 --- a/tests/test_attest_gcp_reasoning_compute.py +++ b/tests/test_attest_gcp_reasoning_compute.py @@ -38,7 +38,7 @@ def compute_identity() -> dict: "zone_resource": "projects/123456789/zones/europe-west6-a", "network_resource": "projects/123456789/networks/teleo-staging-net", "private_ip": "10.61.0.2", - "service_account": "teleo-runtime@teleo-501523.iam.gserviceaccount.com", + "service_account": attestation.DEFAULT_RESTORE_SERVICE_ACCOUNT, "expected_private_network": "projects/teleo-501523/global/networks/teleo-staging-net", "checks": { "project": True, @@ -58,13 +58,14 @@ def test_attestation_binds_reasoning_hash_to_live_metadata(tmp_path: Path, monke receipt_path = reasoning_receipt(tmp_path / "reasoning.json") observed = {} - def fake_identity(project, instance, zone, network, *, timeout): + def fake_identity(project, instance, zone, network, service_account, *, timeout): observed.update( { "project": project, "instance": instance, "zone": zone, "network": network, + "service_account": service_account, "timeout": timeout, } ) @@ -88,6 +89,7 @@ def test_attestation_binds_reasoning_hash_to_live_metadata(tmp_path: Path, monke "instance": "teleo-prod-1", "zone": "europe-west6-a", "network": "projects/teleo-501523/global/networks/teleo-staging-net", + "service_account": attestation.DEFAULT_RESTORE_SERVICE_ACCOUNT, "timeout": 10.0, } diff --git a/tests/test_capture_vps_canonical_postgres_snapshot.py b/tests/test_capture_vps_canonical_postgres_snapshot.py index 762d14e..fed31b8 100644 --- a/tests/test_capture_vps_canonical_postgres_snapshot.py +++ b/tests/test_capture_vps_canonical_postgres_snapshot.py @@ -48,6 +48,11 @@ def test_manifest_hash_order_is_collation_independent() -> None: assert 'collate "C"' in manifest assert "'server_address', host(inet_server_addr())" in manifest + assert "'kb_gate_owner', 'kb_apply', 'kb_review'" in manifest + assert "'kind', 'role_memberships'" in manifest + assert "'kind', 'object_ownership'" in manifest + assert "'kind', 'object_acl'" in manifest + assert "aclexplode(attribute.attacl)" in manifest def test_toc_counts_database_objects(tmp_path: Path) -> None: diff --git a/tests/test_private_receipt_io.py b/tests/test_private_receipt_io.py new file mode 100644 index 0000000..4df55cb --- /dev/null +++ b/tests/test_private_receipt_io.py @@ -0,0 +1,29 @@ +from __future__ import annotations + +import json +from pathlib import Path + +from ops.private_receipt_io import print_private_receipt_summary, write_private_json + + +def test_stdout_summary_never_replays_private_receipt_payload(tmp_path: Path, capsys) -> None: + output = tmp_path / "private" / "receipt.json" + payload = { + "artifact": "private_test_receipt", + "status": "pass", + "private_claim_body": "must-not-appear-on-stdout", + "database_password": "must-not-appear-on-stdout-either", + } + + write_private_json(output, payload) + print_private_receipt_summary(output, payload) + + stdout = capsys.readouterr().out + summary = json.loads(stdout) + assert "must-not-appear" not in stdout + assert summary["artifact"] == "private_test_receipt" + assert summary["status"] == "pass" + assert summary["output_name"] == output.name + assert str(output.parent) not in stdout + assert len(summary["output_sha256"]) == 64 + assert output.stat().st_mode & 0o777 == 0o600 diff --git a/tests/test_restore_gcp_generated_postgres_snapshot.py b/tests/test_restore_gcp_generated_postgres_snapshot.py index 5c152b9..e1c9535 100644 --- a/tests/test_restore_gcp_generated_postgres_snapshot.py +++ b/tests/test_restore_gcp_generated_postgres_snapshot.py @@ -49,7 +49,27 @@ def manifest_rows(database: str, *, target: bool, row_count: int = 1) -> list[di "inherit": True, "replication": False, "bypass_rls": False, - } + }, + { + "name": "kb_gate_owner", + "can_login": False, + "superuser": False, + "create_db": False, + "create_role": False, + "inherit": True, + "replication": False, + "bypass_rls": False, + }, + { + "name": "kb_review", + "can_login": True, + "superuser": False, + "create_db": False, + "create_role": False, + "inherit": True, + "replication": False, + "bypass_rls": False, + }, ], }, ] @@ -65,6 +85,9 @@ def manifest_rows(database: str, *, target: bool, row_count: int = 1) -> list[di "triggers", "types", "policies", + "role_memberships", + "object_ownership", + "object_acl", ) ) rows.extend( @@ -97,6 +120,8 @@ def args_for(tmp_path: Path, **overrides) -> argparse.Namespace: source_manifest = tmp_path / "source-manifest.jsonl" source_manifest.write_text(manifest_text("teleo", target=False), encoding="utf-8") manifest_sql = Path("ops/postgres_parity_manifest.sql").resolve() + ssl_root_cert = tmp_path / "cloudsql-server-ca.pem" + ssl_root_cert.write_text("test-only-ca\n", encoding="utf-8") source_context_sha256 = "4" * 64 snapshot = { "exported_snapshot_id": "00000003-000001A2-1", @@ -166,6 +191,9 @@ def args_for(tmp_path: Path, **overrides) -> argparse.Namespace: "expected_private_network": restore.DEFAULT_PRIVATE_NETWORK, "expected_compute_instance": restore.DEFAULT_COMPUTE_INSTANCE, "expected_compute_zone": restore.DEFAULT_COMPUTE_ZONE, + "expected_restore_service_account": restore.DEFAULT_RESTORE_SERVICE_ACCOUNT, + "ssl_server_name": "teleo-pgvector-standby.private.teleo.internal", + "ssl_root_cert": ssl_root_cert, "expected_source_ssh_target": restore.DEFAULT_SOURCE_SSH_TARGET, "expected_source_container": restore.DEFAULT_SOURCE_CONTAINER, "expected_source_database": restore.DEFAULT_SOURCE_DATABASE, @@ -203,7 +231,7 @@ def install_restore_fakes( monkeypatch.setattr( restore, "gce_compute_identity", - lambda project, instance, zone, expected_private_network, **_kwargs: { + lambda project, instance, zone, expected_private_network, expected_service_account, **_kwargs: { "project": project, "project_number": "785938879453", "instance": instance, @@ -212,7 +240,7 @@ def install_restore_fakes( "expected_private_network": expected_private_network, "private_ip": "10.61.0.2", "network_resource": "projects/123456789/networks/teleo-staging-net", - "service_account": "teleo-runtime@teleo-501523.iam.gserviceaccount.com", + "service_account": expected_service_account, "checks": { "project": True, "project_number": True, @@ -492,6 +520,9 @@ def test_cleanup_is_bound_to_passing_restore_receipt( "expected_private_network", "expected_compute_instance", "expected_compute_zone", + "expected_restore_service_account", + "ssl_server_name", + "ssl_root_cert", "expected_source_ssh_target", "expected_source_container", "expected_source_database", @@ -534,6 +565,9 @@ def test_cleanup_rejects_target_topology_drift_before_credentials( "expected_private_network", "expected_compute_instance", "expected_compute_zone", + "expected_restore_service_account", + "ssl_server_name", + "ssl_root_cert", "expected_source_ssh_target", "expected_source_container", "expected_source_database", @@ -584,6 +618,9 @@ def test_cleanup_retains_failure_receipt_when_post_drop_readback_fails( "expected_private_network", "expected_compute_instance", "expected_compute_zone", + "expected_restore_service_account", + "ssl_server_name", + "ssl_root_cert", "expected_source_ssh_target", "expected_source_container", "expected_source_database", @@ -631,7 +668,10 @@ def test_restore_dump_uses_tls_no_owner_no_acl_and_keeps_secret_out_of_argv( assert captured["command"][0] == "pg_restore" assert captured["command"][1] == "-d" - assert "sslmode=require" in captured["command"][2] + assert "sslmode=verify-full" in captured["command"][2] + assert "hostaddr='10.61.0.3'" in captured["command"][2] + assert "host='teleo-pgvector-standby.private.teleo.internal'" in captured["command"][2] + assert f"sslrootcert='{args.ssl_root_cert}'" in captured["command"][2] assert "--no-owner" in captured["command"] assert "--no-acl" in captured["command"] assert "--exit-on-error" in captured["command"] @@ -695,7 +735,7 @@ def test_gce_compute_identity_is_bound_to_metadata_instance_zone_and_network( "instance/zone": "projects/123456789/zones/europe-west6-a", "instance/network-interfaces/0/network": "projects/785938879453/networks/teleo-staging-net", "instance/network-interfaces/0/ip": "10.61.0.2", - "instance/service-accounts/default/email": "teleo-runtime@teleo-501523.iam.gserviceaccount.com", + "instance/service-accounts/default/email": restore.DEFAULT_RESTORE_SERVICE_ACCOUNT, } monkeypatch.setattr(restore, "gce_metadata_value", lambda path, **_kwargs: metadata[path]) @@ -704,6 +744,7 @@ def test_gce_compute_identity_is_bound_to_metadata_instance_zone_and_network( "teleo-prod-1", "europe-west6-a", restore.DEFAULT_PRIVATE_NETWORK, + restore.DEFAULT_RESTORE_SERVICE_ACCOUNT, timeout=2.0, ) @@ -718,6 +759,7 @@ def test_gce_compute_identity_is_bound_to_metadata_instance_zone_and_network( "different-instance", "europe-west6-a", restore.DEFAULT_PRIVATE_NETWORK, + restore.DEFAULT_RESTORE_SERVICE_ACCOUNT, timeout=2.0, ) @@ -729,6 +771,7 @@ def test_gce_compute_identity_is_bound_to_metadata_instance_zone_and_network( "teleo-prod-1", "europe-west6-a", restore.DEFAULT_PRIVATE_NETWORK, + restore.DEFAULT_RESTORE_SERVICE_ACCOUNT, timeout=2.0, ) @@ -739,6 +782,19 @@ def test_gce_compute_identity_is_bound_to_metadata_instance_zone_and_network( "teleo-prod-1", "europe-west6-a", "projects/other-project/global/networks/teleo-staging-net", + restore.DEFAULT_RESTORE_SERVICE_ACCOUNT, + timeout=2.0, + ) + + metadata["instance/network-interfaces/0/network"] = "projects/785938879453/networks/teleo-staging-net" + metadata["instance/service-accounts/default/email"] = "project-owner@teleo-501523.iam.gserviceaccount.com" + with pytest.raises(restore.RestoreError, match="service_account"): + restore.gce_compute_identity( + "teleo-501523", + "teleo-prod-1", + "europe-west6-a", + restore.DEFAULT_PRIVATE_NETWORK, + restore.DEFAULT_RESTORE_SERVICE_ACCOUNT, timeout=2.0, ) diff --git a/tests/test_run_local_canonical_postgres_rebuild.py b/tests/test_run_local_canonical_postgres_rebuild.py index 4bd4860..b03cb69 100644 --- a/tests/test_run_local_canonical_postgres_rebuild.py +++ b/tests/test_run_local_canonical_postgres_rebuild.py @@ -137,6 +137,9 @@ def manifest_rows() -> list[dict]: {"kind": "schemas", "items": ["kb_stage", "public"]}, {"kind": "extensions", "items": []}, {"kind": "application_roles", "items": [role]}, + {"kind": "role_memberships", "items": []}, + {"kind": "object_ownership", "items": []}, + {"kind": "object_acl", "items": []}, ] rows.extend( {"kind": kind, "items": []} diff --git a/tests/test_run_local_genesis_ledger_rebuild.py b/tests/test_run_local_genesis_ledger_rebuild.py index 4919b1c..865ed98 100644 --- a/tests/test_run_local_genesis_ledger_rebuild.py +++ b/tests/test_run_local_genesis_ledger_rebuild.py @@ -64,6 +64,9 @@ def manifest_rows(database: str = "teleo_fixture") -> list[dict]: {"kind": "schemas", "items": ["kb_stage", "public"]}, {"kind": "extensions", "items": []}, {"kind": "application_roles", "items": [role]}, + {"kind": "role_memberships", "items": []}, + {"kind": "object_ownership", "items": []}, + {"kind": "object_acl", "items": []}, ] rows.extend( {"kind": kind, "items": []} diff --git a/tests/test_verify_gcp_canonical_lifecycle.py b/tests/test_verify_gcp_canonical_lifecycle.py index 28df37b..b2a26da 100644 --- a/tests/test_verify_gcp_canonical_lifecycle.py +++ b/tests/test_verify_gcp_canonical_lifecycle.py @@ -3,8 +3,10 @@ import json from datetime import UTC, datetime from pathlib import Path +from ops.restore_gcp_generated_postgres_snapshot import DEFAULT_RESTORE_SERVICE_ACCOUNT from ops.verify_gcp_canonical_lifecycle import EXPECTED_BLIND_ROW_IDS from ops.verify_gcp_canonical_lifecycle import verify_lifecycle as _verify_lifecycle +from ops.verify_postgres_parity_manifest import REVIEWED_MANIFEST_SQL_SHA256 VERIFY_NOW = datetime(2026, 7, 15, 1, 3, tzinfo=UTC) @@ -45,6 +47,9 @@ def lifecycle_receipts(tmp_path: Path) -> dict[str, Path]: "triggers", "types", "policies", + "role_memberships", + "object_ownership", + "object_acl", ), 1, ) @@ -57,8 +62,10 @@ def lifecycle_receipts(tmp_path: Path) -> dict[str, Path]: "source_total_rows": 52167, "target_total_rows": 52167, "table_mismatches": [], + "table_hash_problems": [], "structural_hashes": structural_hashes, "required_extension_mismatches": {}, + "target_extra_extensions": [], "application_role_mismatches": {}, "target_extra_application_roles": [], "performance": [{"query": "count_claims", "source_ms": 1.0, "target_ms": 2.0, "source_ratio": 0.4}], @@ -90,7 +97,12 @@ def lifecycle_receipts(tmp_path: Path) -> dict[str, Path]: "production_db_mutated": False, "telegram_send_attempted": False, "dump": {"sha256": "a" * 64}, - "manifest": {"sha256": "b" * 64, "table_count": 39, "total_rows": 52167}, + "manifest": { + "sha256": "b" * 64, + "manifest_sql_sha256": REVIEWED_MANIFEST_SQL_SHA256, + "table_count": 39, + "total_rows": 52167, + }, "provenance_binding": {"algorithm": "sha256", "payload": {"bound": True}, "sha256": "c" * 64}, } source_path = write_json(tmp_path / "source.json", source) @@ -146,6 +158,9 @@ def lifecycle_receipts(tmp_path: Path) -> dict[str, Path]: "private_network": "projects/teleo-501523/global/networks/teleo-staging-net", "compute_instance": "teleo-prod-1", "compute_zone": "europe-west6-a", + "restore_service_account": DEFAULT_RESTORE_SERVICE_ACCOUNT, + "ssl_server_name": "teleo-pgvector-standby.private.teleo.internal", + "ssl_root_cert_sha256": "9" * 64, "source": source["source"], "password_secret": "gcp-teleo-pgvector-standby-postgres-password", "service": "leoclean-gcp-prod-parallel.service", @@ -155,6 +170,10 @@ def lifecycle_receipts(tmp_path: Path) -> dict[str, Path]: "connectivity": { "private_connectivity": True, "ssl": True, + "sslmode": "verify-full", + "ssl_server_name": "teleo-pgvector-standby.private.teleo.internal", + "ssl_root_cert_sha256": "9" * 64, + "server_identity_verified": True, "server_address": "10.61.0.3", "source_compute": "teleo-prod-1", }, @@ -184,7 +203,7 @@ def lifecycle_receipts(tmp_path: Path) -> dict[str, Path]: "expected_private_network": "projects/teleo-501523/global/networks/teleo-staging-net", "network_resource": "projects/123456789/networks/teleo-staging-net", "private_ip": "10.61.0.2", - "service_account": "teleo-runtime@teleo-501523.iam.gserviceaccount.com", + "service_account": DEFAULT_RESTORE_SERVICE_ACCOUNT, "checks": { "project": True, "project_number": True, @@ -231,6 +250,10 @@ def lifecycle_receipts(tmp_path: Path) -> dict[str, Path]: "source_compute_zone": "europe-west6-a", "source_compute_private_ip": "10.61.0.2", "server_address": "10.61.0.3", + "sslmode": "verify-full", + "ssl_server_name": "teleo-pgvector-standby.private.teleo.internal", + "ssl_root_cert_sha256": "9" * 64, + "server_identity_verified": True, "project": "teleo-501523", "cloudsql_instance": "teleo-pgvector-standby", "private_network": "projects/teleo-501523/global/networks/teleo-staging-net", @@ -347,6 +370,10 @@ def lifecycle_receipts(tmp_path: Path) -> dict[str, Path]: "transaction_read_only": "on", "default_transaction_read_only": "on", "ssl": True, + "sslmode": "verify-full", + "ssl_server_name": "teleo-pgvector-standby.private.teleo.internal", + "ssl_root_cert_sha256": "9" * 64, + "server_identity_verified": True, "server_address": "10.61.0.3", }, } @@ -403,6 +430,7 @@ def lifecycle_receipts(tmp_path: Path) -> dict[str, Path]: "run_id": "canonical-current-test", "capture_receipt_sha256": sha256(source_path), "manifest_sha256": "b" * 64, + "manifest_sql_sha256": REVIEWED_MANIFEST_SQL_SHA256, "capture_authorization_ref": "codex-delegation:test1234", "captured_at_utc": "2026-07-15T00:59:00+00:00", "table_count": 39, @@ -414,6 +442,7 @@ def lifecycle_receipts(tmp_path: Path) -> dict[str, Path]: "run_id": "canonical-current-postflight", "capture_receipt_sha256": sha256(current_source_path), "manifest_sha256": "b" * 64, + "manifest_sql_sha256": REVIEWED_MANIFEST_SQL_SHA256, "capture_authorization_ref": "codex-delegation:postflight1234", "captured_at_utc": "2026-07-15T01:02:00+00:00", "table_count": 39, @@ -438,7 +467,7 @@ def lifecycle_receipts(tmp_path: Path) -> dict[str, Path]: "completed_at_utc": "2026-07-15T01:01:30+00:00", "run_id": run_id, "target_database": target_db, - "database": {"clone_database_remaining": 0}, + "database": {"existed_before": True, "clone_database_remaining": 0}, "live_service": {"before": healthy_service, "after": healthy_service, "unchanged": True}, "restore_receipt": {"sha256": sha256(restore_path)}, "execution_contract": restore["execution_contract"], @@ -460,14 +489,19 @@ def lifecycle_receipts(tmp_path: Path) -> dict[str, Path]: } -def test_bound_live_lifecycle_passes(tmp_path: Path) -> None: +def test_caller_authored_bundle_is_preparation_only_even_when_all_checks_pass(tmp_path: Path) -> None: paths = lifecycle_receipts(tmp_path) result = verify_lifecycle(**paths) - assert result["status"] == "pass" + assert result["status"] == "prepared_external_attestation_required" assert result["failed_checks"] == [] assert result["checks"]["clone_absent"] is True + assert result["current_tier"] == "T2_structural_receipt_bundle" + assert result["required_tier"] == "T3_live_private_gcp_staging" + assert result["t3_receipt_ready"] is False + assert result["accepted_by_this_api"] is False + assert "current private GCP staging" not in result["strongest_claim_allowed"] def test_local_parity_or_missing_cleanup_fails_closed(tmp_path: Path) -> None: @@ -488,6 +522,18 @@ def test_local_parity_or_missing_cleanup_fails_closed(tmp_path: Path) -> None: assert {"parity_pass", "clone_absent"} <= set(result["failed_checks"]) +def test_cleanup_cannot_pass_when_the_named_clone_never_existed(tmp_path: Path) -> None: + paths = lifecycle_receipts(tmp_path) + cleanup = json.loads(paths["cleanup_path"].read_text()) + cleanup["database"]["existed_before"] = False + write_json(paths["cleanup_path"], cleanup) + + result = verify_lifecycle(**paths) + + assert result["status"] == "fail" + assert "clone_absent" in result["failed_checks"] + + def test_mismatched_source_or_reasoning_receipt_fails_closed(tmp_path: Path) -> None: paths = lifecycle_receipts(tmp_path) restore = json.loads(paths["restore_path"].read_text()) @@ -529,6 +575,28 @@ def test_reasoning_compute_attestation_must_bind_live_metadata(tmp_path: Path) - assert "reasoning_compute_identity_bound" in result["failed_checks"] +def test_project_owner_shaped_service_account_cannot_pass_preparation(tmp_path: Path) -> None: + paths = lifecycle_receipts(tmp_path) + project_owner = "project-owner@teleo-501523.iam.gserviceaccount.com" + restore = json.loads(paths["restore_path"].read_text()) + restore["target"]["compute"]["service_account"] = project_owner + restore["execution_contract"]["restore_service_account"] = project_owner + write_json(paths["restore_path"], restore) + attestation = json.loads(paths["reasoning_compute_path"].read_text()) + attestation["compute"]["service_account"] = project_owner + write_json(paths["reasoning_compute_path"], attestation) + cleanup = json.loads(paths["cleanup_path"].read_text()) + cleanup["target"]["compute"]["service_account"] = project_owner + cleanup["execution_contract"]["restore_service_account"] = project_owner + cleanup["restore_receipt"]["sha256"] = sha256(paths["restore_path"]) + write_json(paths["cleanup_path"], cleanup) + + result = verify_lifecycle(**paths) + + assert result["status"] == "fail" + assert "restore_compute_contract" in result["failed_checks"] + + def test_source_postflight_must_be_newer_than_cleanup_completion(tmp_path: Path) -> None: paths = lifecycle_receipts(tmp_path) cleanup = json.loads(paths["cleanup_path"].read_text()) @@ -578,6 +646,21 @@ def test_reasoning_must_retrieve_expected_claim_sources_and_preserve_handle(tmp_ assert result["reasoning_producer_checks"]["reply_uses_only_m3taversal_handle"] is False +def test_reasoning_wrapper_must_authenticate_the_cloudsql_server(tmp_path: Path) -> None: + paths = lifecycle_receipts(tmp_path) + reasoning = json.loads(paths["reasoning_path"].read_text()) + for invocation in reasoning["wrapper_tool_proof"]["invocations"]: + invocation["database_identity"]["sslmode"] = "require" + invocation["database_identity"]["server_identity_verified"] = False + write_json(paths["reasoning_path"], reasoning) + + result = verify_lifecycle(**paths) + + assert result["status"] == "fail" + assert "reasoning_producer_evidence" in result["failed_checks"] + assert result["reasoning_producer_checks"]["wrapper_invocations"] is False + + def test_lifecycle_requires_restore_parity_reasoning_cleanup_chronology(tmp_path: Path) -> None: paths = lifecycle_receipts(tmp_path) cleanup = json.loads(paths["cleanup_path"].read_text()) diff --git a/tests/test_verify_postgres_parity_manifest.py b/tests/test_verify_postgres_parity_manifest.py index d016aee..37f5045 100644 --- a/tests/test_verify_postgres_parity_manifest.py +++ b/tests/test_verify_postgres_parity_manifest.py @@ -28,7 +28,23 @@ def manifest_rows( {"kind": "extensions", "items": [{"name": "plpgsql", "version": "1.0"}]}, { "kind": "application_roles", - "items": [{"name": "kb_apply", "can_login": True, "superuser": False}], + "items": [ + {"name": "kb_apply", "can_login": True, "superuser": False}, + {"name": "kb_gate_owner", "can_login": False, "superuser": False}, + {"name": "kb_review", "can_login": True, "superuser": False}, + ], + }, + { + "kind": "role_memberships", + "items": [{"role": "kb_gate_owner", "member": "kb_apply", "admin_option": False}], + }, + { + "kind": "object_ownership", + "items": [{"object_type": "schema", "schema": "public", "name": "public", "owner": "kb_gate_owner"}], + }, + { + "kind": "object_acl", + "items": [{"object_type": "schema", "name": "public", "grantee": "kb_apply", "privilege_type": "USAGE"}], }, {"kind": "columns", "items": [{"schema": "public", "table": "claims", "name": "id"}]}, {"kind": "constraints", "items": [{"schema": "public", "table": "claims", "name": "claims_pkey"}]}, @@ -94,6 +110,8 @@ def test_matching_local_manifests_pass(tmp_path: Path) -> None: assert len(payload["target_manifest_sha256"]) == 64 assert payload["private_connectivity"]["status"] == "not_applicable_local_restore" assert (tmp_path / "result.json").stat().st_mode & 0o777 == 0o600 + assert '"source_database"' not in completed.stdout + assert '"output_sha256"' in completed.stdout def test_row_hash_and_role_mismatches_fail(tmp_path: Path) -> None: @@ -103,7 +121,32 @@ def test_row_hash_and_role_mismatches_fail(tmp_path: Path) -> None: assert completed.returncode == 1 assert "table_content_mismatches=1" in payload["problems"] + assert "application_role_mismatches=3" in payload["problems"] + + +def test_missing_row_hash_and_extra_roles_or_extensions_fail(tmp_path: Path) -> None: + target = manifest_rows(database="teleo_copy", rowset_md5="") + target[2]["items"].append({"name": "unreviewed", "version": "1.0"}) + target[3]["items"].append({"name": "project_owner", "can_login": True, "superuser": True}) + + completed, payload = run_verifier(tmp_path, manifest_rows(database="teleo"), target) + + assert completed.returncode == 1 + assert "table_hash_problems=1" in payload["problems"] + assert "extension_mismatches=1" in payload["problems"] assert "application_role_mismatches=1" in payload["problems"] + assert payload["details"]["target_extra_extensions"] == ["unreviewed"] + assert payload["details"]["target_extra_application_roles"] == ["project_owner"] + + +def test_authorization_ownership_or_acl_drift_fails(tmp_path: Path) -> None: + target = manifest_rows(database="teleo_copy") + target[5]["items"][0]["owner"] = "postgres" + + completed, payload = run_verifier(tmp_path, manifest_rows(database="teleo"), target) + + assert completed.returncode == 1 + assert "object_ownership_hash_mismatch" in payload["problems"] def test_gcp_scope_requires_private_connectivity_receipt(tmp_path: Path) -> None: @@ -132,6 +175,10 @@ def test_gcp_scope_accepts_matching_private_connectivity_receipt(tmp_path: Path) "target_database": "teleo_gcp_copy", "server_address": "10.61.0.3", "ssl": True, + "sslmode": "verify-full", + "ssl_server_name": "teleo-pgvector-standby.private.teleo.internal", + "ssl_root_cert_sha256": "a" * 64, + "server_identity_verified": True, }, ) @@ -155,6 +202,10 @@ def test_gcp_scope_rejects_receipt_for_a_different_server(tmp_path: Path) -> Non "target_database": "teleo_gcp_copy", "server_address": "10.61.0.4", "ssl": True, + "sslmode": "verify-full", + "ssl_server_name": "teleo-pgvector-standby.private.teleo.internal", + "ssl_root_cert_sha256": "a" * 64, + "server_identity_verified": True, }, ) @@ -176,6 +227,10 @@ def test_gcp_scope_rejects_public_or_non_tls_target_manifest(tmp_path: Path) -> "target_database": "teleo_gcp_copy", "server_address": "34.1.2.3", "ssl": False, + "sslmode": "verify-full", + "ssl_server_name": "teleo-pgvector-standby.private.teleo.internal", + "ssl_root_cert_sha256": "a" * 64, + "server_identity_verified": True, }, ) diff --git a/tests/test_verify_vps_canonical_snapshot_delta.py b/tests/test_verify_vps_canonical_snapshot_delta.py index 451c6f6..61e264c 100644 --- a/tests/test_verify_vps_canonical_snapshot_delta.py +++ b/tests/test_verify_vps_canonical_snapshot_delta.py @@ -8,6 +8,7 @@ from ops.capture_vps_canonical_postgres_snapshot import ( SOURCE_SNAPSHOT_RECEIPT_SCHEMA, build_provenance_binding, ) +from ops.verify_postgres_parity_manifest import REVIEWED_MANIFEST_SQL_SHA256 from ops.verify_vps_canonical_snapshot_delta import verify_delta, write_private_json @@ -21,7 +22,17 @@ def manifest_text(*, claims: int = 1, constraint_name: str = "claims_pkey") -> s }, {"kind": "schemas", "items": ["kb_stage", "public"]}, {"kind": "extensions", "items": [{"name": "pgcrypto", "version": "1.3"}]}, - {"kind": "application_roles", "items": [{"name": "kb_apply", "can_login": True}]}, + { + "kind": "application_roles", + "items": [ + {"name": "kb_apply", "can_login": True}, + {"name": "kb_gate_owner", "can_login": False}, + {"name": "kb_review", "can_login": True}, + ], + }, + {"kind": "role_memberships", "items": []}, + {"kind": "object_ownership", "items": []}, + {"kind": "object_acl", "items": []}, {"kind": "columns", "items": []}, {"kind": "constraints", "items": [{"name": constraint_name}]}, {"kind": "indexes", "items": []}, @@ -56,7 +67,7 @@ def capture_fixture( manifest.write_text(manifest_text(claims=claims, constraint_name=constraint_name), encoding="utf-8") dump_sha256 = hashlib.sha256(f"{name}-dump".encode()).hexdigest() manifest_sha256 = hashlib.sha256(manifest.read_bytes()).hexdigest() - manifest_sql_sha256 = "2" * 64 + manifest_sql_sha256 = REVIEWED_MANIFEST_SQL_SHA256 source_context_sha256 = "3" * 64 source = { "ssh_target": "root@77.42.65.182", @@ -239,6 +250,44 @@ def test_postflight_rejects_unchanged_but_unhealthy_service(tmp_path: Path) -> N ) +def test_postflight_rejects_a_different_manifest_algorithm_even_when_rebound(tmp_path: Path) -> None: + baseline_receipt, baseline_manifest = capture_fixture( + tmp_path, + name="canonical-baseline", + captured_at="2026-07-15T01:00:00+00:00", + authorization_ref="codex-delegation:baseline123", + ) + current_receipt, current_manifest = capture_fixture( + tmp_path, + name="canonical-postflight", + captured_at="2026-07-15T02:00:00+00:00", + authorization_ref="codex-delegation:postflight123", + ) + current = json.loads(current_receipt.read_text()) + current["manifest"]["manifest_sql_sha256"] = "f" * 64 + current["provenance_binding"] = build_provenance_binding( + run_id=current["run_id"], + authorization_ref=current["capture_authorization_ref"], + source=current["source"], + snapshot=current["snapshot"], + dump_sha256=current["dump"]["sha256"], + manifest_sql_sha256="f" * 64, + source_manifest_sha256=current["manifest"]["sha256"], + source_context_sha256=current["source_context"]["sha256"], + ) + current_receipt.write_text(json.dumps(current, sort_keys=True) + "\n", encoding="utf-8") + + with pytest.raises(ValueError, match="manifest_sql_reviewed"): + verify_delta( + baseline_receipt_path=baseline_receipt, + baseline_manifest_path=baseline_manifest, + baseline_authorization_ref="codex-delegation:baseline123", + current_receipt_path=current_receipt, + current_manifest_path=current_manifest, + current_authorization_ref="codex-delegation:postflight123", + ) + + def test_delta_receipt_writer_keeps_private_mode(tmp_path: Path) -> None: output = tmp_path / "private" / "delta.json"