Harden private Observatory runtime identity

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"):
+        if self.iam_database_user != EXPECTED_IAM_DATABASE_USER:
-            raise ValueError("DB_IAM_USER must be the scoped Cloud SQL service-account user")
+            raise ValueError(f"DB_IAM_USER must be {EXPECTED_IAM_DATABASE_USER}")
-        if any(character.isspace() for character in self.iam_database_user):
-            raise ValueError("DB_IAM_USER must not contain whitespace")
         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")

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,20 +80,12 @@ class CloudSqlClaimReader:
 
         from google.cloud.sql.connector import Connector, IPTypes
 
-        self._connector = Connector(refresh_strategy="LAZY")
+        self._connector, self._connection_factory = _private_connection_factory(
-
+            settings,
-        def connect() -> Any:
+            connector_type=Connector,
-            return self._connector.connect(
+            private_ip=IPTypes.PRIVATE,
-                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
-
     def close(self) -> None:
         if self._connector is not None:
             self._connector.close()

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
+  if principal <> 'sa-observatory-read-adapter@teleo-501523.iam' then
-    raise exception 'OBSERVATORY_DB_IAM_USER is not the scoped GCP service-account database user';
+    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;

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.config import (
-from observatory_read_adapter.db import CloudSqlClaimReader
+    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()

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