teleo/teleo-infrastructure
8
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions 1

Merge pull request #170 from living-ip/codex/leo-observatory-private-staging-followup-w2-20260715

Browse source 
Harden private Observatory staging preflight
This commit is contained in:
twentyOne2x 2026-07-16 12:06:41 +02:00 committed by GitHub
commit 42c32d5242
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
10 changed files with 505 additions and 52 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

20
.github/workflows/gcp-observatory-read-adapter.yml vendored
View file

@ -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: |

6
docs/observatory-read-adapter-gcp-staging.md
View file

@ -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

11
observatory_read_adapter/config.py
View file

@ -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")

43
observatory_read_adapter/db.py
View file

@ -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,20 +80,12 @@ 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._connector, self._connection_factory = _private_connection_factory(
settings,
connector_type=Connector,
private_ip=IPTypes.PRIVATE,
)
self._connection_factory = connect
def close(self) -> None:
if self._connector is not None:
self._connector.close()

146
ops/check_observatory_read_adapter_gcp_preflight.py
View file

@ -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__":

4
ops/observatory_read_role.sql
View file

@ -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;

66
ops/plan_observatory_read_adapter_gcp.py
View file

@ -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",

60
tests/test_observatory_read_adapter.py
View file

@ -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()

177
tests/test_observatory_read_adapter_gcp_plan.py
View file

@ -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,

22
tests/test_observatory_read_role_postgres.py
View file

@ -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",