feat(messaging): optional identity verification for push subscriptions

products/messaging/backend/api/push_identity_tokens.py

@@ -0,0 +1,85 @@
+"""
+Signed identity tokens for push subscription registration.
+
+A push device token (FCM registration token / APNs device token) is a delivery *address*, not a
+credential: FCM/APNs hand a token to any app instance that registers, and an attacker owns their own
+token legitimately. So possession of a token proves "deliver to this device" — never "this device
+belongs to user X". Binding a token to a `distinct_id` therefore needs proof that the caller is
+allowed to act for that `distinct_id`. The public project token can't provide it: it is embedded in
+the mobile app and world-readable, so anyone can present it and claim any `distinct_id`.
+
+Following the pattern proven by Braze's "SDK Authentication", the customer's backend — the only party
+that actually authenticated the end user — mints a short-lived token asserting the user's
+`distinct_id`, signed with the project's secret API key. PostHog re-verifies the signature at
+registration time. An attacker holding only the public project token cannot forge it.
+
+We use symmetric HMAC (HS256) keyed by `Team.secret_api_token` rather than an asymmetric key pair:
+PostHog verifies its own ingestion, so there is no third-party verifier that would need a public key,
+and the secret already ships with built-in rotation via `secret_api_token_backup`. Verification
+accepts either the current or the backup secret so a key rotation doesn't reject in-flight tokens.
+"""
+
+from datetime import UTC, datetime, timedelta
+
+import jwt
+
+from posthog.models.team.team import Team
+
+PUSH_IDENTITY_TOKEN_AUDIENCE = "posthog:push_identity"
+_ALGORITHM = "HS256"
+
+# Short TTL: the token only needs to survive the round trip from the customer's backend, through the
+# app, to the registration call. Keeping it small bounds the replay window (a replay can only re-assert
+# the same (distinct_id, app_id) binding the legitimate user already holds, so the value is low anyway).
+DEFAULT_TTL = timedelta(minutes=5)
+
+
+def sign_push_identity_token(
+    secret_api_token: str,
+    distinct_id: str,
+    app_id: str,
+    ttl: timedelta = DEFAULT_TTL,
+) -> str:
+    """Mint a signed identity token.
+
+    This is the reference implementation of what the *customer's backend* runs after it has
+    authenticated the end user. It is not called by PostHog's own ingestion (which only verifies);
+    it lives here so the signing and verification rules stay in one place and the tests can exercise
+    the real round trip.
+    """
+    return jwt.encode(
+        {
+            "sub": distinct_id,
+            "app_id": app_id,
+            "aud": PUSH_IDENTITY_TOKEN_AUDIENCE,
+            "exp": datetime.now(UTC) + ttl,
+        },
+        secret_api_token,
+        algorithm=_ALGORITHM,
+    )
+
+
+def verify_push_identity_token(token: str, team: Team, distinct_id: str, app_id: str) -> bool:
+    """Return True iff `token` is a valid, unexpired identity assertion for exactly this
+    `(distinct_id, app_id)`, signed by the team's current or backup secret API key.
+
+    Binding the claim to `app_id` as well as `distinct_id` stops a token minted for one app being
+    replayed to register a device under a different app in the same project.
+    """
+    candidate_secrets = [secret for secret in (team.secret_api_token, team.secret_api_token_backup) if secret]
+    for secret in candidate_secrets:
+        try:
+            payload = jwt.decode(
+                token,
+                secret,
+                algorithms=[_ALGORITHM],
+                audience=PUSH_IDENTITY_TOKEN_AUDIENCE,
+                # Require exp explicitly: PyJWT only checks expiry when the claim is present, so without
+                # this a token minted (by an external signer) with no exp would never expire.
+                options={"require": ["exp"]},
+            )
+        except jwt.InvalidTokenError:
+            continue
+        if payload.get("sub") == distinct_id and payload.get("app_id") == app_id:
+            return True
+    return False

products/messaging/backend/api/push_subscriptions.py

@@ -4,6 +4,7 @@
 from django.http import HttpResponse, JsonResponse
 from django.views.decorators.csrf import csrf_exempt
 
