zhongwei/gh-hermeticormus-alqvimia-contador-plugins-payment-processing
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit

Browse Source
This commit is contained in:
Zhongwei Li
2025-11-29 18:34:57 +08:00
commit 544aff54bf
8 changed files with 2073 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
.claude-plugin/plugin.json Normal file
View File

@@ -0,0 +1,18 @@
{
"name": "payment-processing",
"description": "Payment gateway integration with Stripe, PayPal, checkout flow implementation, subscription billing, and PCI compliance",
"version": "1.2.1",
"author": {
"name": "Seth Hobson",
"url": "https://github.com/wshobson"
},
"skills": [
"./skills/billing-automation",
"./skills/paypal-integration",
"./skills/pci-compliance",
"./skills/stripe-integration"
],
"agents": [
"./agents/payment-integration.md"
]
}

3
README.md Normal file
View File

@@ -0,0 +1,3 @@
# payment-processing
Payment gateway integration with Stripe, PayPal, checkout flow implementation, subscription billing, and PCI compliance

57
agents/payment-integration.md Normal file
View File

@@ -0,0 +1,57 @@
---
name: payment-integration
description: Integrate Stripe, PayPal, and payment processors. Handles checkout flows, subscriptions, webhooks, and PCI compliance. Use PROACTIVELY when implementing payments, billing, or subscription features.
model: haiku
---
You are a payment integration specialist focused on secure, reliable payment processing.
## Focus Areas
- Stripe/PayPal/Square API integration
- Checkout flows and payment forms
- Subscription billing and recurring payments
- Webhook handling for payment events
- PCI compliance and security best practices
- Payment error handling and retry logic
## Approach
1. Security first - never log sensitive card data
2. Implement idempotency for all payment operations
3. Handle all edge cases (failed payments, disputes, refunds)
4. Test mode first, with clear migration path to production
5. Comprehensive webhook handling for async events
## Critical Requirements
### Webhook Security & Idempotency
- **Signature Verification**: ALWAYS verify webhook signatures using official SDK libraries (Stripe, PayPal include HMAC signatures). Never process unverified webhooks.
- **Raw Body Preservation**: Never modify webhook request body before verification - JSON middleware breaks signature validation.
- **Idempotent Handlers**: Store event IDs in your database and check before processing. Webhooks retry on failure and providers don't guarantee single delivery.
- **Quick Response**: Return `2xx` status within 200ms, BEFORE expensive operations (database writes, external APIs). Timeouts trigger retries and duplicate processing.
- **Server Validation**: Re-fetch payment status from provider API. Never trust webhook payload or client response alone.
### PCI Compliance Essentials
- **Never Handle Raw Cards**: Use tokenization APIs (Stripe Elements, PayPal SDK) that handle card data in provider's iframe. NEVER store, process, or transmit raw card numbers.
- **Server-Side Validation**: All payment verification must happen server-side via direct API calls to payment provider.
- **Environment Separation**: Test credentials must fail in production. Misconfigured gateways commonly accept test cards on live sites.
## Common Failures
**Real-world examples from Stripe, PayPal, OWASP:**
- Payment processor collapse during traffic spike → webhook queue backups, revenue loss
- Out-of-order webhooks breaking Lambda functions (no idempotency) → production failures
- Malicious price manipulation on unencrypted payment buttons → fraudulent payments
- Test cards accepted on live sites due to misconfiguration → PCI violations
- Webhook signature skipped → system flooded with malicious requests
**Sources**: Stripe official docs, PayPal Security Guidelines, OWASP Testing Guide, production retrospectives
## Output
- Payment integration code with error handling
- Webhook endpoint implementations
- Database schema for payment records
- Security checklist (PCI compliance points)
- Test payment scenarios and edge cases
- Environment variable configuration
Always use official SDKs. Include both server-side and client-side code where needed.

61
plugin.lock.json Normal file
View File

@@ -0,0 +1,61 @@
{
"$schema": "internal://schemas/plugin.lock.v1.json",
"pluginId": "gh:HermeticOrmus/Alqvimia-Contador:plugins/payment-processing",
"normalized": {
"repo": null,
"ref": "refs/tags/v20251128.0",
"commit": "c4322aace44273b7e4705283fc6ba88725429732",
"treeHash": "366e46374ac52f7bd556b066d74307faf0f60166d0b51a6d576a7b5eaa68669b",
"generatedAt": "2025-11-28T10:10:43.472972Z",
"toolVersion": "publish_plugins.py@0.2.0"
},
"origin": {
"remote": "git@github.com:zhongweili/42plugin-data.git",
"branch": "master",
"commit": "aa1497ed0949fd50e99e70d6324a29c5b34f9390",
"repoRoot": "/Users/zhongweili/projects/openmind/42plugin-data"
},
"manifest": {
"name": "payment-processing",
"description": "Payment gateway integration with Stripe, PayPal, checkout flow implementation, subscription billing, and PCI compliance",
"version": "1.2.1"
},
"content": {
"files": [
{
"path": "README.md",
"sha256": "cda54985e1dbaa2160e85d1ac5d0e21c6c35cc84acfce7076cb0a1a4ef323b8b"
},
{
"path": "agents/payment-integration.md",
"sha256": "5563fbb97a75ac6ce276ac4423fa083acb33d3cc0bffdcb0dc3872c36af6d58f"
},
{
"path": ".claude-plugin/plugin.json",
"sha256": "e3d60cf176c13e9a45665685ee08a83f4b000d89c396a9e4f362a4575fd328e4"
},
{
"path": "skills/stripe-integration/SKILL.md",
"sha256": "c424fb2edf330858cc28b6416abdc09eb1233a0cff98231fa917fba401c3fe6b"
},
{
"path": "skills/billing-automation/SKILL.md",
"sha256": "d8cfd13532ef4573539de5b3404c5cb794e9988c96f54c158de37da91ff91caf"
},
{
"path": "skills/paypal-integration/SKILL.md",
"sha256": "88802f7ef05d07ff15f8e8e0d2a37714599dc7629aac1d21e4ecba06cdbd6ce0"
},
{
"path": "skills/pci-compliance/SKILL.md",
"sha256": "9c50102176f3b0108358299aa7de9168aec84ab1c1fb88b7beaf12ef58e78ee4"
}
],
"dirSha256": "366e46374ac52f7bd556b066d74307faf0f60166d0b51a6d576a7b5eaa68669b"
},
"security": {
"scannedAt": null,
"scannerVersion": null,
"flags": []
}
}

559
skills/billing-automation/SKILL.md Normal file
View File

