company_site/financial/stripe_billing.py

@@ -25,6 +25,16 @@ from .models import BillingCustomer, Invoice, RecurringSubscription
 
 logger = logging.getLogger(__name__)
 
+
+def _optional_pk(value) -> int | None:
+    """Local row ids are integers. Another site's metadata must not crash the webhook."""
+    if value is None or value == "":
+        return None
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        return None
+
 STRIPE_API_BASE = "https://api.stripe.com/v1"
 
 
@@ -411,18 +421,22 @@ def apply_subscription_event(stripe_sub: dict) -> RecurringSubscription | None:
 
 
 def apply_checkout_session_completed(session: dict) -> RecurringSubscription | None:
-    """Attach subscription id after Checkout completes."""
+    """Attach subscription id after Checkout completes. Unknown sessions are ignored."""
+    session_id = session.get("id") or ""
     if session.get("mode") != "subscription":
+        logger.info("Ignoring checkout session %s with mode %s", session_id, session.get("mode") or "")
         return None
     local_id = (session.get("metadata") or {}).get("local_subscription_id")
     sub = None
-    if local_id:
-        sub = RecurringSubscription.objects.filter(pk=local_id).first()
-    if sub is None and session.get("id"):
+    pk = _optional_pk(local_id)
+    if pk is not None:
+        sub = RecurringSubscription.objects.filter(pk=pk).first()
+    if sub is None and session_id:
         sub = RecurringSubscription.objects.filter(
-            checkout_session_id=session["id"]
+            checkout_session_id=session_id
         ).first()
     if sub is None:
+        logger.info("Ignoring checkout session %s with no local subscription", session_id)
         return None
 
     stripe_sub_id = session.get("subscription")

company_site/financial/tests.py

@@ -218,6 +218,40 @@ class InvoiceAccessTests(TestCase):
         self.assertEqual(invoice.status, Invoice.Status.PAID)
         self.assertEqual(invoice.hosted_invoice_url, "https://pay.stripe.com/test")
 
+    def test_webhook_ignores_checkout_from_other_site(self):
+        from django.test import override_settings
+
+        from financial.models import RecurringSubscription
+        from financial.stripe_billing import apply_checkout_session_completed
+
+        session = {
+            "id": "cs_other_site",
+            "mode": "subscription",
+            "customer": "cus_other",
+            "subscription": "sub_other",
+            "metadata": {
+                "user_id": "999999",
+                "purpose": "personal",
+                "local_subscription_id": "not-ours",
+            },
+        }
+        before = RecurringSubscription.objects.count()
+        self.assertIsNone(apply_checkout_session_completed(session))
+        payload = (
+            '{"id":"evt_other_1","type":"checkout.session.completed",'
+            '"data":{"object":{"id":"cs_other_site","mode":"subscription",'
+            '"metadata":{"user_id":"999999","purpose":"personal",'
+            '"local_subscription_id":"not-ours"}}}}'
+        )
+        with override_settings(STRIPE_WEBHOOK_SECRET=""):
+            response = self.client_http.post(
+                reverse("stripe_webhook"),
+                data=payload,
+                content_type="application/json",
+            )
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(RecurringSubscription.objects.count(), before)
+
     def test_apply_invoice_webhook_ignores_zero_dollar_paid(self):
         from financial.models import BillingCustomer, Invoice
         from financial.stripe_billing import apply_stripe_invoice_event