diff --git a/api/app/settings/common.py b/api/app/settings/common.py index 5adde3332224..43e6e2da71f2 100644 --- a/api/app/settings/common.py +++ b/api/app/settings/common.py @@ -58,6 +58,11 @@ SECRET_KEY = env("DJANGO_SECRET_KEY", default=get_random_secret_key()) +WAREHOUSE_CREDENTIALS_SECRET = env( + "WAREHOUSE_CREDENTIALS_SECRET", + default=SECRET_KEY, +) + HOSTED_SEATS_LIMIT = env.int("HOSTED_SEATS_LIMIT", default=0) MAX_PROJECTS_IN_FREE_PLAN = 1 @@ -361,6 +366,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": [ diff --git a/api/app/settings/test.py b/api/app/settings/test.py index c659649528a0..c6e4954a94be 100644 --- a/api/app/settings/test.py +++ b/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" diff --git a/api/core/fields.py b/api/core/fields.py new file mode 100644 index 000000000000..ef07f7caae31 --- /dev/null +++ b/api/core/fields.py @@ -0,0 +1,46 @@ +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: + secret: str = settings.WAREHOUSE_CREDENTIALS_SECRET + digest = hashlib.sha256(secret.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) diff --git a/api/core/network.py b/api/core/network.py new file mode 100644 index 000000000000..40baad2facba --- /dev/null +++ b/api/core/network.py @@ -0,0 +1,34 @@ +import ipaddress +import socket + +# RFC 6598 carrier-grade NAT; not caught by ipaddress.is_private. +_SHARED_ADDRESS_SPACE = ipaddress.ip_network("100.64.0.0/10") + + +def is_internal_address( + hostname: str, + *, + include_shared: bool = False, +) -> bool: + """Return True if the hostname is, or resolves to, an internal network + address: loopback, RFC 1918 private, link-local, reserved, or multicast. + When *include_shared* is True, also blocks RFC 6598 (100.64.0.0/10). + Unresolvable hostnames are not considered internal.""" + try: + ips = [ipaddress.ip_address(hostname)] + except ValueError: + 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 + or (include_shared and ip in _SHARED_ADDRESS_SPACE) + for ip in ips + ) diff --git a/api/experimentation/metrics.py b/api/experimentation/metrics.py new file mode 100644 index 000000000000..8c9d14d1771e --- /dev/null +++ b/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"], + ) +) diff --git a/api/experimentation/migrations/0010_warehouse_connection_credentials_and_status_detail.py b/api/experimentation/migrations/0010_warehouse_connection_credentials_and_status_detail.py new file mode 100644 index 000000000000..0f117529259a --- /dev/null +++ b/api/experimentation/migrations/0010_warehouse_connection_credentials_and_status_detail.py @@ -0,0 +1,22 @@ +from django.db import migrations, models + +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), + ), + migrations.AddField( + model_name="warehouseconnection", + name="status_detail", + field=models.CharField(blank=True, max_length=255, null=True), + ), + ] diff --git a/api/experimentation/models.py b/api/experimentation/models.py index a53b16fa28db..75501d1d5048 100644 --- a/api/experimentation/models.py +++ b/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; diff --git a/api/experimentation/serializers.py b/api/experimentation/serializers.py index 92ca8cd0ace4..d4c60a2d2fc5 100644 --- a/api/experimentation/serializers.py +++ b/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, @@ -40,7 +41,10 @@ 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) + config = serializers.JSONField(required=False, allow_null=True) + credentials = serializers.JSONField( + required=False, allow_null=True, write_only=True + ) total_events_received = serializers.SerializerMethodField() unique_events_count = serializers.SerializerMethodField() @@ -50,33 +54,52 @@ 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 + ) + + validate_credentials(attrs, warehouse_type, self.instance) # type: ignore[arg-type] - if "config" not in attrs and self.instance is not None: + 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): + stored = ( + getattr(self.instance, "config", None) + if self.instance is not None + and not type_changed + and isinstance(getattr(self.instance, "config", None), dict) + else None + ) + attrs["config"] = config_validator(config or {}, stored=stored) 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 +128,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() diff --git a/api/experimentation/services.py b/api/experimentation/services.py index c304f2878b1a..49fdbdcf7890 100644 --- a/api/experimentation/services.py +++ b/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: @@ -788,6 +795,74 @@ def mark_warehouse_pending_connection( return connection +class InternalAddressError(Exception): + pass + + +def _describe_verification_error(error: Exception) -> str: + if isinstance(error, clickhouse_errors.ServerException): + # 516 = AUTHENTICATION_FAILED (not in clickhouse_driver.errors.ErrorCodes) + if error.code == 516: + return "Authentication failed." + if error.code == clickhouse_errors.ErrorCodes.UNKNOWN_DATABASE: + 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"], include_shared=True): + 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, diff --git a/api/experimentation/types.py b/api/experimentation/types.py index 996384956c6c..6122499cb6b8 100644 --- a/api/experimentation/types.py +++ b/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 diff --git a/api/experimentation/views.py b/api/experimentation/views.py index 897b3d0546e0..36fbdf244e93 100644 --- a/api/experimentation/views.py +++ b/api/experimentation/views.py @@ -15,6 +15,7 @@ from rest_framework.request import Request from rest_framework.response import Response from rest_framework.serializers import BaseSerializer +from rest_framework.throttling import BaseThrottle, ScopedRateThrottle from rest_framework.viewsets import GenericViewSet from app.pagination import CustomPagination @@ -62,6 +63,7 @@ mark_warehouse_pending_connection, refresh_warehouse_connection_status, transition_experiment_status, + verify_clickhouse_connection, ) from experimentation.tasks import ( compute_experiment_exposures, @@ -89,6 +91,17 @@ class WarehouseConnectionViewSet( lookup_field = "id" lookup_url_kwarg = "connection_id" + def get_throttles(self) -> list[BaseThrottle]: + if self.action in ( + "create", + "update", + "partial_update", + "test_warehouse_connection", + ): + self.throttle_scope = "warehouse_connection_write" + return [*super().get_throttles(), ScopedRateThrottle()] + return super().get_throttles() + def perform_create(self, serializer: BaseSerializer[WarehouseConnection]) -> None: connection: WarehouseConnection = serializer.save( environment=self._get_environment() @@ -96,12 +109,19 @@ def perform_create(self, serializer: BaseSerializer[WarehouseConnection]) -> Non create_warehouse_audit_log( connection, self._get_user(self.request), action="created" ) + if connection.warehouse_type == WarehouseType.CLICKHOUSE: + verify_clickhouse_connection(connection) def perform_update(self, serializer: BaseSerializer[WarehouseConnection]) -> None: connection: WarehouseConnection = serializer.save() create_warehouse_audit_log( connection, self._get_user(self.request), action="updated" ) + if connection.warehouse_type == WarehouseType.CLICKHOUSE and ( + "config" in serializer.validated_data + or "credentials" in serializer.validated_data + ): + verify_clickhouse_connection(connection) def perform_destroy(self, instance: WarehouseConnection) -> None: create_warehouse_audit_log( @@ -130,9 +150,14 @@ def retrieve(self, request: Request, *args: object, **kwargs: object) -> Respons @action(detail=True, methods=["post"], url_path="test-warehouse-connection") def test_warehouse_connection(self, request: Request, **kwargs: object) -> Response: connection: WarehouseConnection = self.get_object() + if connection.warehouse_type == WarehouseType.CLICKHOUSE: + verify_clickhouse_connection(connection) + return Response(self.get_serializer(connection).data) if connection.warehouse_type != WarehouseType.FLAGSMITH: return Response( - {"detail": "Test events are only supported for Flagsmith warehouses."}, + { + "detail": "Connection testing is not supported for this warehouse type." + }, status=status.HTTP_400_BAD_REQUEST, ) mark_warehouse_pending_connection(connection) diff --git a/api/experimentation/warehouse_validation.py b/api/experimentation/warehouse_validation.py new file mode 100644 index 000000000000..aa01e9450bf5 --- /dev/null +++ b/api/experimentation/warehouse_validation.py @@ -0,0 +1,128 @@ +from typing import Any, Callable, cast + +from rest_framework import serializers + +from core.network import is_internal_address +from experimentation.models import WarehouseConnection, WarehouseType +from experimentation.types import ( + CLICKHOUSE_DEFAULTS, + SNOWFLAKE_DEFAULTS, + ClickHouseConfig, + ClickHouseCredentials, + SnowflakeConfig, +) + + +def validate_clickhouse_credentials( + credentials: dict[str, Any], +) -> ClickHouseCredentials: + if not isinstance(credentials, dict): + raise serializers.ValidationError({"credentials": "Must be an object."}) + password = credentials.get("password") + if not password or not isinstance(password, str): + raise serializers.ValidationError( + {"credentials": {"password": "This field is required."}} + ) + return {"password": password} + + +def validate_clickhouse_config( + config: dict[str, Any], + *, + stored: dict[str, Any] | None = None, +) -> ClickHouseConfig: + if not isinstance(config, dict): + raise serializers.ValidationError({"config": "Must be an object."}) + if unknown_keys := set(config) - set(CLICKHOUSE_DEFAULTS): + raise serializers.ValidationError( + {"config": {key: "Unknown field." for key in sorted(unknown_keys)}} + ) + base = stored if stored is not None else dict(CLICKHOUSE_DEFAULTS) + merged: dict[str, Any] = {**base, **config} + if not merged["host"] or not isinstance(merged["host"], str): + raise serializers.ValidationError( + {"config": {"host": "This field is required."}} + ) + if is_internal_address(merged["host"], include_shared=True): + raise serializers.ValidationError( + { + "config": { + "host": ( + "Host must not target internal or private network addresses." + ) + } + } + ) + port = merged["port"] + if isinstance(port, bool) or not isinstance(port, int) or not (1 <= port <= 65535): + raise serializers.ValidationError( + {"config": {"port": "Enter a valid port number (1-65535)."}} + ) + for key in ("database", "username"): + if not merged[key] or not isinstance(merged[key], str): + raise serializers.ValidationError( + {"config": {key: "Must be a non-empty string."}} + ) + if not isinstance(merged["secure"], bool): + raise serializers.ValidationError({"config": {"secure": "Must be a boolean."}}) + return cast(ClickHouseConfig, merged) + + +def validate_snowflake_config( + config: dict[str, Any], + *, + stored: dict[str, Any] | None = None, +) -> SnowflakeConfig: + if not isinstance(config, dict): + raise serializers.ValidationError({"config": "Must be an object."}) + if unknown_keys := set(config) - set(SNOWFLAKE_DEFAULTS): + raise serializers.ValidationError( + {"config": {key: "Unknown field." for key in sorted(unknown_keys)}} + ) + for key, value in config.items(): + if not isinstance(value, str): + raise serializers.ValidationError({"config": {key: "Must be a string."}}) + base = stored if stored is not None else dict(SNOWFLAKE_DEFAULTS) + merged: SnowflakeConfig = { + **base, # type: ignore[typeddict-item] + **config, + } + if not merged.get("account_identifier"): + raise serializers.ValidationError( + {"config": {"account_identifier": "This field is required."}} + ) + return merged + + +CONFIG_VALIDATORS: dict[str, Callable[..., Any]] = { + WarehouseType.SNOWFLAKE: validate_snowflake_config, + WarehouseType.CLICKHOUSE: validate_clickhouse_config, +} + +CREDENTIAL_VALIDATORS: dict[str, Callable[[dict[str, Any]], Any]] = { + WarehouseType.CLICKHOUSE: validate_clickhouse_credentials, +} + + +def validate_credentials( + attrs: dict[str, Any], + warehouse_type: str, + instance: WarehouseConnection | None, +) -> None: + validator = CREDENTIAL_VALIDATORS.get(warehouse_type) + credentials: dict[str, Any] | None = attrs.get("credentials") + if validator is None: + if credentials is not None: + raise serializers.ValidationError( + {"credentials": "Only ClickHouse connections accept credentials."} + ) + if instance is not None and instance.credentials is not None: + attrs["credentials"] = None + return + if ( + "credentials" not in attrs + and instance is not None + and instance.warehouse_type == warehouse_type + ): + return + attrs["credentials"] = validator(credentials or {}) diff --git a/api/tests/unit/core/test_fields.py b/api/tests/unit/core/test_fields.py new file mode 100644 index 000000000000..d891fa667a60 --- /dev/null +++ b/api/tests/unit/core/test_fields.py @@ -0,0 +1,59 @@ +import pytest +from pytest_django.fixtures import SettingsWrapper +from pytest_structlog import StructuredLogCapture + +from core.fields import EncryptedJSONField + + +def test_get_prep_value__json_value__returns_ciphertext_that_roundtrips() -> None: + # Given + field = EncryptedJSONField() + value = {"password": "hunter2"} + + # When + stored = field.get_prep_value(value) + + # Then + assert stored is not None + assert "hunter2" not in stored + assert field.from_db_value(stored, None, None) == value + + +def test_field_methods__none__returns_none() -> None: + # Given + field = EncryptedJSONField() + + # When & Then + assert field.get_prep_value(None) is None + assert field.from_db_value(None, None, None) is None + + +def test_from_db_value__secret_key_changed__returns_none_and_logs( + settings: SettingsWrapper, + log: StructuredLogCapture, +) -> None: + # Given + settings.WAREHOUSE_CREDENTIALS_SECRET = "old-secret" + field = EncryptedJSONField() + stored = field.get_prep_value({"password": "hunter2"}) + settings.WAREHOUSE_CREDENTIALS_SECRET = "new-secret" + + # When + value = field.from_db_value(stored, None, None) + + # Then + assert value is None + assert { + "level": "warning", + "event": "encrypted_field.decrypt_failed", + } in [{"level": e["level"], "event": e["event"]} for e in log.events] + + +def test_get_lookup__non_isnull__raises_not_implemented() -> None: + # Given + field = EncryptedJSONField() + + # When & Then + with pytest.raises(NotImplementedError): + field.get_lookup("exact") + assert field.get_lookup("isnull") is not None diff --git a/api/tests/unit/experimentation/conftest.py b/api/tests/unit/experimentation/conftest.py index ad9c74cce4b7..04d814c42ce6 100644 --- a/api/tests/unit/experimentation/conftest.py +++ b/api/tests/unit/experimentation/conftest.py @@ -91,3 +91,23 @@ def experiment_with_rollout( ), ) return experiment + + +@pytest.fixture() +def clickhouse_connection( + environment: Environment, +) -> WarehouseConnection: + connection: WarehouseConnection = WarehouseConnection.objects.create( + environment=environment, + warehouse_type=WarehouseType.CLICKHOUSE, + name="Production ClickHouse", + config={ + "host": "ch.acme-corp.example", + "port": 9440, + "database": "acme_dwh", + "username": "acme_svc", + "secure": True, + }, + credentials={"password": "hunter2"}, + ) + return connection diff --git a/api/tests/unit/experimentation/test_models.py b/api/tests/unit/experimentation/test_models.py index ba3ae0c173fe..a058e872010e 100644 --- a/api/tests/unit/experimentation/test_models.py +++ b/api/tests/unit/experimentation/test_models.py @@ -3,6 +3,7 @@ from datetime import timezone as dt_timezone import pytest +from django.db import connection as django_db_connection from django.utils import timezone from pytest_mock import MockerFixture @@ -241,3 +242,22 @@ def test_experiment_results__is_final__reflects_window_coverage( # When / Then the row is final only once it covers the experiment's end results = ExperimentResults(experiment=experiment, as_of=as_of) assert results.is_final is expected + + +def test_warehouse_connection_credentials__saved__ciphertext_in_db_and_roundtrips( + clickhouse_connection: WarehouseConnection, +) -> None: + # Given + + # When + with django_db_connection.cursor() as cursor: + cursor.execute( + "SELECT credentials FROM experimentation_warehouseconnection WHERE id = %s", + [clickhouse_connection.id], + ) + raw = cursor.fetchone()[0] + + # Then + assert "hunter2" not in raw + clickhouse_connection.refresh_from_db() + assert clickhouse_connection.credentials == {"password": "hunter2"} diff --git a/api/tests/unit/experimentation/test_services.py b/api/tests/unit/experimentation/test_services.py index 81f5561291d5..9ee8437f86ca 100644 --- a/api/tests/unit/experimentation/test_services.py +++ b/api/tests/unit/experimentation/test_services.py @@ -3,8 +3,10 @@ from unittest.mock import MagicMock import pytest +from clickhouse_driver import errors as clickhouse_errors from django.db.models import Q from flag_engine.segments.constants import PERCENTAGE_SPLIT +from prometheus_client import REGISTRY from pytest_django.fixtures import SettingsWrapper from pytest_mock import MockerFixture from pytest_structlog import StructuredLogCapture @@ -36,6 +38,11 @@ WarehouseType, ) from experimentation.results_query import _MetricSlot +from experimentation.services import ( + InternalAddressError, + _describe_verification_error, + verify_clickhouse_connection, +) from experimentation.stats import VariantStats from features.feature_types import MULTIVARIATE from features.models import Feature, FeatureState @@ -2005,3 +2012,182 @@ def variant_assignment() -> dict[str, int]: # Then every already-enrolled identity keeps the variant it was first # assigned; tuning the rollout must not re-randomise the split. assert before == after + + +def _verification_count(result: str) -> float: + return ( + REGISTRY.get_sample_value( + "flagsmith_experimentation_warehouse_connection_verifications_total", + {"result": result}, + ) + or 0.0 + ) + + +def test_verify_clickhouse_connection__reachable__sets_connected( + clickhouse_connection: WarehouseConnection, + log: StructuredLogCapture, + mocker: MockerFixture, +) -> None: + # Given + mock_client = mocker.patch("experimentation.services.Client") + success_count_before = _verification_count("success") + clickhouse_connection.status_detail = "stale detail" + clickhouse_connection.save() + + # When + verify_clickhouse_connection(clickhouse_connection) + + # Then + clickhouse_connection.refresh_from_db() + assert clickhouse_connection.status == WarehouseConnectionStatus.CONNECTED + assert clickhouse_connection.status_detail is None + mock_client.assert_called_once_with( + "ch.acme-corp.example", + port=9440, + user="acme_svc", + password="hunter2", + database="acme_dwh", + secure=True, + connect_timeout=5, + send_receive_timeout=5, + ) + mock_client.return_value.execute.assert_called_once_with("SELECT 1") + mock_client.return_value.disconnect.assert_called_once_with() + assert _verification_count("success") == success_count_before + 1 + assert { + "level": "info", + "event": "connection.verification_succeeded", + "environment__id": clickhouse_connection.environment_id, + "organisation__id": (clickhouse_connection.environment.project.organisation_id), + } in log.events + + +@pytest.mark.parametrize( + "credentials, execute_side_effect, expected_detail", + [ + ( + {"password": "hunter2"}, + Exception("connection refused"), + "Verification failed.", + ), + (None, None, "Stored connection details are incomplete."), + ], + ids=["driver_error", "missing_credentials"], +) +def test_verify_clickhouse_connection__failure__sets_errored_with_detail( + clickhouse_connection: WarehouseConnection, + credentials: dict[str, str] | None, + execute_side_effect: Exception | None, + expected_detail: str, + log: StructuredLogCapture, + mocker: MockerFixture, +) -> None: + # Given + mock_client = mocker.patch("experimentation.services.Client") + mock_client.return_value.execute.side_effect = execute_side_effect + clickhouse_connection.credentials = credentials + clickhouse_connection.save() + failure_count_before = _verification_count("failure") + + # When + verify_clickhouse_connection(clickhouse_connection) + + # Then + clickhouse_connection.refresh_from_db() + assert clickhouse_connection.status == WarehouseConnectionStatus.ERRORED + assert clickhouse_connection.status_detail == expected_detail + assert _verification_count("failure") == failure_count_before + 1 + assert any( + event["event"] == "connection.verification_failed" for event in log.events + ) + + +def test_verify_clickhouse_connection__internal_host__sets_errored_without_connecting( + clickhouse_connection: WarehouseConnection, + mocker: MockerFixture, +) -> None: + # Given + mock_client = mocker.patch("experimentation.services.Client") + clickhouse_connection.config = { + **(clickhouse_connection.config or {}), + "host": "10.0.0.5", + } + clickhouse_connection.save() + + # When + verify_clickhouse_connection(clickhouse_connection) + + # Then + clickhouse_connection.refresh_from_db() + assert clickhouse_connection.status == WarehouseConnectionStatus.ERRORED + assert ( + clickhouse_connection.status_detail + == "Host must not target internal or private network addresses." + ) + mock_client.assert_not_called() + + +@pytest.mark.parametrize( + "error,expected_detail", + [ + ( + clickhouse_errors.ServerException("Authentication failed", code=516), + "Authentication failed.", + ), + ( + clickhouse_errors.ServerException( + "Database not_a_real_db does not exist", code=81 + ), + "Database does not exist.", + ), + ( + clickhouse_errors.ServerException("Some other server error", code=999), + "The ClickHouse server rejected the request.", + ), + ( + clickhouse_errors.SocketTimeoutError("(10.255.255.1:9000)"), + "The connection timed out.", + ), + ( + TimeoutError("timed out"), + "The connection timed out.", + ), + ( + clickhouse_errors.NetworkError("Connection refused"), + "Could not connect to the host.", + ), + ( + InternalAddressError("10.0.0.5"), + "Host must not target internal or private network addresses.", + ), + ( + KeyError("host"), + "Stored connection details are incomplete.", + ), + ( + ValueError("unexpected"), + "Verification failed.", + ), + ], + ids=[ + "server_exception_auth_failure", + "server_exception_unknown_database", + "server_exception_other", + "socket_timeout_error", + "builtin_timeout_error", + "network_error", + "internal_address_error", + "key_error", + "generic_exception", + ], +) +def test_describe_verification_error__known_error_types__returns_expected_detail( + error: Exception, + expected_detail: str, +) -> None: + # Given / When + detail = _describe_verification_error(error) + + # Then + assert detail == expected_detail diff --git a/api/tests/unit/experimentation/test_views.py b/api/tests/unit/experimentation/test_views.py index dc6068edb1d5..cbcb9aae5c6b 100644 --- a/api/tests/unit/experimentation/test_views.py +++ b/api/tests/unit/experimentation/test_views.py @@ -1,9 +1,12 @@ +import socket + import pytest from django.urls import reverse from pytest_django.fixtures import SettingsWrapper from pytest_mock import MockerFixture from rest_framework import status from rest_framework.test import APIClient +from rest_framework.throttling import ScopedRateThrottle from audit.models import AuditLog from audit.related_object_type import RelatedObjectType @@ -15,6 +18,7 @@ WarehouseConnectionStatus, WarehouseType, ) +from experimentation.views import WarehouseConnectionViewSet from tests.types import EnableFeaturesFixture pytestmark = pytest.mark.django_db @@ -632,6 +636,43 @@ def test_patch__snowflake_update_config__returns_200( assert data["config"]["warehouse"] == "BIG_WH" +def test_patch__snowflake_partial_config__preserves_stored_account_identifier( + admin_client: APIClient, + environment: Environment, + enable_features: EnableFeaturesFixture, + warehouse_connection_url: str, +) -> None: + # Given + enable_features("experimentation_warehouse_connection") + create_response = admin_client.post( + warehouse_connection_url, + data={ + "warehouse_type": "snowflake", + "name": "My Snowflake", + "config": {"account_identifier": "xy12345.us-east-1"}, + }, + format="json", + ) + connection_id = create_response.json()["id"] + url = reverse( + "api-v1:environments:experimentation:warehouse-connections-detail", + args=[environment.api_key, connection_id], + ) + + # When + response = admin_client.patch( + url, + data={"config": {"warehouse": "BIG_WH"}}, + format="json", + ) + + # Then + assert response.status_code == status.HTTP_200_OK + data = response.json() + assert data["config"]["account_identifier"] == "xy12345.us-east-1" + assert data["config"]["warehouse"] == "BIG_WH" + + def test_patch__snowflake_update_name__returns_200( admin_client: APIClient, environment: Environment, @@ -1047,6 +1088,41 @@ def test_test_warehouse_connection__non_admin__returns_403( assert response.status_code == status.HTTP_403_FORBIDDEN +@pytest.mark.parametrize( + "action", + ["create", "update", "partial_update", "test_warehouse_connection"], +) +def test_get_throttles__write_actions__returns_scoped_throttle(action: str) -> None: + # Given + view = WarehouseConnectionViewSet() + view.action = action + + # When + throttles = view.get_throttles() + + # Then + assert any(isinstance(t, ScopedRateThrottle) for t in throttles) + assert view.throttle_scope == "warehouse_connection_write" + + +@pytest.mark.parametrize( + "action", + ["list", "retrieve", "destroy"], +) +def test_get_throttles__other_actions__returns_view_default_throttles( + action: str, +) -> None: + # Given + view = WarehouseConnectionViewSet() + view.action = action + + # When + throttles = view.get_throttles() + + # Then + assert [type(throttle) for throttle in throttles] == list(view.throttle_classes) + + def test_get__clickhouse_unconfigured__returns_200_without_stats( admin_client: APIClient, environment: Environment, @@ -1126,3 +1202,552 @@ def test_get__clickhouse_errors__returns_200_without_stats( assert data["unique_events_count"] is None # connection stays pending (GET never writes) assert data["status"] == "pending_connection" + + +def test_post__clickhouse_minimal_payload__applies_defaults_and_generates_name( + admin_client: APIClient, + enable_features: EnableFeaturesFixture, + environment: Environment, + mocker: MockerFixture, + warehouse_connection_url: str, +) -> None: + # Given + enable_features("experimentation_warehouse_connection") + mocker.patch("experimentation.services.Client") + + # When + response = admin_client.post( + warehouse_connection_url, + data={ + "warehouse_type": "clickhouse", + "config": {"host": "ch.example.com"}, + "credentials": {"password": "hunter2"}, + }, + format="json", + ) + + # Then + assert response.status_code == status.HTTP_201_CREATED + assert response.json()["config"] == { + "host": "ch.example.com", + "port": 9440, + "database": "flagsmith", + "username": "default", + "secure": True, + } + assert response.json()["name"] == f"ClickHouse Warehouse - {environment.name}" + assert "credentials" not in response.json() + + +@pytest.mark.parametrize( + "data, error_path", + [ + pytest.param( + {"warehouse_type": "clickhouse", "credentials": {"password": "hunter2"}}, + ("config", "host"), + id="missing_host", + ), + pytest.param( + { + "warehouse_type": "clickhouse", + "config": {"host": "127.0.0.1"}, + "credentials": {"password": "hunter2"}, + }, + ("config", "host"), + id="loopback_host", + ), + pytest.param( + { + "warehouse_type": "clickhouse", + "config": {"host": "10.0.0.1"}, + "credentials": {"password": "hunter2"}, + }, + ("config", "host"), + id="private_host", + ), + pytest.param( + { + "warehouse_type": "clickhouse", + "config": {"host": "169.254.169.254"}, + "credentials": {"password": "hunter2"}, + }, + ("config", "host"), + id="link_local_host", + ), + pytest.param( + { + "warehouse_type": "clickhouse", + "config": {"host": "ch.example.com", "port": 0}, + "credentials": {"password": "hunter2"}, + }, + ("config", "port"), + id="port_zero", + ), + pytest.param( + { + "warehouse_type": "clickhouse", + "config": {"host": "ch.example.com", "port": 65536}, + "credentials": {"password": "hunter2"}, + }, + ("config", "port"), + id="port_above_range", + ), + pytest.param( + { + "warehouse_type": "clickhouse", + "config": {"host": "ch.example.com", "port": "not-a-port"}, + "credentials": {"password": "hunter2"}, + }, + ("config", "port"), + id="port_not_a_number", + ), + pytest.param( + { + "warehouse_type": "clickhouse", + "config": {"host": "ch.example.com", "port": True}, + "credentials": {"password": "hunter2"}, + }, + ("config", "port"), + id="port_boolean", + ), + pytest.param( + {"warehouse_type": "clickhouse", "config": {"host": "ch.example.com"}}, + ("credentials", "password"), + id="missing_password", + ), + pytest.param( + { + "warehouse_type": "clickhouse", + "config": {"host": "ch.example.com"}, + "credentials": "hunter2", + }, + ("credentials",), + id="non_dict_credentials", + ), + pytest.param( + { + "warehouse_type": "clickhouse", + "config": "not-a-dict", + "credentials": {"password": "hunter2"}, + }, + ("config",), + id="non_dict_config", + ), + pytest.param( + {"warehouse_type": "flagsmith", "credentials": {"password": "hunter2"}}, + ("credentials",), + id="flagsmith_with_credentials", + ), + pytest.param( + { + "warehouse_type": "snowflake", + "config": {"account_identifier": "xy12345.us-east-1"}, + "credentials": {"password": "hunter2"}, + }, + ("credentials",), + id="snowflake_with_credentials", + ), + pytest.param( + {"warehouse_type": "flagsmith", "credentials": {}}, + ("credentials",), + id="flagsmith_empty_credentials", + ), + pytest.param( + { + "warehouse_type": "clickhouse", + "config": {"host": "ch.example.com", "password": "oops"}, + "credentials": {"password": "hunter2"}, + }, + ("config", "password"), + id="unknown_config_key", + ), + pytest.param( + { + "warehouse_type": "clickhouse", + "config": {"host": "ch.example.com", "secure": "false"}, + "credentials": {"password": "hunter2"}, + }, + ("config", "secure"), + id="secure_not_boolean", + ), + pytest.param( + { + "warehouse_type": "clickhouse", + "config": {"host": "ch.example.com", "database": 123}, + "credentials": {"password": "hunter2"}, + }, + ("config", "database"), + id="database_not_string", + ), + pytest.param( + {"warehouse_type": "snowflake", "config": "not-a-dict"}, + ("config",), + id="snowflake_non_dict_config", + ), + pytest.param( + { + "warehouse_type": "snowflake", + "config": {"account_identifier": "xy12345", "extra": "bad"}, + }, + ("config", "extra"), + id="snowflake_unknown_key", + ), + pytest.param( + { + "warehouse_type": "snowflake", + "config": {"account_identifier": "xy12345", "warehouse": 123}, + }, + ("config", "warehouse"), + id="snowflake_non_string_value", + ), + pytest.param( + { + "warehouse_type": "clickhouse", + "config": {"host": "100.64.0.1"}, + "credentials": {"password": "hunter2"}, + }, + ("config", "host"), + id="shared_address_space_host", + ), + ], +) +def test_post__invalid_payload__returns_400( + admin_client: APIClient, + data: dict[str, object], + enable_features: EnableFeaturesFixture, + error_path: tuple[str, ...], + warehouse_connection_url: str, +) -> None: + # Given + enable_features("experimentation_warehouse_connection") + + # When + response = admin_client.post(warehouse_connection_url, data=data, format="json") + + # Then + assert response.status_code == status.HTTP_400_BAD_REQUEST + errors = response.json() + for key in error_path[:-1]: + errors = errors[key] + assert error_path[-1] in errors + + +def test_post__clickhouse_hostname_resolves_to_internal_ip__returns_400( + admin_client: APIClient, + enable_features: EnableFeaturesFixture, + mocker: MockerFixture, + warehouse_connection_url: str, +) -> None: + # Given + enable_features("experimentation_warehouse_connection") + mocker.patch( + "core.network.socket.getaddrinfo", + return_value=[(socket.AF_INET, None, None, None, ("192.168.1.100", 0))], + ) + + # When + response = admin_client.post( + warehouse_connection_url, + data={ + "warehouse_type": "clickhouse", + "config": {"host": "internal.example.com"}, + "credentials": {"password": "hunter2"}, + }, + format="json", + ) + + # Then + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert "host" in response.json()["config"] + + +def test_patch__flagsmith_to_clickhouse_without_credentials__returns_400( + admin_client: APIClient, + enable_features: EnableFeaturesFixture, + environment: Environment, + warehouse_connection: WarehouseConnection, +) -> None: + # Given + enable_features("experimentation_warehouse_connection") + url = reverse( + "api-v1:environments:experimentation:warehouse-connections-detail", + args=[environment.api_key, warehouse_connection.id], + ) + + # When + response = admin_client.patch( + url, + data={ + "warehouse_type": "clickhouse", + "config": {"host": "ch.example.com"}, + }, + format="json", + ) + + # Then + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert "password" in response.json()["credentials"] + + +@pytest.mark.parametrize( + "execute_side_effect, expected_status, expected_detail", + [ + (None, "connected", None), + (Exception("unreachable"), "errored", "Verification failed."), + ], + ids=["reachable", "unreachable"], +) +def test_post__clickhouse_verification_outcome__returns_201_with_status( + admin_client: APIClient, + enable_features: EnableFeaturesFixture, + execute_side_effect: Exception | None, + expected_detail: str | None, + expected_status: str, + mocker: MockerFixture, + warehouse_connection_url: str, +) -> None: + # Given + enable_features("experimentation_warehouse_connection") + mock_client = mocker.patch("experimentation.services.Client") + mock_client.return_value.execute.side_effect = execute_side_effect + + # When + response = admin_client.post( + warehouse_connection_url, + data={ + "warehouse_type": "clickhouse", + "name": "Production ClickHouse", + "config": {"host": "ch.example.com"}, + "credentials": {"password": "hunter2"}, + }, + format="json", + ) + + # Then + assert response.status_code == status.HTTP_201_CREATED + assert response.json()["status"] == expected_status + assert response.json()["status_detail"] == expected_detail + assert "credentials" not in response.json() + mock_client.return_value.execute.assert_called_once_with("SELECT 1") + + +def test_get__clickhouse__credentials_not_in_response( + admin_client: APIClient, + clickhouse_connection: WarehouseConnection, + enable_features: EnableFeaturesFixture, + warehouse_connection_url: str, +) -> None: + # Given + enable_features("experimentation_warehouse_connection") + + # When + response = admin_client.get(warehouse_connection_url) + + # Then + assert response.status_code == status.HTTP_200_OK + (data,) = response.json() + assert "credentials" not in data + assert "password" not in str(data) + + +def test_patch__clickhouse_config_without_credentials__keeps_stored_password( + admin_client: APIClient, + clickhouse_connection: WarehouseConnection, + enable_features: EnableFeaturesFixture, + environment: Environment, + mocker: MockerFixture, +) -> None: + # Given + enable_features("experimentation_warehouse_connection") + mock_client = mocker.patch("experimentation.services.Client") + url = reverse( + "api-v1:environments:experimentation:warehouse-connections-detail", + args=[environment.api_key, clickhouse_connection.id], + ) + + # When + response = admin_client.patch( + url, + data={"config": {"host": "ch.example.com", "port": 9000}}, + format="json", + ) + + # Then + assert response.status_code == status.HTTP_200_OK + clickhouse_connection.refresh_from_db() + assert clickhouse_connection.credentials == {"password": "hunter2"} + assert clickhouse_connection.config is not None + assert clickhouse_connection.config.get("database") == "acme_dwh" + assert clickhouse_connection.config.get("port") == 9000 + assert mock_client.call_args.kwargs["password"] == "hunter2" + assert mock_client.call_args.kwargs["port"] == 9000 + + +def test_patch__clickhouse_name_only__does_not_reverify( + admin_client: APIClient, + clickhouse_connection: WarehouseConnection, + enable_features: EnableFeaturesFixture, + environment: Environment, + mocker: MockerFixture, +) -> None: + # Given + enable_features("experimentation_warehouse_connection") + mock_client = mocker.patch("experimentation.services.Client") + url = reverse( + "api-v1:environments:experimentation:warehouse-connections-detail", + args=[environment.api_key, clickhouse_connection.id], + ) + + # When + response = admin_client.patch(url, data={"name": "Renamed"}, format="json") + + # Then + assert response.status_code == status.HTTP_200_OK + mock_client.assert_not_called() + + +def test_put__clickhouse_name_only__preserves_config_and_credentials( + admin_client: APIClient, + clickhouse_connection: WarehouseConnection, + enable_features: EnableFeaturesFixture, + environment: Environment, + mocker: MockerFixture, +) -> None: + # Given + enable_features("experimentation_warehouse_connection") + mocker.patch("experimentation.services.Client") + url = reverse( + "api-v1:environments:experimentation:warehouse-connections-detail", + args=[environment.api_key, clickhouse_connection.id], + ) + + # When + response = admin_client.put( + url, + data={ + "warehouse_type": "clickhouse", + "name": "Renamed", + }, + format="json", + ) + + # Then + assert response.status_code == status.HTTP_200_OK + clickhouse_connection.refresh_from_db() + assert clickhouse_connection.config is not None + assert clickhouse_connection.credentials == {"password": "hunter2"} + + +def test_test_warehouse_connection__clickhouse__reverifies_and_returns_status( + admin_client: APIClient, + clickhouse_connection: WarehouseConnection, + enable_features: EnableFeaturesFixture, + environment: Environment, + mocker: MockerFixture, +) -> None: + # Given + enable_features("experimentation_warehouse_connection") + clickhouse_connection.status = WarehouseConnectionStatus.ERRORED + clickhouse_connection.save() + mocker.patch("experimentation.services.Client") + url = reverse( + "api-v1:environments:experimentation:" + "warehouse-connections-test-warehouse-connection", + args=[environment.api_key, clickhouse_connection.id], + ) + + # When + response = admin_client.post(url, format="json") + + # Then + assert response.status_code == status.HTTP_200_OK + assert response.json()["status"] == "connected" + clickhouse_connection.refresh_from_db() + assert clickhouse_connection.status == WarehouseConnectionStatus.CONNECTED + + +def test_patch__clickhouse_to_flagsmith__resets_connection_state( + admin_client: APIClient, + clickhouse_connection: WarehouseConnection, + enable_features: EnableFeaturesFixture, + environment: Environment, +) -> None: + # Given + enable_features("experimentation_warehouse_connection") + clickhouse_connection.status = WarehouseConnectionStatus.ERRORED + clickhouse_connection.status_detail = "Authentication failed." + clickhouse_connection.save() + url = reverse( + "api-v1:environments:experimentation:warehouse-connections-detail", + args=[environment.api_key, clickhouse_connection.id], + ) + + # When + response = admin_client.patch( + url, + data={"warehouse_type": "flagsmith"}, + format="json", + ) + + # Then + assert response.status_code == status.HTTP_200_OK + clickhouse_connection.refresh_from_db() + assert clickhouse_connection.credentials is None + assert clickhouse_connection.config is None + assert clickhouse_connection.status == WarehouseConnectionStatus.CREATED + assert clickhouse_connection.status_detail is None + + +def test_patch__flagsmith_to_clickhouse_without_config__returns_400( + admin_client: APIClient, + enable_features: EnableFeaturesFixture, + environment: Environment, + warehouse_connection: WarehouseConnection, +) -> None: + # Given + enable_features("experimentation_warehouse_connection") + url = reverse( + "api-v1:environments:experimentation:warehouse-connections-detail", + args=[environment.api_key, warehouse_connection.id], + ) + + # When + response = admin_client.patch( + url, + data={ + "warehouse_type": "clickhouse", + "credentials": {"password": "hunter2"}, + }, + format="json", + ) + + # Then + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert "host" in response.json()["config"] + + +def test_patch__clickhouse_null_credentials__returns_400( + admin_client: APIClient, + clickhouse_connection: WarehouseConnection, + enable_features: EnableFeaturesFixture, + environment: Environment, +) -> None: + # Given + enable_features("experimentation_warehouse_connection") + url = reverse( + "api-v1:environments:experimentation:warehouse-connections-detail", + args=[environment.api_key, clickhouse_connection.id], + ) + + # When + response = admin_client.patch( + url, + data={"credentials": None}, + format="json", + ) + + # Then + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert "password" in response.json()["credentials"] + clickhouse_connection.refresh_from_db() + assert clickhouse_connection.credentials == {"password": "hunter2"} diff --git a/api/tests/unit/webhooks/test_unit_webhooks.py b/api/tests/unit/webhooks/test_unit_webhooks.py index 620d210a23e6..411e629ea4fe 100644 --- a/api/tests/unit/webhooks/test_unit_webhooks.py +++ b/api/tests/unit/webhooks/test_unit_webhooks.py @@ -373,7 +373,7 @@ def test_send_test_webhook__200_response_from_webhook__returns_correct_response( mock_response.text = "success" mock_post.return_value = mock_response mocker.patch( - "webhooks.fields.socket.getaddrinfo", + "core.network.socket.getaddrinfo", return_value=[(socket.AF_INET, None, None, None, ("93.184.216.34", 0))], ) @@ -421,7 +421,7 @@ def test_send_test_webhook__various_2xx_status_codes__returns_success( mock_response.text = "success" mock_post.return_value = mock_response mocker.patch( - "webhooks.fields.socket.getaddrinfo", + "core.network.socket.getaddrinfo", return_value=[(socket.AF_INET, None, None, None, ("93.184.216.34", 0))], ) @@ -471,7 +471,7 @@ def test_send_test_webhook__various_error_status_codes__returns_correct_response mock_response.text = external_api_error_text mock_post.return_value = mock_response mocker.patch( - "webhooks.fields.socket.getaddrinfo", + "core.network.socket.getaddrinfo", return_value=[(socket.AF_INET, None, None, None, ("93.184.216.34", 0))], ) @@ -540,7 +540,7 @@ def test_send_test_webhook__various_secrets__sends_correct_payload( mock_response.ok = True mock_post.return_value = mock_response mocker.patch( - "webhooks.fields.socket.getaddrinfo", + "core.network.socket.getaddrinfo", return_value=[(socket.AF_INET, None, None, None, ("93.184.216.34", 0))], ) @@ -583,7 +583,7 @@ def test_send_test_webhook__request_exception__returns_error_response( "Some internal exception details that should not be exposed!" ) mocker.patch( - "webhooks.fields.socket.getaddrinfo", + "core.network.socket.getaddrinfo", return_value=[(socket.AF_INET, None, None, None, ("93.184.216.34", 0))], ) diff --git a/api/tests/unit/webhooks/test_webhooks_fields.py b/api/tests/unit/webhooks/test_webhooks_fields.py index f2763a2c85b6..5706a8337279 100644 --- a/api/tests/unit/webhooks/test_webhooks_fields.py +++ b/api/tests/unit/webhooks/test_webhooks_fields.py @@ -58,7 +58,7 @@ def test_no_ssrf_url_field__hostname_resolving_to_private_ip__raises_validation_ ) -> None: # Given — a hostname that resolves to an RFC1918 address with mock.patch( - "webhooks.fields.socket.getaddrinfo", + "core.network.socket.getaddrinfo", return_value=[(socket.AF_INET, None, None, None, ("192.168.1.100", 0))], ): # When / Then @@ -73,7 +73,7 @@ def test_no_ssrf_url_field__hostname_resolving_to_private_ipv6__raises_validatio ) -> None: # Given — an AAAA-only hostname resolving to a private IPv6 address with mock.patch( - "webhooks.fields.socket.getaddrinfo", + "core.network.socket.getaddrinfo", return_value=[(socket.AF_INET6, None, None, None, ("fc00::1", 0, 0, 0))], ): # When / Then @@ -108,7 +108,7 @@ def test_no_ssrf_url_field__unresolvable_hostname__returns_value( ) -> None: # Given — the hostname cannot be resolved; URL format is still valid with mock.patch( - "webhooks.fields.socket.getaddrinfo", + "core.network.socket.getaddrinfo", side_effect=socket.gaierror, ): # When diff --git a/api/webhooks/fields.py b/api/webhooks/fields.py index 8729c1262b0a..13ffe8fa8b9b 100644 --- a/api/webhooks/fields.py +++ b/api/webhooks/fields.py @@ -1,9 +1,9 @@ -import ipaddress -import socket from urllib.parse import urlparse from rest_framework import serializers +from core.network import is_internal_address + class NoSSRFURLField(serializers.URLField): """ @@ -27,26 +27,5 @@ def run_validators(self, value: str) -> None: super().run_validators(value) hostname = urlparse(value).hostname or "" - - 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: - # Unresolvable hostname; leave it to the URL validator. - return - - for ip in ips: - if ( - ip.is_loopback - or ip.is_private - or ip.is_link_local - or ip.is_reserved - or ip.is_multicast - ): - self.fail("internal_address") + if is_internal_address(hostname): + self.fail("internal_address") diff --git a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md index 40fe1dd98bb8..9583b11095c3 100644 --- a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md +++ b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md @@ -71,6 +71,14 @@ Attributes: - `feature.count` - `organisation.id` +### `core.encrypted_field.decrypt_failed` + +Logged at `warning` from: + - `api/core/fields.py:37` + +Attributes: + - `exc_info` + ### `dynamodb.environment_document_compressed` Logged at `info` from: @@ -533,7 +541,7 @@ Attributes: ### `warehouse.connection.connected` Logged at `info` from: - - `api/experimentation/services.py:803` + - `api/experimentation/services.py:878` Attributes: - `environment.id` @@ -542,7 +550,26 @@ Attributes: ### `warehouse.connection.test_event_sent` Logged at `info` from: - - `api/experimentation/services.py:783` + - `api/experimentation/services.py:790` + +Attributes: + - `environment.id` + - `organisation.id` + +### `warehouse.connection.verification_failed` + +Logged at `warning` from: + - `api/experimentation/services.py:854` + +Attributes: + - `environment.id` + - `exc_info` + - `organisation.id` + +### `warehouse.connection.verification_succeeded` + +Logged at `info` from: + - `api/experimentation/services.py:863` Attributes: - `environment.id` @@ -551,7 +578,7 @@ Attributes: ### `warehouse.srm.overallocated` Logged at `error` from: - - `api/experimentation/services.py:405` + - `api/experimentation/services.py:412` Attributes: - `environment.id` @@ -561,7 +588,7 @@ Attributes: ### `warehouse.srm.unkeyed_variant` Logged at `error` from: - - `api/experimentation/services.py:391` + - `api/experimentation/services.py:398` Attributes: - `environment.id` diff --git a/docs/docs/deployment-self-hosting/observability/_metrics-catalogue.md b/docs/docs/deployment-self-hosting/observability/_metrics-catalogue.md index 649284cdac6b..ffe186d0d105 100644 --- a/docs/docs/deployment-self-hosting/observability/_metrics-catalogue.md +++ b/docs/docs/deployment-self-hosting/observability/_metrics-catalogue.md @@ -34,6 +34,15 @@ Counter. Results of cache retrieval for environment document. `result` label is either `hit` or `miss`. +Labels: + - `result` + +### `flagsmith_experimentation_warehouse_connection_verifications` + +Counter. + +Outcomes of connection verification attempts against customers' own data warehouses. `result` label is either `success` or `failure`. + Labels: - `result` diff --git a/infrastructure/aws/production/ecs-task-definition-admin-api.json b/infrastructure/aws/production/ecs-task-definition-admin-api.json index 32605f19ac9c..08df7794c8e9 100644 --- a/infrastructure/aws/production/ecs-task-definition-admin-api.json +++ b/infrastructure/aws/production/ecs-task-definition-admin-api.json @@ -247,6 +247,10 @@ "name": "DJANGO_SECRET_KEY", "valueFrom": "arn:aws:secretsmanager:eu-west-2:084060095745:secret:ECS-API-LxUiIQ:DJANGO_SECRET_KEY::" }, + { + "name": "WAREHOUSE_CREDENTIALS_SECRET", + "valueFrom": "arn:aws:secretsmanager:eu-west-2:084060095745:secret:ECS-API-LxUiIQ:DJANGO_SECRET_KEY::" + }, { "name": "E2E_TEST_AUTH_TOKEN", "valueFrom": "arn:aws:secretsmanager:eu-west-2:084060095745:secret:ECS-API-LxUiIQ:E2E_TEST_AUTH_TOKEN::" diff --git a/infrastructure/aws/production/ecs-task-definition-sdk-api.json b/infrastructure/aws/production/ecs-task-definition-sdk-api.json index 7e928544ac72..5ba0e4ee7aae 100644 --- a/infrastructure/aws/production/ecs-task-definition-sdk-api.json +++ b/infrastructure/aws/production/ecs-task-definition-sdk-api.json @@ -252,6 +252,10 @@ "name": "DJANGO_SECRET_KEY", "valueFrom": "arn:aws:secretsmanager:eu-west-2:084060095745:secret:ECS-API-LxUiIQ:DJANGO_SECRET_KEY::" }, + { + "name": "WAREHOUSE_CREDENTIALS_SECRET", + "valueFrom": "arn:aws:secretsmanager:eu-west-2:084060095745:secret:ECS-API-LxUiIQ:DJANGO_SECRET_KEY::" + }, { "name": "E2E_TEST_AUTH_TOKEN", "valueFrom": "arn:aws:secretsmanager:eu-west-2:084060095745:secret:ECS-API-LxUiIQ:E2E_TEST_AUTH_TOKEN::" diff --git a/infrastructure/aws/production/ecs-task-definition-task-processor.json b/infrastructure/aws/production/ecs-task-definition-task-processor.json index a476a1bf12e4..bfc4ea7dcd56 100644 --- a/infrastructure/aws/production/ecs-task-definition-task-processor.json +++ b/infrastructure/aws/production/ecs-task-definition-task-processor.json @@ -212,6 +212,10 @@ "name": "DJANGO_SECRET_KEY", "valueFrom": "arn:aws:secretsmanager:eu-west-2:084060095745:secret:ECS-API-LxUiIQ:DJANGO_SECRET_KEY::" }, + { + "name": "WAREHOUSE_CREDENTIALS_SECRET", + "valueFrom": "arn:aws:secretsmanager:eu-west-2:084060095745:secret:ECS-API-LxUiIQ:DJANGO_SECRET_KEY::" + }, { "name": "E2E_TEST_AUTH_TOKEN", "valueFrom": "arn:aws:secretsmanager:eu-west-2:084060095745:secret:ECS-API-LxUiIQ:E2E_TEST_AUTH_TOKEN::" diff --git a/infrastructure/aws/staging/ecs-task-definition-admin-api.json b/infrastructure/aws/staging/ecs-task-definition-admin-api.json index 7afb69722b3b..9e5085cec168 100644 --- a/infrastructure/aws/staging/ecs-task-definition-admin-api.json +++ b/infrastructure/aws/staging/ecs-task-definition-admin-api.json @@ -248,6 +248,10 @@ "name": "DJANGO_SECRET_KEY", "valueFrom": "arn:aws:secretsmanager:eu-west-2:302456015006:secret:ECS-API-heAdoB:DJANGO_SECRET_KEY::" }, + { + "name": "WAREHOUSE_CREDENTIALS_SECRET", + "valueFrom": "arn:aws:secretsmanager:eu-west-2:302456015006:secret:ECS-API-heAdoB:DJANGO_SECRET_KEY::" + }, { "name": "E2E_TEST_AUTH_TOKEN", "valueFrom": "arn:aws:secretsmanager:eu-west-2:302456015006:secret:ECS-API-heAdoB:E2E_TEST_AUTH_TOKEN::" diff --git a/infrastructure/aws/staging/ecs-task-definition-sdk-api.json b/infrastructure/aws/staging/ecs-task-definition-sdk-api.json index d3b609fb34dc..c37116f73c92 100644 --- a/infrastructure/aws/staging/ecs-task-definition-sdk-api.json +++ b/infrastructure/aws/staging/ecs-task-definition-sdk-api.json @@ -259,6 +259,10 @@ "name": "DJANGO_SECRET_KEY", "valueFrom": "arn:aws:secretsmanager:eu-west-2:302456015006:secret:ECS-API-heAdoB:DJANGO_SECRET_KEY::" }, + { + "name": "WAREHOUSE_CREDENTIALS_SECRET", + "valueFrom": "arn:aws:secretsmanager:eu-west-2:302456015006:secret:ECS-API-heAdoB:DJANGO_SECRET_KEY::" + }, { "name": "E2E_TEST_AUTH_TOKEN", "valueFrom": "arn:aws:secretsmanager:eu-west-2:302456015006:secret:ECS-API-heAdoB:E2E_TEST_AUTH_TOKEN::" diff --git a/infrastructure/aws/staging/ecs-task-definition-task-processor.json b/infrastructure/aws/staging/ecs-task-definition-task-processor.json index 7c1ed3f07f94..8e3acf02fb88 100644 --- a/infrastructure/aws/staging/ecs-task-definition-task-processor.json +++ b/infrastructure/aws/staging/ecs-task-definition-task-processor.json @@ -202,6 +202,10 @@ "name": "DJANGO_SECRET_KEY", "valueFrom": "arn:aws:secretsmanager:eu-west-2:302456015006:secret:ECS-API-heAdoB:DJANGO_SECRET_KEY::" }, + { + "name": "WAREHOUSE_CREDENTIALS_SECRET", + "valueFrom": "arn:aws:secretsmanager:eu-west-2:302456015006:secret:ECS-API-heAdoB:DJANGO_SECRET_KEY::" + }, { "name": "GITHUB_CLIENT_SECRET", "valueFrom": "arn:aws:secretsmanager:eu-west-2:302456015006:secret:ECS-API-heAdoB:GITHUB_CLIENT_SECRET::" diff --git a/openapi.yaml b/openapi.yaml index 9c04337341d9..b27f5a3cef04 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -24889,6 +24889,11 @@ components: allOf: - $ref: '#/components/schemas/WarehouseConnectionStatusEnum' readOnly: true + status_detail: + type: + - string + - 'null' + readOnly: true name: type: string maxLength: 255 @@ -24896,6 +24901,11 @@ components: oneOf: - {} - type: 'null' + credentials: + oneOf: + - {} + - type: 'null' + writeOnly: true created_at: type: string format: date-time @@ -27807,6 +27817,11 @@ components: allOf: - $ref: '#/components/schemas/WarehouseConnectionStatusEnum' readOnly: true + status_detail: + type: + - string + - 'null' + readOnly: true name: type: string maxLength: 255 @@ -27814,6 +27829,11 @@ components: oneOf: - {} - type: 'null' + credentials: + oneOf: + - {} + - type: 'null' + writeOnly: true created_at: type: string format: date-time