@@ -0,0 +1,559 @@
---
name: billing-automation
description: Build automated billing systems for recurring payments, invoicing, subscription lifecycle, and dunning management. Use when implementing subscription billing, automating invoicing, or managing recurring payment systems.
---
# Billing Automation
Master automated billing systems including recurring billing, invoice generation, dunning management, proration, and tax calculation.
## When to Use This Skill
- Implementing SaaS subscription billing
- Automating invoice generation and delivery
- Managing failed payment recovery (dunning)
- Calculating prorated charges for plan changes
- Handling sales tax, VAT, and GST
- Processing usage-based billing
- Managing billing cycles and renewals
## Core Concepts
### 1. Billing Cycles
**Common Intervals:**
- Monthly (most common for SaaS)
- Annual (discounted long-term)
- Quarterly
- Weekly
- Custom (usage-based, per-seat)
### 2. Subscription States
```
trial → active → past_due → canceled
→ paused → resumed
```
### 3. Dunning Management
Automated process to recover failed payments through:
- Retry schedules
- Customer notifications
- Grace periods
- Account restrictions
### 4. Proration
Adjusting charges when:
- Upgrading/downgrading mid-cycle
- Adding/removing seats
- Changing billing frequency
## Quick Start
```python
from billing import BillingEngine, Subscription
# Initialize billing engine
billing = BillingEngine()
# Create subscription
subscription = billing.create_subscription(
customer_id="cus_123",
plan_id="plan_pro_monthly",
billing_cycle_anchor=datetime.now(),
trial_days=14
)
# Process billing cycle
billing.process_billing_cycle(subscription.id)
```
## Subscription Lifecycle Management
```python
from datetime import datetime, timedelta
from enum import Enum
class SubscriptionStatus(Enum):
TRIAL = "trial"
ACTIVE = "active"
PAST_DUE = "past_due"
CANCELED = "canceled"
PAUSED = "paused"
class Subscription:
def __init__(self, customer_id, plan, billing_cycle_day=None):
self.id = generate_id()
self.customer_id = customer_id
self.plan = plan
self.status = SubscriptionStatus.TRIAL
self.current_period_start = datetime.now()
self.current_period_end = self.current_period_start + timedelta(days=plan.trial_days or 30)
self.billing_cycle_day = billing_cycle_day or self.current_period_start.day
self.trial_end = datetime.now() + timedelta(days=plan.trial_days) if plan.trial_days else None
def start_trial(self, trial_days):
"""Start trial period."""
self.status = SubscriptionStatus.TRIAL
self.trial_end = datetime.now() + timedelta(days=trial_days)
self.current_period_end = self.trial_end
def activate(self):
"""Activate subscription after trial or immediately."""
self.status = SubscriptionStatus.ACTIVE
self.current_period_start = datetime.now()
self.current_period_end = self.calculate_next_billing_date()
def mark_past_due(self):
"""Mark subscription as past due after failed payment."""
self.status = SubscriptionStatus.PAST_DUE
# Trigger dunning workflow
def cancel(self, at_period_end=True):
"""Cancel subscription."""
if at_period_end:
self.cancel_at_period_end = True
# Will cancel when current period ends
else:
self.status = SubscriptionStatus.CANCELED
self.canceled_at = datetime.now()
def calculate_next_billing_date(self):
"""Calculate next billing date based on interval."""
if self.plan.interval == 'month':
return self.current_period_start + timedelta(days=30)
elif self.plan.interval == 'year':
return self.current_period_start + timedelta(days=365)
elif self.plan.interval == 'week':
return self.current_period_start + timedelta(days=7)
```
## Billing Cycle Processing
```python
class BillingEngine:
def process_billing_cycle(self, subscription_id):
"""Process billing for a subscription."""
subscription = self.get_subscription(subscription_id)
# Check if billing is due
if datetime.now() < subscription.current_period_end:
return
# Generate invoice
invoice = self.generate_invoice(subscription)
# Attempt payment
payment_result = self.charge_customer(
subscription.customer_id,
invoice.total
)
if payment_result.success:
# Payment successful
invoice.mark_paid()
subscription.advance_billing_period()
self.send_invoice(invoice)
else:
# Payment failed
subscription.mark_past_due()
self.start_dunning_process(subscription, invoice)
def generate_invoice(self, subscription):
"""Generate invoice for billing period."""
invoice = Invoice(
customer_id=subscription.customer_id,
subscription_id=subscription.id,
period_start=subscription.current_period_start,
period_end=subscription.current_period_end
)
# Add subscription line item
invoice.add_line_item(
description=subscription.plan.name,
amount=subscription.plan.amount,
quantity=subscription.quantity or 1
)
# Add usage-based charges if applicable
if subscription.has_usage_billing:
usage_charges = self.calculate_usage_charges(subscription)
invoice.add_line_item(
description="Usage charges",
amount=usage_charges
)
# Calculate tax
tax = self.calculate_tax(invoice.subtotal, subscription.customer)
invoice.tax = tax
invoice.finalize()
return invoice
def charge_customer(self, customer_id, amount):
"""Charge customer using saved payment method."""
customer = self.get_customer(customer_id)
try:
# Charge using payment processor
charge = stripe.Charge.create(
customer=customer.stripe_id,
amount=int(amount * 100), # Convert to cents
currency='usd'
)
return PaymentResult(success=True, transaction_id=charge.id)
except stripe.error.CardError as e:
return PaymentResult(success=False, error=str(e))
```
## Dunning Management
```python
class DunningManager:
"""Manage failed payment recovery."""
def __init__(self):
self.retry_schedule = [
{'days': 3, 'email_template': 'payment_failed_first'},
{'days': 7, 'email_template': 'payment_failed_reminder'},
{'days': 14, 'email_template': 'payment_failed_final'}
]
def start_dunning_process(self, subscription, invoice):
"""Start dunning process for failed payment."""
dunning_attempt = DunningAttempt(
subscription_id=subscription.id,
invoice_id=invoice.id,
attempt_number=1,
next_retry=datetime.now() + timedelta(days=3)
)
# Send initial failure notification
self.send_dunning_email(subscription, 'payment_failed_first')
# Schedule retries
self.schedule_retries(dunning_attempt)
def retry_payment(self, dunning_attempt):
"""Retry failed payment."""
subscription = self.get_subscription(dunning_attempt.subscription_id)
invoice = self.get_invoice(dunning_attempt.invoice_id)
# Attempt payment again
result = self.charge_customer(subscription.customer_id, invoice.total)
if result.success:
# Payment succeeded
invoice.mark_paid()
subscription.status = SubscriptionStatus.ACTIVE
self.send_dunning_email(subscription, 'payment_recovered')
dunning_attempt.mark_resolved()
else:
# Still failing
dunning_attempt.attempt_number += 1
if dunning_attempt.attempt_number < len(self.retry_schedule):
# Schedule next retry
next_retry_config = self.retry_schedule[dunning_attempt.attempt_number]
dunning_attempt.next_retry = datetime.now() + timedelta(days=next_retry_config['days'])
self.send_dunning_email(subscription, next_retry_config['email_template'])
else:
# Exhausted retries, cancel subscription
subscription.cancel(at_period_end=False)
self.send_dunning_email(subscription, 'subscription_canceled')
def send_dunning_email(self, subscription, template):
"""Send dunning notification to customer."""
customer = self.get_customer(subscription.customer_id)
email_content = self.render_template(template, {
'customer_name': customer.name,
'amount_due': subscription.plan.amount,
'update_payment_url': f"https://app.example.com/billing"
})
send_email(
to=customer.email,
subject=email_content['subject'],
body=email_content['body']
)
```
## Proration
```python
class ProrationCalculator:
"""Calculate prorated charges for plan changes."""
@staticmethod
def calculate_proration(old_plan, new_plan, period_start, period_end, change_date):
"""Calculate proration for plan change."""
# Days in current period
total_days = (period_end - period_start).days
# Days used on old plan
days_used = (change_date - period_start).days
# Days remaining on new plan
days_remaining = (period_end - change_date).days
# Calculate prorated amounts
unused_amount = (old_plan.amount / total_days) * days_remaining
new_plan_amount = (new_plan.amount / total_days) * days_remaining
# Net charge/credit
proration = new_plan_amount - unused_amount
return {
'old_plan_credit': -unused_amount,
'new_plan_charge': new_plan_amount,
'net_proration': proration,
'days_used': days_used,
'days_remaining': days_remaining
}
@staticmethod
def calculate_seat_proration(current_seats, new_seats, price_per_seat, period_start, period_end, change_date):
"""Calculate proration for seat changes."""
total_days = (period_end - period_start).days
days_remaining = (period_end - change_date).days
# Additional seats charge
additional_seats = new_seats - current_seats
prorated_amount = (additional_seats * price_per_seat / total_days) * days_remaining
return {
'additional_seats': additional_seats,
'prorated_charge': max(0, prorated_amount), # No refund for removing seats mid-cycle
'effective_date': change_date
}
```
## Tax Calculation
```python
class TaxCalculator:
"""Calculate sales tax, VAT, GST."""
def __init__(self):
# Tax rates by region
self.tax_rates = {
'US_CA': 0.0725, # California sales tax
'US_NY': 0.04, # New York sales tax
'GB': 0.20, # UK VAT
'DE': 0.19, # Germany VAT
'FR': 0.20, # France VAT
'AU': 0.10, # Australia GST
}
def calculate_tax(self, amount, customer):
"""Calculate applicable tax."""
# Determine tax jurisdiction
jurisdiction = self.get_tax_jurisdiction(customer)
if not jurisdiction:
return 0
# Get tax rate
tax_rate = self.tax_rates.get(jurisdiction, 0)
# Calculate tax
tax = amount * tax_rate
return {
'tax_amount': tax,
'tax_rate': tax_rate,
'jurisdiction': jurisdiction,
'tax_type': self.get_tax_type(jurisdiction)
}
def get_tax_jurisdiction(self, customer):
"""Determine tax jurisdiction based on customer location."""
if customer.country == 'US':
# US: Tax based on customer state
return f"US_{customer.state}"
elif customer.country in ['GB', 'DE', 'FR']:
# EU: VAT
return customer.country
elif customer.country == 'AU':
# Australia: GST
return 'AU'
else:
return None
def get_tax_type(self, jurisdiction):
"""Get type of tax for jurisdiction."""
if jurisdiction.startswith('US_'):
return 'Sales Tax'
elif jurisdiction in ['GB', 'DE', 'FR']:
return 'VAT'
elif jurisdiction == 'AU':
return 'GST'
return 'Tax'
def validate_vat_number(self, vat_number, country):
"""Validate EU VAT number."""
# Use VIES API for validation
# Returns True if valid, False otherwise
pass
```
## Invoice Generation
```python
class Invoice:
def __init__(self, customer_id, subscription_id=None):
self.id = generate_invoice_number()
self.customer_id = customer_id
self.subscription_id = subscription_id
self.status = 'draft'
self.line_items = []
self.subtotal = 0
self.tax = 0
self.total = 0
self.created_at = datetime.now()
def add_line_item(self, description, amount, quantity=1):
"""Add line item to invoice."""
line_item = {
'description': description,
'unit_amount': amount,
'quantity': quantity,
'total': amount * quantity
}
self.line_items.append(line_item)
self.subtotal += line_item['total']
def finalize(self):
"""Finalize invoice and calculate total."""
self.total = self.subtotal + self.tax
self.status = 'open'
self.finalized_at = datetime.now()
def mark_paid(self):
"""Mark invoice as paid."""
self.status = 'paid'
self.paid_at = datetime.now()
def to_pdf(self):
"""Generate PDF invoice."""
from reportlab.pdfgen import canvas
# Generate PDF
# Include: company info, customer info, line items, tax, total
pass
def to_html(self):
"""Generate HTML invoice."""
template = """
<!DOCTYPE html>
<html>
<head><title>Invoice #{invoice_number}</title></head>
<body>
<h1>Invoice #{invoice_number}</h1>
<p>Date: {date}</p>
<h2>Bill To:</h2>
<p>{customer_name}<br>{customer_address}</p>
<table>
<tr><th>Description</th><th>Quantity</th><th>Amount</th></tr>
{line_items}
</table>
<p>Subtotal: ${subtotal}</p>
<p>Tax: ${tax}</p>
<h3>Total: ${total}</h3>
</body>
</html>
"""
return template.format(
invoice_number=self.id,
date=self.created_at.strftime('%Y-%m-%d'),
customer_name=self.customer.name,
customer_address=self.customer.address,
line_items=self.render_line_items(),
subtotal=self.subtotal,
tax=self.tax,
total=self.total
)
```
## Usage-Based Billing
```python
class UsageBillingEngine:
"""Track and bill for usage."""
def track_usage(self, customer_id, metric, quantity):
"""Track usage event."""
UsageRecord.create(
customer_id=customer_id,
metric=metric,
quantity=quantity,
timestamp=datetime.now()
)
def calculate_usage_charges(self, subscription, period_start, period_end):
"""Calculate charges for usage in billing period."""
usage_records = UsageRecord.get_for_period(
subscription.customer_id,
period_start,
period_end
)
total_usage = sum(record.quantity for record in usage_records)
# Tiered pricing
if subscription.plan.pricing_model == 'tiered':
charge = self.calculate_tiered_pricing(total_usage, subscription.plan.tiers)
# Per-unit pricing
elif subscription.plan.pricing_model == 'per_unit':
charge = total_usage * subscription.plan.unit_price
# Volume pricing
elif subscription.plan.pricing_model == 'volume':
charge = self.calculate_volume_pricing(total_usage, subscription.plan.tiers)
return charge
def calculate_tiered_pricing(self, total_usage, tiers):
"""Calculate cost using tiered pricing."""
charge = 0
remaining = total_usage
for tier in sorted(tiers, key=lambda x: x['up_to']):
tier_usage = min(remaining, tier['up_to'] - tier['from'])
charge += tier_usage * tier['unit_price']
remaining -= tier_usage
if remaining <= 0:
break
return charge
```
## Resources
- **references/billing-cycles.md**: Billing cycle management
- **references/dunning-management.md**: Failed payment recovery
- **references/proration.md**: Prorated charge calculations
- **references/tax-calculation.md**: Tax/VAT/GST handling
- **references/invoice-lifecycle.md**: Invoice state management
- **assets/billing-state-machine.yaml**: Billing workflow
- **assets/invoice-template.html**: Invoice templates
- **assets/dunning-policy.yaml**: Dunning configuration
## Best Practices
1. **Automate Everything**: Minimize manual intervention
2. **Clear Communication**: Notify customers of billing events
3. **Flexible Retry Logic**: Balance recovery with customer experience
4. **Accurate Proration**: Fair calculation for plan changes
5. **Tax Compliance**: Calculate correct tax for jurisdiction
6. **Audit Trail**: Log all billing events
7. **Graceful Degradation**: Handle edge cases without breaking
## Common Pitfalls
- **Incorrect Proration**: Not accounting for partial periods
- **Missing Tax**: Forgetting to add tax to invoices
- **Aggressive Dunning**: Canceling too quickly
- **No Notifications**: Not informing customers of failures
- **Hardcoded Cycles**: Not supporting custom billing dates