+from prometheus_client import Counter
 from rest_framework import status
 from rest_framework.request import Request
 
@@ -20,6 +21,18 @@
 from posthog.utils import decompress, load_data_from_request
 from posthog.utils_cors import cors_response
 
+from products.messaging.backend.api.push_identity_tokens import verify_push_identity_token
+
+# Identity verification is opt-in per integration via config["push_identity_verification"]:
+#   "disabled" (default) — no token required; anyone with the public project token can register.
+#   "optional"           — a token is verified and recorded when present, but never required.
+#   "required"           — registration/unregistration is rejected without a valid identity token.
+PUSH_IDENTITY_VERIFICATION_COUNTER = Counter(
+    "push_subscription_identity_verification",
+    "Outcome of push subscription identity token verification.",
+    labelnames=["mode", "operation", "outcome"],
+)
+
 VALID_PLATFORMS = ("android", "ios")
 
 # A device registration payload is a handful of short string fields (distinct_id, device_token,
@@ -33,16 +46,30 @@
 _encrypted_fields = EncryptedFieldMixin()
 
 
-# Resolve the integration from the app_id alone, not the device platform. An app_id is either a
+# Verification-mode precedence. An app_id can match more than one integration — config identifiers
+# (project_id / bundle_id) aren't covered by a uniqueness constraint — so mode resolution must fail
+# closed: take the strictest mode across every match so a lax duplicate can't downgrade a sibling's
+# `required` policy. Unknown/garbage values sort to 0 (treated as disabled).
+_VERIFICATION_MODE_PRECEDENCE = {"disabled": 0, "optional": 1, "required": 2}
+
+
+# Resolve integrations from the app_id alone, not the device platform. An app_id is either a
 # Firebase project_id or an APNs bundle_id, so a device can register with either provider regardless
 # of its OS — e.g. an iOS device delivering through Firebase registers with the Firebase project_id.
 # (The client still sends its platform, but it's metadata, not what selects the provider.)
-def _find_integration(team_id: int, app_id: str) -> Integration | None:
-    return (
+def _find_integrations(team_id: int, app_id: str) -> list[Integration]:
+    return list(
         Integration.objects.filter(team_id=team_id)
         .filter(Q(kind="firebase", config__project_id=app_id) | Q(kind="apns", config__bundle_id=app_id))
-        .only("id")
-        .first()
+        .only("id", "config")
+    )
+
+
+def _strictest_verification_mode(integrations: list[Integration]) -> str:
+    return max(
+        (integration.config.get("push_identity_verification", "disabled") for integration in integrations),
+        key=lambda mode: _VERIFICATION_MODE_PRECEDENCE.get(mode, 0),
+        default="disabled",
     )
 
 
@@ -176,8 +203,8 @@ def push_subscriptions(request: Request):
             ),
         )
 
-    integration = _find_integration(team.id, app_id)
-    if not integration:
+    integrations = _find_integrations(team.id, app_id)
+    if not integrations:
         return cors_response(
             request,
             generate_exception_response(
@@ -190,6 +217,31 @@ def push_subscriptions(request: Request):
             ),
         )
 
+    operation = "register" if request.method == "POST" else "unregister"
+    verification_mode = _strictest_verification_mode(integrations)
+    if verification_mode in ("optional", "required"):
+        identity_token = data.get("identity_token")
+        verified = isinstance(identity_token, str) and verify_push_identity_token(
+            identity_token, team, distinct_id, app_id
+        )
+        PUSH_IDENTITY_VERIFICATION_COUNTER.labels(
+            mode=verification_mode,
+            operation=operation,
+            outcome="verified" if verified else "unverified",
+        ).inc()
+        if not verified and verification_mode == "required":
+            return cors_response(
+                request,
+                generate_exception_response(
+                    "push_subscriptions",
+                    "A valid identity token is required for this device. Your backend must mint one for "
+                    "the signed-in user with the project's secret API key.",
+                    type="authentication_error",
+                    code="identity_verification_failed",
+                    status_code=status.HTTP_401_UNAUTHORIZED,
+                ),
+            )
+
     property_key = f"$device_push_subscription_{app_id}"
 
     # $unset of an absent property is a no-op, so DELETE (logout) is idempotent. device_token is

