feat: ClickHouse bring-your-own-warehouse connections

api/app/settings/common.py

@@ -361,6 +361,7 @@
         "invite": "10/min",
         "user": USER_THROTTLE_RATE,
         "influx_query": "5/min",
+        "warehouse_connection_write": "10/min",
     },
     "DEFAULT_FILTER_BACKENDS": ["django_filters.rest_framework.DjangoFilterBackend"],
     "DEFAULT_RENDERER_CLASSES": [

api/app/settings/test.py

@@ -27,6 +27,7 @@
     "user": "100000/day",
     "master_api_key": "100000/day",
     "influx_query": "50/min",
+    "warehouse_connection_write": "1000/min",
 }
 
 AWS_SSE_LOGS_BUCKET_NAME = "test_bucket"

api/core/fields.py

@@ -0,0 +1,45 @@
+import base64
+import hashlib
+import json
+from typing import Any
+
+import structlog
+from cryptography.fernet import Fernet, InvalidToken
+from django.conf import settings
+from django.db import models
+
+logger = structlog.get_logger("core")
+
+
+def _get_fernet() -> Fernet:
+    digest = hashlib.sha256(settings.SECRET_KEY.encode()).digest()
+    return Fernet(base64.urlsafe_b64encode(digest))
+
+
+class EncryptedJSONField(models.TextField[Any, Any]):
+    def get_prep_value(self, value: Any) -> str | None:
+        if value is None:
+            return None
+        return _get_fernet().encrypt(json.dumps(value).encode()).decode()
+
+    def from_db_value(
+        self,
+        value: str | None,
+        expression: object,
+        connection: object,
+    ) -> Any:
+        if value is None:
+            return None
+        try:
+            plaintext = _get_fernet().decrypt(value.encode())
+        except InvalidToken:
+            logger.warning("encrypted_field.decrypt_failed", exc_info=True)
+            return None
+        return json.loads(plaintext)
+
+    def get_lookup(self, lookup_name: str) -> Any:
+        if lookup_name != "isnull":
+            raise NotImplementedError(
+                "EncryptedJSONField only supports isnull lookups."
+            )
+        return super().get_lookup(lookup_name)

api/core/network.py

@@ -0,0 +1,26 @@
+import ipaddress
+import socket
+
+
+def is_internal_address(hostname: str) -> bool:
+    """Return True if the hostname is, or resolves to, an internal network
+    address: loopback, RFC 1918 private, link-local, reserved, or multicast.
+    Unresolvable hostnames are not considered internal."""
+    try:
+        ips = [ipaddress.ip_address(hostname)]
+    except ValueError:
+        # hostname is a name rather than a literal IP — resolve it.
+        try:
+            results = socket.getaddrinfo(hostname, None, socket.AF_UNSPEC)
+            ips = [ipaddress.ip_address(str(r[4][0]).split("%")[0]) for r in results]
+        except socket.gaierror:
+            return False
+
+    return any(
+        ip.is_loopback
+        or ip.is_private
+        or ip.is_link_local
+        or ip.is_reserved
+        or ip.is_multicast
+        for ip in ips
+    )

api/experimentation/metrics.py

@@ -0,0 +1,10 @@
+import prometheus_client
+
+flagsmith_experimentation_warehouse_connection_verifications_total = (
+    prometheus_client.Counter(
+        "flagsmith_experimentation_warehouse_connection_verifications_total",
+        "Outcomes of connection verification attempts against customers' own "
+        "data warehouses. `result` label is either `success` or `failure`.",
+        ["result"],
+    )
+)

api/experimentation/migrations/0010_warehouse_connection_credentials.py

@@ -0,0 +1,17 @@
+from django.db import migrations
+
+import core.fields
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("experimentation", "0009_add_rollout_segment"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="warehouseconnection",
+            name="credentials",
+            field=core.fields.EncryptedJSONField(blank=True, null=True),
+        ),
+    ]

api/experimentation/migrations/0011_warehouse_connection_status_detail.py

@@ -0,0 +1,17 @@
+# Generated by Django 5.2.16 on 2026-07-16 09:01
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("experimentation", "0010_warehouse_connection_credentials"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="warehouseconnection",
+            name="status_detail",
+            field=models.CharField(blank=True, max_length=255, null=True),
+        ),
+    ]

api/experimentation/models.py

@@ -12,6 +12,7 @@
     hook,
 )
 
+from core.fields import EncryptedJSONField
 from core.models import SoftDeleteExportableModel
 from environments.models import Environment
 from experimentation.dataclasses import (
@@ -54,10 +55,12 @@ class WarehouseConnection(LifecycleModelMixin, SoftDeleteExportableModel):  # ty
         choices=WarehouseConnectionStatus.choices,
         default=WarehouseConnectionStatus.CREATED,
     )