467
skills/paypal-integration/SKILL.md Normal file
View File

@@ -0,0 +1,467 @@
---
name: paypal-integration
description: Integrate PayPal payment processing with support for express checkout, subscriptions, and refund management. Use when implementing PayPal payments, processing online transactions, or building e-commerce checkout flows.
---
# PayPal Integration
Master PayPal payment integration including Express Checkout, IPN handling, recurring billing, and refund workflows.
## When to Use This Skill
- Integrating PayPal as a payment option
- Implementing express checkout flows
- Setting up recurring billing with PayPal
- Processing refunds and payment disputes
- Handling PayPal webhooks (IPN)
- Supporting international payments
- Implementing PayPal subscriptions
## Core Concepts
### 1. Payment Products
**PayPal Checkout**
- One-time payments
- Express checkout experience
- Guest and PayPal account payments
**PayPal Subscriptions**
- Recurring billing
- Subscription plans
- Automatic renewals
**PayPal Payouts**
- Send money to multiple recipients
- Marketplace and platform payments
### 2. Integration Methods
**Client-Side (JavaScript SDK)**
- Smart Payment Buttons
- Hosted payment flow
- Minimal backend code
**Server-Side (REST API)**
- Full control over payment flow
- Custom checkout UI
- Advanced features
### 3. IPN (Instant Payment Notification)
- Webhook-like payment notifications
- Asynchronous payment updates
- Verification required
## Quick Start
```javascript
// Frontend - PayPal Smart Buttons
<div id="paypal-button-container"></div>
<script src="https://www.paypal.com/sdk/js?client-id=YOUR_CLIENT_ID&currency=USD"></script>
<script>
paypal.Buttons({
createOrder: function(data, actions) {
return actions.order.create({
purchase_units: [{
amount: {
value: '25.00'
}
}]
});
},
onApprove: function(data, actions) {
return actions.order.capture().then(function(details) {
// Payment successful
console.log('Transaction completed by ' + details.payer.name.given_name);
// Send to backend for verification
fetch('/api/paypal/capture', {
method: 'POST',
headers: {'Content-Type': 'application/json'},
body: JSON.stringify({orderID: data.orderID})
});
});
}
}).render('#paypal-button-container');
</script>
```
```python
# Backend - Verify and capture order
from paypalrestsdk import Payment
import paypalrestsdk
paypalrestsdk.configure({
"mode": "sandbox", # or "live"
"client_id": "YOUR_CLIENT_ID",
"client_secret": "YOUR_CLIENT_SECRET"
})
def capture_paypal_order(order_id):
"""Capture a PayPal order."""
payment = Payment.find(order_id)
if payment.execute({"payer_id": payment.payer.payer_info.payer_id}):
# Payment successful
return {
'status': 'success',
'transaction_id': payment.id,
'amount': payment.transactions[0].amount.total
}
else:
# Payment failed
return {
'status': 'failed',
'error': payment.error
}
```
## Express Checkout Implementation
### Server-Side Order Creation
```python
import requests
import json
class PayPalClient:
def __init__(self, client_id, client_secret, mode='sandbox'):
self.client_id = client_id
self.client_secret = client_secret
self.base_url = 'https://api-m.sandbox.paypal.com' if mode == 'sandbox' else 'https://api-m.paypal.com'
self.access_token = self.get_access_token()
def get_access_token(self):
"""Get OAuth access token."""
url = f"{self.base_url}/v1/oauth2/token"
headers = {"Accept": "application/json", "Accept-Language": "en_US"}
response = requests.post(
url,
headers=headers,
data={"grant_type": "client_credentials"},
auth=(self.client_id, self.client_secret)
)
return response.json()['access_token']
def create_order(self, amount, currency='USD'):
"""Create a PayPal order."""
url = f"{self.base_url}/v2/checkout/orders"
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {self.access_token}"
}
payload = {
"intent": "CAPTURE",
"purchase_units": [{
"amount": {
"currency_code": currency,
"value": str(amount)
}
}]
}
response = requests.post(url, headers=headers, json=payload)
return response.json()
def capture_order(self, order_id):
"""Capture payment for an order."""
url = f"{self.base_url}/v2/checkout/orders/{order_id}/capture"
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {self.access_token}"
}
response = requests.post(url, headers=headers)
return response.json()
def get_order_details(self, order_id):
"""Get order details."""
url = f"{self.base_url}/v2/checkout/orders/{order_id}"
headers = {
"Authorization": f"Bearer {self.access_token}"
}
response = requests.get(url, headers=headers)
return response.json()
```
## IPN (Instant Payment Notification) Handling
### IPN Verification and Processing
```python
from flask import Flask, request
import requests
from urllib.parse import parse_qs
app = Flask(__name__)
@app.route('/ipn', methods=['POST'])
def handle_ipn():
"""Handle PayPal IPN notifications."""
# Get IPN message
ipn_data = request.form.to_dict()
# Verify IPN with PayPal
if not verify_ipn(ipn_data):
return 'IPN verification failed', 400
# Process IPN based on transaction type
payment_status = ipn_data.get('payment_status')
txn_type = ipn_data.get('txn_type')
if payment_status == 'Completed':
handle_payment_completed(ipn_data)
elif payment_status == 'Refunded':
handle_refund(ipn_data)
elif payment_status == 'Reversed':
handle_chargeback(ipn_data)
return 'IPN processed', 200
def verify_ipn(ipn_data):
"""Verify IPN message authenticity."""
# Add 'cmd' parameter
verify_data = ipn_data.copy()
verify_data['cmd'] = '_notify-validate'
# Send back to PayPal for verification
paypal_url = 'https://ipnpb.sandbox.paypal.com/cgi-bin/webscr' # or production URL
response = requests.post(paypal_url, data=verify_data)
return response.text == 'VERIFIED'
def handle_payment_completed(ipn_data):
"""Process completed payment."""
txn_id = ipn_data.get('txn_id')
payer_email = ipn_data.get('payer_email')
mc_gross = ipn_data.get('mc_gross')
item_name = ipn_data.get('item_name')
# Check if already processed (prevent duplicates)
if is_transaction_processed(txn_id):
return
# Update database
# Send confirmation email
# Fulfill order
print(f"Payment completed: {txn_id}, Amount: ${mc_gross}")
def handle_refund(ipn_data):
"""Handle refund."""
parent_txn_id = ipn_data.get('parent_txn_id')
mc_gross = ipn_data.get('mc_gross')
# Process refund in your system
print(f"Refund processed: {parent_txn_id}, Amount: ${mc_gross}")
def handle_chargeback(ipn_data):
"""Handle payment reversal/chargeback."""
txn_id = ipn_data.get('txn_id')
reason_code = ipn_data.get('reason_code')
# Handle chargeback
print(f"Chargeback: {txn_id}, Reason: {reason_code}")
```
## Subscription/Recurring Billing
### Create Subscription Plan
```python
def create_subscription_plan(name, amount, interval='MONTH'):
"""Create a subscription plan."""
client = PayPalClient(CLIENT_ID, CLIENT_SECRET)
url = f"{client.base_url}/v1/billing/plans"
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {client.access_token}"
}
payload = {
"product_id": "PRODUCT_ID", # Create product first
"name": name,
"billing_cycles": [{
"frequency": {
"interval_unit": interval,
"interval_count": 1
},
"tenure_type": "REGULAR",
"sequence": 1,
"total_cycles": 0, # Infinite
"pricing_scheme": {
"fixed_price": {
"value": str(amount),
"currency_code": "USD"
}
}
}],
"payment_preferences": {
"auto_bill_outstanding": True,
"setup_fee": {
"value": "0",
"currency_code": "USD"
},
"setup_fee_failure_action": "CONTINUE",
"payment_failure_threshold": 3
}
}
response = requests.post(url, headers=headers, json=payload)
return response.json()
def create_subscription(plan_id, subscriber_email):
"""Create a subscription for a customer."""
client = PayPalClient(CLIENT_ID, CLIENT_SECRET)
url = f"{client.base_url}/v1/billing/subscriptions"
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {client.access_token}"
}
payload = {
"plan_id": plan_id,
"subscriber": {
"email_address": subscriber_email
},
"application_context": {
"return_url": "https://yourdomain.com/subscription/success",
"cancel_url": "https://yourdomain.com/subscription/cancel"
}
}
response = requests.post(url, headers=headers, json=payload)
subscription = response.json()
# Get approval URL
for link in subscription.get('links', []):
if link['rel'] == 'approve':
return {
'subscription_id': subscription['id'],
'approval_url': link['href']
}
```
## Refund Workflows
```python
def create_refund(capture_id, amount=None, note=None):
"""Create a refund for a captured payment."""
client = PayPalClient(CLIENT_ID, CLIENT_SECRET)
url = f"{client.base_url}/v2/payments/captures/{capture_id}/refund"
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {client.access_token}"
}
payload = {}
if amount:
payload["amount"] = {
"value": str(amount),
"currency_code": "USD"
}
if note:
payload["note_to_payer"] = note
response = requests.post(url, headers=headers, json=payload)
return response.json()
def get_refund_details(refund_id):
"""Get refund details."""
client = PayPalClient(CLIENT_ID, CLIENT_SECRET)
url = f"{client.base_url}/v2/payments/refunds/{refund_id}"
headers = {
"Authorization": f"Bearer {client.access_token}"
}
response = requests.get(url, headers=headers)
return response.json()
```
## Error Handling
```python
class PayPalError(Exception):
"""Custom PayPal error."""
pass
def handle_paypal_api_call(api_function):
"""Wrapper for PayPal API calls with error handling."""
try:
result = api_function()
return result
except requests.exceptions.RequestException as e:
# Network error
raise PayPalError(f"Network error: {str(e)}")
except Exception as e:
# Other errors
raise PayPalError(f"PayPal API error: {str(e)}")
# Usage
try:
order = handle_paypal_api_call(lambda: client.create_order(25.00))
except PayPalError as e:
# Handle error appropriately
log_error(e)
```
## Testing
```python
# Use sandbox credentials
SANDBOX_CLIENT_ID = "..."
SANDBOX_SECRET = "..."
# Test accounts
# Create test buyer and seller accounts at developer.paypal.com
def test_payment_flow():
"""Test complete payment flow."""
client = PayPalClient(SANDBOX_CLIENT_ID, SANDBOX_SECRET, mode='sandbox')
# Create order
order = client.create_order(10.00)
assert 'id' in order
# Get approval URL
approval_url = next((link['href'] for link in order['links'] if link['rel'] == 'approve'), None)
assert approval_url is not None
# After approval (manual step with test account)
# Capture order
# captured = client.capture_order(order['id'])
# assert captured['status'] == 'COMPLETED'
```
## Resources
- **references/express-checkout.md**: Express Checkout implementation guide
- **references/ipn-handling.md**: IPN verification and processing
- **references/refund-workflows.md**: Refund handling patterns
- **references/billing-agreements.md**: Recurring billing setup
- **assets/paypal-client.py**: Production PayPal client
- **assets/ipn-processor.py**: IPN webhook processor
- **assets/recurring-billing.py**: Subscription management
## Best Practices
1. **Always Verify IPN**: Never trust IPN without verification
2. **Idempotent Processing**: Handle duplicate IPN notifications
3. **Error Handling**: Implement robust error handling
4. **Logging**: Log all transactions and errors
5. **Test Thoroughly**: Use sandbox extensively
6. **Webhook Backup**: Don't rely solely on client-side callbacks
7. **Currency Handling**: Always specify currency explicitly
## Common Pitfalls
- **Not Verifying IPN**: Accepting IPN without verification
- **Duplicate Processing**: Not checking for duplicate transactions
- **Wrong Environment**: Mixing sandbox and production URLs/credentials
- **Missing Webhooks**: Not handling all payment states
- **Hardcoded Values**: Not making configurable for different environments

