Ack Stripe checkouts that belong to another site (#36)
Unit Tests / test (push) Successful in 9s
Unit Tests / test (push) Successful in 9s
## Summary - Checkouts from the other site on this Stripe account have no local subscription. The webhook already returned 200 for those; it now logs the ignore. - A non-numeric `local_subscription_id` no longer crashes the lookup. ## Test plan - [ ] `uv run python company_site/manage.py test financial.tests.InvoiceAccessTests.test_webhook_ignores_checkout_from_other_site` - [ ] A checkout that matches a local subscription still marks it active. - [ ] An unrelated checkout posted to the company webhook returns 200. Reviewed-on: #36
This commit was merged in pull request #36.
This commit is contained in:
1 parent
5956d0432b
commit
37917c51d2
2 files changed
+53
-5
No files matched your search
@@ -25,6 +25,16 @@ from .models import BillingCustomer, Invoice, RecurringSubscription
|
|||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
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"
|
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:
|
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":
|
if session.get("mode") != "subscription":
|
||||||
|
logger.info("Ignoring checkout session %s with mode %s", session_id, session.get("mode") or "")
|
||||||
return None
|
return None
|
||||||
local_id = (session.get("metadata") or {}).get("local_subscription_id")
|
local_id = (session.get("metadata") or {}).get("local_subscription_id")
|
||||||
sub = None
|
sub = None
|
||||||
if local_id:
|
pk = _optional_pk(local_id)
|
||||||
sub = RecurringSubscription.objects.filter(pk=local_id).first()
|
if pk is not None:
|
||||||
if sub is None and session.get("id"):
|
sub = RecurringSubscription.objects.filter(pk=pk).first()
|
||||||
|
if sub is None and session_id:
|
||||||
sub = RecurringSubscription.objects.filter(
|
sub = RecurringSubscription.objects.filter(
|
||||||
checkout_session_id=session["id"]
|
checkout_session_id=session_id
|
||||||
).first()
|
).first()
|
||||||
if sub is None:
|
if sub is None:
|
||||||
|
logger.info("Ignoring checkout session %s with no local subscription", session_id)
|
||||||
return None
|
return None
|
||||||
|
|
||||||
stripe_sub_id = session.get("subscription")
|
stripe_sub_id = session.get("subscription")
|
||||||
|
|||||||
@@ -218,6 +218,40 @@ class InvoiceAccessTests(TestCase):
|
|||||||
self.assertEqual(invoice.status, Invoice.Status.PAID)
|
self.assertEqual(invoice.status, Invoice.Status.PAID)
|
||||||
self.assertEqual(invoice.hosted_invoice_url, "https://pay.stripe.com/test")
|
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):
|
def test_apply_invoice_webhook_ignores_zero_dollar_paid(self):
|
||||||
from financial.models import BillingCustomer, Invoice
|
from financial.models import BillingCustomer, Invoice
|
||||||
from financial.stripe_billing import apply_stripe_invoice_event
|
from financial.stripe_billing import apply_stripe_invoice_event
|
||||||
|
|||||||
Reference in new issue
Block a user