+    status_detail = models.CharField(max_length=255, null=True, blank=True)
     name = models.CharField(max_length=255)
     config: models.JSONField[dict[str, object] | None, dict[str, object] | None] = (
         models.JSONField(null=True, blank=True)
     )
+    credentials = EncryptedJSONField(null=True, blank=True)
     created_at = models.DateTimeField(auto_now_add=True)
 
     # Populated at serialization time for flagsmith connections from ClickHouse;

api/experimentation/serializers.py

@@ -17,16 +17,17 @@
     ExperimentStatus,
     Metric,
     WarehouseConnection,
+    WarehouseConnectionStatus,
     WarehouseType,
 )
 from experimentation.services import (
     apply_experiment_rollout,
     get_experiment_rollout,
 )
-from experimentation.types import (
-    SNOWFLAKE_DEFAULTS,
-    MetricExperimentResult,
-    SnowflakeConfig,
+from experimentation.types import MetricExperimentResult
+from experimentation.warehouse_validation import (
+    CONFIG_VALIDATORS,
+    validate_credentials,
 )
 from features.feature_states.serializers import (
     FeatureValueSerializer,
@@ -41,6 +42,9 @@
 class WarehouseConnectionSerializer(serializers.ModelSerializer):  # type: ignore[type-arg]
     name = serializers.CharField(max_length=255, required=False)
     config = serializers.JSONField(default=None, required=False, allow_null=True)
+    credentials = serializers.JSONField(
+        default=None, required=False, allow_null=True, write_only=True
+    )
     total_events_received = serializers.SerializerMethodField()
     unique_events_count = serializers.SerializerMethodField()
 
@@ -50,33 +54,45 @@ class Meta:
             "id",
             "warehouse_type",
             "status",
+            "status_detail",
             "name",
             "config",
+            "credentials",
             "created_at",
             "total_events_received",
             "unique_events_count",
         )
-        read_only_fields = ("id", "status", "created_at")
+        read_only_fields = ("id", "status", "status_detail", "created_at")
 
     def validate(self, attrs: dict[str, Any]) -> dict[str, Any]:
         warehouse_type: str = attrs.get(
             "warehouse_type",
             getattr(self.instance, "warehouse_type", ""),
         )
+        type_changed = (
+            self.instance is not None
+            and getattr(self.instance, "warehouse_type", "") != warehouse_type
+        )
 
-        if "config" not in attrs and self.instance is not None:
+        validate_credentials(attrs, warehouse_type, self.instance)  # type: ignore[arg-type]
+
+        if "config" not in attrs and self.instance is not None and not type_changed:
             return attrs
 
         config: dict[str, Any] | None = attrs.get("config")
 
-        if warehouse_type == WarehouseType.SNOWFLAKE:
-            attrs["config"] = self._validate_snowflake_config(config or {})
+        if config_validator := CONFIG_VALIDATORS.get(warehouse_type):
+            attrs["config"] = config_validator(config or {})
         elif warehouse_type == WarehouseType.FLAGSMITH:
             if config:
                 raise serializers.ValidationError(
                     {"config": "Flagsmith warehouse does not accept configuration."}
                 )
             attrs["config"] = None
+
+        if type_changed:
+            attrs["status"] = WarehouseConnectionStatus.CREATED
+            attrs["status_detail"] = None
         return attrs
 
     def create(
@@ -105,19 +121,6 @@ def _generate_name(warehouse_type: str, environment: Environment) -> str:
         label = WarehouseType(warehouse_type).label
         return f"{label} Warehouse - {environment.name}"
 
-    @staticmethod
-    def _validate_snowflake_config(config: dict[str, Any]) -> SnowflakeConfig:
-        account_identifier = config.get("account_identifier", "")
-        if not account_identifier:
-            raise serializers.ValidationError(
-                {"config": {"account_identifier": "This field is required."}}
-            )
-        merged: SnowflakeConfig = {
-            **SNOWFLAKE_DEFAULTS,
-            **config,  # type: ignore[typeddict-item]
-        }
-        return merged
-
 
 class MetricSerializer(serializers.ModelSerializer):  # type: ignore[type-arg]
     experiments = serializers.SerializerMethodField()

api/experimentation/services.py

@@ -6,6 +6,7 @@
 
 import structlog
 from clickhouse_driver import Client
+from clickhouse_driver import errors as clickhouse_errors
 from clickhouse_driver.util.helpers import parse_url
 from django.conf import settings
 from django.db import transaction
@@ -17,6 +18,7 @@
 from audit.models import AuditLog
 from audit.related_object_type import RelatedObjectType
 from core.dataclasses import AuthorData
+from core.network import is_internal_address
 from environments.tasks import rebuild_environment_document
 from experimentation.constants import (
     CONTROL_VARIANT_KEY,
@@ -40,6 +42,9 @@
     RolloutSpec,
     WarehouseEventStats,
 )
+from experimentation.metrics import (
+    flagsmith_experimentation_warehouse_connection_verifications_total,
+)
 from experimentation.models import (
     VALID_STATUS_TRANSITIONS,
     Experiment,
@@ -56,6 +61,7 @@
     compare_to_control,
     srm_p_value,
 )
+from experimentation.types import ClickHouseConfig, ClickHouseCredentials
 from features.models import FeatureState
 from features.value_types import BOOLEAN, INTEGER, STRING
 from features.versioning.dataclasses import FlagChangeSet, MultivariateValueChangeSet
@@ -88,6 +94,7 @@
 
 CLICKHOUSE_CONNECT_TIMEOUT_SECONDS = 5
 CLICKHOUSE_QUERY_TIMEOUT_SECONDS = 30
+CLICKHOUSE_VERIFY_TIMEOUT_SECONDS = 5
 
 
 def is_warehouse_feature_enabled(organisation: Organisation) -> bool:
@@ -787,6 +794,73 @@ def mark_warehouse_pending_connection(
     return connection
 
 
+class InternalAddressError(Exception):
+    pass
+
+
+def _describe_verification_error(error: Exception) -> str:
+    if isinstance(error, clickhouse_errors.ServerException):
+        if error.code == 516:
+            return "Authentication failed."
+        if error.code == 81:
+            return "Database does not exist."
+        return "The ClickHouse server rejected the request."
+    if isinstance(error, (clickhouse_errors.SocketTimeoutError, TimeoutError)):
+        return "The connection timed out."
+    if isinstance(error, clickhouse_errors.NetworkError):
+        return "Could not connect to the host."
+    if isinstance(error, InternalAddressError):
+        return "Host must not target internal or private network addresses."
+    if isinstance(error, KeyError):
+        return "Stored connection details are incomplete."
+    return "Verification failed."
+
+
+def verify_clickhouse_connection(connection: WarehouseConnection) -> None:
+    """Run SELECT 1 against the customer's ClickHouse and set the status to
+    connected or errored; never raises."""
+    log = logger.bind(environment__id=connection.environment_id)
+    try:
+        log = log.bind(organisation__id=connection.environment.project.organisation_id)
+        config = typing.cast(ClickHouseConfig, connection.config or {})
+        credentials = typing.cast(ClickHouseCredentials, connection.credentials or {})
+        # Re-check right before connecting: DNS may resolve differently than it
+        # did at validation time, and rows may predate host validation.
+        if is_internal_address(config["host"]):
+            raise InternalAddressError(config["host"])
+        client = Client(
+            config["host"],
+            port=config["port"],
+            user=config["username"],
+            password=credentials["password"],
+            database=config["database"],
+            secure=config["secure"],
+            connect_timeout=CLICKHOUSE_CONNECT_TIMEOUT_SECONDS,
+            send_receive_timeout=CLICKHOUSE_VERIFY_TIMEOUT_SECONDS,
+        )
+        try:
+            client.execute("SELECT 1")
+        finally:
+            client.disconnect()
+    except Exception as error:
+        connection.status = WarehouseConnectionStatus.ERRORED
+        connection.status_detail = _describe_verification_error(error)
+        connection.save(update_fields=["status", "status_detail"])
+        flagsmith_experimentation_warehouse_connection_verifications_total.labels(
+            result="failure"
+        ).inc()
+        log.warning("connection.verification_failed", exc_info=True)
+        return
+
+    connection.status = WarehouseConnectionStatus.CONNECTED
+    connection.status_detail = None
+    connection.save(update_fields=["status", "status_detail"])
+    flagsmith_experimentation_warehouse_connection_verifications_total.labels(
+        result="success"
+    ).inc()
+    log.info("connection.verification_succeeded")
+
+
 def refresh_warehouse_connection_status(
     connection: WarehouseConnection,
     stats: WarehouseEventStats,

api/experimentation/types.py

@@ -44,3 +44,24 @@ class SnowflakeConfig(TypedDict):
     "role": "FLAGSMITH_LOADER",
     "user": "FLAGSMITH_SERVICE",
 }
+
+
+class ClickHouseConfig(TypedDict):
+    host: str
+    port: int
+    database: str
+    username: str
+    secure: bool
+
+
+CLICKHOUSE_DEFAULTS: ClickHouseConfig = {
+    "host": "",
+    "port": 9440,
+    "database": "flagsmith",
+    "username": "default",
+    "secure": True,
+}
+
+
+class ClickHouseCredentials(TypedDict):
+    password: str