feat(messaging): configure push identity verification in channel setup

posthog/api/integration.py

@@ -416,6 +416,24 @@ def create(self, validated_data: Any) -> Any:
         team_id = self.context["team_id"]
         kind = validated_data["kind"]
 
+        # Setting push identity verification is a security policy change, not a credential upload, so it
+        # needs the same admin bar as editing an integration. Without this a plain member could add a
+        # *new* APNs integration carrying someone else's bundle_id under a different Apple team id —
+        # which sidesteps the overwrite check below, since it creates rather than overwrites — and set
+        # `required` on it. The push endpoint resolves the strictest mode across every integration
+        # matching a bundle_id, so that would start rejecting the real app's device registrations.
+        #
+        # `disabled` is gated too, not just the enabling modes. The overwrite check below reads the
+        # existing ids before the write, so a member racing the first setup of an integration would
+        # still be classified as a create and could land `disabled` over the policy an admin had just
+        # written. Omitting the key entirely stays open to members and is what connecting a channel
+        # without touching the policy does — that path preserves whatever is already stored.
+        requested_verification = (validated_data.get("config") or {}).get("push_identity_verification")
+        if requested_verification and not github_callback_state.has_team_management_access(
+            self.context["request"].user, self.context["get_team"]()
+        ):
+            raise PermissionDenied("Changing push identity verification requires project admin access.")
+
         # `create` is a POST with upsert semantics: each kind's helper does an `update_or_create`
         # keyed on (team, kind, integration_id), so re-submitting the same resource overwrites the
         # existing integration instead of adding a new one. Adding is allowed for any project
@@ -471,7 +489,12 @@ def _build_integration(self, validated_data: Any) -> Any:
                 key_info = config.get("key_info")
                 if not key_info:
                     raise ValidationError("Firebase service account key must be provided")
-            instance = FirebaseIntegration.integration_from_key(key_info, team_id, request.user)
+            instance = FirebaseIntegration.integration_from_key(
+                key_info,
+                team_id,
+                request.user,
+                push_identity_verification=(validated_data.get("config") or {}).get("push_identity_verification"),
+            )
             return instance
 
         elif validated_data["kind"] == "email":
@@ -683,6 +706,7 @@ def _build_integration(self, validated_data: Any) -> Any:
                 team_id=team_id,
                 created_by=request.user,
                 environment=environment,
+                push_identity_verification=config.get("push_identity_verification"),
             )
             return instance
 

posthog/api/test/test_integration.py

