diff --git a/.github/workflows/gcp-observatory-read-adapter.yml b/.github/workflows/gcp-observatory-read-adapter.yml index ceb3b59..a1a3dae 100644 --- a/.github/workflows/gcp-observatory-read-adapter.yml +++ b/.github/workflows/gcp-observatory-read-adapter.yml @@ -34,7 +34,7 @@ env: DB_IAM_USER: sa-observatory-read-adapter@teleo-501523.iam API_KEY_SECRET: observatory-read-api-key-staging BUILD_WORKLOAD_IDENTITY_PROVIDER: projects/785938879453/locations/global/workloadIdentityPools/github-actions/providers/living-ip-github - DEPLOY_WORKLOAD_IDENTITY_PROVIDER: projects/785938879453/locations/global/workloadIdentityPools/github-actions/providers/living-ip-github + DEPLOY_WORKLOAD_IDENTITY_PROVIDER: projects/785938879453/locations/global/workloadIdentityPools/github-actions/providers/observatory-read-adapter-main EXPECTED_WORKFLOW_REF: living-ip/teleo-infrastructure/.github/workflows/gcp-observatory-read-adapter.yml@refs/heads/main jobs: @@ -55,8 +55,16 @@ jobs: - name: Run focused adapter tests run: | python -m pip install -e '.[dev]' - python -m pytest tests/test_observatory_read_adapter.py -q - python -m ruff check observatory_read_adapter tests/test_observatory_read_adapter.py + python -m pytest \ + tests/test_observatory_read_adapter.py \ + tests/test_observatory_read_adapter_gcp_plan.py \ + -q + python -m ruff check \ + observatory_read_adapter \ + ops/check_observatory_read_adapter_gcp_preflight.py \ + ops/plan_observatory_read_adapter_gcp.py \ + tests/test_observatory_read_adapter.py \ + tests/test_observatory_read_adapter_gcp_plan.py - id: auth uses: google-github-actions/auth@v3 @@ -106,6 +114,12 @@ jobs: steps: - uses: actions/checkout@v4 + - name: Initialize fail-closed preflight receipt + run: | + python3 ops/check_observatory_read_adapter_gcp_preflight.py \ + --initialize-fail-closed \ + --output observatory-read-adapter-preflight.json + - name: Enforce reviewed main-only staging path shell: bash run: | diff --git a/docs/observatory-read-adapter-gcp-staging.md b/docs/observatory-read-adapter-gcp-staging.md index 23837c1..3eba8f7 100644 --- a/docs/observatory-read-adapter-gcp-staging.md +++ b/docs/observatory-read-adapter-gcp-staging.md @@ -84,6 +84,12 @@ they do not route production traffic. `run.operations.get`; it contains no delete permission. Do not add `run.admin`, `run.developer`, `cloudsql.admin`, `secretmanager.admin`, `compute.admin`, Editor, or Owner to either lane identity. + The deployer binding uses the provider-unique + `attribute.observatory_workflow` principal set. A pool-wide subject binding + or the general `living-ip-github` provider is diagnostic-only and must not + authenticate the preflight or deploy jobs. The preflight lists every + provider in `github-actions`, including soft-deleted providers, and blocks + if any non-dedicated provider maps that attribute. 3. Add the dedicated runtime service account as a Cloud SQL IAM service-account user on `teleo-pgvector-standby`. 4. From the existing private GCP VM database-admin path, run diff --git a/observatory_read_adapter/config.py b/observatory_read_adapter/config.py index b01b5ac..d114df4 100644 --- a/observatory_read_adapter/config.py +++ b/observatory_read_adapter/config.py @@ -1,6 +1,7 @@ from __future__ import annotations import os +import re from dataclasses import dataclass, field EXPECTED_PROJECT = "teleo-501523" @@ -9,6 +10,8 @@ EXPECTED_INSTANCE = "teleo-pgvector-standby" EXPECTED_CONNECTION_NAME = f"{EXPECTED_PROJECT}:{EXPECTED_REGION}:{EXPECTED_INSTANCE}" EXPECTED_DATABASE = "teleo_canonical" EXPECTED_AUTHORIZATION_ROLE = "kb_observatory_read" +EXPECTED_IAM_DATABASE_USER = f"sa-observatory-read-adapter@{EXPECTED_PROJECT}.iam" +SERVICE_REVISION_PATTERN = re.compile(r"^[0-9a-f]{40}$") @dataclass(frozen=True) @@ -46,10 +49,8 @@ class Settings: raise ValueError(f"DB_NAME must be {EXPECTED_DATABASE}") if self.authorization_role != EXPECTED_AUTHORIZATION_ROLE: raise ValueError(f"DB_AUTHORIZATION_ROLE must be {EXPECTED_AUTHORIZATION_ROLE}") - if not self.iam_database_user.endswith(f"@{EXPECTED_PROJECT}.iam"): - raise ValueError("DB_IAM_USER must be the scoped Cloud SQL service-account user") - if any(character.isspace() for character in self.iam_database_user): - raise ValueError("DB_IAM_USER must not contain whitespace") + if self.iam_database_user != EXPECTED_IAM_DATABASE_USER: + raise ValueError(f"DB_IAM_USER must be {EXPECTED_IAM_DATABASE_USER}") if ( len(self.api_key) < 32 or len(self.api_key) > 512 @@ -57,5 +58,7 @@ class Settings: or any(character.isspace() for character in self.api_key) ): raise ValueError("OBSERVATORY_API_KEY must be 32-512 non-whitespace ASCII characters") + if SERVICE_REVISION_PATTERN.fullmatch(self.service_revision) is None: + raise ValueError("SERVICE_REVISION must be a lowercase 40-character Git commit SHA") if not 1 <= self.port <= 65535: raise ValueError("PORT is outside the valid TCP port range") diff --git a/observatory_read_adapter/db.py b/observatory_read_adapter/db.py index 8ed89c4..426a913 100644 --- a/observatory_read_adapter/db.py +++ b/observatory_read_adapter/db.py @@ -8,6 +8,8 @@ from typing import Any from .config import Settings ConnectionFactory = Callable[[], Any] +ConnectorType = Callable[..., Any] +DATABASE_CONNECT_TIMEOUT_SECONDS = 8 class ReaderContractError(RuntimeError): @@ -43,6 +45,31 @@ def _rows(cursor: Any) -> list[dict[str, Any]]: ] +def _private_connection_factory( + settings: Settings, + *, + connector_type: ConnectorType, + private_ip: Any, +) -> tuple[Any, ConnectionFactory]: + connector = connector_type( + refresh_strategy="LAZY", + timeout=DATABASE_CONNECT_TIMEOUT_SECONDS, + ) + + def connect() -> Any: + return connector.connect( + settings.instance_connection_name, + "pg8000", + user=settings.iam_database_user, + db=settings.database, + enable_iam_auth=True, + ip_type=private_ip, + timeout=DATABASE_CONNECT_TIMEOUT_SECONDS, + ) + + return connector, connect + + class CloudSqlClaimReader: def __init__(self, settings: Settings, *, connection_factory: ConnectionFactory | None = None): self.settings = settings @@ -53,19 +80,11 @@ class CloudSqlClaimReader: from google.cloud.sql.connector import Connector, IPTypes - self._connector = Connector(refresh_strategy="LAZY") - - def connect() -> Any: - return self._connector.connect( - settings.instance_connection_name, - "pg8000", - user=settings.iam_database_user, - db=settings.database, - enable_iam_auth=True, - ip_type=IPTypes.PRIVATE, - ) - - self._connection_factory = connect + self._connector, self._connection_factory = _private_connection_factory( + settings, + connector_type=Connector, + private_ip=IPTypes.PRIVATE, + ) def close(self) -> None: if self._connector is not None: diff --git a/ops/check_observatory_read_adapter_gcp_preflight.py b/ops/check_observatory_read_adapter_gcp_preflight.py index 8c75e43..6711bf8 100644 --- a/ops/check_observatory_read_adapter_gcp_preflight.py +++ b/ops/check_observatory_read_adapter_gcp_preflight.py @@ -26,6 +26,9 @@ SECRET = "observatory-read-api-key-staging" REPOSITORY = "teleo" WIF_POOL = "github-actions" DEPLOY_WIF_PROVIDER = "observatory-read-adapter-main" +DEPLOY_WIF_PROVIDER_RESOURCE = ( + f"projects/{PROJECT_NUMBER}/locations/global/workloadIdentityPools/{WIF_POOL}/providers/{DEPLOY_WIF_PROVIDER}" +) DEPLOY_WORKFLOW_REF = ( "living-ip/teleo-infrastructure/.github/workflows/" "gcp-observatory-read-adapter.yml@refs/heads/main" @@ -36,9 +39,8 @@ DEPLOY_WIF_ATTRIBUTE_CONDITION = ( f"assertion.workflow_ref=='{DEPLOY_WORKFLOW_REF}'" ) DEPLOY_WIF_MEMBER = ( - f"principal://iam.googleapis.com/projects/{PROJECT_NUMBER}/locations/global/" - f"workloadIdentityPools/{WIF_POOL}/subject/" - "repo:living-ip/teleo-infrastructure:ref:refs/heads/main" + f"principalSet://iam.googleapis.com/projects/{PROJECT_NUMBER}/locations/global/" + f"workloadIdentityPools/{WIF_POOL}/attribute.observatory_workflow/{DEPLOY_WORKFLOW_REF}" ) PREFLIGHT_ROLE = f"projects/{PROJECT}/roles/observatoryStagingPreflight" DEPLOY_ROLE = f"projects/{PROJECT}/roles/observatoryStagingDeployer" @@ -52,6 +54,7 @@ PREFLIGHT_PERMISSIONS = { "iam.serviceAccounts.getIamPolicy", "iam.roles.get", "iam.workloadIdentityPoolProviders.get", + "iam.workloadIdentityPoolProviders.list", "resourcemanager.projects.get", "resourcemanager.projects.getIamPolicy", "run.operations.get", @@ -215,6 +218,22 @@ def member_roles(policy: dict[str, object], member: str) -> set[str]: return roles +def role_has_exact_unconditional_members( + policy: dict[str, object], + role: str, + expected_members: set[str], +) -> bool: + bindings = policy.get("bindings", []) + if not isinstance(bindings, list): + return False + role_bindings = [binding for binding in bindings if isinstance(binding, dict) and binding.get("role") == role] + if len(role_bindings) != 1: + return False + binding = role_bindings[0] + members = binding.get("members", []) + return isinstance(members, list) and set(members) == expected_members and binding.get("condition") in (None, {}) + + def project_iam_is_exact(value: str) -> bool: policy = json.loads(value) deployer = f"serviceAccount:{DEPLOY_SERVICE_ACCOUNT}" @@ -277,6 +296,7 @@ def wif_provider_is_exact(value: str) -> bool: mapping = payload.get("attributeMapping", {}) expected_mapping = { "google.subject": "assertion.sub", + "attribute.observatory_workflow": "assertion.workflow_ref", "attribute.repository": "assertion.repository", "attribute.ref": "assertion.ref", "attribute.workflow_ref": "assertion.workflow_ref", @@ -288,9 +308,42 @@ def wif_provider_is_exact(value: str) -> bool: ) +def wif_provider_inventory_is_isolated(value: str) -> bool: + providers = json.loads(value) + if not isinstance(providers, list) or not providers: + return False + + dedicated: list[dict[str, object]] = [] + provider_names: set[str] = set() + for provider in providers: + if not isinstance(provider, dict): + return False + name = provider.get("name") + if not isinstance(name, str) or not name: + return False + provider_id = name.rsplit("/", 1)[-1] + if provider_id in provider_names: + return False + provider_names.add(provider_id) + + mapping = provider.get("attributeMapping", {}) + if not isinstance(mapping, dict): + return False + if provider_id == DEPLOY_WIF_PROVIDER: + dedicated.append(provider) + elif "attribute.observatory_workflow" in mapping: + return False + + return len(dedicated) == 1 and wif_provider_is_exact(json.dumps(dedicated[0])) + + def deployer_service_account_policy_is_exact(value: str) -> bool: policy = json.loads(value) - return policy_has_binding(policy, "roles/iam.workloadIdentityUser", DEPLOY_WIF_MEMBER) + return role_has_exact_unconditional_members( + policy, + "roles/iam.workloadIdentityUser", + {DEPLOY_WIF_MEMBER}, + ) def runtime_service_account_policy_is_exact(value: str) -> bool: @@ -459,6 +512,26 @@ def build_checks(image_ref: str) -> list[Check]: wif_provider_is_exact, "main_and_exact_workflow_only", ), + command_check( + "wif_pool_provider_attribute_isolation", + [ + "gcloud", + "iam", + "workload-identity-pools", + "providers", + "list", + "--project", + PROJECT, + "--location", + "global", + "--workload-identity-pool", + WIF_POOL, + "--show-deleted", + "--format=json", + ], + wif_provider_inventory_is_isolated, + "dedicated_provider_is_sole_observatory_workflow_mapper", + ), command_check( "preflight_custom_role", [ @@ -508,7 +581,7 @@ def build_checks(image_ref: str) -> list[Check]: "--format=json", ], deployer_service_account_policy_is_exact, - "exact_main_subject_workload_identity_user", + "exact_workflow_attribute_principal_set_only", ), command_check( "runtime_act_as_policy", @@ -633,23 +706,23 @@ def build_checks(image_ref: str) -> list[Check]: return checks -def main() -> int: - parser = argparse.ArgumentParser() - parser.add_argument("--image-ref", required=True) - parser.add_argument("--output", type=Path) - args = parser.parse_args() - - checks = build_checks(args.image_ref) +def build_receipt( + checks: list[Check], + *, + current_tier: str, + claim_ceiling: str, +) -> dict[str, object]: pass_count = sum(check.status == "pass" for check in checks) blocked_count = len(checks) - pass_count - receipt = { + return { "schema": "livingip.observatory-read-adapter-gcp-preflight.v1", "project": PROJECT, "required_tier": "T3_live_readonly", - "current_tier": "T2_authenticated_gcp_preflight", + "current_tier": current_tier, "ready_for_staging_deploy": blocked_count == 0, - "claim_ceiling": "Preflight only; T3 requires the live private Cloud SQL claim and negative receipts.", + "claim_ceiling": claim_ceiling, "identity": DEPLOY_SERVICE_ACCOUNT, + "workload_identity_provider": DEPLOY_WIF_PROVIDER_RESOURCE, "checks": [asdict(check) for check in checks], "pass_count": pass_count, "blocked_count": blocked_count, @@ -657,12 +730,49 @@ def main() -> int: "database_role_live_verification_deferred_to_fail_closed_canary": True, "production_repoint_executed": False, } + + +def write_receipt(receipt: dict[str, object], output: Path | None) -> None: rendered = json.dumps(receipt, indent=2, sort_keys=True) + "\n" - if args.output: - args.output.write_text(rendered, encoding="utf-8") + if output: + output.write_text(rendered, encoding="utf-8") else: print(rendered, end="") - return 0 if blocked_count == 0 else 1 + + +def main() -> int: + parser = argparse.ArgumentParser() + mode = parser.add_mutually_exclusive_group(required=True) + mode.add_argument("--image-ref") + mode.add_argument("--initialize-fail-closed", action="store_true") + parser.add_argument("--output", type=Path) + args = parser.parse_args() + + if args.initialize_fail_closed: + receipt = build_receipt( + [ + Check( + "identity_authentication_and_gcp_preflight", + "blocked", + "dedicated staging identity authentication and GCP preflight did not complete", + ) + ], + current_tier="T2_identity_preflight_not_completed", + claim_ceiling=( + "Fail-closed initialization only; no authenticated GCP preflight or staging deploy has completed." + ), + ) + write_receipt(receipt, args.output) + return 0 + + checks = build_checks(args.image_ref) + receipt = build_receipt( + checks, + current_tier="T2_authenticated_gcp_preflight", + claim_ceiling="Preflight only; T3 requires the live private Cloud SQL claim and negative receipts.", + ) + write_receipt(receipt, args.output) + return 0 if receipt["ready_for_staging_deploy"] else 1 if __name__ == "__main__": diff --git a/ops/observatory_read_role.sql b/ops/observatory_read_role.sql index 0536237..34f036e 100644 --- a/ops/observatory_read_role.sql +++ b/ops/observatory_read_role.sql @@ -12,8 +12,8 @@ begin if current_database() <> 'teleo_canonical' then raise exception 'observatory_read_role must run only in teleo_canonical'; end if; - if principal !~ '^[a-z0-9-]+@teleo-501523[.]iam$' then - raise exception 'OBSERVATORY_DB_IAM_USER is not the scoped GCP service-account database user'; + if principal <> 'sa-observatory-read-adapter@teleo-501523.iam' then + raise exception 'OBSERVATORY_DB_IAM_USER is not the exact dedicated Observatory database user'; end if; if not exists (select 1 from pg_catalog.pg_roles where rolname = principal) then raise exception 'Cloud SQL IAM database user does not exist: %', principal; diff --git a/ops/plan_observatory_read_adapter_gcp.py b/ops/plan_observatory_read_adapter_gcp.py index 1bd236c..c24172f 100644 --- a/ops/plan_observatory_read_adapter_gcp.py +++ b/ops/plan_observatory_read_adapter_gcp.py @@ -13,12 +13,16 @@ REGION = "europe-west6" GITHUB_REPOSITORY = "living-ip/teleo-infrastructure" WIF_POOL = "github-actions" DEPLOY_WIF_PROVIDER = "observatory-read-adapter-main" +DEPLOY_WIF_PROVIDER_RESOURCE = ( + f"projects/{PROJECT_NUMBER}/locations/global/workloadIdentityPools/{WIF_POOL}/providers/{DEPLOY_WIF_PROVIDER}" +) DEPLOY_WORKFLOW_REF = ( f"{GITHUB_REPOSITORY}/.github/workflows/gcp-observatory-read-adapter.yml@refs/heads/main" ) DEPLOY_WIF_ATTRIBUTE_MAPPING = ( "google.subject=assertion.sub,attribute.repository=assertion.repository," - "attribute.ref=assertion.ref,attribute.workflow_ref=assertion.workflow_ref" + "attribute.ref=assertion.ref,attribute.workflow_ref=assertion.workflow_ref," + "attribute.observatory_workflow=assertion.workflow_ref" ) DEPLOY_WIF_ATTRIBUTE_CONDITION = ( "assertion.repository=='living-ip/teleo-infrastructure' && " @@ -51,6 +55,7 @@ PREFLIGHT_PERMISSIONS = ( "iam.serviceAccounts.getIamPolicy", "iam.roles.get", "iam.workloadIdentityPoolProviders.get", + "iam.workloadIdentityPoolProviders.list", "resourcemanager.projects.get", "resourcemanager.projects.getIamPolicy", "run.operations.get", @@ -93,6 +98,13 @@ def shell_join(parts: list[str]) -> str: def github_wif_member() -> str: + return ( + f"principalSet://iam.googleapis.com/projects/{PROJECT_NUMBER}/locations/global/" + f"workloadIdentityPools/{WIF_POOL}/attribute.observatory_workflow/{DEPLOY_WORKFLOW_REF}" + ) + + +def legacy_pool_wide_github_wif_member() -> str: return ( f"principal://iam.googleapis.com/projects/{PROJECT_NUMBER}/locations/global/" f"workloadIdentityPools/{WIF_POOL}/subject/repo:{GITHUB_REPOSITORY}:ref:refs/heads/main" @@ -268,6 +280,35 @@ def build_plan() -> dict[str, object]: ) commands.extend( [ + ( + "DEPLOYER_POLICY_JSON=$(gcloud iam service-accounts get-iam-policy " + + shlex.quote(DEPLOY_SERVICE_ACCOUNT) + + " --project " + + shlex.quote(PROJECT) + + " --format=json)" + ), + ( + "if jq -e --arg legacy " + + shlex.quote(legacy_pool_wide_github_wif_member()) + + " 'any(.bindings[]?; .role == \"roles/iam.workloadIdentityUser\" " + "and ((.members // []) | index($legacy)))' <<<\"${DEPLOYER_POLICY_JSON}\" >/dev/null; then " + + shell_join( + [ + "gcloud", + "iam", + "service-accounts", + "remove-iam-policy-binding", + DEPLOY_SERVICE_ACCOUNT, + "--project", + PROJECT, + "--member", + legacy_pool_wide_github_wif_member(), + "--role", + "roles/iam.workloadIdentityUser", + ] + ) + + "; fi" + ), project_binding(deployer_member, PREFLIGHT_ROLE), project_binding(deployer_member, DEPLOY_ROLE), shell_join( @@ -398,10 +439,23 @@ def build_plan() -> dict[str, object]: "deployer": "workflow-specific main-only WIF plus a five-permission Cloud Run role, repository read, exact runtime actAs, exact secret access", "runtime": "exact Cloud SQL instance connect/login plus exact secret access", }, - "github_wif_provider": ( - f"projects/{PROJECT_NUMBER}/locations/global/workloadIdentityPools/{WIF_POOL}/providers/{DEPLOY_WIF_PROVIDER}" - ), + "github_wif_provider": DEPLOY_WIF_PROVIDER_RESOURCE, "github_wif_member": github_wif_member(), + "legacy_pool_wide_github_wif_member_removed": legacy_pool_wide_github_wif_member(), + "workflow_authentication": { + "provider": DEPLOY_WIF_PROVIDER_RESOURCE, + "service_account": DEPLOY_SERVICE_ACCOUNT, + "principal_set": github_wif_member(), + "fallback_provider_allowed": False, + "required_condition": DEPLOY_WIF_ATTRIBUTE_CONDITION, + }, + "provider_isolation": { + "pool": WIF_POOL, + "dedicated_provider": DEPLOY_WIF_PROVIDER, + "exclusive_attribute": "attribute.observatory_workflow", + "all_providers_must_be_enumerated": True, + "non_dedicated_provider_mapping_allowed": False, + }, "required_apis": list(REQUIRED_APIS), "preflight_custom_role": { "name": PREFLIGHT_ROLE, @@ -436,7 +490,8 @@ def build_plan() -> dict[str, object]: "post_apply_checks": [ f"gcloud projects describe {PROJECT} --format=value(projectId)", "python3 ops/check_gcp_infra_readiness.py", - "Require the adapter preflight to verify the dedicated WIF provider, custom roles, IAM bindings, private Cloud SQL, secret, and immutable image.", + "Require the adapter preflight to enumerate every provider in the pool and prove only the dedicated provider maps attribute.observatory_workflow.", + "Require the adapter preflight to verify custom roles, IAM bindings, private Cloud SQL, secret, and immutable image.", ( "gh workflow run gcp-observatory-read-adapter.yml --repo " f"{GITHUB_REPOSITORY} --ref main -f action=preflight_staging" @@ -446,6 +501,7 @@ def build_plan() -> dict[str, object]: "forbidden_deployer_roles": list(FORBIDDEN_DEPLOYER_ROLES), "forbidden_actions": [ "grant deploy or infra roles to sa-artifact-builder", + "authenticate the deploy or canary jobs through the general living-ip-github provider", "reuse a pre-existing image tag instead of building the current workflow revision", "add a public Cloud SQL address", "expose a secret value in Git, logs, or a receipt", diff --git a/tests/test_observatory_read_adapter.py b/tests/test_observatory_read_adapter.py index f259145..3a7589c 100644 --- a/tests/test_observatory_read_adapter.py +++ b/tests/test_observatory_read_adapter.py @@ -7,8 +7,16 @@ from pathlib import Path from aiohttp.test_utils import TestClient, TestServer from observatory_read_adapter.app import create_app -from observatory_read_adapter.config import EXPECTED_CONNECTION_NAME, Settings -from observatory_read_adapter.db import CloudSqlClaimReader +from observatory_read_adapter.config import ( + EXPECTED_CONNECTION_NAME, + EXPECTED_IAM_DATABASE_USER, + Settings, +) +from observatory_read_adapter.db import ( + DATABASE_CONNECT_TIMEOUT_SECONDS, + CloudSqlClaimReader, + _private_connection_factory, +) CLAIM_ID = "d0000000-0000-0000-0000-000000000005" API_KEY = "observatory-test-api-key-000000000000000000000000" @@ -20,10 +28,10 @@ def settings(**overrides: object) -> Settings: "project_id": "teleo-501523", "instance_connection_name": EXPECTED_CONNECTION_NAME, "database": "teleo_canonical", - "iam_database_user": "sa-observatory-read-adapter@teleo-501523.iam", + "iam_database_user": EXPECTED_IAM_DATABASE_USER, "authorization_role": "kb_observatory_read", "api_key": API_KEY, - "service_revision": "test-revision", + "service_revision": "a" * 40, } values.update(overrides) return Settings(**values) @@ -225,12 +233,15 @@ class FakeConnection: self.closed = True -def test_settings_fail_closed_on_wrong_database_role_or_short_key() -> None: +def test_settings_fail_closed_on_wrong_runtime_contract() -> None: for overrides in ( {"database": "teleo"}, {"authorization_role": "leoclean_kb_runtime"}, {"api_key": "too-short"}, {"instance_connection_name": "other:region:instance"}, + {"iam_database_user": "sa-unrelated@teleo-501523.iam"}, + {"service_revision": "unknown"}, + {"service_revision": "A" * 40}, ): candidate = settings(**overrides) try: @@ -241,6 +252,45 @@ def test_settings_fail_closed_on_wrong_database_role_or_short_key() -> None: raise AssertionError(f"unsafe settings accepted: {overrides}") +def test_cloud_sql_connector_is_private_iam_only_and_bounded() -> None: + connector_calls: list[dict[str, object]] = [] + connect_calls: list[tuple[tuple[object, ...], dict[str, object]]] = [] + connection = object() + + class FakeConnector: + def __init__(self, **kwargs: object) -> None: + connector_calls.append(kwargs) + + def connect(self, *args: object, **kwargs: object) -> object: + connect_calls.append((args, kwargs)) + return connection + + def close(self) -> None: + return None + + private_ip = object() + connector, connection_factory = _private_connection_factory( + settings(), + connector_type=FakeConnector, + private_ip=private_ip, + ) + assert isinstance(connector, FakeConnector) + assert connection_factory() is connection + assert connector_calls == [{"refresh_strategy": "LAZY", "timeout": DATABASE_CONNECT_TIMEOUT_SECONDS}] + assert connect_calls == [ + ( + (EXPECTED_CONNECTION_NAME, "pg8000"), + { + "user": EXPECTED_IAM_DATABASE_USER, + "db": "teleo_canonical", + "enable_iam_auth": True, + "ip_type": private_ip, + "timeout": DATABASE_CONNECT_TIMEOUT_SECONDS, + }, + ) + ] + + def test_authenticated_claim_response_and_negative_http_paths() -> None: async def exercise() -> None: reader = FakeReader() diff --git a/tests/test_observatory_read_adapter_gcp_plan.py b/tests/test_observatory_read_adapter_gcp_plan.py index 023c942..4294964 100644 --- a/tests/test_observatory_read_adapter_gcp_plan.py +++ b/tests/test_observatory_read_adapter_gcp_plan.py @@ -38,12 +38,32 @@ def test_plan_splits_build_deploy_and_runtime_identities() -> None: 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 ( + "attribute.observatory_workflow/living-ip/teleo-infrastructure/.github/workflows/" + "gcp-observatory-read-adapter.yml@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 plan["workflow_authentication"] == { + "provider": plan["github_wif_provider"], + "service_account": "sa-observatory-deployer@teleo-501523.iam.gserviceaccount.com", + "principal_set": plan["github_wif_member"], + "fallback_provider_allowed": False, + "required_condition": plan["conditions"]["deployer_wif"], + } + assert plan["provider_isolation"] == { + "pool": "github-actions", + "dedicated_provider": "observatory-read-adapter-main", + "exclusive_attribute": "attribute.observatory_workflow", + "all_providers_must_be_enumerated": True, + "non_dedicated_provider_mapping_allowed": False, + } + assert "iam.workloadIdentityPoolProviders.list" in plan["preflight_custom_role"]["permissions"] + assert "remove-iam-policy-binding" in commands + assert plan["legacy_pool_wide_github_wif_member_removed"] in commands assert "roles/run.serviceAgent" in commands assert "teleo-pgvector-standby" in plan["conditions"]["cloud_sql"] assert set(plan["deploy_custom_role"]["permissions"]) == { @@ -84,7 +104,10 @@ def test_workflow_requires_dedicated_preflight_before_staging_deploy() -> None: 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/living-ip-github") + assert workflow["env"]["DEPLOY_WORKLOAD_IDENTITY_PROVIDER"].endswith( + "/providers/observatory-read-adapter-main" + ) + assert workflow["env"]["DEPLOY_WORKLOAD_IDENTITY_PROVIDER"] == preflight.DEPLOY_WIF_PROVIDER_RESOURCE assert workflow["jobs"]["deploy-staging"]["needs"] == ["build", "preflight-staging"] assert "action=preflight_staging" not in text assert "inputs.action != 'build_only'" in text @@ -97,6 +120,8 @@ def test_workflow_requires_dedicated_preflight_before_staging_deploy() -> None: assert "API_KEY_VERSION" in text assert "EXPECTED_IMAGE_REF" in text assert "check_observatory_read_adapter_gcp_preflight.py" in text + assert "--initialize-fail-closed" in text + assert text.index("--initialize-fail-closed") < text.index("workload_identity_provider: ${{ env.DEPLOY_WORKLOAD_IDENTITY_PROVIDER }}") 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 @@ -107,6 +132,37 @@ def test_workflow_requires_dedicated_preflight_before_staging_deploy() -> None: assert "rm -f positive.json" in text +def test_fail_closed_preflight_initialization_retains_a_blocked_receipt(tmp_path: Path) -> None: + output = tmp_path / "preflight.json" + subprocess.run( + [ + "python3", + "ops/check_observatory_read_adapter_gcp_preflight.py", + "--initialize-fail-closed", + "--output", + str(output), + ], + cwd=ROOT, + check=True, + ) + + receipt = json.loads(output.read_text(encoding="utf-8")) + assert receipt["ready_for_staging_deploy"] is False + assert receipt["current_tier"] == "T2_identity_preflight_not_completed" + assert receipt["identity"] == "sa-observatory-deployer@teleo-501523.iam.gserviceaccount.com" + assert receipt["workload_identity_provider"].endswith("/providers/observatory-read-adapter-main") + assert receipt["pass_count"] == 0 + assert receipt["blocked_count"] == 1 + assert receipt["checks"] == [ + { + "name": "identity_authentication_and_gcp_preflight", + "status": "blocked", + "detail": "dedicated staging identity authentication and GCP preflight did not complete", + } + ] + assert receipt["secret_values_retained"] is False + + def test_preflight_private_cloud_sql_and_secret_validators() -> None: instance = { "name": "teleo-pgvector-standby", @@ -221,10 +277,15 @@ def test_preflight_end_to_end_snapshot_retains_no_secret(monkeypatch) -> None: ] } provider = { + "name": ( + "projects/785938879453/locations/global/workloadIdentityPools/github-actions/providers/" + "observatory-read-adapter-main" + ), "state": "ACTIVE", "attributeCondition": preflight.DEPLOY_WIF_ATTRIBUTE_CONDITION, "attributeMapping": { "google.subject": "assertion.sub", + "attribute.observatory_workflow": "assertion.workflow_ref", "attribute.repository": "assertion.repository", "attribute.ref": "assertion.ref", "attribute.workflow_ref": "assertion.workflow_ref", @@ -244,6 +305,24 @@ def test_preflight_end_to_end_snapshot_retains_no_secret(monkeypatch) -> None: stdout = "ENABLED\n" elif "workload-identity-pools providers describe" in joined: stdout = json.dumps(provider) + elif "workload-identity-pools providers list" in joined: + stdout = json.dumps( + [ + provider, + { + "name": ( + "projects/785938879453/locations/global/workloadIdentityPools/" + "github-actions/providers/living-ip-github" + ), + "state": "ACTIVE", + "attributeMapping": { + "google.subject": "assertion.sub", + "attribute.repository": "assertion.repository", + "attribute.ref": "assertion.ref", + }, + }, + ] + ) elif "iam roles describe observatoryStagingPreflight" in joined: stdout = json.dumps({"includedPermissions": sorted(preflight.PREFLIGHT_PERMISSIONS)}) elif "iam roles describe observatoryStagingDeployer" in joined: @@ -304,6 +383,100 @@ def test_preflight_end_to_end_snapshot_retains_no_secret(monkeypatch) -> None: assert secret_value not in json.dumps([check.__dict__ for check in checks]) +def test_deployer_wif_binding_rejects_pool_wide_or_additional_impersonators() -> None: + exact = { + "bindings": [ + { + "role": "roles/iam.workloadIdentityUser", + "members": [preflight.DEPLOY_WIF_MEMBER], + } + ] + } + assert preflight.deployer_service_account_policy_is_exact(json.dumps(exact)) is True + + old_pool_wide_subject = ( + "principal://iam.googleapis.com/projects/785938879453/locations/global/" + "workloadIdentityPools/github-actions/subject/" + "repo:living-ip/teleo-infrastructure:ref:refs/heads/main" + ) + exact["bindings"][0]["members"].append(old_pool_wide_subject) + assert preflight.deployer_service_account_policy_is_exact(json.dumps(exact)) is False + + +def test_dedicated_provider_requires_provider_unique_workflow_mapping() -> None: + provider = { + "name": ( + "projects/785938879453/locations/global/workloadIdentityPools/github-actions/providers/" + "observatory-read-adapter-main" + ), + "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", + }, + } + assert preflight.wif_provider_is_exact(json.dumps(provider)) is False + provider["attributeMapping"]["attribute.observatory_workflow"] = "assertion.workflow_ref" + assert preflight.wif_provider_is_exact(json.dumps(provider)) is True + + +@pytest.mark.parametrize( + "malicious_mapping", + [ + "assertion.workflow_ref", + "'living-ip/teleo-infrastructure/.github/workflows/" + "gcp-observatory-read-adapter.yml@refs/heads/main'", + ], +) +def test_provider_inventory_rejects_any_second_observatory_mapper(malicious_mapping: str) -> None: + dedicated = { + "name": ( + "projects/785938879453/locations/global/workloadIdentityPools/github-actions/providers/" + "observatory-read-adapter-main" + ), + "state": "ACTIVE", + "attributeCondition": preflight.DEPLOY_WIF_ATTRIBUTE_CONDITION, + "attributeMapping": { + "google.subject": "assertion.sub", + "attribute.observatory_workflow": "assertion.workflow_ref", + "attribute.repository": "assertion.repository", + "attribute.ref": "assertion.ref", + "attribute.workflow_ref": "assertion.workflow_ref", + }, + } + general = { + "name": ( + "projects/785938879453/locations/global/workloadIdentityPools/github-actions/providers/" + "living-ip-github" + ), + "state": "ACTIVE", + "attributeMapping": { + "google.subject": "assertion.sub", + "attribute.repository": "assertion.repository", + }, + } + assert preflight.wif_provider_inventory_is_isolated(json.dumps([dedicated, general])) is True + + general["attributeMapping"]["attribute.observatory_workflow"] = malicious_mapping + assert preflight.wif_provider_inventory_is_isolated(json.dumps([dedicated, general])) is False + + +def test_provider_inventory_fails_closed_on_missing_or_duplicate_dedicated_provider() -> None: + general = { + "name": ( + "projects/785938879453/locations/global/workloadIdentityPools/github-actions/providers/" + "living-ip-github" + ), + "state": "ACTIVE", + "attributeMapping": {"google.subject": "assertion.sub"}, + } + assert preflight.wif_provider_inventory_is_isolated(json.dumps([general])) is False + assert preflight.wif_provider_inventory_is_isolated(json.dumps([general, general])) is False + + def test_malformed_successful_readback_becomes_blocked_check(monkeypatch) -> None: monkeypatch.setattr( preflight, diff --git a/tests/test_observatory_read_role_postgres.py b/tests/test_observatory_read_role_postgres.py index c062b1d..97f0921 100644 --- a/tests/test_observatory_read_role_postgres.py +++ b/tests/test_observatory_read_role_postgres.py @@ -13,12 +13,14 @@ ROLE_SQL = ROOT / "ops" / "observatory_read_role.sql" POSTGRES_IMAGE = "postgres:16-alpine" DATABASE = "teleo_canonical" IAM_USER = "sa-observatory-read-adapter@teleo-501523.iam" +OTHER_IAM_USER = "sa-unrelated@teleo-501523.iam" TEST_PASSWORD = "generated-test-only-password" BOOTSTRAP_SQL = f""" begin; create schema kb_stage; create role "{IAM_USER}" login inherit password '{TEST_PASSWORD}'; +create role "{OTHER_IAM_USER}" login inherit password '{TEST_PASSWORD}'; create table public.claims (id uuid primary key); create table public.sources (id uuid primary key); create table public.claim_evidence (id uuid primary key); @@ -98,6 +100,26 @@ def test_observatory_role_is_read_only_and_exactly_allowlisted() -> None: else: assert bootstrap is not None raise AssertionError(f"ephemeral Postgres bootstrap failed: {bootstrap.stderr[-1000:]}") + wrong_principal = run( + [ + "docker", + "exec", + "-e", + f"OBSERVATORY_DB_IAM_USER={OTHER_IAM_USER}", + "-i", + container, + "psql", + "-U", + "postgres", + "-d", + DATABASE, + ], + input_text=ROLE_SQL.read_text(encoding="utf-8"), + check=False, + ) + assert wrong_principal.returncode != 0 + assert "not the exact dedicated Observatory database user" in wrong_principal.stderr + migration = run( [ "docker",