from __future__ import annotations import json import shutil import subprocess from pathlib import Path import pytest import yaml from ops import check_observatory_read_adapter_gcp_preflight as preflight ROOT = Path(__file__).resolve().parents[1] def run_plan(*args: str) -> str: return subprocess.check_output( ["python3", "ops/plan_observatory_read_adapter_gcp.py", *args], cwd=ROOT, text=True, ) def test_plan_splits_build_deploy_and_runtime_identities() -> None: plan = json.loads(run_plan()) assert plan["current_tier"] == "T2_unexecuted_least_privilege_packet" assert plan["identities"] == { "build": "sa-artifact-builder@teleo-501523.iam.gserviceaccount.com", "deploy_and_canary": "sa-observatory-deployer@teleo-501523.iam.gserviceaccount.com", "runtime": "sa-observatory-read-adapter@teleo-501523.iam.gserviceaccount.com", "database_principal": "sa-observatory-read-adapter@teleo-501523.iam", } commands = "\n".join(plan["commands"]) assert "roles/iam.workloadIdentityUser" in commands assert "roles/iam.serviceAccountUser" in commands assert "roles/cloudsql.client" in commands assert "roles/cloudsql.instanceUser" in commands assert "roles/artifactregistry.reader" in commands assert "roles/secretmanager.secretAccessor" in commands assert "subject/repo:living-ip/teleo-infrastructure:ref:refs/heads/main" in plan["github_wif_member"] assert plan["conditions"]["deployer_wif"].endswith( "assertion.workflow_ref=='living-ip/teleo-infrastructure/.github/workflows/" "gcp-observatory-read-adapter.yml@refs/heads/main'" ) assert plan["github_wif_provider"].endswith("/providers/observatory-read-adapter-main") assert "roles/run.serviceAgent" in commands assert "teleo-pgvector-standby" in plan["conditions"]["cloud_sql"] assert set(plan["deploy_custom_role"]["permissions"]) == { "run.operations.get", "run.services.create", "run.services.get", "run.services.setIamPolicy", "run.services.update", } assert "roles/run.admin" not in commands assert "roles/run.developer" not in commands assert "sa-artifact-builder" not in commands assert not set(plan["forbidden_deployer_roles"]) & set(commands.split()) assert plan["production_repoint_executed"] is False def test_shell_packet_is_unexecuted_and_preserves_private_boundaries() -> None: shell = run_plan("--format", "shell") assert shell.startswith("# Unexecuted staging packet") assert "set -euo pipefail" in shell assert "run.googleapis.com" in shell assert "--type CLOUD_IAM_SERVICE_ACCOUNT" in shell assert "openssl rand -hex 32 | tr -d '\\n'" in shell assert "--data-file=-" in shell assert "grep -Fxq sa-observatory-read-adapter@teleo-501523.iam" in shell assert "users create sa-observatory-read-adapter@teleo-501523.iam" in shell assert "add-iam-policy-binding teleo" in shell assert "roles/artifactregistry.writer" not in shell assert "roles/cloudsql.admin" not in shell assert "Vercel" not in shell def test_workflow_requires_dedicated_preflight_before_staging_deploy() -> None: workflow_path = ROOT / ".github" / "workflows" / "gcp-observatory-read-adapter.yml" workflow = yaml.safe_load(workflow_path.read_text(encoding="utf-8")) text = workflow_path.read_text(encoding="utf-8") assert workflow["env"]["BUILD_SERVICE_ACCOUNT"].startswith("sa-artifact-builder@") assert workflow["env"]["DEPLOY_SERVICE_ACCOUNT"].startswith("sa-observatory-deployer@") assert workflow["env"]["DEPLOY_WORKLOAD_IDENTITY_PROVIDER"].endswith( "/providers/observatory-read-adapter-main" ) assert workflow["jobs"]["deploy-staging"]["needs"] == ["build", "preflight-staging"] assert "action=preflight_staging" not in text assert "inputs.action != 'build_only'" in text assert "EXPECTED_WORKFLOW_REF" in text assert "${GITHUB_SHA}-${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}" in text assert "docker images describe" in text assert "2>/dev/null || true" not in text assert "--no-invoker-iam-check" in text assert "--allow-unauthenticated" not in text assert "API_KEY_VERSION" in text assert "EXPECTED_IMAGE_REF" in text assert "check_observatory_read_adapter_gcp_preflight.py" in text assert 'service_account: ${{ env.BUILD_SERVICE_ACCOUNT }}' in text assert text.count('service_account: ${{ env.DEPLOY_SERVICE_ACCOUNT }}') == 2 assert "positive-redacted.json" in text assert "redact_observatory_read_adapter_receipt.jq" in text assert "database_principal == $db_iam_user" in text assert "canonical_state_unchanged_after_post: true" in text assert "cmp <(jq -S . positive.json) <(jq -S . after-write-attempt.json)" in text assert "rm -f positive.json" in text def test_preflight_private_cloud_sql_and_secret_validators() -> None: instance = { "name": "teleo-pgvector-standby", "region": "europe-west6", "databaseVersion": "POSTGRES_16", "settings": { "databaseFlags": [{"name": "cloudsql.iam_authentication", "value": "on"}], "ipConfiguration": { "ipv4Enabled": False, "privateNetwork": "projects/teleo-501523/global/networks/teleo-staging-net", }, }, "ipAddresses": [{"type": "PRIVATE", "ipAddress": "10.61.0.3"}], } assert preflight.cloud_sql_is_private_and_iam_enabled(json.dumps(instance)) is True instance["settings"]["ipConfiguration"]["ipv4Enabled"] = True instance["ipAddresses"].append({"type": "PRIMARY", "ipAddress": "203.0.113.1"}) assert preflight.cloud_sql_is_private_and_iam_enabled(json.dumps(instance)) is False assert preflight.cloud_sql_iam_user_exists( json.dumps( [ { "name": "sa-observatory-read-adapter@teleo-501523.iam", "type": "CLOUD_IAM_SERVICE_ACCOUNT", } ] ) ) assert preflight.subnet_is_exact( json.dumps( { "name": "teleo-staging-europe-west6", "region": "https://www.googleapis.com/compute/v1/projects/teleo-501523/regions/europe-west6", "network": "https://www.googleapis.com/compute/v1/projects/teleo-501523/global/networks/teleo-staging-net", } ) ) assert preflight.secret_is_valid("a" * 32) is True assert preflight.secret_is_valid("too-short") is False assert preflight.secret_is_valid("a" * 32 + "\n") is False assert preflight.secret_is_valid(" " + "a" * 32) is False assert "gha-creds" not in preflight.sanitize_error("/tmp/work/gha-creds-deadbeef.json denied") def test_preflight_end_to_end_snapshot_retains_no_secret(monkeypatch) -> None: instance = { "name": "teleo-pgvector-standby", "region": "europe-west6", "databaseVersion": "POSTGRES_16", "settings": { "databaseFlags": [{"name": "cloudsql.iam_authentication", "value": "on"}], "ipConfiguration": { "ipv4Enabled": False, "privateNetwork": "projects/teleo-501523/global/networks/teleo-staging-net", }, }, "ipAddresses": [{"type": "PRIVATE", "ipAddress": "10.61.0.3"}], } secret_value = "never-retain-this-secret-value-000000000000" deployer = "serviceAccount:sa-observatory-deployer@teleo-501523.iam.gserviceaccount.com" runtime = "serviceAccount:sa-observatory-read-adapter@teleo-501523.iam.gserviceaccount.com" project_policy = { "bindings": [ {"role": preflight.PREFLIGHT_ROLE, "members": [deployer]}, {"role": preflight.DEPLOY_ROLE, "members": [deployer]}, { "role": "roles/cloudsql.client", "members": [runtime], "condition": { "title": "observatory-exact-cloudsql-instance", "expression": ( "resource.name == 'projects/teleo-501523/instances/teleo-pgvector-standby' " "&& resource.service == 'sqladmin.googleapis.com'" ), }, }, { "role": "roles/cloudsql.instanceUser", "members": [runtime], "condition": { "title": "observatory-exact-cloudsql-instance", "expression": ( "resource.name == 'projects/teleo-501523/instances/teleo-pgvector-standby' " "&& resource.service == 'sqladmin.googleapis.com'" ), }, }, { "role": "roles/run.serviceAgent", "members": [ "serviceAccount:service-785938879453@serverless-robot-prod.iam.gserviceaccount.com" ], }, ] } deployer_policy = { "bindings": [ {"role": "roles/iam.workloadIdentityUser", "members": [preflight.DEPLOY_WIF_MEMBER]} ] } runtime_policy = { "bindings": [{"role": "roles/iam.serviceAccountUser", "members": [deployer]}] } artifact_policy = { "bindings": [{"role": "roles/artifactregistry.reader", "members": [deployer]}] } secret_policy = { "bindings": [ {"role": "roles/secretmanager.secretAccessor", "members": [deployer, runtime]}, {"role": "roles/secretmanager.viewer", "members": [deployer]}, ] } provider = { "state": "ACTIVE", "attributeCondition": preflight.DEPLOY_WIF_ATTRIBUTE_CONDITION, "attributeMapping": { "google.subject": "assertion.sub", "attribute.repository": "assertion.repository", "attribute.ref": "assertion.ref", "attribute.workflow_ref": "assertion.workflow_ref", }, } def fake_run(command: list[str]) -> subprocess.CompletedProcess[str]: joined = " ".join(command) stdout = "" stderr = "" returncode = 0 if "auth list" in joined: stdout = "sa-observatory-deployer@teleo-501523.iam.gserviceaccount.com\n" elif "projects describe" in joined: stdout = "teleo-501523\n" elif command[:3] == ["gcloud", "services", "describe"]: stdout = "ENABLED\n" elif "workload-identity-pools providers describe" in joined: stdout = json.dumps(provider) elif "iam roles describe observatoryStagingPreflight" in joined: stdout = json.dumps({"includedPermissions": sorted(preflight.PREFLIGHT_PERMISSIONS)}) elif "iam roles describe observatoryStagingDeployer" in joined: stdout = json.dumps({"includedPermissions": sorted(preflight.DEPLOY_PERMISSIONS)}) elif "projects get-iam-policy" in joined: stdout = json.dumps(project_policy) elif f"service-accounts get-iam-policy {preflight.DEPLOY_SERVICE_ACCOUNT}" in joined: stdout = json.dumps(deployer_policy) elif f"service-accounts get-iam-policy {preflight.RUNTIME_SERVICE_ACCOUNT}" in joined: stdout = json.dumps(runtime_policy) elif "artifacts repositories get-iam-policy" in joined: stdout = json.dumps(artifact_policy) elif "secrets get-iam-policy" in joined: stdout = json.dumps(secret_policy) elif "run services describe" in joined: returncode = 1 stderr = "NOT_FOUND: service not found" elif "service-accounts describe" in joined: stdout = "sa-observatory-read-adapter@teleo-501523.iam.gserviceaccount.com\n" elif "networks subnets describe" in joined: stdout = json.dumps( { "name": "teleo-staging-europe-west6", "region": "https://www.googleapis.com/compute/v1/projects/teleo-501523/regions/europe-west6", "network": "https://www.googleapis.com/compute/v1/projects/teleo-501523/global/networks/teleo-staging-net", } ) elif "networks describe" in joined: stdout = "teleo-staging-net\n" elif "sql instances describe" in joined: stdout = json.dumps(instance) elif "sql users list" in joined: stdout = json.dumps( [ { "name": "sa-observatory-read-adapter@teleo-501523.iam", "type": "CLOUD_IAM_SERVICE_ACCOUNT", } ] ) elif "secrets versions list" in joined: stdout = "7\n" elif "secrets versions access" in joined: stdout = secret_value elif "artifacts docker images describe" in joined: stdout = "{}" return subprocess.CompletedProcess(command, returncode, stdout=stdout, stderr=stderr) monkeypatch.setattr(preflight, "run", fake_run) image_ref = ( "europe-west6-docker.pkg.dev/teleo-501523/teleo/observatory-read-adapter:" "33399bd9acfa09e4e5409873e86c8c46ac694a8d-29389090997-1@sha256:" + "a" * 64 ) checks = preflight.build_checks(image_ref) assert checks assert {check.status for check in checks} == {"pass"} assert secret_value not in json.dumps([check.__dict__ for check in checks]) def test_malformed_successful_readback_becomes_blocked_check(monkeypatch) -> None: monkeypatch.setattr( preflight, "run", lambda command: subprocess.CompletedProcess(command, 0, stdout="[]", stderr=""), ) check = preflight.command_check( "malformed", ["gcloud", "sql", "instances", "describe"], preflight.cloud_sql_is_private_and_iam_enabled, "unexpected", ) assert check.status == "blocked" assert check.detail == "invalid_readback:AttributeError" @pytest.mark.skipif(shutil.which("jq") is None, reason="jq is required") def test_receipt_redaction_is_an_explicit_allowlist() -> None: payload = { "schema": "livingip.observatory-canonical-claim.v1", "read_only": True, "provenance": { "gcp_project": "teleo-501523", "cloud_sql_instance": "teleo-501523:europe-west6:teleo-pgvector-standby", "database": "teleo_canonical", "database_principal": "sa-observatory-read-adapter@teleo-501523.iam", "authorization_role": "kb_observatory_read", "transaction_read_only": True, "write_privileges_denied": True, "service_revision": "a" * 40, "future_sensitive_field": "must-not-survive", }, "canonical": { "claim": {"id": "claim-1", "type": "fact", "status": "live", "text": "secret claim text"}, "evidence": [{"excerpt": "secret excerpt", "source_url": "https://private.invalid"}], "edges": [{"direction": "incoming"}, {"direction": "outgoing"}], }, "proposal_ledger": { "distinct_from_canonical": True, "status_counts": {"approved": 1}, "related": [{"proposal_text": "secret proposal"}], }, } result = subprocess.run( ["jq", "-f", str(ROOT / "ops" / "redact_observatory_read_adapter_receipt.jq")], input=json.dumps(payload), text=True, capture_output=True, check=True, ) redacted = json.loads(result.stdout) rendered = json.dumps(redacted) assert set(redacted["provenance"]) == { "gcp_project", "cloud_sql_instance", "database", "database_principal", "authorization_role", "transaction_read_only", "write_privileges_denied", "service_revision", } assert redacted["canonical"]["evidence_count"] == 1 assert redacted["proposal_ledger"]["related_count"] == 1 for sensitive in ( "must-not-survive", "secret claim text", "secret excerpt", "https://private.invalid", "secret proposal", ): assert sensitive not in rendered