fix(data-warehouse): gate url_pattern writes on node-role tables

products/data_warehouse/backend/direct_clickhouse.py

@@ -60,6 +60,8 @@ def upsert_direct_clickhouse_table(
     existing_table.options = options
     existing_table.deleted = False
     existing_table.deleted_at = None
+    # DIRECT_CLICKHOUSE_URL_PATTERN is a fixed sentinel, not request input, so this upsert is a
+    # trusted writer of a credential-less table's URL.
     existing_table.save(
         update_fields=[
             "name",
@@ -70,7 +72,8 @@ def upsert_direct_clickhouse_table(
             "deleted",
             "deleted_at",
             "updated_at",
-        ]
+        ],
+        internally_computed_url_pattern=True,
     )
     return existing_table
 

products/data_warehouse/backend/direct_mysql.py

@@ -62,6 +62,8 @@ def upsert_direct_mysql_table(
     existing_table.options = options
     existing_table.deleted = False
     existing_table.deleted_at = None
+    # DIRECT_MYSQL_URL_PATTERN is a fixed sentinel, not request input, so this upsert is a
+    # trusted writer of a credential-less table's URL.
     existing_table.save(
         update_fields=[
             "name",
@@ -72,7 +74,8 @@ def upsert_direct_mysql_table(
             "deleted",
             "deleted_at",
             "updated_at",
-        ]
+        ],
+        internally_computed_url_pattern=True,
     )
     return existing_table
 

products/data_warehouse/backend/direct_postgres.py

@@ -70,6 +70,8 @@ def upsert_direct_postgres_table(
     existing_table.options = options
     existing_table.deleted = False
     existing_table.deleted_at = None
+    # DIRECT_POSTGRES_URL_PATTERN is a fixed sentinel, not request input, so this upsert is a
+    # trusted writer of a credential-less table's URL.
     existing_table.save(
         update_fields=[
             "name",
@@ -80,7 +82,8 @@ def upsert_direct_postgres_table(
             "deleted",
             "deleted_at",
             "updated_at",
-        ]
+        ],
+        internally_computed_url_pattern=True,
     )
     return existing_table
 

products/data_warehouse/backend/direct_redshift.py

@@ -71,6 +71,8 @@ def upsert_direct_redshift_table(
     existing_table.options = options
     existing_table.deleted = False
     existing_table.deleted_at = None
+    # DIRECT_REDSHIFT_URL_PATTERN is a fixed sentinel, not request input, so this upsert is a
+    # trusted writer of a credential-less table's URL.
     existing_table.save(
         update_fields=[
             "name",
@@ -81,7 +83,8 @@ def upsert_direct_redshift_table(
             "deleted",
             "deleted_at",
             "updated_at",
-        ]
+        ],
+        internally_computed_url_pattern=True,
     )
     return existing_table
 

products/data_warehouse/backend/direct_snowflake.py

@@ -67,6 +67,8 @@ def upsert_direct_snowflake_table(
     existing_table.options = options
     existing_table.deleted = False
     existing_table.deleted_at = None
+    # DIRECT_SNOWFLAKE_URL_PATTERN is a fixed sentinel, not request input, so this upsert is a
+    # trusted writer of a credential-less table's URL.
     existing_table.save(
         update_fields=[
             "name",
@@ -77,7 +79,8 @@ def upsert_direct_snowflake_table(
             "deleted",
             "deleted_at",
             "updated_at",
-        ]
+        ],
+        internally_computed_url_pattern=True,
     )
     return existing_table
 

products/data_warehouse/backend/presentation/views/table.py

@@ -803,7 +803,10 @@ def file(self, request: request.Request, *args: Any, **kwargs: Any) -> response.
 
                 # Try to determine columns from the file
                 table.columns = table.get_columns()
-                table.save()
+                # team_id comes from routing and safe_filename is sanitized (no path separators), so
+                # the URL above is always scoped to this team's own managed/ prefix, never taken
+                # verbatim from request input the way the PATCH endpoint's url_pattern field is.
+                table.save(internally_computed_url_pattern=True)
 
                 # Validate columns in background
                 from posthog.tasks.warehouse import validate_data_warehouse_table_columns

products/demo/backend/logic/products/hedgebox/matrix.py

@@ -2433,7 +2433,11 @@ def _register_demo_data_warehouse_table(
             existing_table.deleted_at = None
             if existing_table.created_by_id is None:
                 existing_table.created_by = user
-            existing_table.save()
+            # url_pattern is computed above from source_team_id/table_name, not request input, and
+            # credential is a real value from get_or_create_datawarehouse_credential (never None) -
+            # but the guard reads the row's prior DB state, so a stale credential-less row from
+            # before this function existed would still trip it without this declared explicitly.
+            existing_table.save(internally_computed_url_pattern=True)
             return
 
         DataWarehouseTable.objects.create(

products/engineering_analytics/backend/management/commands/seed_engineering_analytics.py

@@ -1145,7 +1145,11 @@ def _upsert_schema_table(
             existing.options = {**(existing.options or {}), "csv_allow_double_quotes": True}
             existing.deleted = False
             existing.deleted_at = None
-            existing.save()
+            # url_pattern is computed above from team/table_name, not request input, and credential
+            # is a real value from get_or_create_datawarehouse_credential (never None) - but the
+            # guard reads the row's prior DB state, so a stale credential-less row would still trip
+            # it without this declared explicitly.
+            existing.save(internally_computed_url_pattern=True)
             table = existing
         else:
             table = DataWarehouseTable.objects.create(

products/warehouse_sources/backend/admin/data_warehouse_table_admin.py

@@ -1,4 +1,5 @@
 from django.contrib import admin
+from django.http import HttpRequest
 from django.urls import reverse
 from django.utils.html import format_html
 
@@ -24,6 +25,15 @@ class DataWarehouseTableAdmin(admin.ModelAdmin):
     readonly_fields = ("credential", "external_data_source")
     ordering = ("-created_at",)
 
+    def has_add_permission(self, request: HttpRequest) -> bool:
+        # A table created here would have no credential (this form has no way to set one - see
+        # readonly_fields) and an unrestricted url_pattern, which is exactly the combination
+        # DataWarehouseTable.clean()/save() exist to refuse on every other write path. Those checks
+        # can't cover creation (a brand-new row has no prior state to compare against), so the
+        # invariant depends entirely on the creator computing url_pattern itself rather than taking
+        # it from form input - true for upload/pipeline sync, never true for a raw admin add form.
+        return False
+
     @admin.display(description="Team")
     def team_link(self, obj: DataWarehouseTable):
         return format_html(

products/warehouse_sources/backend/models/table.py

@@ -2,10 +2,12 @@
 import sys
 import time
 import subprocess
+from collections.abc import Iterable
 from io import StringIO
 from typing import TYPE_CHECKING, Any, NotRequired, Optional, TypedDict, cast
 from uuid import UUID
 
+from django.core.exceptions import ValidationError
 from django.db import models
 from django.db.models import Q
 from django.utils import timezone
@@ -325,6 +327,46 @@ class TableFormat(models.TextChoices):
     class Meta:
         db_table = "posthog_datawarehousetable"
 
+    def save(self, *args: Any, internally_computed_url_pattern: bool = False, **kwargs: Any) -> None:
+        if not internally_computed_url_pattern:
+            self._reject_client_supplied_url_pattern_change(kwargs.get("update_fields"))
+        super().save(*args, **kwargs)
+
+    def clean(self) -> None:
+        # Django admin's changeform validates via full_clean() (form.is_valid() -> clean()) before
+        # ModelAdmin.save_model() ever calls save() - and unlike DRF's perform_update, save_model
+        # doesn't translate a save()-raised ValidationError into a form error, so it would surface
+        # as an unhandled 500. Running the same check here lets admin catch it as a normal field
+        # error. save()'s check stays the enforcement of record for every other caller (DRF, a
+        # management command, a future endpoint), since nothing but ModelForm calls full_clean().
+        super().clean()
+        self._reject_client_supplied_url_pattern_change(update_fields=None)
+
+    def _reject_client_supplied_url_pattern_change(self, update_fields: Iterable[str] | None) -> None:
+        """Block a url_pattern change on a table with no credential, unless the caller declares the
+        new value was computed by PostHog's own code rather than taken from request input.
+
+        A table with no credential is read by ClickHouse under the cluster's own S3 role rather than
+        a key the team supplied, so its url_pattern is only safe because PostHog chose it. Pipeline
+        syncs, saved-query materialization, and the direct-connection upsert helpers all legitimately
+        rewrite this field on such tables using values they compute themselves; those call
+        ``save(internally_computed_url_pattern=True)`` to say so. Every other caller, present or
+        future, is refused by default rather than trusted to have remembered a check elsewhere.
+        """
+        if self._state.adding:
+            return
+        if update_fields is not None and "url_pattern" not in update_fields:
+            return
+        prior = type(self).raw_objects.filter(pk=self.pk).values_list("credential_id", "url_pattern").first()
+        if prior is None:
+            return
+        prior_credential_id, prior_url_pattern = prior
+        if prior_credential_id is None and prior_url_pattern != self.url_pattern:
+            raise ValidationError(
+                "This table has no credential, so its URL is only safe because PostHog set it. "
+                "Add an access key and secret before pointing it at a different location."
+            )
+
     @property
     def name_chain(self) -> list[str]:
         return self.name.split(".")
@@ -988,4 +1030,6 @@ def acreate_datawarehousetable(**kwargs):
 
 @database_sync_to_async
 def asave_datawarehousetable(table: DataWarehouseTable) -> None:
-    table.save()
+    # Saved-query materialization is the only caller: it computes url_pattern itself from the
+    # backing DataWarehouseSavedQuery rather than taking it from request input.
+    table.save(internally_computed_url_pattern=True)

products/warehouse_sources/backend/temporal/data_imports/pipelines/pipeline_sync.py

@@ -227,8 +227,13 @@ def _validate_and_update():
                 # get_count() above can retry against a degraded ClickHouse cluster for minutes, long
                 # enough for the pooled Postgres connection to be recycled underneath us. Retry once
                 # on a fresh connection rather than let this escape as error-tracking noise.
+                # new_url_pattern above is derived from the job's own destination folder, not from
+                # request input, so this sync is a trusted writer of a credential-less table's URL.
                 retry_on_db_connection_drop(
-                    lambda: table.save(update_fields=["format", "url_pattern", "queryable_folder", "row_count"])
+                    lambda: table.save(
+                        update_fields=["format", "url_pattern", "queryable_folder", "row_count"],
+                        internally_computed_url_pattern=True,
+                    )
                 )
 
             if not table_created:
@@ -389,8 +394,13 @@ def _register():
                 # get_count() above can retry against a degraded ClickHouse cluster for minutes, long
                 # enough for the pooled Postgres connection to be recycled underneath us. Retry once
                 # on a fresh connection rather than let this escape as error-tracking noise.
+                # new_url_pattern above is derived from the job's own destination folder, not from
+                # request input, so this sync is a trusted writer of a credential-less table's URL.
                 retry_on_db_connection_drop(
-                    lambda: table.save(update_fields=["format", "url_pattern", "queryable_folder", "row_count"])
+                    lambda: table.save(
+                        update_fields=["format", "url_pattern", "queryable_folder", "row_count"],
+                        internally_computed_url_pattern=True,
+                    )
                 )
             else:
                 logger.debug(f"Creating CDC companion table: {companion_table_name}")

products/warehouse_sources/backend/tests/test_data_warehouse_table_admin.py

@@ -0,0 +1,57 @@
+from posthog.test.base import BaseTest
+
+from django.contrib.admin import AdminSite
+from django.test import RequestFactory
+from django.urls import reverse
+
+from posthog.admin import register_all_admin
+
+from products.warehouse_sources.backend.admin.data_warehouse_table_admin import DataWarehouseTableAdmin
+from products.warehouse_sources.backend.models.table import DataWarehouseTable
+
+
+class TestDataWarehouseTableAdmin(BaseTest):
+    def setUp(self) -> None:
+        super().setUp()
+        # is_superuser is a read-only alias for is_staff in this codebase (no separate superuser
+        # concept), and Django's default auth backend grants every permission to a superuser.
+        self.user.is_staff = True
+        self.user.save()
+        self.admin = DataWarehouseTableAdmin(DataWarehouseTable, AdminSite())
+
+    def test_has_add_permission_is_false(self) -> None:
+        # This form has no way to set a credential (see readonly_fields), so any table it created
+        # would carry the same credential-less-plus-attacker-chosen-url_pattern combination
+        # DataWarehouseTable.clean()/save() exist to refuse - and those checks can't cover creation,
+        # since a brand-new row has no prior state to compare against. Blocking add here is the
+        # closest equivalent for a surface that isn't the product's own controlled creation paths.
+        request = RequestFactory().get("/")
+        request.user = self.user
+
+        assert self.admin.has_add_permission(request) is False
+
+    def test_add_view_rejects_a_credential_less_table_pointed_at_another_teams_data(self) -> None:
+        register_all_admin()
+        self.client.force_login(self.user)
+        add_url = reverse(f"admin:{DataWarehouseTable._meta.app_label}_{DataWarehouseTable._meta.model_name}_add")
+
+        # row_count/size_in_s3_mib/columns are filled in so this is otherwise a fully valid
+        # submission - without has_add_permission, it 302-redirects to the changelist and creates
+        # the row (verified by temporarily reverting the fix locally), so the 403 below is coming
+        # from the permission check this test exists to pin, not from incidental form invalidity.
+        response = self.client.post(
+            add_url,
+            {
+                "team": self.team.pk,
+                "name": "attacker_table",
+                "format": "CSVWithNames",
+                "url_pattern": "https://s3.us-east-1.amazonaws.com/ph-warehouse/file_uploads/team_999/*.csv",
+                "options": "{}",
+                "row_count": "0",
+                "size_in_s3_mib": "0",
+                "columns": "{}",
+            },
+        )
+
+        assert response.status_code == 403
+        assert not DataWarehouseTable.objects.filter(name="attacker_table").exists()