products/messaging/backend/api/test/test_push_identity_tokens.py

@@ -0,0 +1,70 @@
+from datetime import timedelta
+
+from django.test import SimpleTestCase
+
+import jwt
+from parameterized import parameterized
+
+from posthog.models.team.team import Team
+
+from products.messaging.backend.api.push_identity_tokens import (
+    PUSH_IDENTITY_TOKEN_AUDIENCE,
+    sign_push_identity_token,
+    verify_push_identity_token,
+)
+
+# Realistic length (>= 32 bytes) — matches a real phs_ secret and avoids PyJWT's short-key warning.
+CURRENT_SECRET = "phs_current_secret_0123456789abcdef0123"
+BACKUP_SECRET = "phs_backup_secret_0123456789abcdef01234"
+DISTINCT_ID = "user-1"
+APP_ID = "my-firebase-project"
+
+
+class TestPushIdentityTokens(SimpleTestCase):
+    def _team(self, secret: str | None = CURRENT_SECRET, backup: str | None = None) -> Team:
+        return Team(secret_api_token=secret, secret_api_token_backup=backup)
+
+    def test_verifies_a_token_signed_with_the_current_secret(self):
+        token = sign_push_identity_token(CURRENT_SECRET, DISTINCT_ID, APP_ID)
+        assert verify_push_identity_token(token, self._team(), DISTINCT_ID, APP_ID) is True
+
+    def test_verifies_a_token_signed_with_the_backup_secret_after_rotation(self):
+        token = sign_push_identity_token(BACKUP_SECRET, DISTINCT_ID, APP_ID)
+        team = self._team(secret=CURRENT_SECRET, backup=BACKUP_SECRET)
+        assert verify_push_identity_token(token, team, DISTINCT_ID, APP_ID) is True
+
+    @parameterized.expand(
+        [
+            ("wrong_distinct_id", "someone-else", APP_ID),
+            ("wrong_app_id", DISTINCT_ID, "other-app"),
+        ]
+    )
+    def test_rejects_a_token_whose_claims_do_not_match_the_registration(self, _name, sub, app_id):
+        # The rebind guard: a token minted for one (distinct_id, app_id) cannot authorize a different one.
+        token = sign_push_identity_token(CURRENT_SECRET, sub, app_id)
+        assert verify_push_identity_token(token, self._team(), DISTINCT_ID, APP_ID) is False
+
+    def test_rejects_a_token_signed_with_a_different_secret(self):
+        token = sign_push_identity_token("phs_attacker_secret_0123456789abcdef012", DISTINCT_ID, APP_ID)
+        assert verify_push_identity_token(token, self._team(), DISTINCT_ID, APP_ID) is False
+
+    def test_rejects_an_expired_token(self):
+        token = sign_push_identity_token(CURRENT_SECRET, DISTINCT_ID, APP_ID, ttl=timedelta(seconds=-1))
+        assert verify_push_identity_token(token, self._team(), DISTINCT_ID, APP_ID) is False
+
+    def test_rejects_a_token_with_no_exp_claim(self):
+        # An external signer (customer backend / SDK) could omit exp; without requiring it a token would
+        # never expire, so the verifier must reject it even though the signature is valid.
+        token = jwt.encode(
+            {"sub": DISTINCT_ID, "app_id": APP_ID, "aud": PUSH_IDENTITY_TOKEN_AUDIENCE},
+            CURRENT_SECRET,
+            algorithm="HS256",
+        )
+        assert verify_push_identity_token(token, self._team(), DISTINCT_ID, APP_ID) is False
+
+    def test_rejects_a_malformed_token(self):
+        assert verify_push_identity_token("not-a-jwt", self._team(), DISTINCT_ID, APP_ID) is False
+
+    def test_rejects_when_the_team_has_no_secret_configured(self):
+        token = sign_push_identity_token(CURRENT_SECRET, DISTINCT_ID, APP_ID)
+        assert verify_push_identity_token(token, self._team(secret=None), DISTINCT_ID, APP_ID) is False