466
skills/pci-compliance/SKILL.md Normal file
View File

@@ -0,0 +1,466 @@
---
name: pci-compliance
description: Implement PCI DSS compliance requirements for secure handling of payment card data and payment systems. Use when securing payment processing, achieving PCI compliance, or implementing payment card security measures.
---
# PCI Compliance
Master PCI DSS (Payment Card Industry Data Security Standard) compliance for secure payment processing and handling of cardholder data.
## When to Use This Skill
- Building payment processing systems
- Handling credit card information
- Implementing secure payment flows
- Conducting PCI compliance audits
- Reducing PCI compliance scope
- Implementing tokenization and encryption
- Preparing for PCI DSS assessments
## PCI DSS Requirements (12 Core Requirements)
### Build and Maintain Secure Network
1. Install and maintain firewall configuration
2. Don't use vendor-supplied defaults for passwords
### Protect Cardholder Data
3. Protect stored cardholder data
4. Encrypt transmission of cardholder data across public networks
### Maintain Vulnerability Management
5. Protect systems against malware
6. Develop and maintain secure systems and applications
### Implement Strong Access Control
7. Restrict access to cardholder data by business need-to-know
8. Identify and authenticate access to system components
9. Restrict physical access to cardholder data
### Monitor and Test Networks
10. Track and monitor all access to network resources and cardholder data
11. Regularly test security systems and processes
### Maintain Information Security Policy
12. Maintain a policy that addresses information security
## Compliance Levels
**Level 1**: > 6 million transactions/year (annual ROC required)
**Level 2**: 1-6 million transactions/year (annual SAQ)
**Level 3**: 20,000-1 million e-commerce transactions/year
**Level 4**: < 20,000 e-commerce or < 1 million total transactions
## Data Minimization (Never Store)
```python
# NEVER STORE THESE
PROHIBITED_DATA = {
'full_track_data': 'Magnetic stripe data',
'cvv': 'Card verification code/value',
'pin': 'PIN or PIN block'
}
# CAN STORE (if encrypted)
ALLOWED_DATA = {
'pan': 'Primary Account Number (card number)',
'cardholder_name': 'Name on card',
'expiration_date': 'Card expiration',
'service_code': 'Service code'
}
class PaymentData:
"""Safe payment data handling."""
def __init__(self):
self.prohibited_fields = ['cvv', 'cvv2', 'cvc', 'pin']
def sanitize_log(self, data):
"""Remove sensitive data from logs."""
sanitized = data.copy()
# Mask PAN
if 'card_number' in sanitized:
card = sanitized['card_number']
sanitized['card_number'] = f"{card[:6]}{'*' * (len(card) - 10)}{card[-4:]}"
# Remove prohibited data
for field in self.prohibited_fields:
sanitized.pop(field, None)
return sanitized
def validate_no_prohibited_storage(self, data):
"""Ensure no prohibited data is being stored."""
for field in self.prohibited_fields:
if field in data:
raise SecurityError(f"Attempting to store prohibited field: {field}")
```
## Tokenization
### Using Payment Processor Tokens
```python
import stripe
class TokenizedPayment:
"""Handle payments using tokens (no card data on server)."""
@staticmethod
def create_payment_method_token(card_details):
"""Create token from card details (client-side only)."""
# THIS SHOULD ONLY BE DONE CLIENT-SIDE WITH STRIPE.JS
# NEVER send card details to your server
"""
// Frontend JavaScript
const stripe = Stripe('pk_...');
const {token, error} = await stripe.createToken({
card: {
number: '4242424242424242',
exp_month: 12,
exp_year: 2024,
cvc: '123'
}
});
// Send token.id to server (NOT card details)
"""
pass
@staticmethod
def charge_with_token(token_id, amount):
"""Charge using token (server-side)."""
# Your server only sees the token, never the card number
stripe.api_key = "sk_..."
charge = stripe.Charge.create(
amount=amount,
currency="usd",
source=token_id, # Token instead of card details
description="Payment"
)
return charge
@staticmethod
def store_payment_method(customer_id, payment_method_token):
"""Store payment method as token for future use."""
stripe.Customer.modify(
customer_id,
source=payment_method_token
)
# Store only customer_id and payment_method_id in your database
# NEVER store actual card details
return {
'customer_id': customer_id,
'has_payment_method': True
# DO NOT store: card number, CVV, etc.
}
```
### Custom Tokenization (Advanced)
```python
import secrets
from cryptography.fernet import Fernet
class TokenVault:
"""Secure token vault for card data (if you must store it)."""
def __init__(self, encryption_key):
self.cipher = Fernet(encryption_key)
self.vault = {} # In production: use encrypted database
def tokenize(self, card_data):
"""Convert card data to token."""
# Generate secure random token
token = secrets.token_urlsafe(32)
# Encrypt card data
encrypted = self.cipher.encrypt(json.dumps(card_data).encode())
# Store token -> encrypted data mapping
self.vault[token] = encrypted
return token
def detokenize(self, token):
"""Retrieve card data from token."""
encrypted = self.vault.get(token)
if not encrypted:
raise ValueError("Token not found")
# Decrypt
decrypted = self.cipher.decrypt(encrypted)
return json.loads(decrypted.decode())
def delete_token(self, token):
"""Remove token from vault."""
self.vault.pop(token, None)
```
## Encryption
### Data at Rest
```python
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
import os
class EncryptedStorage:
"""Encrypt data at rest using AES-256-GCM."""
def __init__(self, encryption_key):
"""Initialize with 256-bit key."""
self.key = encryption_key # Must be 32 bytes
def encrypt(self, plaintext):
"""Encrypt data."""
# Generate random nonce
nonce = os.urandom(12)
# Encrypt
aesgcm = AESGCM(self.key)
ciphertext = aesgcm.encrypt(nonce, plaintext.encode(), None)
# Return nonce + ciphertext
return nonce + ciphertext
def decrypt(self, encrypted_data):
"""Decrypt data."""
# Extract nonce and ciphertext
nonce = encrypted_data[:12]
ciphertext = encrypted_data[12:]
# Decrypt
aesgcm = AESGCM(self.key)
plaintext = aesgcm.decrypt(nonce, ciphertext, None)
return plaintext.decode()
# Usage
storage = EncryptedStorage(os.urandom(32))
encrypted_pan = storage.encrypt("4242424242424242")
# Store encrypted_pan in database
```
### Data in Transit
```python
# Always use TLS 1.2 or higher
# Flask/Django example
app.config['SESSION_COOKIE_SECURE'] = True # HTTPS only
app.config['SESSION_COOKIE_HTTPONLY'] = True
app.config['SESSION_COOKIE_SAMESITE'] = 'Strict'
# Enforce HTTPS
from flask_talisman import Talisman
Talisman(app, force_https=True)
```
## Access Control
```python
from functools import wraps
from flask import session
def require_pci_access(f):
"""Decorator to restrict access to cardholder data."""
@wraps(f)
def decorated_function(*args, **kwargs):
user = session.get('user')
# Check if user has PCI access role
if not user or 'pci_access' not in user.get('roles', []):
return {'error': 'Unauthorized access to cardholder data'}, 403
# Log access attempt
audit_log(
user=user['id'],
action='access_cardholder_data',
resource=f.__name__
)
return f(*args, **kwargs)
return decorated_function
@app.route('/api/payment-methods')
@require_pci_access
def get_payment_methods():
"""Retrieve payment methods (restricted access)."""
# Only accessible to users with pci_access role
pass
```
## Audit Logging
```python
import logging
from datetime import datetime
class PCIAuditLogger:
"""PCI-compliant audit logging."""
def __init__(self):
self.logger = logging.getLogger('pci_audit')
# Configure to write to secure, append-only log
def log_access(self, user_id, resource, action, result):
"""Log access to cardholder data."""
entry = {
'timestamp': datetime.utcnow().isoformat(),
'user_id': user_id,
'resource': resource,
'action': action,
'result': result,
'ip_address': request.remote_addr
}
self.logger.info(json.dumps(entry))
def log_authentication(self, user_id, success, method):
"""Log authentication attempt."""
entry = {
'timestamp': datetime.utcnow().isoformat(),
'user_id': user_id,
'event': 'authentication',
'success': success,
'method': method,
'ip_address': request.remote_addr
}
self.logger.info(json.dumps(entry))
# Usage
audit = PCIAuditLogger()
audit.log_access(user_id=123, resource='payment_methods', action='read', result='success')
```
## Security Best Practices
### Input Validation
```python
import re
def validate_card_number(card_number):
"""Validate card number format (Luhn algorithm)."""
# Remove spaces and dashes
card_number = re.sub(r'[\s-]', '', card_number)
# Check if all digits
if not card_number.isdigit():
return False
# Luhn algorithm
def luhn_checksum(card_num):
def digits_of(n):
return [int(d) for d in str(n)]
digits = digits_of(card_num)
odd_digits = digits[-1::-2]
even_digits = digits[-2::-2]
checksum = sum(odd_digits)
for d in even_digits:
checksum += sum(digits_of(d * 2))
return checksum % 10
return luhn_checksum(card_number) == 0
def sanitize_input(user_input):
"""Sanitize user input to prevent injection."""
# Remove special characters
# Validate against expected format
# Escape for database queries
pass
```
## PCI DSS SAQ (Self-Assessment Questionnaire)
### SAQ A (Least Requirements)
- E-commerce using hosted payment page
- No card data on your systems
- ~20 questions
### SAQ A-EP
- E-commerce with embedded payment form
- Uses JavaScript to handle card data
- ~180 questions
### SAQ D (Most Requirements)
- Store, process, or transmit card data
- Full PCI DSS requirements
- ~300 questions
## Compliance Checklist
```python
PCI_COMPLIANCE_CHECKLIST = {
'network_security': [
'Firewall configured and maintained',
'No vendor default passwords',
'Network segmentation implemented'
],
'data_protection': [
'No storage of CVV, track data, or PIN',
'PAN encrypted when stored',
'PAN masked when displayed',
'Encryption keys properly managed'
],
'vulnerability_management': [
'Anti-virus installed and updated',
'Secure development practices',
'Regular security patches',
'Vulnerability scanning performed'
],
'access_control': [
'Access restricted by role',
'Unique IDs for all users',
'Multi-factor authentication',
'Physical security measures'
],
'monitoring': [
'Audit logs enabled',
'Log review process',
'File integrity monitoring',
'Regular security testing'
],
'policy': [
'Security policy documented',
'Risk assessment performed',
'Security awareness training',
'Incident response plan'
]
}
```
## Resources
- **references/data-minimization.md**: Never store prohibited data
- **references/tokenization.md**: Tokenization strategies
- **references/encryption.md**: Encryption requirements
- **references/access-control.md**: Role-based access
- **references/audit-logging.md**: Comprehensive logging
- **assets/pci-compliance-checklist.md**: Complete checklist
- **assets/encrypted-storage.py**: Encryption utilities
- **scripts/audit-payment-system.sh**: Compliance audit script
## Common Violations
1. **Storing CVV**: Never store card verification codes
2. **Unencrypted PAN**: Card numbers must be encrypted at rest
3. **Weak Encryption**: Use AES-256 or equivalent
4. **No Access Controls**: Restrict who can access cardholder data
5. **Missing Audit Logs**: Must log all access to payment data
6. **Insecure Transmission**: Always use TLS 1.2+
7. **Default Passwords**: Change all default credentials
8. **No Security Testing**: Regular penetration testing required
## Reducing PCI Scope
1. **Use Hosted Payments**: Stripe Checkout, PayPal, etc.
2. **Tokenization**: Replace card data with tokens
3. **Network Segmentation**: Isolate cardholder data environment
4. **Outsource**: Use PCI-compliant payment processors
5. **No Storage**: Never store full card details
By minimizing systems that touch card data, you reduce compliance burden significantly.

