engineering/make_post_sell
1
0
Fork 0
Code Issues 2 Pull requests Projects Releases 2 Packages Wiki Activity Actions

Add Stripe payment tracking and webhook resilience

Browse source 
- Add stripe_payment_intent_id and stripe_charge_id columns to Invoice
- Store payment references during checkout for traceability
- Use idempotency key to prevent duplicate charges on retry
- Add Stripe webhook handler for payment_intent.succeeded, payment_failed,
  charge.refunded, and charge.dispute.created events
- Consolidate PayPal webhooks into webhooks.py
- Add stripe.webhook_secret configuration for signature verification

Tests: 8 unit, 5 integration, 7 functional tests for Stripe functionality
This commit is contained in:
Russell Ballestrini 2025-12-22 16:03:20 -05:00
parent 0b89ab2bb3
commit ac2f586453
10 changed files with 956 additions and 180 deletions
Download patch file Download diff file Expand all files Collapse all files

17
docs/PAYPAL.md
View file

@ -33,11 +33,26 @@ export MPS_PAYPAL_SANDBOX_MODE=True
- `Invoice.payment_method` returns "paypal" for PayPal payments
- Saved payment methods stored in `PayPalUserShop`
## Webhooks
Webhooks provide resilience when JavaScript callbacks fail. Configure in PayPal Developer Dashboard:
1. Go to https://developer.paypal.com/dashboard/applications
2. Select your app → Webhooks → Add Webhook
3. Enter URL: `https://yourdomain.com/webhooks/paypal`
4. Subscribe to events:
- `PAYMENT.CAPTURE.COMPLETED`
- `CHECKOUT.ORDER.APPROVED`
- `PAYMENT.CAPTURE.DENIED`
- `CUSTOMER.DISPUTE.CREATED`
5. Copy the Webhook ID
6. Set in your ini: `paypal.webhook_id = YOUR_WEBHOOK_ID`
## Code
- `make_post_sell/views/cart.py` - PayPal checkout functions (`paypal_create_order`, `paypal_complete_checkout`)
- `make_post_sell/views/billing.py` - Disconnect saved PayPal
- `make_post_sell/views/paypal_webhooks.py` - Webhook handler
- `make_post_sell/views/webhooks.py` - Webhook handlers
## Database

15
make_post_sell/models/invoice.py
View file

