From bd0a549285599e463c66eba861851a26b512c847 Mon Sep 17 00:00:00 2001 From: Gagan Trivedi Date: Fri, 21 Aug 2026 13:39:03 +0530 Subject: [PATCH 01/10] feat(cohorts): add Mixpanel cohort sync webhook --- api/api/openapi.py | 4 +- api/cohorts/authentication.py | 31 +- .../migrations/0004_mixpanel_source.py | 43 ++ api/cohorts/models.py | 14 + api/cohorts/serializers.py | 21 + api/cohorts/services.py | 63 ++- api/cohorts/sync_urls.py | 8 +- api/cohorts/sync_views.py | 122 ++++- api/tests/unit/cohorts/conftest.py | 24 + api/tests/unit/cohorts/test_services.py | 73 ++- api/tests/unit/cohorts/test_sync_views.py | 449 ++++++++++++++++++ .../observability/_events-catalogue.md | 51 +- sdk/openapi.yaml | 2 +- 13 files changed, 863 insertions(+), 42 deletions(-) create mode 100644 api/cohorts/migrations/0004_mixpanel_source.py diff --git a/api/api/openapi.py b/api/api/openapi.py index 2784dd1270b4..1c6b9a7e3750 100644 --- a/api/api/openapi.py +++ b/api/api/openapi.py @@ -175,7 +175,9 @@ def get_security_definition( "scheme": "bearer", "description": ( "For cohort sync endpoints called by an external cohort " - "source, such as Amplitude." + "source, such as Amplitude. Sources that can only send " + "Basic credentials (e.g. Mixpanel) pass the key as the " + "password, with any username." ), } diff --git a/api/cohorts/authentication.py b/api/cohorts/authentication.py index 175a29b2f005..fc608c5c8ec5 100644 --- a/api/cohorts/authentication.py +++ b/api/cohorts/authentication.py @@ -1,3 +1,4 @@ +import base64 import typing from contextlib import suppress @@ -9,17 +10,43 @@ class CohortSyncKeyAuthentication(authentication.BaseAuthentication): + """ + Accepts a cohort sync key sent either as a Bearer token or as the + password of Basic credentials. Amplitude sends Bearer; Mixpanel's + webhook setup only offers a username/password form, so its customers + enter any username and the key as the password. The username is + ignored. + """ + def authenticate( self, request: Request ) -> tuple[AnonymousUser, CohortSyncKey] | None: header = request.headers.get("Authorization", "") - if not header.startswith("Bearer "): + if header.startswith("Bearer "): + raw_key = header.removeprefix("Bearer ") + elif header.startswith("Basic "): + try: + decoded = base64.b64decode( + header.removeprefix("Basic "), validate=True + ).decode() + except ValueError: + # Covers malformed base64, header bytes outside ASCII, and + # decoded credentials that are not valid UTF-8. + raise exceptions.AuthenticationFailed("Invalid Basic credentials.") + # Split at the first colon, so a key containing colons survives. + _, _, raw_key = decoded.partition(":") + else: return None + if "\x00" in raw_key: + # Postgres refuses to run a query containing a NUL character, so + # the key lookup below would crash instead of returning 401. + raise exceptions.AuthenticationFailed("Valid cohort sync key not found.") + with suppress(CohortSyncKey.DoesNotExist): key = typing.cast( CohortSyncKey, - CohortSyncKey.objects.get_from_key(header.removeprefix("Bearer ")), + CohortSyncKey.objects.get_from_key(raw_key), ) if not key.has_expired: # No person is acting here, so no user is returned: the key diff --git a/api/cohorts/migrations/0004_mixpanel_source.py b/api/cohorts/migrations/0004_mixpanel_source.py new file mode 100644 index 000000000000..8d2368c4d6ab --- /dev/null +++ b/api/cohorts/migrations/0004_mixpanel_source.py @@ -0,0 +1,43 @@ +# Generated by Django 5.2.16 on 2026-08-20 10:06 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ("cohorts", "0003_cohort_sync_key"), + ("environments", "0039_use_no_ssrf_url_field"), + ("segments", "0032_add_segment_rules_data"), + ] + + operations = [ + migrations.AddField( + model_name="cohort", + name="external_id", + field=models.CharField(blank=True, max_length=255, null=True), + ), + migrations.AlterField( + model_name="cohort", + name="source_type", + field=models.CharField( + choices=[ + ("csv", "CSV"), + ("amplitude", "Amplitude"), + ("mixpanel", "Mixpanel"), + ], + default="csv", + max_length=50, + ), + ), + migrations.AddConstraint( + model_name="cohort", + constraint=models.UniqueConstraint( + condition=models.Q( + ("deleted_at__isnull", True), ("external_id__isnull", False) + ), + fields=("environment", "source_type", "external_id"), + name="unique_active_cohort_per_source_external_id", + ), + ), + ] diff --git a/api/cohorts/models.py b/api/cohorts/models.py index b15927b6fcd0..f59e55489b29 100644 --- a/api/cohorts/models.py +++ b/api/cohorts/models.py @@ -8,6 +8,7 @@ class CohortSourceType(models.TextChoices): CSV = "csv", "CSV" AMPLITUDE = "amplitude", "Amplitude" + MIXPANEL = "mixpanel", "Mixpanel" class Cohort(SoftDeleteExportableModel): @@ -26,6 +27,10 @@ class Cohort(SoftDeleteExportableModel): choices=CohortSourceType.choices, default=CohortSourceType.CSV, ) + # The cohort's identifier in the external source (e.g. Mixpanel's cohort + # ID). Set for sources that push to us under their own identifier; null + # for sources that adopt ours (Amplitude) and for CSV cohorts. + external_id = models.CharField(max_length=255, null=True, blank=True) version = models.PositiveIntegerField(default=0) created_at = models.DateTimeField(auto_now_add=True) # Deletion drains memberships from the identity store first; the cohort is @@ -45,6 +50,15 @@ class Meta: condition=models.Q(deleted_at__isnull=True), name="unique_active_cohort_per_segment", ), + # One Mixpanel cohort must map to one active cohort per + # environment: without this, two simultaneous first-sync requests + # would each create their own cohort and split the members + # between them. + models.UniqueConstraint( + fields=["environment", "source_type", "external_id"], + condition=models.Q(deleted_at__isnull=True, external_id__isnull=False), + name="unique_active_cohort_per_source_external_id", + ), ] diff --git a/api/cohorts/serializers.py b/api/cohorts/serializers.py index da5bc7e7d7f7..d919d0ff0636 100644 --- a/api/cohorts/serializers.py +++ b/api/cohorts/serializers.py @@ -98,6 +98,27 @@ def _validate_identifier_byte_length(value: str) -> None: ) +class MixpanelMemberSerializer(serializers.Serializer[None]): + # Length mirrors CohortMembership.identifier. + mixpanel_distinct_id = serializers.CharField(max_length=2000) + + +class MixpanelParametersSerializer(serializers.Serializer[None]): + mixpanel_cohort_id = serializers.CharField(max_length=255) + mixpanel_cohort_name = serializers.CharField(max_length=2000) + # An empty page is valid: a first sync of an empty cohort has no members. + members = MixpanelMemberSerializer(many=True, allow_empty=True) + + +class MixpanelWebhookSerializer(serializers.Serializer[None]): + # "members" carries the full membership on the first sync; + # "add_members"/"remove_members" carry changes since the last sync. + action = serializers.ChoiceField( + choices=["members", "add_members", "remove_members"] + ) + parameters = MixpanelParametersSerializer() + + class CohortSyncMembersSerializer(serializers.Serializer[None]): user_ids = serializers.ListField( child=serializers.CharField(validators=[_validate_identifier_byte_length]), diff --git a/api/cohorts/services.py b/api/cohorts/services.py index 4bb1f5ac74b6..972905d3404d 100644 --- a/api/cohorts/services.py +++ b/api/cohorts/services.py @@ -3,7 +3,7 @@ import typing import structlog -from django.db import transaction +from django.db import IntegrityError, transaction from django.db.models import F, QuerySet from django.utils import timezone from flag_engine.segments.constants import IS_SET @@ -116,6 +116,7 @@ def create_cohort( name: str, description: str | None = None, source_type: CohortSourceType = CohortSourceType.CSV, + external_id: str | None = None, ) -> Cohort: project = environment.project # Mirrors the segment limit enforced by SegmentSerializer, which cohort @@ -137,7 +138,10 @@ def create_cohort( ) rule = SegmentRule.objects.create(segment=segment, type=SegmentRule.ALL_RULE) cohort: Cohort = Cohort.objects.create( - environment=environment, segment=segment, source_type=source_type + environment=environment, + segment=segment, + source_type=source_type, + external_id=external_id, ) Condition.objects.create( rule=rule, @@ -161,10 +165,16 @@ def create_cohort_for_source( environment: "Environment", name: str, source_type: CohortSourceType, + external_id: str | None = None, ) -> Cohort: """Create a cohort on behalf of an external source, where no Flagsmith user is acting.""" - cohort = create_cohort(environment=environment, name=name, source_type=source_type) + cohort = create_cohort( + environment=environment, + name=name, + source_type=source_type, + external_id=external_id, + ) # Nothing records a user for these calls, so the audit log that Flagsmith # derives from historical records is skipped — and with it the environment # document rebuild that makes the new segment visible to SDKs. Write the @@ -182,6 +192,53 @@ def create_cohort_for_source( return cohort +def get_cohort_for_source( + *, + environment: "Environment", + source_type: CohortSourceType, + external_id: str, +) -> Cohort | None: + cohort: Cohort | None = Cohort.objects.filter( + environment=environment, + source_type=source_type, + external_id=external_id, + deletion_requested_at__isnull=True, + ).first() + return cohort + + +def get_or_create_cohort_for_source( + *, + environment: "Environment", + name: str, + source_type: CohortSourceType, + external_id: str, +) -> Cohort | None: + if cohort := get_cohort_for_source( + environment=environment, source_type=source_type, external_id=external_id + ): + return cohort + try: + return create_cohort_for_source( + environment=environment, + name=name, + source_type=source_type, + external_id=external_id, + ) + except IntegrityError: + # Two situations end up here. A simultaneous first-sync request + # created the cohort between our lookup and our insert — use the one + # it created. Or the cohort was deleted in Flagsmith and is still + # draining memberships from identity data: the lookup doesn't see it, + # but it still occupies the external ID — nothing usable exists, so + # return None. + if cohort := get_cohort_for_source( + environment=environment, source_type=source_type, external_id=external_id + ): + return cohort + return None + + def add_cohort_members(cohort: Cohort, identifiers: "typing.Iterable[str]") -> None: from cohorts.tasks import apply_cohort_membership_deltas diff --git a/api/cohorts/sync_urls.py b/api/cohorts/sync_urls.py index 6afc517a0b6c..3b599f178e1f 100644 --- a/api/cohorts/sync_urls.py +++ b/api/cohorts/sync_urls.py @@ -1,6 +1,7 @@ +from django.urls import path from rest_framework.routers import SimpleRouter -from cohorts.sync_views import AmplitudeCohortSyncViewSet +from cohorts.sync_views import AmplitudeCohortSyncViewSet, MixpanelCohortSyncView app_name = "cohort-sync" @@ -8,4 +9,7 @@ router = SimpleRouter() router.register(r"amplitude/lists", AmplitudeCohortSyncViewSet, basename="amplitude") -urlpatterns = router.urls +urlpatterns = [ + path("mixpanel/webhook/", MixpanelCohortSyncView.as_view(), name="mixpanel"), + *router.urls, +] diff --git a/api/cohorts/sync_views.py b/api/cohorts/sync_views.py index 85f9d484c9ad..3c16a29838f9 100644 --- a/api/cohorts/sync_views.py +++ b/api/cohorts/sync_views.py @@ -1,12 +1,15 @@ +import json import typing import uuid as uuid_module +import structlog from drf_spectacular.utils import extend_schema, extend_schema_view, inline_serializer from rest_framework import serializers, viewsets from rest_framework.decorators import action -from rest_framework.exceptions import NotFound +from rest_framework.exceptions import NotFound, ParseError from rest_framework.request import Request from rest_framework.response import Response +from rest_framework.views import APIView from cohorts import services from cohorts.authentication import CohortSyncKeyAuthentication @@ -15,12 +18,20 @@ from cohorts.serializers import ( AmplitudeListSerializer, CohortSyncMembersSerializer, + MixpanelWebhookSerializer, ) _LIST_RESPONSE = inline_serializer( "AmplitudeListResponse", {"list_id": serializers.UUIDField()} ) +_MIXPANEL_RESPONSE = inline_serializer( + "MixpanelWebhookResponse", + {"action": serializers.CharField(), "status": serializers.CharField()}, +) + +logger = structlog.get_logger("cohorts") + @extend_schema_view( create=extend_schema( @@ -83,3 +94,112 @@ def _get_cohort(self, request: Request, pk: str) -> Cohort: if cohort is None: raise NotFound("List not found.") return cohort + + +class MixpanelCohortSyncView(APIView): + """ + The receiving end of Mixpanel's Custom Webhook cohort destination: + https://docs.mixpanel.com/docs/cohort-sync/webhooks + + Mixpanel POSTs every message to this one URL and reads the outcome from + the response body, which must repeat the action alongside a + success/failure status. + """ + + authentication_classes = [CohortSyncKeyAuthentication] + permission_classes = [HasCohortSyncKey] + + @extend_schema( + description=( + "Called by Mixpanel every sync cycle with the cohort's full " + "membership (`members`) or the changes since the last sync " + "(`add_members`/`remove_members`)." + ), + request=MixpanelWebhookSerializer, + responses={200: _MIXPANEL_RESPONSE}, + ) + def post(self, request: Request) -> Response: + serializer = MixpanelWebhookSerializer(data=request.data) + if not serializer.is_valid(): + return self._failure( + request, + message=( + f"Invalid payload: {json.dumps(serializer.errors, default=str)}" + ), + code=400, + ) + + data = serializer.validated_data + webhook_action: str = data["action"] + parameters = data["parameters"] + identifiers = [ + member["mixpanel_distinct_id"] for member in parameters["members"] + ] + environment = typing.cast(CohortSyncKey, request.auth).environment + + if webhook_action == "members": + # A large first sync arrives as several requests, each one page + # of members. Every page only adds; removals can't be detected + # without seeing all pages at once. + cohort_or_none = services.get_or_create_cohort_for_source( + environment=environment, + name=parameters["mixpanel_cohort_name"], + source_type=CohortSourceType.MIXPANEL, + external_id=parameters["mixpanel_cohort_id"], + ) + if cohort_or_none is None: + return self._failure( + request, message="Cohort is being deleted.", code=404 + ) + services.add_cohort_members(cohort_or_none, identifiers) + else: + cohort_or_none = services.get_cohort_for_source( + environment=environment, + source_type=CohortSourceType.MIXPANEL, + external_id=parameters["mixpanel_cohort_id"], + ) + if cohort_or_none is None: + # A 404 makes Mixpanel pause the sync and email the customer, + # which is what should happen when the cohort was deleted in + # Flagsmith but Mixpanel is still syncing it. + return self._failure(request, message="Cohort not found.", code=404) + if webhook_action == "add_members": + services.add_cohort_members(cohort_or_none, identifiers) + else: + services.remove_cohort_members(cohort_or_none, identifiers) + + return Response({"action": webhook_action, "status": "success"}) + + def handle_exception(self, exc: Exception) -> Response: + if isinstance(exc, ParseError): + # A body that isn't valid JSON raises before post() runs, so the + # response is shaped here to keep the envelope Mixpanel expects. + return self._failure(self.request, message="Invalid payload.", code=400) + return super().handle_exception(exc) + + def _failure(self, request: Request, *, message: str, code: int) -> Response: + logger.warning( + "sync_webhook.rejected", + source="mixpanel", + action=self._echo_action(request), + environment__id=typing.cast(CohortSyncKey, request.auth).environment_id, + error__message=message, + error__code=code, + ) + return Response( + { + "action": self._echo_action(request), + "status": "failure", + "error": {"message": message, "code": code}, + }, + status=code, + ) + + def _echo_action(self, request: Request) -> str | None: + # Mixpanel expects the response to name the action it sent, even on + # failure; None when the request was too malformed to carry one. + if isinstance(request.data, dict) and isinstance( + action_value := request.data.get("action"), str + ): + return action_value + return None diff --git a/api/tests/unit/cohorts/conftest.py b/api/tests/unit/cohorts/conftest.py index d013cc4d4751..dabd18addb6e 100644 --- a/api/tests/unit/cohorts/conftest.py +++ b/api/tests/unit/cohorts/conftest.py @@ -65,6 +65,30 @@ def amplitude_cohort( return cohort +@pytest.fixture() +def postgres_cohort_sync_key( + environment: Environment, +) -> typing.Tuple[CohortSyncKey, str]: + return typing.cast( + typing.Tuple[CohortSyncKey, str], + CohortSyncKey.objects.create_key(name="postgres key", environment=environment), + ) + + +@pytest.fixture() +def mixpanel_cohort(environment: Environment) -> Cohort: + segment = Segment.objects.create( + name="mixpanel segment", project=environment.project + ) + cohort: Cohort = Cohort.objects.create( + environment=environment, + segment=segment, + source_type=CohortSourceType.MIXPANEL, + external_id="mp-42", + ) + return cohort + + @pytest.fixture() def edge_cohort( dynamo_enabled_project: Project, diff --git a/api/tests/unit/cohorts/test_services.py b/api/tests/unit/cohorts/test_services.py index 35d8ad1201a6..21141549570c 100644 --- a/api/tests/unit/cohorts/test_services.py +++ b/api/tests/unit/cohorts/test_services.py @@ -1,24 +1,32 @@ import io import pytest +from django.db import IntegrityError +from django.utils import timezone from flag_engine.segments.constants import IS_SET from pytest_mock import MockerFixture from pytest_structlog import StructuredLogCapture from rest_framework.exceptions import ValidationError -from cohorts.models import Cohort, CohortMembership, CohortMembershipState +from cohorts.models import ( + Cohort, + CohortMembership, + CohortMembershipState, + CohortSourceType, +) from cohorts.services import ( _batched, apply_pending_memberships, create_cohort, delete_cohort, extract_identifiers_from_csv, + get_or_create_cohort_for_source, sync_cohort_memberships_from_csv, ) from environments.dynamodb import DynamoIdentityWrapper from environments.identities.models import Identity from environments.models import Environment -from segments.models import SegmentManagedBy, SegmentRule +from segments.models import Segment, SegmentManagedBy, SegmentRule @pytest.mark.parametrize( @@ -542,3 +550,64 @@ def test_sync_cohort_memberships_from_csv__edge_cohort__applies_traits( assert document["system_traits"] == {edge_cohort.system_trait_key: True} membership = CohortMembership.objects.get(cohort=edge_cohort) assert membership.state == CohortMembershipState.APPLIED + + +def test_get_or_create_cohort_for_source__simultaneous_creation__returns_other_requests_cohort( + environment: Environment, + mocker: MockerFixture, +) -> None: + # Given - creating the cohort fails because another request created its + # own cohort between our lookup and our insert + def create_winning_cohort_and_conflict(**kwargs: object) -> Cohort: + segment = Segment.objects.create( + name="Power users", project=environment.project + ) + Cohort.objects.create( + environment=environment, + segment=segment, + source_type=CohortSourceType.MIXPANEL, + external_id="mp-42", + ) + raise IntegrityError("unique_active_cohort_per_source_external_id") + + mocker.patch( + "cohorts.services.create_cohort_for_source", + side_effect=create_winning_cohort_and_conflict, + ) + + # When + cohort = get_or_create_cohort_for_source( + environment=environment, + name="Power users", + source_type=CohortSourceType.MIXPANEL, + external_id="mp-42", + ) + + # Then + assert cohort == Cohort.objects.get(external_id="mp-42") + + +def test_get_or_create_cohort_for_source__deletion_requested_same_external_id__returns_none( + environment: Environment, +) -> None: + # Given - a cohort with the same external ID is awaiting deletion, so it + # is invisible to the lookup but still occupies the external ID + segment = Segment.objects.create(name="Power users", project=environment.project) + Cohort.objects.create( + environment=environment, + segment=segment, + source_type=CohortSourceType.MIXPANEL, + external_id="mp-42", + deletion_requested_at=timezone.now(), + ) + + # When + cohort = get_or_create_cohort_for_source( + environment=environment, + name="Power users", + source_type=CohortSourceType.MIXPANEL, + external_id="mp-42", + ) + + # Then + assert cohort is None diff --git a/api/tests/unit/cohorts/test_sync_views.py b/api/tests/unit/cohorts/test_sync_views.py index a9e654b49dc8..335421526e0f 100644 --- a/api/tests/unit/cohorts/test_sync_views.py +++ b/api/tests/unit/cohorts/test_sync_views.py @@ -1,3 +1,4 @@ +import base64 import typing import pytest @@ -5,6 +6,7 @@ from django.utils import timezone from flag_engine.segments.constants import IS_SET from pytest_django.fixtures import SettingsWrapper +from pytest_structlog import StructuredLogCapture from rest_framework import status from rest_framework.test import APIClient @@ -18,6 +20,7 @@ CohortSyncKey, ) from environments.dynamodb import DynamoIdentityWrapper +from environments.identities.models import Identity from environments.models import Environment from projects.models import Project from segments.models import Segment @@ -468,3 +471,449 @@ def test_amplitude_create_list__segment_limit_reached__returns_400( "The project has reached the maximum allowed segments limit." ] assert not Cohort.objects.exists() + + +def _basic_auth_client(plaintext_key: str) -> APIClient: + credentials = base64.b64encode(f"flagsmith:{plaintext_key}".encode()).decode() + client = APIClient() + client.credentials(HTTP_AUTHORIZATION=f"Basic {credentials}") + return client + + +def test_mixpanel_webhook__members_action_unknown_cohort__creates_cohort_and_memberships( + postgres_cohort_sync_key: _KeyAndPlaintext, +) -> None: + # Given + key, plaintext = postgres_cohort_sync_key + client = _basic_auth_client(plaintext) + url = reverse("api-v1:cohort-sync:mixpanel") + + # When + response = client.post( + url, + data={ + "action": "members", + "parameters": { + "mixpanel_cohort_id": "mp-42", + "mixpanel_cohort_name": "Power users", + "members": [ + {"mixpanel_distinct_id": "user-1"}, + {"mixpanel_distinct_id": "user-2"}, + ], + }, + }, + format="json", + ) + + # Then + assert response.status_code == status.HTTP_200_OK + assert response.json() == {"action": "members", "status": "success"} + cohort = Cohort.objects.get( + environment=key.environment, + source_type=CohortSourceType.MIXPANEL, + external_id="mp-42", + ) + assert cohort.segment.name == "Power users" + assert sorted( + CohortMembership.objects.filter(cohort=cohort).values_list( + "identifier", "state" + ) + ) == [ + ("user-1", CohortMembershipState.APPLIED), + ("user-2", CohortMembershipState.APPLIED), + ] + + +def test_mixpanel_webhook__members_action_existing_cohort__adds_to_it( + postgres_cohort_sync_key: _KeyAndPlaintext, + mixpanel_cohort: Cohort, +) -> None: + # Given + _, plaintext = postgres_cohort_sync_key + client = _basic_auth_client(plaintext) + url = reverse("api-v1:cohort-sync:mixpanel") + + # When + response = client.post( + url, + data={ + "action": "members", + "parameters": { + "mixpanel_cohort_id": "mp-42", + "mixpanel_cohort_name": "Power users", + "members": [{"mixpanel_distinct_id": "user-1"}], + }, + }, + format="json", + ) + + # Then + assert response.status_code == status.HTTP_200_OK + assert Cohort.objects.filter(source_type=CohortSourceType.MIXPANEL).count() == 1 + membership = CohortMembership.objects.get(cohort=mixpanel_cohort) + assert (membership.identifier, membership.state) == ( + "user-1", + CohortMembershipState.APPLIED, + ) + + +def test_mixpanel_webhook__add_members_action__sets_system_trait( + postgres_cohort_sync_key: _KeyAndPlaintext, + mixpanel_cohort: Cohort, +) -> None: + # Given + _, plaintext = postgres_cohort_sync_key + client = _basic_auth_client(plaintext) + url = reverse("api-v1:cohort-sync:mixpanel") + + # When + response = client.post( + url, + data={ + "action": "add_members", + "parameters": { + "mixpanel_cohort_id": "mp-42", + "mixpanel_cohort_name": "Power users", + "members": [{"mixpanel_distinct_id": "user-1"}], + }, + }, + format="json", + ) + + # Then + assert response.status_code == status.HTTP_200_OK + assert response.json() == {"action": "add_members", "status": "success"} + identity = Identity.objects.get( + environment=mixpanel_cohort.environment, identifier="user-1" + ) + assert identity.system_traits == {mixpanel_cohort.system_trait_key: True} + + +def test_mixpanel_webhook__remove_members_action__unsets_system_trait( + postgres_cohort_sync_key: _KeyAndPlaintext, + mixpanel_cohort: Cohort, +) -> None: + # Given + Identity.objects.create( + environment=mixpanel_cohort.environment, + identifier="member", + system_traits={mixpanel_cohort.system_trait_key: True}, + ) + CohortMembership.objects.create( + cohort=mixpanel_cohort, + identifier="member", + state=CohortMembershipState.APPLIED, + ) + _, plaintext = postgres_cohort_sync_key + client = _basic_auth_client(plaintext) + url = reverse("api-v1:cohort-sync:mixpanel") + + # When + response = client.post( + url, + data={ + "action": "remove_members", + "parameters": { + "mixpanel_cohort_id": "mp-42", + "mixpanel_cohort_name": "Power users", + "members": [{"mixpanel_distinct_id": "member"}], + }, + }, + format="json", + ) + + # Then + assert response.status_code == status.HTTP_200_OK + assert response.json() == {"action": "remove_members", "status": "success"} + assert not CohortMembership.objects.filter(cohort=mixpanel_cohort).exists() + identity = Identity.objects.get( + environment=mixpanel_cohort.environment, identifier="member" + ) + assert identity.system_traits == {} + + +def test_mixpanel_webhook__add_members_unknown_cohort__returns_404_failure( + postgres_cohort_sync_key: _KeyAndPlaintext, + log: StructuredLogCapture, +) -> None: + # Given + key, plaintext = postgres_cohort_sync_key + client = _basic_auth_client(plaintext) + url = reverse("api-v1:cohort-sync:mixpanel") + + # When + response = client.post( + url, + data={ + "action": "add_members", + "parameters": { + "mixpanel_cohort_id": "unknown", + "mixpanel_cohort_name": "Power users", + "members": [{"mixpanel_distinct_id": "user-1"}], + }, + }, + format="json", + ) + + # Then + assert response.status_code == status.HTTP_404_NOT_FOUND + assert response.json() == { + "action": "add_members", + "status": "failure", + "error": {"message": "Cohort not found.", "code": 404}, + } + assert log.events == [ + { + "level": "warning", + "event": "sync_webhook.rejected", + "source": "mixpanel", + "action": "add_members", + "environment__id": key.environment_id, + "error__message": "Cohort not found.", + "error__code": 404, + } + ] + + +def test_mixpanel_webhook__deletion_requested_cohort__returns_404_failure( + postgres_cohort_sync_key: _KeyAndPlaintext, + mixpanel_cohort: Cohort, +) -> None: + # Given + mixpanel_cohort.deletion_requested_at = timezone.now() + mixpanel_cohort.save() + _, plaintext = postgres_cohort_sync_key + client = _basic_auth_client(plaintext) + url = reverse("api-v1:cohort-sync:mixpanel") + + # When + response = client.post( + url, + data={ + "action": "remove_members", + "parameters": { + "mixpanel_cohort_id": "mp-42", + "mixpanel_cohort_name": "Power users", + "members": [{"mixpanel_distinct_id": "user-1"}], + }, + }, + format="json", + ) + + # Then + assert response.status_code == status.HTTP_404_NOT_FOUND + assert response.json()["status"] == "failure" + + +def test_mixpanel_webhook__missing_parameters__returns_400_failure( + postgres_cohort_sync_key: _KeyAndPlaintext, +) -> None: + # Given + _, plaintext = postgres_cohort_sync_key + client = _basic_auth_client(plaintext) + url = reverse("api-v1:cohort-sync:mixpanel") + + # When + response = client.post(url, data={"action": "members"}, format="json") + + # Then + assert response.status_code == status.HTTP_400_BAD_REQUEST + body = response.json() + assert body["action"] == "members" + assert body["status"] == "failure" + assert body["error"]["code"] == 400 + assert "parameters" in body["error"]["message"] + + +def test_mixpanel_webhook__non_object_payload__returns_400_failure( + postgres_cohort_sync_key: _KeyAndPlaintext, +) -> None: + # Given + _, plaintext = postgres_cohort_sync_key + client = _basic_auth_client(plaintext) + url = reverse("api-v1:cohort-sync:mixpanel") + + # When + response = client.post(url, data=["not", "an", "object"], format="json") + + # Then + assert response.status_code == status.HTTP_400_BAD_REQUEST + body = response.json() + assert body["action"] is None + assert body["status"] == "failure" + assert body["error"]["code"] == 400 + + +def test_mixpanel_webhook__unparseable_body__returns_400_failure( + postgres_cohort_sync_key: _KeyAndPlaintext, +) -> None: + # Given + _, plaintext = postgres_cohort_sync_key + client = _basic_auth_client(plaintext) + url = reverse("api-v1:cohort-sync:mixpanel") + + # When + response = client.post(url, data="{not json", content_type="application/json") + + # Then + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert response.json() == { + "action": None, + "status": "failure", + "error": {"message": "Invalid payload.", "code": 400}, + } + + +def test_mixpanel_webhook__members_action_while_cohort_draining__returns_404_failure( + postgres_cohort_sync_key: _KeyAndPlaintext, + mixpanel_cohort: Cohort, +) -> None: + # Given - the cohort was deleted in Flagsmith and is still draining + mixpanel_cohort.deletion_requested_at = timezone.now() + mixpanel_cohort.save() + _, plaintext = postgres_cohort_sync_key + client = _basic_auth_client(plaintext) + url = reverse("api-v1:cohort-sync:mixpanel") + + # When + response = client.post( + url, + data={ + "action": "members", + "parameters": { + "mixpanel_cohort_id": "mp-42", + "mixpanel_cohort_name": "Power users", + "members": [{"mixpanel_distinct_id": "user-1"}], + }, + }, + format="json", + ) + + # Then + assert response.status_code == status.HTTP_404_NOT_FOUND + assert response.json() == { + "action": "members", + "status": "failure", + "error": {"message": "Cohort is being deleted.", "code": 404}, + } + assert Cohort.objects.filter(source_type=CohortSourceType.MIXPANEL).count() == 1 + + +def test_mixpanel_webhook__other_environment_key__returns_404_failure( + mixpanel_cohort: Cohort, +) -> None: + # Given - a key scoped to a different environment than the cohort's + other_environment = Environment.objects.create( + name="Other environment", project=mixpanel_cohort.environment.project + ) + _, plaintext = CohortSyncKey.objects.create_key( + name="other key", environment=other_environment + ) + client = _basic_auth_client(plaintext) + url = reverse("api-v1:cohort-sync:mixpanel") + + # When + response = client.post( + url, + data={ + "action": "add_members", + "parameters": { + "mixpanel_cohort_id": "mp-42", + "mixpanel_cohort_name": "Power users", + "members": [{"mixpanel_distinct_id": "user-1"}], + }, + }, + format="json", + ) + + # Then + assert response.status_code == status.HTTP_404_NOT_FOUND + assert not CohortMembership.objects.filter(cohort=mixpanel_cohort).exists() + + +def test_mixpanel_webhook__empty_members_page__returns_success( + postgres_cohort_sync_key: _KeyAndPlaintext, +) -> None: + # Given + key, plaintext = postgres_cohort_sync_key + client = _basic_auth_client(plaintext) + url = reverse("api-v1:cohort-sync:mixpanel") + + # When + response = client.post( + url, + data={ + "action": "members", + "parameters": { + "mixpanel_cohort_id": "mp-42", + "mixpanel_cohort_name": "Empty cohort", + "members": [], + }, + }, + format="json", + ) + + # Then + assert response.status_code == status.HTTP_200_OK + cohort = Cohort.objects.get(environment=key.environment, external_id="mp-42") + assert not CohortMembership.objects.filter(cohort=cohort).exists() + + +def test_mixpanel_webhook__non_ascii_basic_credentials__returns_401( + db: None, +) -> None: + # Given - a header byte outside ASCII, which base64 decoding rejects + client = APIClient() + client.credentials(HTTP_AUTHORIZATION="Basic dXNlcjprÿZXk=") + url = reverse("api-v1:cohort-sync:mixpanel") + + # When + response = client.post(url, data={"action": "members"}, format="json") + + # Then + assert response.status_code == status.HTTP_401_UNAUTHORIZED + + +def test_mixpanel_webhook__nul_byte_in_basic_password__returns_401( + db: None, +) -> None: + # Given - valid base64 whose decoded password contains a NUL character + credentials = base64.b64encode(b"user:\x00key").decode() + client = APIClient() + client.credentials(HTTP_AUTHORIZATION=f"Basic {credentials}") + url = reverse("api-v1:cohort-sync:mixpanel") + + # When + response = client.post(url, data={"action": "members"}, format="json") + + # Then + assert response.status_code == status.HTTP_401_UNAUTHORIZED + + +def test_mixpanel_webhook__malformed_basic_credentials__returns_401( + db: None, +) -> None: + # Given + client = APIClient() + client.credentials(HTTP_AUTHORIZATION="Basic not-base64!!") + url = reverse("api-v1:cohort-sync:mixpanel") + + # When + response = client.post(url, data={"action": "members"}, format="json") + + # Then + assert response.status_code == status.HTTP_401_UNAUTHORIZED + + +def test_mixpanel_webhook__unknown_key_in_basic_password__returns_401( + db: None, +) -> None: + # Given + client = _basic_auth_client("not-a-key") + url = reverse("api-v1:cohort-sync:mixpanel") + + # When + response = client.post(url, data={"action": "members"}, format="json") + + # Then + assert response.status_code == status.HTTP_401_UNAUTHORIZED diff --git a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md index 8c508716727e..2dc0ca9307c5 100644 --- a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md +++ b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md @@ -74,7 +74,7 @@ Attributes: ### `cohorts.cohort.created` Logged at `info` from: - - `api/cohorts/services.py:148` + - `api/cohorts/services.py:115` Attributes: - `cohort.id` @@ -86,7 +86,7 @@ Attributes: ### `cohorts.cohort.deleted` Logged at `info` from: - - `api/cohorts/services.py:396` + - `api/cohorts/services.py:276` Attributes: - `cohort.id` @@ -95,39 +95,16 @@ Attributes: ### `cohorts.cohort.deletion_requested` Logged at `info` from: - - `api/cohorts/services.py:380` + - `api/cohorts/services.py:260` Attributes: - `cohort.id` - `environment.id` -### `cohorts.csv.synced` - -Logged at `info` from: - - `api/cohorts/services.py:352` - -Attributes: - - `adds.count` - - `cohort.id` - - `cohort.version` - - `environment.id` - - `removes.count` - - `unchanged.count` - -### `cohorts.membership.adds_received` - -Logged at `info` from: - - `api/cohorts/services.py:205` - -Attributes: - - `cohort.id` - - `deltas.count` - - `environment.id` - ### `cohorts.membership.applied` Logged at `info` from: - - `api/cohorts/services.py:103` + - `api/cohorts/services.py:77` Attributes: - `adds.count` @@ -152,17 +129,31 @@ Logged at `warning` from: Attributes: - `cohort.id` -### `cohorts.membership.removals_received` +### `cohorts.membership.deltas_received` Logged at `info` from: - - `api/cohorts/services.py:230` + - `api/cohorts/services.py:225` + - `api/cohorts/services.py:244` Attributes: + - `action` - `cohort.id` - `deltas.count` - `environment.id` - `members.matched` +### `cohorts.sync_webhook.rejected` + +Logged at `warning` from: + - `api/cohorts/sync_views.py:181` + +Attributes: + - `action` + - `environment.id` + - `error.code` + - `error.message` + - `source` + ### `core.encrypted_field.decrypt_failed` Logged at `warning` from: @@ -667,7 +658,7 @@ Attributes: ### `segments.serializers.segment_revision_created` Logged at `info` from: - - `api/segments/serializers.py:215` + - `api/segments/serializers.py:185` Attributes: - `revision_id` diff --git a/sdk/openapi.yaml b/sdk/openapi.yaml index dc12e071dac2..eb7b6086cef8 100644 --- a/sdk/openapi.yaml +++ b/sdk/openapi.yaml @@ -564,7 +564,7 @@ components: Cohort Sync Key: type: http scheme: bearer - description: 'For cohort sync endpoints called by an external cohort source, such as Amplitude.' + description: 'For cohort sync endpoints called by an external cohort source, such as Amplitude. Sources that can only send Basic credentials (e.g. Mixpanel) pass the key as the password, with any username.' Environment API Key: type: apiKey in: header From 7dd9c572e76e44064be526ddd639eb21b9213244 Mon Sep 17 00:00:00 2001 From: Gagan Trivedi Date: Fri, 21 Aug 2026 16:45:58 +0530 Subject: [PATCH 02/10] fix(cohorts): harden sync key auth and drop the external ID constraint --- api/cohorts/authentication.py | 11 +--- .../migrations/0004_mixpanel_source.py | 10 --- api/cohorts/models.py | 32 +++++---- api/cohorts/services.py | 34 +--------- api/cohorts/sync_views.py | 12 ++-- api/tests/unit/cohorts/test_services.py | 65 +------------------ api/tests/unit/cohorts/test_sync_views.py | 35 ---------- .../observability/_events-catalogue.md | 8 +-- 8 files changed, 32 insertions(+), 175 deletions(-) diff --git a/api/cohorts/authentication.py b/api/cohorts/authentication.py index fc608c5c8ec5..6ef0b9c78569 100644 --- a/api/cohorts/authentication.py +++ b/api/cohorts/authentication.py @@ -1,5 +1,4 @@ import base64 -import typing from contextlib import suppress from django.contrib.auth.models import AnonymousUser @@ -38,16 +37,8 @@ def authenticate( else: return None - if "\x00" in raw_key: - # Postgres refuses to run a query containing a NUL character, so - # the key lookup below would crash instead of returning 401. - raise exceptions.AuthenticationFailed("Valid cohort sync key not found.") - with suppress(CohortSyncKey.DoesNotExist): - key = typing.cast( - CohortSyncKey, - CohortSyncKey.objects.get_from_key(raw_key), - ) + key = CohortSyncKey.objects.get_from_key(raw_key) if not key.has_expired: # No person is acting here, so no user is returned: the key # alone carries authority, and audit trails record the source diff --git a/api/cohorts/migrations/0004_mixpanel_source.py b/api/cohorts/migrations/0004_mixpanel_source.py index 8d2368c4d6ab..1cdb5e5260e4 100644 --- a/api/cohorts/migrations/0004_mixpanel_source.py +++ b/api/cohorts/migrations/0004_mixpanel_source.py @@ -30,14 +30,4 @@ class Migration(migrations.Migration): max_length=50, ), ), - migrations.AddConstraint( - model_name="cohort", - constraint=models.UniqueConstraint( - condition=models.Q( - ("deleted_at__isnull", True), ("external_id__isnull", False) - ), - fields=("environment", "source_type", "external_id"), - name="unique_active_cohort_per_source_external_id", - ), - ), ] diff --git a/api/cohorts/models.py b/api/cohorts/models.py index f59e55489b29..1efd43e478ff 100644 --- a/api/cohorts/models.py +++ b/api/cohorts/models.py @@ -1,5 +1,7 @@ +import typing + from django.db import models -from rest_framework_api_key.models import AbstractAPIKey +from rest_framework_api_key.models import AbstractAPIKey, APIKeyManager from cohorts.constants import COHORT_SYSTEM_TRAIT_KEY_PREFIX from core.models import SoftDeleteExportableModel @@ -27,9 +29,10 @@ class Cohort(SoftDeleteExportableModel): choices=CohortSourceType.choices, default=CohortSourceType.CSV, ) - # The cohort's identifier in the external source (e.g. Mixpanel's cohort - # ID). Set for sources that push to us under their own identifier; null - # for sources that adopt ours (Amplitude) and for CSV cohorts. + # The cohort's identifier in the external source. Mixpanel pushes under + # its own cohort ID, so we store it to route later requests; Amplitude + # uses the ID we hand back at list creation, and CSV cohorts have no + # external system, so both leave this null. external_id = models.CharField(max_length=255, null=True, blank=True) version = models.PositiveIntegerField(default=0) created_at = models.DateTimeField(auto_now_add=True) @@ -50,19 +53,22 @@ class Meta: condition=models.Q(deleted_at__isnull=True), name="unique_active_cohort_per_segment", ), - # One Mixpanel cohort must map to one active cohort per - # environment: without this, two simultaneous first-sync requests - # would each create their own cohort and split the members - # between them. - models.UniqueConstraint( - fields=["environment", "source_type", "external_id"], - condition=models.Q(deleted_at__isnull=True, external_id__isnull=False), - name="unique_active_cohort_per_source_external_id", - ), ] +class CohortSyncKeyManager(APIKeyManager): + def get_from_key(self, key: str) -> "CohortSyncKey": + if "\x00" in key: + # A NUL can't travel in a raw header, but base64 credentials can + # decode to one, and the database driver refuses to build a query + # containing it. No real key holds one, so treat it as absent. + raise self.model.DoesNotExist("Key contains a NUL character.") + return typing.cast("CohortSyncKey", super().get_from_key(key)) + + class CohortSyncKey(AbstractAPIKey): + objects: typing.ClassVar[CohortSyncKeyManager] = CohortSyncKeyManager() + environment = models.ForeignKey( "environments.Environment", on_delete=models.CASCADE, diff --git a/api/cohorts/services.py b/api/cohorts/services.py index 972905d3404d..10f6bb03fe43 100644 --- a/api/cohorts/services.py +++ b/api/cohorts/services.py @@ -3,7 +3,7 @@ import typing import structlog -from django.db import IntegrityError, transaction +from django.db import transaction from django.db.models import F, QuerySet from django.utils import timezone from flag_engine.segments.constants import IS_SET @@ -207,38 +207,6 @@ def get_cohort_for_source( return cohort -def get_or_create_cohort_for_source( - *, - environment: "Environment", - name: str, - source_type: CohortSourceType, - external_id: str, -) -> Cohort | None: - if cohort := get_cohort_for_source( - environment=environment, source_type=source_type, external_id=external_id - ): - return cohort - try: - return create_cohort_for_source( - environment=environment, - name=name, - source_type=source_type, - external_id=external_id, - ) - except IntegrityError: - # Two situations end up here. A simultaneous first-sync request - # created the cohort between our lookup and our insert — use the one - # it created. Or the cohort was deleted in Flagsmith and is still - # draining memberships from identity data: the lookup doesn't see it, - # but it still occupies the external ID — nothing usable exists, so - # return None. - if cohort := get_cohort_for_source( - environment=environment, source_type=source_type, external_id=external_id - ): - return cohort - return None - - def add_cohort_members(cohort: Cohort, identifiers: "typing.Iterable[str]") -> None: from cohorts.tasks import apply_cohort_membership_deltas diff --git a/api/cohorts/sync_views.py b/api/cohorts/sync_views.py index 3c16a29838f9..42a3fa714fc1 100644 --- a/api/cohorts/sync_views.py +++ b/api/cohorts/sync_views.py @@ -141,17 +141,17 @@ def post(self, request: Request) -> Response: # A large first sync arrives as several requests, each one page # of members. Every page only adds; removals can't be detected # without seeing all pages at once. - cohort_or_none = services.get_or_create_cohort_for_source( + cohort = services.get_cohort_for_source( + environment=environment, + source_type=CohortSourceType.MIXPANEL, + external_id=parameters["mixpanel_cohort_id"], + ) or services.create_cohort_for_source( environment=environment, name=parameters["mixpanel_cohort_name"], source_type=CohortSourceType.MIXPANEL, external_id=parameters["mixpanel_cohort_id"], ) - if cohort_or_none is None: - return self._failure( - request, message="Cohort is being deleted.", code=404 - ) - services.add_cohort_members(cohort_or_none, identifiers) + services.add_cohort_members(cohort, identifiers) else: cohort_or_none = services.get_cohort_for_source( environment=environment, diff --git a/api/tests/unit/cohorts/test_services.py b/api/tests/unit/cohorts/test_services.py index 21141549570c..8fb2f6d9e0c5 100644 --- a/api/tests/unit/cohorts/test_services.py +++ b/api/tests/unit/cohorts/test_services.py @@ -1,7 +1,6 @@ import io import pytest -from django.db import IntegrityError from django.utils import timezone from flag_engine.segments.constants import IS_SET from pytest_mock import MockerFixture @@ -12,7 +11,6 @@ Cohort, CohortMembership, CohortMembershipState, - CohortSourceType, ) from cohorts.services import ( _batched, @@ -20,13 +18,12 @@ create_cohort, delete_cohort, extract_identifiers_from_csv, - get_or_create_cohort_for_source, sync_cohort_memberships_from_csv, ) from environments.dynamodb import DynamoIdentityWrapper from environments.identities.models import Identity from environments.models import Environment -from segments.models import Segment, SegmentManagedBy, SegmentRule +from segments.models import SegmentManagedBy, SegmentRule @pytest.mark.parametrize( @@ -551,63 +548,3 @@ def test_sync_cohort_memberships_from_csv__edge_cohort__applies_traits( membership = CohortMembership.objects.get(cohort=edge_cohort) assert membership.state == CohortMembershipState.APPLIED - -def test_get_or_create_cohort_for_source__simultaneous_creation__returns_other_requests_cohort( - environment: Environment, - mocker: MockerFixture, -) -> None: - # Given - creating the cohort fails because another request created its - # own cohort between our lookup and our insert - def create_winning_cohort_and_conflict(**kwargs: object) -> Cohort: - segment = Segment.objects.create( - name="Power users", project=environment.project - ) - Cohort.objects.create( - environment=environment, - segment=segment, - source_type=CohortSourceType.MIXPANEL, - external_id="mp-42", - ) - raise IntegrityError("unique_active_cohort_per_source_external_id") - - mocker.patch( - "cohorts.services.create_cohort_for_source", - side_effect=create_winning_cohort_and_conflict, - ) - - # When - cohort = get_or_create_cohort_for_source( - environment=environment, - name="Power users", - source_type=CohortSourceType.MIXPANEL, - external_id="mp-42", - ) - - # Then - assert cohort == Cohort.objects.get(external_id="mp-42") - - -def test_get_or_create_cohort_for_source__deletion_requested_same_external_id__returns_none( - environment: Environment, -) -> None: - # Given - a cohort with the same external ID is awaiting deletion, so it - # is invisible to the lookup but still occupies the external ID - segment = Segment.objects.create(name="Power users", project=environment.project) - Cohort.objects.create( - environment=environment, - segment=segment, - source_type=CohortSourceType.MIXPANEL, - external_id="mp-42", - deletion_requested_at=timezone.now(), - ) - - # When - cohort = get_or_create_cohort_for_source( - environment=environment, - name="Power users", - source_type=CohortSourceType.MIXPANEL, - external_id="mp-42", - ) - - # Then - assert cohort is None diff --git a/api/tests/unit/cohorts/test_sync_views.py b/api/tests/unit/cohorts/test_sync_views.py index 335421526e0f..bb9235f60afb 100644 --- a/api/tests/unit/cohorts/test_sync_views.py +++ b/api/tests/unit/cohorts/test_sync_views.py @@ -764,41 +764,6 @@ def test_mixpanel_webhook__unparseable_body__returns_400_failure( } -def test_mixpanel_webhook__members_action_while_cohort_draining__returns_404_failure( - postgres_cohort_sync_key: _KeyAndPlaintext, - mixpanel_cohort: Cohort, -) -> None: - # Given - the cohort was deleted in Flagsmith and is still draining - mixpanel_cohort.deletion_requested_at = timezone.now() - mixpanel_cohort.save() - _, plaintext = postgres_cohort_sync_key - client = _basic_auth_client(plaintext) - url = reverse("api-v1:cohort-sync:mixpanel") - - # When - response = client.post( - url, - data={ - "action": "members", - "parameters": { - "mixpanel_cohort_id": "mp-42", - "mixpanel_cohort_name": "Power users", - "members": [{"mixpanel_distinct_id": "user-1"}], - }, - }, - format="json", - ) - - # Then - assert response.status_code == status.HTTP_404_NOT_FOUND - assert response.json() == { - "action": "members", - "status": "failure", - "error": {"message": "Cohort is being deleted.", "code": 404}, - } - assert Cohort.objects.filter(source_type=CohortSourceType.MIXPANEL).count() == 1 - - def test_mixpanel_webhook__other_environment_key__returns_404_failure( mixpanel_cohort: Cohort, ) -> None: diff --git a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md index 2dc0ca9307c5..2cf596d34a7d 100644 --- a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md +++ b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md @@ -86,7 +86,7 @@ Attributes: ### `cohorts.cohort.deleted` Logged at `info` from: - - `api/cohorts/services.py:276` + - `api/cohorts/services.py:244` Attributes: - `cohort.id` @@ -95,7 +95,7 @@ Attributes: ### `cohorts.cohort.deletion_requested` Logged at `info` from: - - `api/cohorts/services.py:260` + - `api/cohorts/services.py:228` Attributes: - `cohort.id` @@ -132,8 +132,8 @@ Attributes: ### `cohorts.membership.deltas_received` Logged at `info` from: - - `api/cohorts/services.py:225` - - `api/cohorts/services.py:244` + - `api/cohorts/services.py:193` + - `api/cohorts/services.py:212` Attributes: - `action` From f25818fe38646b15a3dadc3626952af93a2a9c11 Mon Sep 17 00:00:00 2001 From: Gagan Trivedi Date: Mon, 24 Aug 2026 09:55:34 +0530 Subject: [PATCH 03/10] docs(cohorts): document Basic auth and failure responses for Mixpanel sync --- api/api/openapi.py | 41 +++++++++++++------ api/cohorts/sync_views.py | 21 +++++++++- .../observability/_events-catalogue.md | 2 +- sdk/openapi.yaml | 6 ++- 4 files changed, 55 insertions(+), 15 deletions(-) diff --git a/api/api/openapi.py b/api/api/openapi.py index 1c6b9a7e3750..fb3d5d91f56a 100644 --- a/api/api/openapi.py +++ b/api/api/openapi.py @@ -165,21 +165,38 @@ def get_security_definition( class CohortSyncKeyAuthenticationExtension(OpenApiAuthenticationExtension): # type: ignore[no-untyped-call] target_class = "cohorts.authentication.CohortSyncKeyAuthentication" - name = "Cohort Sync Key" + name = ["Cohort Sync Key", "Cohort Sync Key (Basic)"] + + def get_security_requirement( + self, auto_schema: openapi.AutoSchema + ) -> list[dict[str, list[Any]]]: + # Separate entries: the caller sends the key with either scheme, + # not both at once. + return [{name: []} for name in self.name] def get_security_definition( self, auto_schema: openapi.AutoSchema | None = None - ) -> dict[str, Any]: - return { - "type": "http", - "scheme": "bearer", - "description": ( - "For cohort sync endpoints called by an external cohort " - "source, such as Amplitude. Sources that can only send " - "Basic credentials (e.g. Mixpanel) pass the key as the " - "password, with any username." - ), - } + ) -> list[dict[str, Any]]: + return [ + { + "type": "http", + "scheme": "bearer", + "description": ( + "For cohort sync endpoints called by an external cohort " + "source, such as Amplitude." + ), + }, + { + "type": "http", + "scheme": "basic", + "description": ( + "For cohort sync endpoints called by an external cohort " + "source that can only send Basic credentials, such as " + "Mixpanel. The key is the password; the username is " + "ignored." + ), + }, + ] # Tag definitions controlling the order and display of sections in the Swagger UI. diff --git a/api/cohorts/sync_views.py b/api/cohorts/sync_views.py index 42a3fa714fc1..6bfdace7d14d 100644 --- a/api/cohorts/sync_views.py +++ b/api/cohorts/sync_views.py @@ -30,6 +30,21 @@ {"action": serializers.CharField(), "status": serializers.CharField()}, ) +_MIXPANEL_FAILURE_RESPONSE = inline_serializer( + "MixpanelWebhookFailureResponse", + { + "action": serializers.CharField(allow_null=True), + "status": serializers.CharField(), + "error": inline_serializer( + "MixpanelWebhookError", + { + "message": serializers.CharField(), + "code": serializers.IntegerField(), + }, + ), + }, +) + logger = structlog.get_logger("cohorts") @@ -116,7 +131,11 @@ class MixpanelCohortSyncView(APIView): "(`add_members`/`remove_members`)." ), request=MixpanelWebhookSerializer, - responses={200: _MIXPANEL_RESPONSE}, + responses={ + 200: _MIXPANEL_RESPONSE, + 400: _MIXPANEL_FAILURE_RESPONSE, + 404: _MIXPANEL_FAILURE_RESPONSE, + }, ) def post(self, request: Request) -> Response: serializer = MixpanelWebhookSerializer(data=request.data) diff --git a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md index 2cf596d34a7d..398ccc3c5874 100644 --- a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md +++ b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md @@ -145,7 +145,7 @@ Attributes: ### `cohorts.sync_webhook.rejected` Logged at `warning` from: - - `api/cohorts/sync_views.py:181` + - `api/cohorts/sync_views.py:200` Attributes: - `action` diff --git a/sdk/openapi.yaml b/sdk/openapi.yaml index eb7b6086cef8..83c4995ef579 100644 --- a/sdk/openapi.yaml +++ b/sdk/openapi.yaml @@ -564,7 +564,11 @@ components: Cohort Sync Key: type: http scheme: bearer - description: 'For cohort sync endpoints called by an external cohort source, such as Amplitude. Sources that can only send Basic credentials (e.g. Mixpanel) pass the key as the password, with any username.' + description: 'For cohort sync endpoints called by an external cohort source, such as Amplitude.' + Cohort Sync Key (Basic): + type: http + scheme: basic + description: 'For cohort sync endpoints called by an external cohort source that can only send Basic credentials, such as Mixpanel. The key is the password; the username is ignored.' Environment API Key: type: apiKey in: header From a34a1ceea13580f1a8ed4df390dd57d16c52a199 Mon Sep 17 00:00:00 2001 From: Gagan Trivedi Date: Tue, 25 Aug 2026 17:39:47 +0530 Subject: [PATCH 04/10] chore: lint fixes and documentation artefacts after rebase --- api/tests/unit/cohorts/test_services.py | 2 - .../observability/_events-catalogue.md | 39 ++++++++++++++----- 2 files changed, 30 insertions(+), 11 deletions(-) diff --git a/api/tests/unit/cohorts/test_services.py b/api/tests/unit/cohorts/test_services.py index 8fb2f6d9e0c5..d8e7b43117e0 100644 --- a/api/tests/unit/cohorts/test_services.py +++ b/api/tests/unit/cohorts/test_services.py @@ -1,7 +1,6 @@ import io import pytest -from django.utils import timezone from flag_engine.segments.constants import IS_SET from pytest_mock import MockerFixture from pytest_structlog import StructuredLogCapture @@ -547,4 +546,3 @@ def test_sync_cohort_memberships_from_csv__edge_cohort__applies_traits( assert document["system_traits"] == {edge_cohort.system_trait_key: True} membership = CohortMembership.objects.get(cohort=edge_cohort) assert membership.state == CohortMembershipState.APPLIED - diff --git a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md index 398ccc3c5874..9f0388a069b3 100644 --- a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md +++ b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md @@ -74,7 +74,7 @@ Attributes: ### `cohorts.cohort.created` Logged at `info` from: - - `api/cohorts/services.py:115` + - `api/cohorts/services.py:152` Attributes: - `cohort.id` @@ -86,7 +86,7 @@ Attributes: ### `cohorts.cohort.deleted` Logged at `info` from: - - `api/cohorts/services.py:244` + - `api/cohorts/services.py:421` Attributes: - `cohort.id` @@ -95,16 +95,39 @@ Attributes: ### `cohorts.cohort.deletion_requested` Logged at `info` from: - - `api/cohorts/services.py:228` + - `api/cohorts/services.py:405` Attributes: - `cohort.id` - `environment.id` +### `cohorts.csv.synced` + +Logged at `info` from: + - `api/cohorts/services.py:377` + +Attributes: + - `adds.count` + - `cohort.id` + - `cohort.version` + - `environment.id` + - `removes.count` + - `unchanged.count` + +### `cohorts.membership.adds_received` + +Logged at `info` from: + - `api/cohorts/services.py:230` + +Attributes: + - `cohort.id` + - `deltas.count` + - `environment.id` + ### `cohorts.membership.applied` Logged at `info` from: - - `api/cohorts/services.py:77` + - `api/cohorts/services.py:103` Attributes: - `adds.count` @@ -129,14 +152,12 @@ Logged at `warning` from: Attributes: - `cohort.id` -### `cohorts.membership.deltas_received` +### `cohorts.membership.removals_received` Logged at `info` from: - - `api/cohorts/services.py:193` - - `api/cohorts/services.py:212` + - `api/cohorts/services.py:255` Attributes: - - `action` - `cohort.id` - `deltas.count` - `environment.id` @@ -658,7 +679,7 @@ Attributes: ### `segments.serializers.segment_revision_created` Logged at `info` from: - - `api/segments/serializers.py:185` + - `api/segments/serializers.py:215` Attributes: - `revision_id` From b7a2a700f85b3a7b2358eb276c41893d3f9e33b5 Mon Sep 17 00:00:00 2001 From: "flagsmith-engineering[bot]" Date: Tue, 25 Aug 2026 12:11:38 +0000 Subject: [PATCH 05/10] chore: Update documentation artefacts --- mcp/src/flagsmith_mcp/openapi.json | 10 ++- openapi.yaml | 129 +++++++++++++++++++++++++++++ 2 files changed, 137 insertions(+), 2 deletions(-) diff --git a/mcp/src/flagsmith_mcp/openapi.json b/mcp/src/flagsmith_mcp/openapi.json index c2a11feeda43..f5e1bb24b355 100644 --- a/mcp/src/flagsmith_mcp/openapi.json +++ b/mcp/src/flagsmith_mcp/openapi.json @@ -7003,11 +7003,12 @@ ] }, "SourceTypeEnum": { - "description": "* `csv` - CSV\n* `amplitude` - Amplitude", + "description": "* `csv` - CSV\n* `amplitude` - Amplitude\n* `mixpanel` - Mixpanel", "type": "string", "enum": [ "csv", - "amplitude" + "amplitude", + "mixpanel" ] }, "StageAction": { @@ -7818,6 +7819,11 @@ "scheme": "bearer", "description": "For cohort sync endpoints called by an external cohort source, such as Amplitude." }, + "Cohort Sync Key (Basic)": { + "type": "http", + "scheme": "basic", + "description": "For cohort sync endpoints called by an external cohort source that can only send Basic credentials, such as Mixpanel. The key is the password; the username is ignored." + }, "Environment API Key": { "type": "apiKey", "in": "header", diff --git a/openapi.yaml b/openapi.yaml index 1620cdd9ac9c..8b8679ac1151 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -1683,6 +1683,7 @@ paths: $ref: '#/components/schemas/AmplitudeListResponse' security: - Cohort Sync Key: [] + - Cohort Sync Key (Basic): [] tags: - Other x-flagsmith-minimum-plan: START_UP @@ -1712,6 +1713,7 @@ paths: description: No response body security: - Cohort Sync Key: [] + - Cohort Sync Key (Basic): [] tags: - Other x-flagsmith-minimum-plan: START_UP @@ -1741,9 +1743,50 @@ paths: description: No response body security: - Cohort Sync Key: [] + - Cohort Sync Key (Basic): [] tags: - Other x-flagsmith-minimum-plan: START_UP + /api/v1/cohort-sync/mixpanel/webhook/: + post: + operationId: api_v1_cohort_sync_mixpanel_webhook_create + description: Called by Mixpanel every sync cycle with the cohort's full membership (`members`) or the changes since the last sync (`add_members`/`remove_members`). + requestBody: + required: true + content: + application/json: + schema: + $ref: '#/components/schemas/MixpanelWebhook' + application/x-www-form-urlencoded: + schema: + $ref: '#/components/schemas/MixpanelWebhook' + multipart/form-data: + schema: + $ref: '#/components/schemas/MixpanelWebhook' + responses: + '200': + description: '' + content: + application/json: + schema: + $ref: '#/components/schemas/MixpanelWebhookResponse' + '400': + description: '' + content: + application/json: + schema: + $ref: '#/components/schemas/MixpanelWebhookFailureResponse' + '404': + description: '' + content: + application/json: + schema: + $ref: '#/components/schemas/MixpanelWebhookFailureResponse' + security: + - Cohort Sync Key: [] + - Cohort Sync Key (Basic): [] + tags: + - Webhooks /api/v1/environment-document/: get: operationId: sdk_v1_environment_document @@ -18726,6 +18769,16 @@ paths: - processor components: schemas: + ActionEnum: + description: |- + * `members` - members + * `add_members` - add_members + * `remove_members` - remove_members + type: string + enum: + - members + - add_members + - remove_members ActionTypeEnum: description: |- * `TOGGLE_FEATURE` - Enable/Disable Feature for the environment @@ -22825,6 +22878,76 @@ components: maxLength: 200 required: - api_key + MixpanelMember: + type: object + properties: + mixpanel_distinct_id: + type: string + maxLength: 2000 + required: + - mixpanel_distinct_id + MixpanelParameters: + type: object + properties: + mixpanel_cohort_id: + type: string + maxLength: 255 + mixpanel_cohort_name: + type: string + maxLength: 2000 + members: + type: array + items: + $ref: '#/components/schemas/MixpanelMember' + required: + - members + - mixpanel_cohort_id + - mixpanel_cohort_name + MixpanelWebhook: + type: object + properties: + action: + $ref: '#/components/schemas/ActionEnum' + parameters: + $ref: '#/components/schemas/MixpanelParameters' + required: + - action + - parameters + MixpanelWebhookError: + type: object + properties: + message: + type: string + code: + type: integer + required: + - code + - message + MixpanelWebhookFailureResponse: + type: object + properties: + action: + type: + - string + - 'null' + status: + type: string + error: + $ref: '#/components/schemas/MixpanelWebhookError' + required: + - action + - error + - status + MixpanelWebhookResponse: + type: object + properties: + action: + type: string + status: + type: string + required: + - action + - status Monitoring: type: object properties: @@ -27481,10 +27604,12 @@ components: description: |- * `csv` - CSV * `amplitude` - Amplitude + * `mixpanel` - Mixpanel type: string enum: - csv - amplitude + - mixpanel StageAction: type: object properties: @@ -29534,6 +29659,10 @@ components: type: http scheme: bearer description: 'For cohort sync endpoints called by an external cohort source, such as Amplitude.' + Cohort Sync Key (Basic): + type: http + scheme: basic + description: 'For cohort sync endpoints called by an external cohort source that can only send Basic credentials, such as Mixpanel. The key is the password; the username is ignored.' Environment API Key: type: apiKey in: header From 83400d62a3584830203ec74383df93d26654f898 Mon Sep 17 00:00:00 2001 From: Gagan Trivedi Date: Tue, 25 Aug 2026 17:55:41 +0530 Subject: [PATCH 06/10] fix(cohorts): apply the identifier byte limit to Mixpanel members --- api/cohorts/serializers.py | 5 ++-- api/tests/unit/cohorts/test_sync_views.py | 31 +++++++++++++++++++++++ 2 files changed, 34 insertions(+), 2 deletions(-) diff --git a/api/cohorts/serializers.py b/api/cohorts/serializers.py index d919d0ff0636..5d1480f5a754 100644 --- a/api/cohorts/serializers.py +++ b/api/cohorts/serializers.py @@ -99,8 +99,9 @@ def _validate_identifier_byte_length(value: str) -> None: class MixpanelMemberSerializer(serializers.Serializer[None]): - # Length mirrors CohortMembership.identifier. - mixpanel_distinct_id = serializers.CharField(max_length=2000) + mixpanel_distinct_id = serializers.CharField( + validators=[_validate_identifier_byte_length] + ) class MixpanelParametersSerializer(serializers.Serializer[None]): diff --git a/api/tests/unit/cohorts/test_sync_views.py b/api/tests/unit/cohorts/test_sync_views.py index bb9235f60afb..1a81ee9c415d 100644 --- a/api/tests/unit/cohorts/test_sync_views.py +++ b/api/tests/unit/cohorts/test_sync_views.py @@ -882,3 +882,34 @@ def test_mixpanel_webhook__unknown_key_in_basic_password__returns_401( # Then assert response.status_code == status.HTTP_401_UNAUTHORIZED + + +def test_mixpanel_webhook__distinct_id_over_1024_bytes__returns_400_failure( + postgres_cohort_sync_key: _KeyAndPlaintext, + mixpanel_cohort: Cohort, +) -> None: + # Given - 512 three-byte characters: few characters, too many bytes + _, plaintext = postgres_cohort_sync_key + client = _basic_auth_client(plaintext) + url = reverse("api-v1:cohort-sync:mixpanel") + + # When + response = client.post( + url, + data={ + "action": "add_members", + "parameters": { + "mixpanel_cohort_id": "mp-42", + "mixpanel_cohort_name": "Power users", + "members": [{"mixpanel_distinct_id": "€" * 512}], + }, + }, + format="json", + ) + + # Then + assert response.status_code == status.HTTP_400_BAD_REQUEST + body = response.json() + assert body["status"] == "failure" + assert "1024 bytes" in body["error"]["message"] + assert not CohortMembership.objects.exists() From 10d6468011aaab31d1d7d7f53d0f1dc91f0b02de Mon Sep 17 00:00:00 2001 From: "flagsmith-engineering[bot]" Date: Tue, 25 Aug 2026 12:27:45 +0000 Subject: [PATCH 07/10] chore: Update documentation artefacts --- openapi.yaml | 1 - 1 file changed, 1 deletion(-) diff --git a/openapi.yaml b/openapi.yaml index 8b8679ac1151..ac62633f8c5a 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -22883,7 +22883,6 @@ components: properties: mixpanel_distinct_id: type: string - maxLength: 2000 required: - mixpanel_distinct_id MixpanelParameters: From e98fd87711aa356a62508bfef0c282d0665d268f Mon Sep 17 00:00:00 2001 From: Gagan Trivedi Date: Wed, 26 Aug 2026 14:39:42 +0530 Subject: [PATCH 08/10] fix(cohorts): cap Mixpanel pages and reject syncs to a deleting cohort --- api/cohorts/serializers.py | 7 +- api/cohorts/services.py | 16 +++ api/cohorts/sync_views.py | 21 +++- api/tests/unit/cohorts/test_sync_views.py | 101 ++++++++++++++++++ .../observability/_events-catalogue.md | 12 +-- 5 files changed, 149 insertions(+), 8 deletions(-) diff --git a/api/cohorts/serializers.py b/api/cohorts/serializers.py index 5d1480f5a754..baa8f5f8251a 100644 --- a/api/cohorts/serializers.py +++ b/api/cohorts/serializers.py @@ -108,7 +108,12 @@ class MixpanelParametersSerializer(serializers.Serializer[None]): mixpanel_cohort_id = serializers.CharField(max_length=255) mixpanel_cohort_name = serializers.CharField(max_length=2000) # An empty page is valid: a first sync of an empty cohort has no members. - members = MixpanelMemberSerializer(many=True, allow_empty=True) + # Mixpanel sends at most 1000 members per message; the cap stops anything + # else from posting an arbitrarily large page. + # The stubs don't know many=True forwards max_length to the list serialiser. + members = MixpanelMemberSerializer( # type: ignore[call-arg] + many=True, allow_empty=True, max_length=1000 + ) class MixpanelWebhookSerializer(serializers.Serializer[None]): diff --git a/api/cohorts/services.py b/api/cohorts/services.py index 10f6bb03fe43..870b5362c1f9 100644 --- a/api/cohorts/services.py +++ b/api/cohorts/services.py @@ -207,6 +207,22 @@ def get_cohort_for_source( return cohort +def cohort_deletion_in_progress( + *, + environment: "Environment", + source_type: CohortSourceType, + external_id: str, +) -> bool: + return bool( + Cohort.objects.filter( + environment=environment, + source_type=source_type, + external_id=external_id, + deletion_requested_at__isnull=False, + ).exists() + ) + + def add_cohort_members(cohort: Cohort, identifiers: "typing.Iterable[str]") -> None: from cohorts.tasks import apply_cohort_membership_deltas diff --git a/api/cohorts/sync_views.py b/api/cohorts/sync_views.py index 6bfdace7d14d..23d592942b3e 100644 --- a/api/cohorts/sync_views.py +++ b/api/cohorts/sync_views.py @@ -6,7 +6,7 @@ from drf_spectacular.utils import extend_schema, extend_schema_view, inline_serializer from rest_framework import serializers, viewsets from rest_framework.decorators import action -from rest_framework.exceptions import NotFound, ParseError +from rest_framework.exceptions import NotFound, ParseError, ValidationError from rest_framework.request import Request from rest_framework.response import Response from rest_framework.views import APIView @@ -157,6 +157,17 @@ def post(self, request: Request) -> Response: environment = typing.cast(CohortSyncKey, request.auth).environment if webhook_action == "members": + if services.cohort_deletion_in_progress( + environment=environment, + source_type=CohortSourceType.MIXPANEL, + external_id=parameters["mixpanel_cohort_id"], + ): + # Recreating the cohort while its memberships are still being + # drained would resurrect it. The 404 pauses the sync and + # emails the customer. + return self._failure( + request, message="Cohort is being deleted.", code=404 + ) # A large first sync arrives as several requests, each one page # of members. Every page only adds; removals can't be detected # without seeing all pages at once. @@ -194,6 +205,14 @@ def handle_exception(self, exc: Exception) -> Response: # A body that isn't valid JSON raises before post() runs, so the # response is shaped here to keep the envelope Mixpanel expects. return self._failure(self.request, message="Invalid payload.", code=400) + if isinstance(exc, ValidationError): + # Raised below the view, e.g. by the segment limit on cohort + # creation; shaped here for the same reason. + return self._failure( + self.request, + message=json.dumps(exc.detail, default=str), + code=400, + ) return super().handle_exception(exc) def _failure(self, request: Request, *, message: str, code: int) -> Response: diff --git a/api/tests/unit/cohorts/test_sync_views.py b/api/tests/unit/cohorts/test_sync_views.py index 1a81ee9c415d..859a3d7742f1 100644 --- a/api/tests/unit/cohorts/test_sync_views.py +++ b/api/tests/unit/cohorts/test_sync_views.py @@ -913,3 +913,104 @@ def test_mixpanel_webhook__distinct_id_over_1024_bytes__returns_400_failure( assert body["status"] == "failure" assert "1024 bytes" in body["error"]["message"] assert not CohortMembership.objects.exists() + + +def test_mixpanel_webhook__over_1000_members__returns_400_failure( + postgres_cohort_sync_key: _KeyAndPlaintext, + mixpanel_cohort: Cohort, +) -> None: + # Given - one more member than Mixpanel's documented batch size + _, plaintext = postgres_cohort_sync_key + client = _basic_auth_client(plaintext) + url = reverse("api-v1:cohort-sync:mixpanel") + + # When + response = client.post( + url, + data={ + "action": "add_members", + "parameters": { + "mixpanel_cohort_id": "mp-42", + "mixpanel_cohort_name": "Power users", + "members": [{"mixpanel_distinct_id": f"user-{i}"} for i in range(1001)], + }, + }, + format="json", + ) + + # Then + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert response.json()["status"] == "failure" + assert not CohortMembership.objects.exists() + + +def test_mixpanel_webhook__members_action_at_segment_limit__returns_400_failure( + postgres_cohort_sync_key: _KeyAndPlaintext, + settings: SettingsWrapper, +) -> None: + # Given - the project already holds as many segments as the plan allows + key, plaintext = postgres_cohort_sync_key + project = key.environment.project + settings.EDGE_ENABLED = True + project.max_segments_allowed = 1 + project.save() + Segment.objects.create(name="existing", project=project) + client = _basic_auth_client(plaintext) + url = reverse("api-v1:cohort-sync:mixpanel") + + # When + response = client.post( + url, + data={ + "action": "members", + "parameters": { + "mixpanel_cohort_id": "mp-99", + "mixpanel_cohort_name": "New cohort", + "members": [{"mixpanel_distinct_id": "user-1"}], + }, + }, + format="json", + ) + + # Then + assert response.status_code == status.HTTP_400_BAD_REQUEST + body = response.json() + assert body["action"] == "members" + assert body["status"] == "failure" + assert "maximum allowed segments" in body["error"]["message"] + assert not Cohort.objects.filter(external_id="mp-99").exists() + + +def test_mixpanel_webhook__members_action_while_cohort_draining__returns_404_failure( + postgres_cohort_sync_key: _KeyAndPlaintext, + mixpanel_cohort: Cohort, +) -> None: + # Given - the cohort was deleted in Flagsmith and is still draining + mixpanel_cohort.deletion_requested_at = timezone.now() + mixpanel_cohort.save() + _, plaintext = postgres_cohort_sync_key + client = _basic_auth_client(plaintext) + url = reverse("api-v1:cohort-sync:mixpanel") + + # When + response = client.post( + url, + data={ + "action": "members", + "parameters": { + "mixpanel_cohort_id": "mp-42", + "mixpanel_cohort_name": "Power users", + "members": [{"mixpanel_distinct_id": "user-1"}], + }, + }, + format="json", + ) + + # Then + assert response.status_code == status.HTTP_404_NOT_FOUND + assert response.json() == { + "action": "members", + "status": "failure", + "error": {"message": "Cohort is being deleted.", "code": 404}, + } + assert Cohort.objects.filter(source_type=CohortSourceType.MIXPANEL).count() == 1 diff --git a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md index 9f0388a069b3..d89dea415a31 100644 --- a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md +++ b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md @@ -86,7 +86,7 @@ Attributes: ### `cohorts.cohort.deleted` Logged at `info` from: - - `api/cohorts/services.py:421` + - `api/cohorts/services.py:437` Attributes: - `cohort.id` @@ -95,7 +95,7 @@ Attributes: ### `cohorts.cohort.deletion_requested` Logged at `info` from: - - `api/cohorts/services.py:405` + - `api/cohorts/services.py:421` Attributes: - `cohort.id` @@ -104,7 +104,7 @@ Attributes: ### `cohorts.csv.synced` Logged at `info` from: - - `api/cohorts/services.py:377` + - `api/cohorts/services.py:393` Attributes: - `adds.count` @@ -117,7 +117,7 @@ Attributes: ### `cohorts.membership.adds_received` Logged at `info` from: - - `api/cohorts/services.py:230` + - `api/cohorts/services.py:246` Attributes: - `cohort.id` @@ -155,7 +155,7 @@ Attributes: ### `cohorts.membership.removals_received` Logged at `info` from: - - `api/cohorts/services.py:255` + - `api/cohorts/services.py:271` Attributes: - `cohort.id` @@ -166,7 +166,7 @@ Attributes: ### `cohorts.sync_webhook.rejected` Logged at `warning` from: - - `api/cohorts/sync_views.py:200` + - `api/cohorts/sync_views.py:219` Attributes: - `action` From 81b0e174a484e549499c7cd5360ceb4e45d5cdd5 Mon Sep 17 00:00:00 2001 From: Gagan Trivedi Date: Wed, 26 Aug 2026 14:44:25 +0530 Subject: [PATCH 09/10] fix(cohorts): plan-gate the Mixpanel webhook --- api/cohorts/sync_views.py | 2 +- api/tests/unit/cohorts/test_sync_views.py | 27 +++++++++++++++++++++++ 2 files changed, 28 insertions(+), 1 deletion(-) diff --git a/api/cohorts/sync_views.py b/api/cohorts/sync_views.py index 23d592942b3e..d9c04f94c3ae 100644 --- a/api/cohorts/sync_views.py +++ b/api/cohorts/sync_views.py @@ -122,7 +122,7 @@ class MixpanelCohortSyncView(APIView): """ authentication_classes = [CohortSyncKeyAuthentication] - permission_classes = [HasCohortSyncKey] + permission_classes = [HasCohortSyncKey, CohortSyncPlanPermission] @extend_schema( description=( diff --git a/api/tests/unit/cohorts/test_sync_views.py b/api/tests/unit/cohorts/test_sync_views.py index 859a3d7742f1..a13f20d49229 100644 --- a/api/tests/unit/cohorts/test_sync_views.py +++ b/api/tests/unit/cohorts/test_sync_views.py @@ -1014,3 +1014,30 @@ def test_mixpanel_webhook__members_action_while_cohort_draining__returns_404_fai "error": {"message": "Cohort is being deleted.", "code": 404}, } assert Cohort.objects.filter(source_type=CohortSourceType.MIXPANEL).count() == 1 + + +@pytest.mark.saas_mode +def test_mixpanel_webhook__saas_free_plan__returns_403( + postgres_cohort_sync_key: _KeyAndPlaintext, +) -> None: + # Given + _, plaintext = postgres_cohort_sync_key + client = _basic_auth_client(plaintext) + url = reverse("api-v1:cohort-sync:mixpanel") + + # When + response = client.post( + url, + data={ + "action": "members", + "parameters": { + "mixpanel_cohort_id": "mp-42", + "mixpanel_cohort_name": "Power users", + "members": [{"mixpanel_distinct_id": "user-1"}], + }, + }, + format="json", + ) + + # Then + assert response.status_code == status.HTTP_403_FORBIDDEN From 0bbd178042df77bd32b1464346e224ef1257ae8c Mon Sep 17 00:00:00 2001 From: "flagsmith-engineering[bot]" Date: Wed, 26 Aug 2026 09:16:04 +0000 Subject: [PATCH 10/10] chore: Update documentation artefacts --- openapi.yaml | 1 + 1 file changed, 1 insertion(+) diff --git a/openapi.yaml b/openapi.yaml index ac62633f8c5a..846cda00783e 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -1787,6 +1787,7 @@ paths: - Cohort Sync Key (Basic): [] tags: - Webhooks + x-flagsmith-minimum-plan: START_UP /api/v1/environment-document/: get: operationId: sdk_v1_environment_document