442
skills/stripe-integration/SKILL.md Normal file
View File

@@ -0,0 +1,442 @@
---
name: stripe-integration
description: Implement Stripe payment processing for robust, PCI-compliant payment flows including checkout, subscriptions, and webhooks. Use when integrating Stripe payments, building subscription systems, or implementing secure checkout flows.
---
# Stripe Integration
Master Stripe payment processing integration for robust, PCI-compliant payment flows including checkout, subscriptions, webhooks, and refunds.
## When to Use This Skill
- Implementing payment processing in web/mobile applications
- Setting up subscription billing systems
- Handling one-time payments and recurring charges
- Processing refunds and disputes
- Managing customer payment methods
- Implementing SCA (Strong Customer Authentication) for European payments
- Building marketplace payment flows with Stripe Connect
## Core Concepts
### 1. Payment Flows
**Checkout Session (Hosted)**
- Stripe-hosted payment page
- Minimal PCI compliance burden
- Fastest implementation
- Supports one-time and recurring payments
**Payment Intents (Custom UI)**
- Full control over payment UI
- Requires Stripe.js for PCI compliance
- More complex implementation
- Better customization options
**Setup Intents (Save Payment Methods)**
- Collect payment method without charging
- Used for subscriptions and future payments
- Requires customer confirmation
### 2. Webhooks
**Critical Events:**
- `payment_intent.succeeded`: Payment completed
- `payment_intent.payment_failed`: Payment failed
- `customer.subscription.updated`: Subscription changed
- `customer.subscription.deleted`: Subscription canceled
- `charge.refunded`: Refund processed
- `invoice.payment_succeeded`: Subscription payment successful
### 3. Subscriptions
**Components:**
- **Product**: What you're selling
- **Price**: How much and how often
- **Subscription**: Customer's recurring payment
- **Invoice**: Generated for each billing cycle
### 4. Customer Management
- Create and manage customer records
- Store multiple payment methods
- Track customer metadata
- Manage billing details
## Quick Start
```python
import stripe
stripe.api_key = "sk_test_..."
# Create a checkout session
session = stripe.checkout.Session.create(
payment_method_types=['card'],
line_items=[{
'price_data': {
'currency': 'usd',
'product_data': {
'name': 'Premium Subscription',
},
'unit_amount': 2000, # $20.00
'recurring': {
'interval': 'month',
},
},
'quantity': 1,
}],
mode='subscription',
success_url='https://yourdomain.com/success?session_id={CHECKOUT_SESSION_ID}',
cancel_url='https://yourdomain.com/cancel',
)
# Redirect user to session.url
print(session.url)
```
## Payment Implementation Patterns
### Pattern 1: One-Time Payment (Hosted Checkout)
```python
def create_checkout_session(amount, currency='usd'):
"""Create a one-time payment checkout session."""
try:
session = stripe.checkout.Session.create(
payment_method_types=['card'],
line_items=[{
'price_data': {
'currency': currency,
'product_data': {
'name': 'Purchase',
'images': ['https://example.com/product.jpg'],
},
'unit_amount': amount, # Amount in cents
},
'quantity': 1,
}],
mode='payment',
success_url='https://yourdomain.com/success?session_id={CHECKOUT_SESSION_ID}',
cancel_url='https://yourdomain.com/cancel',
metadata={
'order_id': 'order_123',
'user_id': 'user_456'
}
)
return session
except stripe.error.StripeError as e:
# Handle error
print(f"Stripe error: {e.user_message}")
raise
```
### Pattern 2: Custom Payment Intent Flow
```python
def create_payment_intent(amount, currency='usd', customer_id=None):
"""Create a payment intent for custom checkout UI."""
intent = stripe.PaymentIntent.create(
amount=amount,
currency=currency,
customer=customer_id,
automatic_payment_methods={
'enabled': True,
},
metadata={
'integration_check': 'accept_a_payment'
}
)
return intent.client_secret # Send to frontend
# Frontend (JavaScript)
"""
const stripe = Stripe('pk_test_...');
const elements = stripe.elements();
const cardElement = elements.create('card');
cardElement.mount('#card-element');
const {error, paymentIntent} = await stripe.confirmCardPayment(
clientSecret,
{
payment_method: {
card: cardElement,
billing_details: {
name: 'Customer Name'
}
}
}
);
if (error) {
// Handle error
} else if (paymentIntent.status === 'succeeded') {
// Payment successful
}
"""
```
### Pattern 3: Subscription Creation
```python
def create_subscription(customer_id, price_id):
"""Create a subscription for a customer."""
try:
subscription = stripe.Subscription.create(
customer=customer_id,
items=[{'price': price_id}],
payment_behavior='default_incomplete',
payment_settings={'save_default_payment_method': 'on_subscription'},
expand=['latest_invoice.payment_intent'],
)
return {
'subscription_id': subscription.id,
'client_secret': subscription.latest_invoice.payment_intent.client_secret
}
except stripe.error.StripeError as e:
print(f"Subscription creation failed: {e}")
raise
```
### Pattern 4: Customer Portal
```python
def create_customer_portal_session(customer_id):
"""Create a portal session for customers to manage subscriptions."""
session = stripe.billing_portal.Session.create(
customer=customer_id,
return_url='https://yourdomain.com/account',
)
return session.url # Redirect customer here
```
## Webhook Handling
### Secure Webhook Endpoint
```python
from flask import Flask, request
import stripe
app = Flask(__name__)
endpoint_secret = 'whsec_...'
@app.route('/webhook', methods=['POST'])
def webhook():
payload = request.data
sig_header = request.headers.get('Stripe-Signature')
try:
event = stripe.Webhook.construct_event(
payload, sig_header, endpoint_secret
)
except ValueError:
# Invalid payload
return 'Invalid payload', 400
except stripe.error.SignatureVerificationError:
# Invalid signature
return 'Invalid signature', 400
# Handle the event
if event['type'] == 'payment_intent.succeeded':
payment_intent = event['data']['object']
handle_successful_payment(payment_intent)
elif event['type'] == 'payment_intent.payment_failed':
payment_intent = event['data']['object']
handle_failed_payment(payment_intent)
elif event['type'] == 'customer.subscription.deleted':
subscription = event['data']['object']
handle_subscription_canceled(subscription)
return 'Success', 200
def handle_successful_payment(payment_intent):
"""Process successful payment."""
customer_id = payment_intent.get('customer')
amount = payment_intent['amount']
metadata = payment_intent.get('metadata', {})
# Update your database
# Send confirmation email
# Fulfill order
print(f"Payment succeeded: {payment_intent['id']}")
def handle_failed_payment(payment_intent):
"""Handle failed payment."""
error = payment_intent.get('last_payment_error', {})
print(f"Payment failed: {error.get('message')}")
# Notify customer
# Update order status
def handle_subscription_canceled(subscription):
"""Handle subscription cancellation."""
customer_id = subscription['customer']
# Update user access
# Send cancellation email
print(f"Subscription canceled: {subscription['id']}")
```
### Webhook Best Practices
```python
import hashlib
import hmac
def verify_webhook_signature(payload, signature, secret):
"""Manually verify webhook signature."""
expected_sig = hmac.new(
secret.encode('utf-8'),
payload,
hashlib.sha256
).hexdigest()
return hmac.compare_digest(signature, expected_sig)
def handle_webhook_idempotently(event_id, handler):
"""Ensure webhook is processed exactly once."""
# Check if event already processed
if is_event_processed(event_id):
return
# Process event
try:
handler()
mark_event_processed(event_id)
except Exception as e:
log_error(e)
# Stripe will retry failed webhooks
raise
```
## Customer Management
```python
def create_customer(email, name, payment_method_id=None):
"""Create a Stripe customer."""
customer = stripe.Customer.create(
email=email,
name=name,
payment_method=payment_method_id,
invoice_settings={
'default_payment_method': payment_method_id
} if payment_method_id else None,
metadata={
'user_id': '12345'
}
)
return customer
def attach_payment_method(customer_id, payment_method_id):
"""Attach a payment method to a customer."""
stripe.PaymentMethod.attach(
payment_method_id,
customer=customer_id
)
# Set as default
stripe.Customer.modify(
customer_id,
invoice_settings={
'default_payment_method': payment_method_id
}
)
def list_customer_payment_methods(customer_id):
"""List all payment methods for a customer."""
payment_methods = stripe.PaymentMethod.list(
customer=customer_id,
type='card'
)
return payment_methods.data
```
## Refund Handling
```python
def create_refund(payment_intent_id, amount=None, reason=None):
"""Create a refund."""
refund_params = {
'payment_intent': payment_intent_id
}
if amount:
refund_params['amount'] = amount # Partial refund
if reason:
refund_params['reason'] = reason # 'duplicate', 'fraudulent', 'requested_by_customer'
refund = stripe.Refund.create(**refund_params)
return refund
def handle_dispute(charge_id, evidence):
"""Update dispute with evidence."""
stripe.Dispute.modify(
charge_id,
evidence={
'customer_name': evidence.get('customer_name'),
'customer_email_address': evidence.get('customer_email'),
'shipping_documentation': evidence.get('shipping_proof'),
'customer_communication': evidence.get('communication'),
}
)
```
## Testing
```python
# Use test mode keys
stripe.api_key = "sk_test_..."
# Test card numbers
TEST_CARDS = {
'success': '4242424242424242',
'declined': '4000000000000002',
'3d_secure': '4000002500003155',
'insufficient_funds': '4000000000009995'
}
def test_payment_flow():
"""Test complete payment flow."""
# Create test customer
customer = stripe.Customer.create(
email="test@example.com"
)
# Create payment intent
intent = stripe.PaymentIntent.create(
amount=1000,
currency='usd',
customer=customer.id,
payment_method_types=['card']
)
# Confirm with test card
confirmed = stripe.PaymentIntent.confirm(
intent.id,
payment_method='pm_card_visa' # Test payment method
)
assert confirmed.status == 'succeeded'
```
## Resources
- **references/checkout-flows.md**: Detailed checkout implementation
- **references/webhook-handling.md**: Webhook security and processing
- **references/subscription-management.md**: Subscription lifecycle
- **references/customer-management.md**: Customer and payment method handling
- **references/invoice-generation.md**: Invoicing and billing
- **assets/stripe-client.py**: Production-ready Stripe client wrapper
- **assets/webhook-handler.py**: Complete webhook processor
- **assets/checkout-config.json**: Checkout configuration templates
## Best Practices
1. **Always Use Webhooks**: Don't rely solely on client-side confirmation
2. **Idempotency**: Handle webhook events idempotently
3. **Error Handling**: Gracefully handle all Stripe errors
4. **Test Mode**: Thoroughly test with test keys before production
5. **Metadata**: Use metadata to link Stripe objects to your database
6. **Monitoring**: Track payment success rates and errors
7. **PCI Compliance**: Never handle raw card data on your server
8. **SCA Ready**: Implement 3D Secure for European payments
## Common Pitfalls
- **Not Verifying Webhooks**: Always verify webhook signatures
- **Missing Webhook Events**: Handle all relevant webhook events
- **Hardcoded Amounts**: Use cents/smallest currency unit
- **No Retry Logic**: Implement retries for API calls
- **Ignoring Test Mode**: Test all edge cases with test cards