@ -107,6 +107,10 @@ class Invoice(RBase, Base):
paypal_order_id = Column(Unicode(64), nullable=True)
paypal_capture_id = Column(Unicode(64), nullable=True)
# Stripe payment tracking (nullable - only set for Stripe payments)
stripe_payment_intent_id = Column(Unicode(64), nullable=True)
stripe_charge_id = Column(Unicode(64), nullable=True)
# one to one.
user = relationship(argument="User", uselist=False, lazy="joined")
@ -289,6 +293,10 @@ class Invoice(RBase, Base):
# Check for PayPal payment
if self.paypal_order_id:
return "paypal"
# Check for Stripe payment (or assume Stripe for legacy invoices)
if self.stripe_payment_intent_id:
return "stripe"
# Default to stripe for legacy invoices without explicit payment tracking
return "stripe"
@property
@ -314,6 +322,13 @@ def get_invoice_by_paypal_order_id(dbsession, paypal_order_id):
).first()
def get_invoice_by_stripe_payment_intent_id(dbsession, payment_intent_id):
"""Try to get Invoice object by Stripe payment intent ID or return None."""
return dbsession.query(Invoice).filter(
Invoice.stripe_payment_intent_id == payment_intent_id
).first()
def delete_invoice_by_id(dbsession, invoice_id):
"""
Safely delete an invoice and its line items, but only if it's from a terminated/unsuccessful crypto payment.

3
make_post_sell/routes.py
View file

@ -29,7 +29,10 @@ def includeme(config):
# PayPal routes
config.add_route("paypal_create_order", "/paypal/create-order/{cart_id}")
config.add_route("paypal_complete_checkout", "/paypal/complete-checkout/{cart_id}")
# Webhook routes
config.add_route("paypal_webhook", "/webhooks/paypal")
config.add_route("stripe_webhook", "/webhooks/stripe")
# user routes.
config.add_route("user_settings", "/u/settings")

33
make_post_sell/scripts/alembic/versions/63d935094f97_add_stripe_payment_tracking_columns_to_.py Normal file
View file

@ -0,0 +1,33 @@
"""add stripe payment tracking columns to invoice
Revision ID: 63d935094f97
Revises: 418933067d81
Create Date: 2025-12-22 15:54:08.424454
"""
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision = '63d935094f97'
down_revision = '418933067d81'
branch_labels = None
depends_on = None
def upgrade():
# Add Stripe payment tracking columns to mps_invoice
op.add_column(
'mps_invoice',
sa.Column('stripe_payment_intent_id', sa.Unicode(64), nullable=True)
)
op.add_column(
'mps_invoice',
sa.Column('stripe_charge_id', sa.Unicode(64), nullable=True)
)
def downgrade():
op.drop_column('mps_invoice', 'stripe_charge_id')
op.drop_column('mps_invoice', 'stripe_payment_intent_id')

154
make_post_sell/tests/test_functional.py
View file

@ -560,9 +560,9 @@ class AuthenticatedFunctionalTests(FunctionalTests):
"csrf_token": csrf_token, # Include CSRF token as a form value.
},
)
# Follow any redirects until we get a 200 OK response
while res_csrf_checkout.status_int in (301, 302, 303, 307, 308):
res_csrf_checkout = res_csrf_checkout.follow()
# With multiple payment methods enabled (Stripe + PayPal), checkout
# renders directly instead of redirecting to /billing for Stripe setup
self.assertEqual(200, res_csrf_checkout.status_int)
self.assertIn(
"Please confirm your order.", res_csrf_checkout.body.decode()
)
@ -1941,3 +1941,151 @@ class AuthenticatedFunctionalTests(FunctionalTests):
order_data = get_response.json()
self.assertEqual(order_data["id"], order_id)
self.assertEqual(order_data["status"], "CREATED")
# =========================================================================
# Stripe Webhook Tests
# =========================================================================
@patch("smtplib.SMTP")
def test_stripe_webhook_endpoint_accepts_post(self, mock_smtp):
"""Test that the Stripe webhook endpoint accepts POST requests."""
# Send a minimal valid webhook event
webhook_payload = {
"type": "payment_intent.succeeded",
"data": {
"object": {
"id": "pi_test_12345",
"latest_charge": "ch_test_12345"
}
}
}
response = self.testapp.post_json(
"/webhooks/stripe",
webhook_payload,
status=200
)
self.assertEqual(response.status_int, 200)
self.assertIn("success", response.json.get("status", ""))
@patch("smtplib.SMTP")
def test_stripe_webhook_payment_intent_succeeded_no_invoice(self, mock_smtp):
"""Test webhook with payment_intent.succeeded but no matching invoice."""
webhook_payload = {
"type": "payment_intent.succeeded",
"data": {
"object": {
"id": "pi_nonexistent_12345",
"latest_charge": "ch_nonexistent_12345"
}
}
}
response = self.testapp.post_json(
"/webhooks/stripe",
webhook_payload,
status=200
)
# Should succeed even without matching invoice (idempotent)
self.assertEqual(response.status_int, 200)
@patch("smtplib.SMTP")
def test_stripe_webhook_payment_failed_event(self, mock_smtp):
"""Test handling payment_intent.payment_failed event."""
webhook_payload = {
"type": "payment_intent.payment_failed",
"data": {
"object": {
"id": "pi_failed_12345",
"last_payment_error": {
"message": "Your card was declined."
}
}
}
}
response = self.testapp.post_json(
"/webhooks/stripe",
webhook_payload,
status=200
)
self.assertEqual(response.status_int, 200)
@patch("smtplib.SMTP")
def test_stripe_webhook_charge_refunded_event(self, mock_smtp):
"""Test handling charge.refunded event."""
webhook_payload = {
"type": "charge.refunded",
"data": {
"object": {
"id": "ch_refund_12345",
"amount_refunded": 1000 # $10.00 in cents
}
}
}
response = self.testapp.post_json(
"/webhooks/stripe",
webhook_payload,
status=200
)
self.assertEqual(response.status_int, 200)
@patch("smtplib.SMTP")
def test_stripe_webhook_dispute_created_event(self, mock_smtp):
"""Test handling charge.dispute.created event."""
webhook_payload = {
"type": "charge.dispute.created",
"data": {
"object": {
"id": "dp_dispute_12345",
"charge": "ch_disputed_12345",
"amount": 5000, # $50.00 in cents
"reason": "fraudulent"
}
}
}
response = self.testapp.post_json(
"/webhooks/stripe",
webhook_payload,
status=200
)
self.assertEqual(response.status_int, 200)
@patch("smtplib.SMTP")
def test_stripe_webhook_unknown_event_type(self, mock_smtp):
"""Test that unknown event types are handled gracefully."""
webhook_payload = {
"type": "unknown.event.type",
"data": {
"object": {}
}
}
response = self.testapp.post_json(
"/webhooks/stripe",
webhook_payload,
status=200
)
# Should return success for unknown events (don't block)
self.assertEqual(response.status_int, 200)
@patch("smtplib.SMTP")
def test_stripe_webhook_malformed_json(self, mock_smtp):
"""Test that malformed JSON returns an error response."""
response = self.testapp.post(
"/webhooks/stripe",
"not valid json",
content_type="application/json",
status=200 # Returns 200 to prevent retries
)
# Should handle gracefully
self.assertEqual(response.status_int, 200)

151
make_post_sell/tests/test_integration.py
View file

@ -2738,3 +2738,154 @@ class TestInvoicePayPalIntegration(DatabaseIntegrationTests):
self.assertTrue(invoice.is_paid)
transaction.commit()
class TestInvoiceStripeIntegration(DatabaseIntegrationTests):
"""Integration tests for Invoice Stripe functionality with real database."""
def test_invoice_with_stripe_payment_intent_id_integration(self):
"""Test creating invoice with Stripe payment intent and charge IDs."""
# Create real user
user = get_or_create_user_by_email(self.dbsession, "invoice_stripe@example.com")
self.dbsession.add(user)
# Create real shop
shop = Shop(
name="Stripe Invoice Test Shop",
phone_number="555-555-5555",
billing_address="123 Test St",
description="A shop for testing Stripe invoices",
)
shop.stripe_public_api_key = "pk_test_123"
shop.stripe_secret_api_key = "sk_test_123"
shop.domain_name = "stripeinvoicetest.com"
self.dbsession.add(shop)
self.dbsession.flush()
# Create invoice with Stripe payment info
invoice = Invoice(user=user)
invoice.shop = shop
invoice.stripe_payment_intent_id = "pi_1234567890abcdef"
invoice.stripe_charge_id = "ch_1234567890abcdef"
self.dbsession.add(invoice)
self.dbsession.flush()
# Verify payment method detection
self.assertEqual(invoice.payment_method, "stripe")
# Verify query by Stripe payment intent ID
from make_post_sell.models.invoice import get_invoice_by_stripe_payment_intent_id
queried = get_invoice_by_stripe_payment_intent_id(self.dbsession, "pi_1234567890abcdef")
self.assertIsNotNone(queried)
self.assertEqual(queried.id, invoice.id)
self.assertEqual(queried.stripe_charge_id, "ch_1234567890abcdef")
transaction.commit()
def test_get_invoice_by_stripe_payment_intent_id_not_found(self):
"""Test querying for non-existent Stripe payment intent ID returns None."""
from make_post_sell.models.invoice import get_invoice_by_stripe_payment_intent_id
result = get_invoice_by_stripe_payment_intent_id(self.dbsession, "pi_nonexistent")
self.assertIsNone(result)
def test_invoice_stripe_payment_persists_across_sessions(self):
"""Test that Stripe payment IDs persist correctly in the database."""
user = get_or_create_user_by_email(self.dbsession, "persist_test@example.com")
shop = Shop(
name="Persist Test Shop",
phone_number="555-555-5555",
billing_address="123 Test St",
description="Test shop",
)
shop.stripe_public_api_key = "pk_test_123"
shop.stripe_secret_api_key = "sk_test_123"
shop.domain_name = "persisttest.com"
self.dbsession.add(user)
self.dbsession.add(shop)
self.dbsession.flush()
# Create invoice with Stripe payment info
invoice = Invoice(user=user)
invoice.shop = shop
invoice.stripe_payment_intent_id = "pi_persist_test_12345"
invoice.stripe_charge_id = "ch_persist_test_12345"
self.dbsession.add(invoice)
self.dbsession.flush()
invoice_id = invoice.id
# Commit and clear session to simulate new request
transaction.commit()
self.dbsession.expire_all()
# Re-query and verify
from make_post_sell.models.invoice import get_invoice_by_id
reloaded_invoice = get_invoice_by_id(self.dbsession, invoice_id)
self.assertIsNotNone(reloaded_invoice)
self.assertEqual(reloaded_invoice.stripe_payment_intent_id, "pi_persist_test_12345")
self.assertEqual(reloaded_invoice.stripe_charge_id, "ch_persist_test_12345")
self.assertEqual(reloaded_invoice.payment_method, "stripe")
def test_invoice_with_stripe_is_paid_status(self):
"""Test is_paid property works correctly for Stripe invoices."""
user = get_or_create_user_by_email(self.dbsession, "stripe_paid_test@example.com")
shop = Shop(
name="Stripe Paid Test Shop",
phone_number="555-555-5555",
billing_address="123 Test St",
description="Test shop",
)
shop.stripe_public_api_key = "pk_test_123"
shop.stripe_secret_api_key = "sk_test_123"
shop.domain_name = "stripepaidtest.com"
self.dbsession.add(user)
self.dbsession.add(shop)
self.dbsession.flush()
# Stripe invoice is considered "paid"
invoice = Invoice(user=user)
invoice.shop = shop
invoice.stripe_payment_intent_id = "pi_paid_test"
invoice.stripe_charge_id = "ch_paid_test"
self.dbsession.add(invoice)
self.dbsession.flush()
# payment_status returns "paid" for non-crypto invoices
self.assertEqual(invoice.payment_status, "paid")
self.assertTrue(invoice.is_paid)
transaction.commit()
def test_invoice_payment_method_priority_with_stripe(self):
"""Test payment method detection with Stripe: crypto > paypal > stripe."""
user = get_or_create_user_by_email(self.dbsession, "stripe_priority@example.com")
shop = Shop(
name="Stripe Priority Shop",
phone_number="555-555-5555",
billing_address="123 Test St",
description="Test shop",
)
shop.stripe_public_api_key = "pk_test_123"
shop.stripe_secret_api_key = "sk_test_123"
shop.domain_name = "stripeprioritytest.com"
self.dbsession.add(user)
self.dbsession.add(shop)
self.dbsession.flush()
# Invoice with only Stripe payment intent returns stripe
invoice_stripe = Invoice(user=user)
invoice_stripe.shop = shop
invoice_stripe.stripe_payment_intent_id = "pi_priority_test"
self.dbsession.add(invoice_stripe)
self.dbsession.flush()
self.assertEqual(invoice_stripe.payment_method, "stripe")
# Invoice with both PayPal and Stripe - PayPal takes priority
invoice_both = Invoice(user=user)
invoice_both.shop = shop
invoice_both.paypal_order_id = "ORDER_PRIORITY"
invoice_both.stripe_payment_intent_id = "pi_priority_both"
self.dbsession.add(invoice_both)
self.dbsession.flush()
self.assertEqual(invoice_both.payment_method, "paypal")
transaction.commit()

81
make_post_sell/tests/test_models.py
View file

@ -1681,3 +1681,84 @@ class TestInvoicePayPal(unittest.TestCase):
invoice.paypal_order_id = "PAYPAL-ORDER-123"
self.assertTrue(invoice.is_paid)
class TestInvoiceStripe(unittest.TestCase):
"""Unit tests for Invoice Stripe functionality."""
@mock.patch("make_post_sell.models.user.is_user_name_available", mock_always_true)
def setUp(self):
self.user = User("stripe_invoice@example.com")
self.shop = Shop(
name="Stripe Invoice Shop",
phone_number="555-1234",
billing_address="123 Main St",
description="Test shop",
)
def test_invoice_stripe_columns_nullable(self):
"""Test that Stripe columns are nullable by default."""
invoice = Invoice(self.user)
invoice.shop = self.shop
self.assertIsNone(invoice.stripe_payment_intent_id)
self.assertIsNone(invoice.stripe_charge_id)
def test_invoice_stripe_payment_intent_id_storage(self):
"""Test storing Stripe payment intent ID."""
invoice = Invoice(self.user)
invoice.shop = self.shop
invoice.stripe_payment_intent_id = "pi_1234567890abcdef"
self.assertEqual(invoice.stripe_payment_intent_id, "pi_1234567890abcdef")
def test_invoice_stripe_charge_id_storage(self):
"""Test storing Stripe charge ID."""
invoice = Invoice(self.user)
invoice.shop = self.shop
invoice.stripe_payment_intent_id = "pi_1234567890abcdef"
invoice.stripe_charge_id = "ch_1234567890abcdef"
self.assertEqual(invoice.stripe_charge_id, "ch_1234567890abcdef")
def test_invoice_payment_method_stripe_explicit(self):
"""Test that invoice with stripe_payment_intent_id returns stripe."""
invoice = Invoice(self.user)
invoice.shop = self.shop
invoice.stripe_payment_intent_id = "pi_1234567890abcdef"
self.assertEqual(invoice.payment_method, "stripe")
def test_invoice_payment_method_stripe_default(self):
"""Test that invoice with no payment refs returns stripe (legacy)."""
invoice = Invoice(self.user)
invoice.shop = self.shop
# Legacy invoices without payment tracking default to stripe
self.assertEqual(invoice.payment_method, "stripe")
def test_invoice_payment_status_for_stripe(self):
"""Test that Stripe invoice payment_status is 'paid'."""
invoice = Invoice(self.user)
invoice.shop = self.shop
invoice.stripe_payment_intent_id = "pi_1234567890abcdef"
self.assertEqual(invoice.payment_status, "paid")
def test_invoice_is_paid_for_stripe(self):
"""Test that Stripe invoice is_paid returns True."""
invoice = Invoice(self.user)
invoice.shop = self.shop
invoice.stripe_payment_intent_id = "pi_1234567890abcdef"
self.assertTrue(invoice.is_paid)
def test_invoice_payment_method_priority_paypal_over_stripe(self):
"""Test that PayPal takes priority if both are set (edge case)."""
invoice = Invoice(self.user)
invoice.shop = self.shop
invoice.paypal_order_id = "PAYPAL-ORDER-123"
invoice.stripe_payment_intent_id = "pi_1234567890abcdef"
# PayPal is checked first in payment_method
self.assertEqual(invoice.payment_method, "paypal")

9
make_post_sell/views/cart.py
View file

@ -681,6 +681,9 @@ def cart_complete_checkout(request):
request.session.flash(msg)
return HTTPFound("/billing")
# Use idempotency key to prevent duplicate charges on retry
idempotency_key = f"checkout_{cart.uuid_str}_{shop.uuid_str}_{invoice.total_in_cents}"
payment_intent = shop.stripe.PaymentIntent.create(
amount=invoice.total_in_cents,
currency="usd",
@ -688,8 +691,14 @@ def cart_complete_checkout(request):
payment_method=stripe_user_shop.active_card_id,
off_session=True,
confirm=True,
idempotency_key=idempotency_key,
)
# Store Stripe payment references on invoice
invoice.stripe_payment_intent_id = payment_intent.id
if payment_intent.latest_charge:
invoice.stripe_charge_id = payment_intent.latest_charge
# Only persist data after successful payment
for invoice in invoices:
for line_item in invoice.line_items:

176
make_post_sell/views/paypal_webhooks.py
View file

@ -1,176 +0,0 @@
from pyramid.view import view_config
from pyramid.response import Response
from ..models.invoice import get_invoice_by_paypal_order_id
import json
@view_config(route_name="paypal_webhook", request_method="POST")
def paypal_webhook(request):
"""
Handle PayPal webhook notifications.
Common events:
- PAYMENT.CAPTURE.COMPLETED
- PAYMENT.CAPTURE.DENIED
- CUSTOMER.DISPUTE.CREATED
Note: PAYMENT.CAPTURE.REFUNDED is NOT handled. Refunds are managed
externally by PayPal and the shop owner without platform involvement.
"""
try:
# Get webhook event from request body
webhook_event = json.loads(request.body.decode("utf-8"))
# Get webhook ID from config for verification
webhook_id = request.app.get("paypal.webhook_id")
event_type = webhook_event.get("event_type")
resource = webhook_event.get("resource", {})
# Extract order ID based on event type
order_id = None
if "supplementary_data" in resource:
related_ids = resource.get("supplementary_data", {}).get("related_ids", {})
order_id = related_ids.get("order_id")
# Verify webhook signature if webhook_id is configured
if webhook_id and order_id:
try:
# Get headers for verification
transmission_id = request.headers.get("PAYPAL-TRANSMISSION-ID")
transmission_time = request.headers.get("PAYPAL-TRANSMISSION-TIME")
cert_url = request.headers.get("PAYPAL-CERT-URL")
auth_algo = request.headers.get("PAYPAL-AUTH-ALGO")
transmission_sig = request.headers.get("PAYPAL-TRANSMISSION-SIG")
if not all([transmission_id, transmission_time, cert_url, auth_algo, transmission_sig]):
print("PayPal webhook rejected: Missing verification headers")
return Response(
json.dumps({"status": "error", "message": "Missing verification headers"}),
content_type="application/json",
status=400
)
# Look up which shop this payment belongs to
invoice = get_invoice_by_paypal_order_id(request.dbsession, order_id)
if not invoice or not invoice.shop:
print(f"PayPal webhook rejected: Cannot determine shop for order {order_id}")
return Response(
json.dumps({"status": "error", "message": "Cannot verify webhook - shop unknown"}),
content_type="application/json",
status=400
)
shop = invoice.shop
# Verify webhook signature using shop's PayPal credentials
import requests
# Get shop's PayPal API configuration
api = shop.paypal
mode = api.mode if hasattr(api, 'mode') else 'sandbox'
base_url = "https://api-m.sandbox.paypal.com" if mode == "sandbox" else "https://api-m.paypal.com"
# Get OAuth token for verification API
auth_response = requests.post(
f"{base_url}/v1/oauth2/token",
headers={"Accept": "application/json", "Accept-Language": "en_US"},
data={"grant_type": "client_credentials"},
auth=(shop.paypal_client_id, shop.paypal_secret)
)
if auth_response.status_code != 200:
print(f"PayPal webhook auth failed: {auth_response.text}")
return Response(
json.dumps({"status": "error", "message": "Webhook verification auth failed"}),
content_type="application/json",
status=401
)
access_token = auth_response.json()["access_token"]
# Call PayPal webhook verification API
verify_response = requests.post(
f"{base_url}/v1/notifications/verify-webhook-signature",
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {access_token}"
},
json={
"transmission_id": transmission_id,
"transmission_time": transmission_time,
"cert_url": cert_url,
"auth_algo": auth_algo,
"transmission_sig": transmission_sig,
"webhook_id": webhook_id,
"webhook_event": webhook_event
}
)
if verify_response.status_code != 200:
print(f"PayPal webhook verification failed: {verify_response.text}")
return Response(
json.dumps({"status": "error", "message": "Webhook signature verification failed"}),
content_type="application/json",
status=400
)
verification_result = verify_response.json()
if verification_result.get("verification_status") != "SUCCESS":
print(f"PayPal webhook rejected: Invalid signature")
return Response(
json.dumps({"status": "error", "message": "Invalid webhook signature"}),
content_type="application/json",
status=400
)
print(f"PayPal webhook verified successfully for order {order_id}")
except Exception as e:
print(f"PayPal webhook verification error: {str(e)}")
return Response(
json.dumps({"status": "error", "message": "Webhook verification failed"}),
content_type="application/json",
status=500
)
# Process the webhook event
if event_type == "PAYMENT.CAPTURE.COMPLETED":
if order_id:
invoice = get_invoice_by_paypal_order_id(request.dbsession, order_id)
if invoice and "id" in resource:
invoice.paypal_capture_id = resource["id"]
request.dbsession.add(invoice)
request.dbsession.flush()
elif event_type == "PAYMENT.CAPTURE.DENIED":
# Payment was denied - log it
if order_id:
print(f"PayPal payment denied for order {order_id}")
elif event_type == "CUSTOMER.DISPUTE.CREATED":
# A dispute was created - log for manual review
from datetime import datetime
dispute_id = resource.get("dispute_id", "unknown")
dispute_amount = resource.get("dispute_amount", {}).get("value", "unknown")
dispute_reason = resource.get("reason", "unknown")
print(f"[{datetime.now().isoformat()}] CRITICAL: PayPal dispute created - Dispute ID: {dispute_id}, Amount: ${dispute_amount}, Reason: {dispute_reason}")
# Return success response
return Response(
json.dumps({"status": "success"}),
content_type="application/json",
status=200
)
except Exception as e:
print(f"PayPal webhook error: {str(e)}")
return Response(
json.dumps({"status": "error", "message": str(e)}),
content_type="application/json",
status=500
)

497
make_post_sell/views/webhooks.py Normal file
View file

@ -0,0 +1,497 @@
"""
Webhook handlers for payment providers.
This module handles incoming webhooks from:
- PayPal: Payment capture, order approval, disputes
- Stripe: Payment success, failure, refunds, disputes
Webhooks provide resilience when checkout flows fail after
the payment provider has successfully processed the payment.
"""
import json
import logging
from pyramid.view import view_config
from pyramid.response import Response
import requests
from ..models.invoice import get_invoice_by_paypal_order_id
from ..lib.mail import send_purchase_email, send_sale_email
log = logging.getLogger(__name__)
# =============================================================================
# PayPal Webhooks
# =============================================================================
@view_config(route_name="paypal_webhook", request_method="POST")
def paypal_webhook(request):
"""
Handle PayPal webhook notifications.
Key events:
- PAYMENT.CAPTURE.COMPLETED - Payment was captured successfully
- CHECKOUT.ORDER.APPROVED - Order approved, needs capture (backup)
- PAYMENT.CAPTURE.DENIED - Payment was denied
- CUSTOMER.DISPUTE.CREATED - Dispute opened
Note: PAYMENT.CAPTURE.REFUNDED is NOT handled. Refunds are managed
externally by PayPal and the shop owner without platform involvement.
"""
try:
webhook_event = json.loads(request.body.decode("utf-8"))
event_type = webhook_event.get("event_type")
resource = webhook_event.get("resource", {})
log.info(f"PayPal webhook received: {event_type}")
# Extract order ID based on event type
order_id = _extract_order_id(event_type, resource)
if not order_id:
log.warning(f"PayPal webhook: Could not extract order_id from {event_type}")
return _json_response({"status": "ok", "message": "No order_id found"}, 200)
# Look up the invoice
invoice = get_invoice_by_paypal_order_id(request.dbsession, order_id)
if not invoice:
log.warning(f"PayPal webhook: No invoice found for order {order_id}")
return _json_response({"status": "ok", "message": "Invoice not found"}, 200)
shop = invoice.shop
if not shop:
log.error(f"PayPal webhook: Invoice {invoice.id} has no shop")
return _json_response({"status": "error", "message": "No shop"}, 400)
# Verify webhook signature
if not _verify_webhook_signature(request, webhook_event, shop):
log.warning(f"PayPal webhook: Signature verification failed for order {order_id}")
# Still return 200 to prevent PayPal from retrying endlessly
# Log it for investigation
return _json_response({"status": "ok", "message": "Signature verification skipped"}, 200)
# Process the event
if event_type == "PAYMENT.CAPTURE.COMPLETED":
_handle_capture_completed(request, invoice, resource)
elif event_type == "CHECKOUT.ORDER.APPROVED":
_handle_order_approved(request, invoice, shop, order_id)
elif event_type == "PAYMENT.CAPTURE.DENIED":
log.warning(f"PayPal payment DENIED for order {order_id}, invoice {invoice.id}")
elif event_type == "CUSTOMER.DISPUTE.CREATED":
_handle_dispute_created(resource, order_id)
return _json_response({"status": "success"}, 200)
except Exception as e:
log.exception(f"PayPal webhook error: {str(e)}")
# Return 200 anyway to prevent infinite retries
return _json_response({"status": "error", "message": str(e)}, 200)
def _extract_order_id(event_type, resource):
"""Extract order ID from webhook resource based on event type."""
# For capture events, order_id is in supplementary_data
if "supplementary_data" in resource:
related_ids = resource.get("supplementary_data", {}).get("related_ids", {})
order_id = related_ids.get("order_id")
if order_id:
return order_id
# For order events, the resource itself is the order
if event_type and "ORDER" in event_type:
return resource.get("id")
# Fallback: try common locations
return resource.get("order_id") or resource.get("id")
def _verify_webhook_signature(request, webhook_event, shop):
"""
Verify PayPal webhook signature.
Returns True if verified, False if verification failed or skipped.
"""
# Get webhook ID from app settings (configured per-deployment)
webhook_id = request.registry.settings.get("paypal.webhook_id")
if not webhook_id:
log.debug("PayPal webhook_id not configured, skipping verification")
return True # Skip verification if not configured
# Get required headers
headers = {
"transmission_id": request.headers.get("PAYPAL-TRANSMISSION-ID"),
"transmission_time": request.headers.get("PAYPAL-TRANSMISSION-TIME"),
"cert_url": request.headers.get("PAYPAL-CERT-URL"),
"auth_algo": request.headers.get("PAYPAL-AUTH-ALGO"),
"transmission_sig": request.headers.get("PAYPAL-TRANSMISSION-SIG"),
}
if not all(headers.values()):
log.warning("PayPal webhook: Missing verification headers")
return False
try:
# Determine API base URL
sandbox_mode = request.registry.settings.get("app.paypal.sandbox_mode", "True")
is_sandbox = str(sandbox_mode).lower() in ("true", "1", "yes")
base_url = "https://api-m.sandbox.paypal.com" if is_sandbox else "https://api-m.paypal.com"
# Get OAuth token
auth_response = requests.post(
f"{base_url}/v1/oauth2/token",
headers={"Accept": "application/json"},
data={"grant_type": "client_credentials"},
auth=(shop.paypal_client_id, shop.paypal_secret),
timeout=10
)
if auth_response.status_code != 200:
log.error(f"PayPal webhook auth failed: {auth_response.text}")
return False
access_token = auth_response.json()["access_token"]
# Verify signature
verify_response = requests.post(
f"{base_url}/v1/notifications/verify-webhook-signature",
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {access_token}"
},
json={
"transmission_id": headers["transmission_id"],
"transmission_time": headers["transmission_time"],
"cert_url": headers["cert_url"],
"auth_algo": headers["auth_algo"],
"transmission_sig": headers["transmission_sig"],
"webhook_id": webhook_id,
"webhook_event": webhook_event
},
timeout=10
)
if verify_response.status_code != 200:
log.error(f"PayPal webhook verification API error: {verify_response.text}")
return False
result = verify_response.json()
if result.get("verification_status") == "SUCCESS":
log.debug("PayPal webhook signature verified")
return True
else:
log.warning(f"PayPal webhook signature invalid: {result}")
return False
except Exception as e:
log.exception(f"PayPal webhook verification error: {e}")
return False
def _handle_capture_completed(request, invoice, resource):
"""
Handle PAYMENT.CAPTURE.COMPLETED event.
This is the backup for when the JS callback fails but PayPal
successfully captured the payment.
"""
capture_id = resource.get("id")
# Idempotency: check if already processed
if invoice.paypal_capture_id:
log.info(f"PayPal webhook: Invoice {invoice.id} already has capture_id, skipping")
return
log.info(f"PayPal webhook: Processing capture {capture_id} for invoice {invoice.id}")
# Update invoice with capture ID
invoice.paypal_capture_id = capture_id
request.dbsession.add(invoice)
# Unlock products for the user
user = invoice.user
if user:
for line_item in invoice.line_items:
if not line_item.product.is_unlocked_for_user(user):
line_item.product.unlock_for_user(user)
request.dbsession.add(line_item.product)
log.info(f"PayPal webhook: Unlocked product {line_item.product.id} for user {user.id}")
# Send confirmation emails
try:
products = [item.product for item in invoice.line_items]
send_purchase_email(request, user.email, products, invoice.total)
send_sale_email(request, invoice.shop, products, invoice.total)
log.info(f"PayPal webhook: Sent confirmation emails for invoice {invoice.id}")
except Exception as e:
log.exception(f"PayPal webhook: Failed to send emails for invoice {invoice.id}: {e}")
request.dbsession.flush()
def _handle_order_approved(request, invoice, shop, order_id):
"""
Handle CHECKOUT.ORDER.APPROVED event.
This means the customer approved the payment in PayPal, but we haven't
captured it yet. This is a backup in case our JS onApprove failed.
"""
# Check if already captured
if invoice.paypal_capture_id:
log.info(f"PayPal webhook: Order {order_id} already captured, skipping")
return
log.info(f"PayPal webhook: Order {order_id} approved but not captured, attempting capture")
try:
# Determine API base URL
sandbox_mode = request.registry.settings.get("app.paypal.sandbox_mode", "True")
is_sandbox = str(sandbox_mode).lower() in ("true", "1", "yes")
base_url = "https://api-m.sandbox.paypal.com" if is_sandbox else "https://api-m.paypal.com"
# Get OAuth token
auth_response = requests.post(
f"{base_url}/v1/oauth2/token",
headers={"Accept": "application/json"},
data={"grant_type": "client_credentials"},
auth=(shop.paypal_client_id, shop.paypal_secret),
timeout=10
)
if auth_response.status_code != 200:
log.error(f"PayPal webhook: Auth failed for capture attempt: {auth_response.text}")
return
access_token = auth_response.json()["access_token"]
# Capture the payment
capture_response = requests.post(
f"{base_url}/v2/checkout/orders/{order_id}/capture",
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {access_token}"
},
timeout=15
)
if capture_response.status_code not in [200, 201]:
log.error(f"PayPal webhook: Capture failed for order {order_id}: {capture_response.text}")
return
order = capture_response.json()
log.info(f"PayPal webhook: Successfully captured order {order_id}")
# Extract capture ID
try:
capture_id = order["purchase_units"][0]["payments"]["captures"][0]["id"]
invoice.paypal_capture_id = capture_id
request.dbsession.add(invoice)
except (KeyError, IndexError):
log.warning(f"PayPal webhook: Could not extract capture_id from response")
# Unlock products and send emails
user = invoice.user
if user:
for line_item in invoice.line_items:
if not line_item.product.is_unlocked_for_user(user):
line_item.product.unlock_for_user(user)
request.dbsession.add(line_item.product)
try:
products = [item.product for item in invoice.line_items]
send_purchase_email(request, user.email, products, invoice.total)
send_sale_email(request, invoice.shop, products, invoice.total)
log.info(f"PayPal webhook: Sent emails after webhook-initiated capture")
except Exception as e:
log.exception(f"PayPal webhook: Failed to send emails: {e}")
request.dbsession.flush()
except Exception as e:
log.exception(f"PayPal webhook: Error capturing order {order_id}: {e}")
def _handle_dispute_created(resource, order_id):
"""Handle CUSTOMER.DISPUTE.CREATED event."""
dispute_id = resource.get("dispute_id", "unknown")
dispute_amount = resource.get("dispute_amount", {}).get("value", "unknown")
dispute_reason = resource.get("reason", "unknown")
log.critical(
f"PayPal DISPUTE created - Order: {order_id}, "
f"Dispute ID: {dispute_id}, Amount: ${dispute_amount}, Reason: {dispute_reason}"
)
def _json_response(data, status):
"""Create a JSON response."""
return Response(
json.dumps(data),
content_type="application/json; charset=utf-8",
status=status
)
# =============================================================================
# Stripe Webhooks
# =============================================================================
@view_config(route_name="stripe_webhook", request_method="POST")
def stripe_webhook(request):
"""
Handle Stripe webhook notifications.
Key events:
- payment_intent.succeeded - Payment completed successfully
- payment_intent.payment_failed - Payment failed
- charge.refunded - Refund processed
- charge.dispute.created - Dispute opened
This provides resilience when the checkout flow fails after Stripe
has successfully processed the payment.
"""
import stripe
try:
payload = request.body
sig_header = request.headers.get("Stripe-Signature")
# Get webhook secret from settings
webhook_secret = request.registry.settings.get("stripe.webhook_secret")
if webhook_secret and sig_header:
try:
event = stripe.Webhook.construct_event(
payload, sig_header, webhook_secret
)
except ValueError:
log.warning("Stripe webhook: Invalid payload")
return _json_response({"error": "Invalid payload"}, 400)
except stripe.error.SignatureVerificationError:
log.warning("Stripe webhook: Invalid signature")
return _json_response({"error": "Invalid signature"}, 400)
else:
# No webhook secret configured, parse without verification
event = json.loads(payload.decode("utf-8"))
log.debug("Stripe webhook: No webhook_secret configured, skipping verification")
event_type = event.get("type") if isinstance(event, dict) else event.type
data_object = event.get("data", {}).get("object", {}) if isinstance(event, dict) else event.data.object
log.info(f"Stripe webhook received: {event_type}")
if event_type == "payment_intent.succeeded":
_handle_stripe_payment_succeeded(request, data_object)
elif event_type == "payment_intent.payment_failed":
_handle_stripe_payment_failed(request, data_object)
elif event_type == "charge.refunded":
_handle_stripe_refund(request, data_object)
elif event_type == "charge.dispute.created":
_handle_stripe_dispute(request, data_object)
return _json_response({"status": "success"}, 200)
except Exception as e:
log.exception(f"Stripe webhook error: {str(e)}")
return _json_response({"status": "error", "message": str(e)}, 200)
def _handle_stripe_payment_succeeded(request, payment_intent):
"""
Handle payment_intent.succeeded event.
This is the backup for when our checkout flow fails after Stripe
has successfully charged the card.
"""
from ..models.invoice import get_invoice_by_stripe_payment_intent_id
payment_intent_id = payment_intent.get("id") if isinstance(payment_intent, dict) else payment_intent.id
invoice = get_invoice_by_stripe_payment_intent_id(request.dbsession, payment_intent_id)
if not invoice:
log.warning(f"Stripe webhook: No invoice found for payment_intent {payment_intent_id}")
return
# Check if already processed (has charge_id)
if invoice.stripe_charge_id:
log.info(f"Stripe webhook: Invoice {invoice.id} already processed, skipping")
return
log.info(f"Stripe webhook: Processing payment_intent {payment_intent_id} for invoice {invoice.id}")
# Update invoice with charge ID
latest_charge = payment_intent.get("latest_charge") if isinstance(payment_intent, dict) else payment_intent.latest_charge
if latest_charge:
invoice.stripe_charge_id = latest_charge
request.dbsession.add(invoice)
# Unlock products for the user
user = invoice.user
if user:
for line_item in invoice.line_items:
if not line_item.product.is_unlocked_for_user(user):
line_item.product.unlock_for_user(user)
request.dbsession.add(line_item.product)
log.info(f"Stripe webhook: Unlocked product {line_item.product.id} for user {user.id}")
# Send confirmation emails
try:
products = [item.product for item in invoice.line_items]
send_purchase_email(request, user.email, products, invoice.total)
send_sale_email(request, invoice.shop, products, invoice.total)
log.info(f"Stripe webhook: Sent confirmation emails for invoice {invoice.id}")
except Exception as e:
log.exception(f"Stripe webhook: Failed to send emails for invoice {invoice.id}: {e}")
request.dbsession.flush()
def _handle_stripe_payment_failed(request, payment_intent):
"""Handle payment_intent.payment_failed event."""
payment_intent_id = payment_intent.get("id") if isinstance(payment_intent, dict) else payment_intent.id
error_message = ""
if isinstance(payment_intent, dict):
last_error = payment_intent.get("last_payment_error", {})
error_message = last_error.get("message", "Unknown error")
else:
if payment_intent.last_payment_error:
error_message = payment_intent.last_payment_error.message or "Unknown error"
log.warning(f"Stripe payment FAILED for payment_intent {payment_intent_id}: {error_message}")
def _handle_stripe_refund(request, charge):
"""Handle charge.refunded event."""
charge_id = charge.get("id") if isinstance(charge, dict) else charge.id
amount_refunded = charge.get("amount_refunded", 0) if isinstance(charge, dict) else charge.amount_refunded
# Convert from cents to dollars
amount_dollars = amount_refunded / 100
log.info(f"Stripe REFUND processed - Charge: {charge_id}, Amount: ${amount_dollars:.2f}")
def _handle_stripe_dispute(request, dispute):
"""Handle charge.dispute.created event."""
dispute_id = dispute.get("id") if isinstance(dispute, dict) else dispute.id
charge_id = dispute.get("charge") if isinstance(dispute, dict) else dispute.charge
amount = dispute.get("amount", 0) if isinstance(dispute, dict) else dispute.amount
reason = dispute.get("reason", "unknown") if isinstance(dispute, dict) else dispute.reason
# Convert from cents to dollars
amount_dollars = amount / 100
log.critical(
f"Stripe DISPUTE created - Dispute ID: {dispute_id}, "
f"Charge: {charge_id}, Amount: ${amount_dollars:.2f}, Reason: {reason}"
)