diff --git a/products/data_warehouse/backend/managed_warehouse_connection.py b/products/data_warehouse/backend/managed_warehouse_connection.py index 97848792073c..972d06932c46 100644 --- a/products/data_warehouse/backend/managed_warehouse_connection.py +++ b/products/data_warehouse/backend/managed_warehouse_connection.py @@ -1,8 +1,15 @@ -"""Expose a managed Duckgres warehouse in the SQL editor as a read-only Postgres connection. +"""Expose a managed Duckgres warehouse in the SQL editor as a project-scoped Postgres connection. Each member team gets a Postgres ``ExternalDataSource`` pointed at the organization's -``DuckgresServer``. Duckgres issues a distinct database login for each project and enforces -read-only access to the project's schemas, so both HogQL and raw SQL stay inside that boundary. +``DuckgresServer``. Duckgres issues a distinct database login for each project and confines it to +that project's schemas, so both HogQL and raw SQL stay inside that boundary. + +The login is Duckgres's ``project_user``: read/write **within** the project's own namespaces. +Writes are the default because a project connection is the user's own warehouse, not a report of +it. The project boundary is identical to the read-only ``project_reader`` it replaces — Duckgres +derives the same namespaces for both modes, and write authorization does not widen the set of +reachable relations, so a project still cannot read or write another project's schemas. + Setup happens in two steps: 1. ``ensure_managed_warehouse_direct_source`` creates the initially empty source row when a team @@ -50,6 +57,13 @@ MANAGED_WAREHOUSE_SOURCE_DESCRIPTION = "Managed warehouse (auto-provisioned)" +# The Duckgres access mode backing a project connection, recorded on the source's +# connection_metadata. It is also the migration marker: a source whose stored kind differs from +# this constant is re-credentialed and its discovered catalog is dropped, which is how sources +# moved off the org root credential and then off the read-only project_reader. Changing this +# value re-runs that handshake for every existing source, so change it only with that intent. +PROJECT_CREDENTIAL_KIND = "project_user" + def _managed_source_queryset(team_id: int) -> QuerySet[ExternalDataSource]: return ExternalDataSource._base_manager.filter( @@ -79,7 +93,7 @@ def _ensure_managed_source_locked( server: DuckgresServer, username: str, password: str, - reader_configured: bool, + credential_configured: bool, ) -> ExternalDataSource | None: # Deliberately includes soft-deleted rows: a re-enabled membership revives its tombstoned # source (the caller's credential pre-read filters deleted=False, so a revived row always @@ -90,9 +104,10 @@ def _ensure_managed_source_locked( if existing is not None: update_fields: list[str] = [] connection_metadata = dict(existing.connection_metadata or {}) - if connection_metadata.get("credential_kind") != "project_reader": - # Old managed sources used the org root credential, so discard any catalog - # entries discovered before Duckgres enforced the project boundary. + if connection_metadata.get("credential_kind") != PROJECT_CREDENTIAL_KIND: + # The source is on a superseded credential — the org root login, or the read-only + # project_reader. Its catalog was discovered as a different principal, so discard + # the entries and let reconcile rediscover them as the current one. now = timezone.now() DataWarehouseTable.raw_objects.filter( team_id=team_id, external_data_source_id=existing.id, deleted=False @@ -104,24 +119,24 @@ def _ensure_managed_source_locked( if existing.access_method != ExternalDataSource.AccessMethod.DIRECT: existing.access_method = ExternalDataSource.AccessMethod.DIRECT update_fields.append("access_method") - if existing.direct_query_enabled != reader_configured: - existing.direct_query_enabled = reader_configured + if existing.direct_query_enabled != credential_configured: + existing.direct_query_enabled = credential_configured update_fields.append("direct_query_enabled") if ( connection_metadata.get("engine") != "duckdb" or connection_metadata.get("system_managed") is not True - or connection_metadata.get("credential_kind") != "project_reader" + or connection_metadata.get("credential_kind") != PROJECT_CREDENTIAL_KIND ): existing.connection_metadata = { **connection_metadata, "engine": "duckdb", "system_managed": True, - "credential_kind": "project_reader", - "reader_configured": reader_configured, + "credential_kind": PROJECT_CREDENTIAL_KIND, + "credential_configured": credential_configured, } update_fields.append("connection_metadata") - elif connection_metadata.get("reader_configured") is not reader_configured: - existing.connection_metadata = {**connection_metadata, "reader_configured": reader_configured} + elif connection_metadata.get("credential_configured") is not credential_configured: + existing.connection_metadata = {**connection_metadata, "credential_configured": credential_configured} update_fields.append("connection_metadata") if existing.deleted: existing.deleted = False @@ -143,12 +158,12 @@ def _ensure_managed_source_locked( description=MANAGED_WAREHOUSE_SOURCE_DESCRIPTION, access_method=ExternalDataSource.AccessMethod.DIRECT, created_via=ExternalDataSource.CreatedVia.WEB, - direct_query_enabled=reader_configured, + direct_query_enabled=credential_configured, connection_metadata={ "engine": "duckdb", "system_managed": True, - "credential_kind": "project_reader", - "reader_configured": reader_configured, + "credential_kind": PROJECT_CREDENTIAL_KIND, + "credential_configured": credential_configured, }, ) @@ -157,7 +172,7 @@ def _membership_table_suffix(*, team_id: int, organization_id: str | UUID) -> st """The team's warehouse table suffix from its duckgres control-plane row. Raises RuntimeError when the control plane can't answer (callers treat that like the - reader handshake being unavailable and retry on a later sweep) and ValueError when the + credential handshake being unavailable and retry on a later sweep) and ValueError when the team has no backfill-enabled row or sits on the legacy shared tables — neither can be exposed as a per-project query connection. """ @@ -176,7 +191,13 @@ def _membership_table_suffix(*, team_id: int, organization_id: str | UUID) -> st def ensure_managed_warehouse_direct_source(*, team_id: int, organization_id: str | UUID) -> ExternalDataSource: - """Create or refresh the team's restricted live-query source from its membership.""" + """Create or refresh the team's project-scoped live-query source from its membership. + + A source already holding a current-kind credential is left alone. One on a superseded kind + (org root, or the read-only project_reader) is re-credentialed here: it gets a fresh username + and password and stays `direct_query_enabled=False` until the Duckgres handshake below + confirms them, so a half-migrated source is never queryable. + """ from posthog.ducklake.models import DuckgresServer # noqa: PLC0415 table_suffix = _membership_table_suffix(team_id=team_id, organization_id=organization_id) @@ -187,24 +208,28 @@ def ensure_managed_warehouse_direct_source(*, team_id: int, organization_id: str existing = _managed_source_queryset(team_id).select_for_update().filter(deleted=False).first() existing_metadata = existing.connection_metadata if existing is not None else None - has_reader_credentials = ( + has_project_credentials = ( existing is not None and isinstance(existing_metadata, dict) - and existing_metadata.get("credential_kind") == "project_reader" + and existing_metadata.get("credential_kind") == PROJECT_CREDENTIAL_KIND and isinstance(existing.job_inputs, dict) and existing.job_inputs.get("user") and existing.job_inputs.get("password") ) - if has_reader_credentials: + if has_project_credentials: assert existing is not None assert isinstance(existing_metadata, dict) assert isinstance(existing.job_inputs, dict) - reader_configured = existing_metadata.get("reader_configured") is True + credential_configured = existing_metadata.get("credential_configured") is True username = str(existing.job_inputs["user"]) password = str(existing.job_inputs["password"]) else: - reader_configured = False - username = f"posthog_team_{team_id}" + credential_configured = False + # Duckgres derives this name from the team and owns it: the read/write project_user + # is `posthog_team__rw`, distinct from the read-only `posthog_team_`. The + # handshake below asserts the name Duckgres returns matches, so a drift in either + # derivation fails loudly instead of silently configuring the wrong login. + username = f"posthog_team_{team_id}_rw" password = secrets.token_urlsafe(32) source = _ensure_managed_source_locked( @@ -212,33 +237,33 @@ def ensure_managed_warehouse_direct_source(*, team_id: int, organization_id: str server=server, username=username, password=password, - reader_configured=reader_configured, + credential_configured=credential_configured, ) if source is None: raise RuntimeError("Failed to create the managed warehouse query source") - if reader_configured: + if credential_configured: return source source_id = source.id - credentials = managed_warehouse.configure_project_reader( + credentials = managed_warehouse.configure_project_user( organization_id=organization_id, team_id=team_id, table_suffix=table_suffix, password=password, ) if credentials != {"username": username, "password": password}: - raise RuntimeError("Managed warehouse reader credentials did not match the requested credentials") + raise RuntimeError("Managed warehouse project credentials did not match the requested credentials") with transaction.atomic(): DuckgresServer.objects.select_for_update().get(organization_id=organization_id) Team.objects.select_for_update().only("id").get(id=team_id, organization_id=organization_id) source = _managed_source_queryset(team_id).select_for_update().filter(id=source_id, deleted=False).first() if source is None or not isinstance(source.job_inputs, dict): - raise RuntimeError("Managed warehouse query source changed while its reader was configured") + raise RuntimeError("Managed warehouse query source changed while its credential was configured") if source.job_inputs.get("user") != username or source.job_inputs.get("password") != password: - raise RuntimeError("Managed warehouse query source changed while its reader was configured") + raise RuntimeError("Managed warehouse query source changed while its credential was configured") connection_metadata = dict(source.connection_metadata or {}) - source.connection_metadata = {**connection_metadata, "reader_configured": True} + source.connection_metadata = {**connection_metadata, "credential_configured": True} source.direct_query_enabled = True source.save(update_fields=["connection_metadata", "direct_query_enabled", "updated_at"]) return source @@ -255,7 +280,7 @@ def reconcile_managed_warehouse_tables(*, team_id: int, organization_id: str | U except RuntimeError: # The credential handshake needs the warehouse control plane; while an org is still # provisioning this fails on every sweep, so skip quietly and let the next run retry. - logger.info("Managed warehouse reader handshake not possible yet", team_id=team_id) + logger.info("Managed warehouse credential handshake not possible yet", team_id=team_id) return with transaction.atomic(): @@ -271,11 +296,11 @@ def reconcile_managed_warehouse_tables(*, team_id: int, organization_id: str | U source_config = dict(source.job_inputs or {}) source_api_version = source.api_version - # The allowlist mirrors the live Duckgres org-team row (the same row its reader policy is + # The allowlist mirrors the live Duckgres org-team row (the same row its access policy is # derived from), so hand-set layouts — legacy overrides like team 2's posthog.events, custom # schema names like devex — stay in sync instead of assuming the suffix-derived scheme. - # Introspection also runs AS the reader, so this filter is defense in depth, not the boundary. - namespaces = managed_warehouse.project_reader_namespaces(organization_id=organization_id, team_id=team_id) + # Introspection also runs AS the project login, so this filter is defense in depth, not the boundary. + namespaces = managed_warehouse.project_namespaces(organization_id=organization_id, team_id=team_id) if namespaces is None: return allowed_schemas, allowed_relations = namespaces @@ -346,7 +371,7 @@ def _managed_sources_for_org(organization_id: str | UUID) -> QuerySet[ExternalDa def update_managed_warehouse_root_password(*, organization_id: str | UUID, password: str) -> None: - """Refresh the internal root writer without changing project reader credentials.""" + """Refresh the internal root credential without touching any project login.""" from posthog.ducklake.models import DuckgresServer # noqa: PLC0415 with transaction.atomic(): diff --git a/products/data_warehouse/backend/presentation/views/managed_warehouse.py b/products/data_warehouse/backend/presentation/views/managed_warehouse.py index 639e9cb37494..7faa28c22bca 100644 --- a/products/data_warehouse/backend/presentation/views/managed_warehouse.py +++ b/products/data_warehouse/backend/presentation/views/managed_warehouse.py @@ -210,10 +210,15 @@ def _row_team_id(row: dict) -> int | None: return None -def configure_project_reader( +def configure_project_user( *, organization_id: UUID | str, team_id: int, table_suffix: str, password: str ) -> dict[str, str]: - """Apply the project's read-only credential, creating its team row only when absent. + """Apply the project's read/write credential, creating its team row only when absent. + + Duckgres issues two team-scoped logins: `project_reader` (read-only) and `project_user` + (read/write). Both are scoped to exactly the same namespaces — the only difference is + whether writes are authorized — so a project connection gets the writable one and the + project boundary is unchanged. The org-team row is Duckgres-owned state that also drives external-writer discovery (viaduck/millpond write targets) and may be hand-set (break-glass edits, legacy layouts). @@ -245,28 +250,29 @@ def configure_project_reader( credential_response = _request( "PUT", organization_id, - f"/teams/{team_id}/project-reader", + f"/teams/{team_id}/project-user", json_body={"password": password}, require_enabled=False, ) if not status.is_success(credential_response.status_code) or not isinstance(credential_response.data, dict): - raise RuntimeError("Failed to create the project's managed warehouse reader") + raise RuntimeError("Failed to create the project's managed warehouse user") username = credential_response.data.get("username") response_password = credential_response.data.get("password") if not isinstance(username, str) or not username or not isinstance(response_password, str) or not response_password: - raise RuntimeError("Managed warehouse reader response did not include credentials") + raise RuntimeError("Managed warehouse user response did not include credentials") return {"username": username, "password": response_password} -def project_reader_namespaces( - *, organization_id: UUID | str, team_id: int -) -> tuple[set[str], set[tuple[str, str]]] | None: - """Return the (whole schemas, legacy posthog-schema tables) the project's reader may see. +def project_namespaces(*, organization_id: UUID | str, team_id: int) -> tuple[set[str], set[tuple[str, str]]] | None: + """Return the (whole schemas, legacy posthog-schema tables) the project's login may see. - Mirrors the Duckgres policy derivation from the org-team row: the reader is granted the row's + Mirrors the Duckgres policy derivation from the org-team row: the login is granted the row's schema_name, its data-imports schema (override or `_data_imports`), the modeled-data schema, and `posthog.` for each non-NULL legacy events/persons override — including overrides that spell the derived default name. None means no enabled row exists (fail closed). + + Mode-independent by design: Duckgres derives the SAME namespaces for `project_reader` and + `project_user`, so this mirror is correct for both and must not gain a mode argument. """ row = _get_project_team_row(organization_id=organization_id, team_id=team_id) if row is None or row.get("enabled") is not True: @@ -427,11 +433,12 @@ def _register_provisioning_team(organization_id: UUID | str, team_id: int) -> No def _ensure_direct_source(team_id: int, organization_id: UUID | str) -> None: - """Best-effort: register the org's managed warehouse as the team's restricted query connection. + """Best-effort: register the org's managed warehouse as the team's scoped query connection. A managed warehouse speaks the Postgres wire protocol, so each member team gets an ExternalDataSource pointed at the org server. Duckgres scopes its credential to the - project and enforces read-only SQL. A failure here must never block onboarding. + project: read/write inside the project's own namespaces, and nothing outside them. + A failure here must never block onboarding. """ try: # Keep the data_warehouse/warehouse_sources stack off this adapter's import path. diff --git a/products/data_warehouse/backend/tests/api/test_managed_warehouse.py b/products/data_warehouse/backend/tests/api/test_managed_warehouse.py index d76db23e926f..e0ab063cae12 100644 --- a/products/data_warehouse/backend/tests/api/test_managed_warehouse.py +++ b/products/data_warehouse/backend/tests/api/test_managed_warehouse.py @@ -33,24 +33,24 @@ def _onboarding_side_effects(): @patch("products.data_warehouse.backend.presentation.views.managed_warehouse._request") -def test_configure_project_reader_creates_a_missing_team_row_before_rotating_credentials( +def test_configure_project_user_creates_a_missing_team_row_before_rotating_credentials( mock_request: MagicMock, ) -> None: organization_id = uuid4() mock_request.side_effect = [ Response({"teams": []}, status=200), Response({"team_id": 42}, status=200), - Response({"username": "posthog_team_42", "password": "reader-password"}, status=200), + Response({"username": "posthog_team_42_rw", "password": "project-password"}, status=200), ] - credentials = managed_warehouse.configure_project_reader( + credentials = managed_warehouse.configure_project_user( organization_id=organization_id, team_id=42, table_suffix="prod", password="caller-managed-password-with-32-characters", ) - assert credentials == {"username": "posthog_team_42", "password": "reader-password"} + assert credentials == {"username": "posthog_team_42_rw", "password": "project-password"} assert mock_request.call_args_list == [ call("GET", organization_id, "/teams", require_enabled=False), call( @@ -70,7 +70,7 @@ def test_configure_project_reader_creates_a_missing_team_row_before_rotating_cre call( "PUT", organization_id, - "/teams/42/project-reader", + "/teams/42/project-user", json_body={"password": "caller-managed-password-with-32-characters"}, require_enabled=False, ), @@ -78,7 +78,7 @@ def test_configure_project_reader_creates_a_missing_team_row_before_rotating_cre @patch("products.data_warehouse.backend.presentation.views.managed_warehouse._request") -def test_configure_project_reader_never_rewrites_an_existing_team_row(mock_request: MagicMock) -> None: +def test_configure_project_user_never_rewrites_an_existing_team_row(mock_request: MagicMock) -> None: # The Duckgres org-team row also drives external-writer discovery (viaduck/millpond), and rows # can be hand-set (break-glass edits, legacy layouts like the dogfood devex/team-2 rows). # Credential setup must therefore never POST over an existing row. @@ -88,23 +88,23 @@ def test_configure_project_reader_never_rewrites_an_existing_team_row(mock_reque {"teams": [{"team_id": 42, "schema_name": "devex", "enabled": True, "events_table_name": "events"}]}, status=200, ), - Response({"username": "posthog_team_42", "password": "reader-password"}, status=200), + Response({"username": "posthog_team_42_rw", "password": "project-password"}, status=200), ] - credentials = managed_warehouse.configure_project_reader( + credentials = managed_warehouse.configure_project_user( organization_id=organization_id, team_id=42, table_suffix="prod", password="caller-managed-password-with-32-characters", ) - assert credentials == {"username": "posthog_team_42", "password": "reader-password"} + assert credentials == {"username": "posthog_team_42_rw", "password": "project-password"} assert mock_request.call_args_list == [ call("GET", organization_id, "/teams", require_enabled=False), call( "PUT", organization_id, - "/teams/42/project-reader", + "/teams/42/project-user", json_body={"password": "caller-managed-password-with-32-characters"}, require_enabled=False, ), @@ -112,14 +112,14 @@ def test_configure_project_reader_never_rewrites_an_existing_team_row(mock_reque @patch("products.data_warehouse.backend.presentation.views.managed_warehouse._request") -def test_configure_project_reader_refuses_a_disabled_team_row(mock_request: MagicMock) -> None: +def test_configure_project_user_refuses_a_disabled_team_row(mock_request: MagicMock) -> None: # `enabled` is an operator-facing serving hold; credential setup must not silently lift it. mock_request.return_value = Response( {"teams": [{"team_id": 42, "schema_name": "team_42", "enabled": False}]}, status=200 ) with pytest.raises(RuntimeError, match="disabled"): - managed_warehouse.configure_project_reader( + managed_warehouse.configure_project_user( organization_id=uuid4(), team_id=42, table_suffix="prod", @@ -130,7 +130,7 @@ def test_configure_project_reader_refuses_a_disabled_team_row(mock_request: Magi @patch("products.data_warehouse.backend.presentation.views.managed_warehouse._request") -def test_project_reader_namespaces_mirror_the_duckgres_team_row(mock_request: MagicMock) -> None: +def test_project_namespaces_mirror_the_duckgres_team_row(mock_request: MagicMock) -> None: # Must match the Duckgres policy derivation: a non-NULL legacy override always grants # posthog. — including overrides that spell the derived default (team 2's # events_table_name="events" -> posthog.events); NULL overrides grant nothing extra. @@ -150,7 +150,7 @@ def test_project_reader_namespaces_mirror_the_duckgres_team_row(mock_request: Ma status=200, ) - namespaces = managed_warehouse.project_reader_namespaces(organization_id=uuid4(), team_id=2) + namespaces = managed_warehouse.project_namespaces(organization_id=uuid4(), team_id=2) assert namespaces == ( {"team_2", "posthog_data_imports_team_2", "shadow_2_models"}, @@ -159,24 +159,24 @@ def test_project_reader_namespaces_mirror_the_duckgres_team_row(mock_request: Ma @patch("products.data_warehouse.backend.presentation.views.managed_warehouse._request") -def test_project_reader_namespaces_derive_imports_and_skip_absent_overrides(mock_request: MagicMock) -> None: +def test_project_namespaces_derive_imports_and_skip_absent_overrides(mock_request: MagicMock) -> None: mock_request.return_value = Response( {"teams": [{"team_id": 7, "schema_name": "team_7", "enabled": True}]}, status=200 ) - namespaces = managed_warehouse.project_reader_namespaces(organization_id=uuid4(), team_id=7) + namespaces = managed_warehouse.project_namespaces(organization_id=uuid4(), team_id=7) assert namespaces == ({"team_7", "team_7_data_imports", "shadow_7_models"}, set()) @patch("products.data_warehouse.backend.presentation.views.managed_warehouse._request") -def test_project_reader_namespaces_fail_closed_without_an_enabled_row(mock_request: MagicMock) -> None: +def test_project_namespaces_fail_closed_without_an_enabled_row(mock_request: MagicMock) -> None: mock_request.return_value = Response( {"teams": [{"team_id": 7, "schema_name": "team_7", "enabled": False}]}, status=200 ) - assert managed_warehouse.project_reader_namespaces(organization_id=uuid4(), team_id=7) is None - assert managed_warehouse.project_reader_namespaces(organization_id=uuid4(), team_id=8) is None + assert managed_warehouse.project_namespaces(organization_id=uuid4(), team_id=7) is None + assert managed_warehouse.project_namespaces(organization_id=uuid4(), team_id=8) is None @patch("products.data_warehouse.backend.presentation.views.managed_warehouse.posthoganalytics.feature_enabled") diff --git a/products/data_warehouse/backend/tests/test_managed_warehouse_connection.py b/products/data_warehouse/backend/tests/test_managed_warehouse_connection.py index 8f76af39a135..07f83bdbff64 100644 --- a/products/data_warehouse/backend/tests/test_managed_warehouse_connection.py +++ b/products/data_warehouse/backend/tests/test_managed_warehouse_connection.py @@ -43,15 +43,15 @@ class _Connection(TypedDict): "password": "pw", } -_PROJECT_READER_PASSWORD = "reader-password-with-at-least-32-characters" +_PROJECT_USER_PASSWORD = "project-password-with-at-least-32-characters" @pytest.fixture(autouse=True) -def _mock_project_reader_credentials(): - def configure_project_reader(*, team_id: int, password: str, **_kwargs: object) -> dict[str, str]: - return {"username": f"posthog_team_{team_id}", "password": password} +def _mock_project_user_credentials(): + def configure_project_user(*, team_id: int, password: str, **_kwargs: object) -> dict[str, str]: + return {"username": f"posthog_team_{team_id}_rw", "password": password} - def project_reader_namespaces(*, team_id: int, **_kwargs: object) -> tuple[set[str], set[tuple[str, str]]]: + def project_namespaces(*, team_id: int, **_kwargs: object) -> tuple[set[str], set[tuple[str, str]]]: # Mirrors the Duckgres row these tests provision (suffix "prod" layout). return ( {f"team_{team_id}", "posthog_data_imports_prod", f"shadow_{team_id}_models"}, @@ -59,11 +59,11 @@ def project_reader_namespaces(*, team_id: int, **_kwargs: object) -> tuple[set[s ) with ( - patch.object(managed_warehouse, "configure_project_reader", side_effect=configure_project_reader) as mocked, - patch.object(managed_warehouse, "project_reader_namespaces", side_effect=project_reader_namespaces), + patch.object(managed_warehouse, "configure_project_user", side_effect=configure_project_user) as mocked, + patch.object(managed_warehouse, "project_namespaces", side_effect=project_namespaces), patch( "products.data_warehouse.backend.managed_warehouse_connection.secrets.token_urlsafe", - return_value=_PROJECT_READER_PASSWORD, + return_value=_PROJECT_USER_PASSWORD, ), ): yield mocked @@ -138,9 +138,9 @@ def test_creates_a_restricted_postgres_query_source_from_the_server(self) -> Non assert source.prefix == MANAGED_WAREHOUSE_SOURCE_PREFIX # job_inputs carry the warehouse connection so live queries reach it. assert source.job_inputs["host"] == _CONNECTION["host"] - assert source.job_inputs["user"] == f"posthog_team_{team.id}" - assert source.job_inputs["password"] == _PROJECT_READER_PASSWORD - assert source.connection_metadata["credential_kind"] == "project_reader" + assert source.job_inputs["user"] == f"posthog_team_{team.id}_rw" + assert source.job_inputs["password"] == _PROJECT_USER_PASSWORD + assert source.connection_metadata["credential_kind"] == "project_user" def test_is_idempotent(self) -> None: # Without dedup, every status poll / re-enable would spawn a duplicate connection. @@ -162,7 +162,7 @@ def test_is_idempotent(self) -> None: assert first.pk == second.pk assert ExternalDataSource.objects.filter(team_id=team.id, prefix=MANAGED_WAREHOUSE_SOURCE_PREFIX).count() == 1 - def test_concurrent_reader_setup_reuses_the_persisted_credential(self) -> None: + def test_concurrent_credential_setup_reuses_the_persisted_credential(self) -> None: org = Organization.objects.create(name="Org") team = Team.objects.create(organization=org) DuckgresServer.objects.create( @@ -176,20 +176,20 @@ def test_concurrent_reader_setup_reuses_the_persisted_credential(self) -> None: _add_membership(team) requested_passwords: list[str] = [] - def configure_project_reader(*, team_id: int, password: str, **_kwargs: object) -> dict[str, str]: + def configure_project_user(*, team_id: int, password: str, **_kwargs: object) -> dict[str, str]: requested_passwords.append(password) if len(requested_passwords) == 1: _ensure(team) - return {"username": f"posthog_team_{team_id}", "password": password} + return {"username": f"posthog_team_{team_id}_rw", "password": password} - with patch.object(managed_warehouse, "configure_project_reader", side_effect=configure_project_reader): + with patch.object(managed_warehouse, "configure_project_user", side_effect=configure_project_user): source = _ensure(team) source.refresh_from_db() - assert requested_passwords == [_PROJECT_READER_PASSWORD, _PROJECT_READER_PASSWORD] + assert requested_passwords == [_PROJECT_USER_PASSWORD, _PROJECT_USER_PASSWORD] assert source.direct_query_enabled is True assert isinstance(source.connection_metadata, dict) - assert source.connection_metadata["reader_configured"] is True + assert source.connection_metadata["credential_configured"] is True assert ExternalDataSource.objects.filter(team=team, prefix=MANAGED_WAREHOUSE_SOURCE_PREFIX).count() == 1 def test_does_not_expose_legacy_shared_tables(self) -> None: @@ -288,6 +288,86 @@ def test_removes_existing_schemas_when_upgrading_a_root_managed_source(self) -> assert managed_source.access_method == ExternalDataSource.AccessMethod.DIRECT assert not ExternalDataSchema.objects.filter(id=schema.id).exists() + def test_upgrades_a_read_only_project_reader_source_to_the_read_write_login(self) -> None: + # Sources provisioned before writes were the default hold the read-only + # `posthog_team_` credential. The kind mismatch must re-credential them onto the + # read/write `posthog_team__rw` login and drop the catalog discovered as the old + # principal, exactly as the earlier root -> project_reader move did. + org = Organization.objects.create(name="Org") + team = Team.objects.create(organization=org) + DuckgresServer.objects.create( + organization=org, + host=_CONNECTION["host"], + port=_CONNECTION["port"], + database=_CONNECTION["database"], + username=_CONNECTION["username"], + password=_CONNECTION["password"], + ) + _add_membership(team) + source = ExternalDataSource.objects.create( + team=team, + source_id="managed-source", + connection_id="managed-connection", + destination_id="managed-destination", + status=ExternalDataSource.Status.RUNNING, + source_type="Postgres", + prefix=MANAGED_WAREHOUSE_SOURCE_PREFIX, + access_method=ExternalDataSource.AccessMethod.DIRECT, + direct_query_enabled=True, + job_inputs={"user": f"posthog_team_{team.id}", "password": "reader-password"}, + connection_metadata={ + "engine": "duckdb", + "system_managed": True, + "credential_kind": "project_reader", + "reader_configured": True, + }, + ) + schema = ExternalDataSchema.objects.create( + team=team, + source=source, + name=f"team_{team.id}.discovered_as_reader", + should_sync=True, + ) + + managed_source = _ensure(team) + + assert managed_source.id == source.id + assert isinstance(managed_source.job_inputs, dict) + assert managed_source.job_inputs["user"] == f"posthog_team_{team.id}_rw" + assert managed_source.job_inputs["password"] == _PROJECT_USER_PASSWORD + assert isinstance(managed_source.connection_metadata, dict) + assert managed_source.connection_metadata["credential_kind"] == "project_user" + assert managed_source.connection_metadata["credential_configured"] is True + # The old principal's catalog is dropped so reconcile rediscovers as the new one. + assert not ExternalDataSchema.objects.filter(id=schema.id).exists() + + def test_leaves_a_source_already_on_the_read_write_login_untouched(self) -> None: + # The upgrade is a one-time handshake, not something every sweep re-runs: a source + # already holding project_user credentials keeps them (re-minting on every poll would + # churn the password and break in-flight queries). + org = Organization.objects.create(name="Org") + team = Team.objects.create(organization=org) + DuckgresServer.objects.create( + organization=org, + host=_CONNECTION["host"], + port=_CONNECTION["port"], + database=_CONNECTION["database"], + username=_CONNECTION["username"], + password=_CONNECTION["password"], + ) + _add_membership(team) + + first = _ensure(team) + assert isinstance(first.job_inputs, dict) + established_password = first.job_inputs["password"] + + second = _ensure(team) + + assert second.id == first.id + assert isinstance(second.job_inputs, dict) + assert second.job_inputs["user"] == f"posthog_team_{team.id}_rw" + assert second.job_inputs["password"] == established_password + def _source_schema(table_name: str, source_schema: str = "posthog") -> SourceSchema: return SourceSchema( @@ -452,7 +532,7 @@ def test_allowlist_follows_a_legacy_row_with_default_named_overrides(self) -> No org, team = self._setup() with patch.object( managed_warehouse, - "project_reader_namespaces", + "project_namespaces", return_value=( {f"team_{team.id}", "posthog_data_imports_team_2", f"shadow_{team.id}_models"}, {("posthog", "events"), ("posthog", "persons")}, @@ -472,7 +552,7 @@ def test_allowlist_follows_a_legacy_row_with_default_named_overrides(self) -> No def test_fails_closed_when_the_team_row_is_missing_or_disabled(self) -> None: org, team = self._setup() - with patch.object(managed_warehouse, "project_reader_namespaces", return_value=None): + with patch.object(managed_warehouse, "project_namespaces", return_value=None): with patch( "products.warehouse_sources.backend.temporal.data_imports.sources.postgres.source.PostgresSource.get_schemas" ) as get_schemas: @@ -611,7 +691,7 @@ def test_update_root_password_only_rotates_the_internal_root_writer(self) -> Non source.refresh_from_db() server.refresh_from_db() - assert source.job_inputs["password"] == _PROJECT_READER_PASSWORD + assert source.job_inputs["password"] == _PROJECT_USER_PASSWORD assert server.password == "rotated" def test_soft_delete_removes_sources_and_their_tables(self) -> None: