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/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/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_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",