@@ -4938,6 +4938,41 @@ def test_invalid_payload_is_rejected(self, _name, payload, mock_task, mock_repor
         mock_report.assert_not_called()
 
 
+class TestPushIdentityVerificationAPI(APIBaseTest):
+    def setUp(self):
+        super().setUp()
+        # Setting the policy requires admin, and the base test user is a plain member.
+        self.organization_membership.level = OrganizationMembership.Level.ADMIN
+        self.organization_membership.save()
+
+    @patch("posthog.models.integration.GoogleRequest")
+    @patch("posthog.models.integration.service_account.Credentials.from_service_account_info")
+    def test_setting_the_mode_reaches_the_integration(self, mock_from_sa, _mock_google_request):
+        # The serializer builds each provider's arguments from named config fields, so a key it doesn't
+        # know about is dropped before it ever reaches the integration. That silently made the setup
+        # UI's toggle inert; this covers the plumbing rather than just the model helper underneath it.
+        credentials = MagicMock()
+        credentials.token = "access-token"
+        credentials.expiry.timestamp.return_value = time.time() + 3600
+        mock_from_sa.return_value = credentials
+
+        response = self.client.post(
+            f"/api/environments/{self.team.pk}/integrations",
+            {
+                "kind": "firebase",
+                "config": {
+                    "key_info": {"type": "service_account", "project_id": "my-firebase-project"},
+                    "push_identity_verification": "required",
+                },
+            },
+            format="json",
+        )
+
+        assert response.status_code == status.HTTP_201_CREATED, response.content
+        integration = Integration.objects.get(team=self.team, kind="firebase")
+        assert integration.config["push_identity_verification"] == "required"
+
+
 class TestIntegrationMembershipPermissions(APIBaseTest):
     def setUp(self):
         super().setUp()
@@ -4967,6 +5002,42 @@ def test_member_cannot_delete_integration(self):
         assert response.status_code == status.HTTP_403_FORBIDDEN, response.content
         assert Integration.objects.filter(id=integration.id).exists()
 
+    @parameterized.expand(["required", "disabled"])
+    def test_member_cannot_set_push_identity_verification(self, mode):
+        # Creating a *new* APNs integration sidesteps the overwrite check below, because a different
+        # Apple team id makes a different integration_id. But the push endpoint resolves the strictest
+        # verification mode across every integration sharing a bundle_id, so a member could otherwise
+        # set `required` on a lookalike and block the real app's device registrations.
+        #
+        # `disabled` is rejected too: the overwrite check reads existing ids before the write, so a
+        # member racing the first setup would look like a create and could land `disabled` over a
+        # policy an admin had just written.
+        Integration.objects.create(
+            team=self.team,
+            kind="apns",
+            integration_id="REALTEAM.com.example.app",
+            config={"bundle_id": "com.example.app", "team_id": "REALTEAM", "key_id": "KEY1"},
+            sensitive_config={},
+        )
+
+        response = self.client.post(
+            f"/api/environments/{self.team.pk}/integrations",
+            {
+                "kind": "apns",
+                "config": {
+                    "signing_key": "-----BEGIN PRIVATE KEY-----\nfake\n-----END PRIVATE KEY-----",
+                    "key_id": "KEY2",
+                    "team_id_apple": "ATTACKER",
+                    "bundle_id": "com.example.app",
+                    "push_identity_verification": mode,
+                },
+            },
+            format="json",
+        )
+
+        assert response.status_code == status.HTTP_403_FORBIDDEN, response.content
+        assert not Integration.objects.filter(team=self.team, integration_id="ATTACKER.com.example.app").exists()
+
     def test_member_cannot_overwrite_existing_integration(self):
         # POST is an upsert (update_or_create keyed on team/kind/integration_id), so re-submitting the
         # same resource edits an existing integration. Members may add a new one, but overwriting an

posthog/models/integration.py

@@ -21,7 +21,7 @@
 
 from django.conf import settings
 from django.core.cache import cache
-from django.db import models, transaction
+from django.db import connection, models, transaction
 from django.db.models import Q
 from django.dispatch import receiver
 from django.http import HttpRequest
@@ -110,6 +110,59 @@ def _decode_jwt_payload(token: str) -> dict | None:
 # `config` key flagging a grant that only the legacy fallback credentials can refresh.
 CONFIG_LEGACY_OAUTH_CLIENT = "oauth_uses_legacy_client"
 
+# `config` key holding the push device-registration identity verification policy, read by the
+# push subscriptions endpoint. Owned by the customer, not by the provider credentials.
+CONFIG_PUSH_IDENTITY_VERIFICATION = "push_identity_verification"
+PUSH_IDENTITY_VERIFICATION_MODES = ("disabled", "optional", "required")
+
+
+def preserved_push_config(
+    team_id: int,
+    kind: str,
+    integration_id: str,
+    push_identity_verification: str | None,
+) -> dict:
+    """Config keys a push credential upsert must carry over rather than drop.
+
+    Connecting a push integration is an upsert, and the provider helpers rebuild `config` from the
+    credentials they were handed. Anything they don't know about would be lost, so rotating a
+    Firebase key or APNs .p8 would silently reset an enabled identity verification policy back to
+    disabled, reopening the device takeover it exists to prevent. Carry the existing value forward
+    unless the caller explicitly sets a new one.
+    """
+    if push_identity_verification is not None and push_identity_verification not in PUSH_IDENTITY_VERIFICATION_MODES:
+        raise ValidationError(
+            f"push_identity_verification must be one of: {', '.join(PUSH_IDENTITY_VERIFICATION_MODES)}"
+        )
+
+    # Serialize concurrent setup of this one integration for the rest of the caller's transaction.
+    # `select_for_update` alone only locks a row that already exists, so two first-time setups could
+    # both read "no policy" and the later write would clobber a policy the earlier one had just set.
+    # An advisory lock covers the not-yet-created case too, keyed on the integration's identity so it
+    # only serializes writers racing for the same integration. Every writer takes it, including one
+    # setting an explicit mode — otherwise it could slip its row in between a preserving writer's read
+    # and write, and have its policy dropped.
+    with connection.cursor() as cursor:
+        cursor.execute("SELECT pg_advisory_xact_lock(%s, hashtext(%s))", [team_id, f"{kind}:{integration_id}"])
+
+    if push_identity_verification is not None:
+        return {CONFIG_PUSH_IDENTITY_VERIFICATION: push_identity_verification}
+
+    existing = (
+        Integration.objects.select_for_update()
+        .filter(team_id=team_id, kind=kind, integration_id=integration_id)
+        .only("config")
+        .first()
+    )
+    existing_mode = (existing.config or {}).get(CONFIG_PUSH_IDENTITY_VERIFICATION) if existing else None
+    # Drop a stored value we don't recognize rather than carrying it forward. The push endpoint already
+    # treats an unknown mode as disabled, so preserving it would keep dead data alive indefinitely, and
+    # raising here would leave a corrupted integration unable to rotate its credentials.
+    if existing_mode not in PUSH_IDENTITY_VERIFICATION_MODES:
+        return {}
+    return {CONFIG_PUSH_IDENTITY_VERIFICATION: existing_mode} if existing_mode else {}
+
+
 # Values for the counter's `reason` label, bucketed from the OAuth error response.
 REFRESH_FAILURE_REASON_INVALID_GRANT = "invalid_grant"
 REFRESH_FAILURE_REASON_INVALID_CLIENT = "invalid_client"
@@ -2206,7 +2259,13 @@ def __init__(self, integration: Integration) -> None:
         self.integration = integration
 
     @classmethod
-    def integration_from_key(cls, key_info: dict, team_id: int, created_by: User | None = None) -> "Integration":
+    def integration_from_key(
+        cls,
+        key_info: dict,
+        team_id: int,
+        created_by: User | None = None,
+        push_identity_verification: str | None = None,
+    ) -> "Integration":
         scope = "https://www.googleapis.com/auth/firebase.messaging"
 
         try:
@@ -2219,23 +2278,26 @@ def integration_from_key(cls, key_info: dict, team_id: int, created_by: User | N
         if not project_id:
             raise ValidationError("Service account key must contain a project_id")
 
-        integration, created = Integration.objects.update_or_create(
-            team_id=team_id,
-            kind="firebase",
-            integration_id=project_id,
-            defaults={
-                "config": {
-                    "project_id": project_id,
-                    "expires_in": credentials.expiry.timestamp() - int(time.time()),
-                    "refreshed_at": int(time.time()),
-                },
-                "sensitive_config": {
-                    "key_info": key_info,
-                    "access_token": credentials.token,
+        # Atomic so `preserved_push_config`'s row lock is held through the upsert that follows it.
+        with transaction.atomic():
+            integration, created = Integration.objects.update_or_create(
+                team_id=team_id,
+                kind="firebase",
+                integration_id=project_id,
+                defaults={
+                    "config": {
+                        **preserved_push_config(team_id, "firebase", project_id, push_identity_verification),
+                        "project_id": project_id,
+                        "expires_in": credentials.expiry.timestamp() - int(time.time()),
+                        "refreshed_at": int(time.time()),
+                    },
+                    "sensitive_config": {
+                        "key_info": key_info,
+                        "access_token": credentials.token,
+                    },
+                    "created_by": created_by,
                 },
-                "created_by": created_by,
-            },
-        )
+            )
 
         if integration.errors:
             integration.errors = ""
@@ -2316,29 +2378,34 @@ def integration_from_key(
         team_id: int,
         created_by: User | None = None,
         environment: str = "production",
+        push_identity_verification: str | None = None,
     ) -> "Integration":
         if not all([signing_key, key_id, team_id_apple, bundle_id]):
             raise ValidationError("All APNS fields are required: signing_key, key_id, team_id_apple, bundle_id")
 
         if environment not in ("production", "sandbox"):
             raise ValidationError("APNS environment must be 'production' or 'sandbox'")
 
-        integration, created = Integration.objects.update_or_create(
-            team_id=team_id,
-            kind="apns",
-            integration_id=f"{team_id_apple}.{bundle_id}",
-            defaults={
-                "config": {
-                    "team_id": team_id_apple,
-                    "bundle_id": bundle_id,
-                    "key_id": key_id,
-                    "environment": environment,
-                },
-                "sensitive_config": {
-                    "signing_key": signing_key,
+        integration_id = f"{team_id_apple}.{bundle_id}"
+        # Atomic so `preserved_push_config`'s row lock is held through the upsert that follows it.
+        with transaction.atomic():
+            integration, created = Integration.objects.update_or_create(
+                team_id=team_id,
+                kind="apns",
+                integration_id=integration_id,
+                defaults={
+                    "config": {
+                        **preserved_push_config(team_id, "apns", integration_id, push_identity_verification),
+                        "team_id": team_id_apple,
+                        "bundle_id": bundle_id,
+                        "key_id": key_id,
+                        "environment": environment,
+                    },
+                    "sensitive_config": {
+                        "signing_key": signing_key,
+                    },
                 },
-            },
-        )
+            )
 
         if created and created_by is not None:
             integration.created_by = created_by

posthog/models/test/test_apple_push_integration.py

@@ -12,13 +12,15 @@ def _create_apple_push_integration(
         key_id: str = "ABC123KEY",
         team_id_apple: str = "TEAM123",
         bundle_id: str = "com.example.app",
+        push_identity_verification: str | None = None,
     ) -> Integration:
         return ApplePushIntegration.integration_from_key(
             signing_key=signing_key,
             key_id=key_id,
             team_id_apple=team_id_apple,
             bundle_id=bundle_id,
             team_id=self.team.id,
+            push_identity_verification=push_identity_verification,
         )
 
     def test_creates_integration(self):
@@ -39,6 +41,18 @@ def test_upserts_on_same_team_and_bundle(self):
         second.refresh_from_db()
         assert second.config["key_id"] == "NEW_KEY_ID"
 
+    def test_reconnecting_preserves_identity_verification(self):
+        # Rotating the .p8 signing key is a routine action that re-upserts the integration. It must
+        # not silently reset the verification policy, which would reopen device takeover.
+        self._create_apple_push_integration(push_identity_verification="required")
+        reconnected = self._create_apple_push_integration(key_id="ROTATED_KEY")
+
+        assert reconnected.config["push_identity_verification"] == "required"
+
+    def test_rejects_an_unknown_identity_verification_mode(self):
+        with self.assertRaises(ValidationError):
+            self._create_apple_push_integration(push_identity_verification="enabled")
+
     def test_separate_integrations_for_different_bundles(self):
         first = self._create_apple_push_integration(bundle_id="com.example.app1")
         second = self._create_apple_push_integration(bundle_id="com.example.app2")

posthog/models/test/test_firebase_integration.py

@@ -36,7 +36,9 @@ def _create_firebase_integration(self, mock_from_sa, mock_google_request, **over
         key_info = overrides.pop("key_info", FAKE_KEY_INFO)
         team_id = overrides.pop("team_id", self.team.id)
         created_by = overrides.pop("created_by", None)
-        return FirebaseIntegration.integration_from_key(key_info, team_id, created_by)
+        return FirebaseIntegration.integration_from_key(
+            key_info, team_id, created_by, push_identity_verification=overrides.pop("push_identity_verification", None)
+        )
 
     def test_creates_integration(self):
         integration = self._create_firebase_integration()
@@ -55,6 +57,18 @@ def test_upserts_on_same_project(self):
 
         assert first.id == second.id
 
+    def test_reconnecting_preserves_identity_verification(self):
+        # Rotating the service account key is a routine action that re-upserts the integration. It
+        # must not silently reset the verification policy, which would reopen device takeover.
+        self._create_firebase_integration(push_identity_verification="required")
+        reconnected = self._create_firebase_integration()
+
+        assert reconnected.config["push_identity_verification"] == "required"
+
+    def test_rejects_an_unknown_identity_verification_mode(self):
+        with self.assertRaises(ValidationError):
+            self._create_firebase_integration(push_identity_verification="enabled")
+
     def test_separate_integrations_for_different_projects(self):
         first = self._create_firebase_integration()