Initial commit
This commit is contained in:
559
skills/billing-automation/SKILL.md
Normal file
559
skills/billing-automation/SKILL.md
Normal 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
467
skills/paypal-integration/SKILL.md
Normal 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¤cy=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
466
skills/pci-compliance/SKILL.md
Normal 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
442
skills/stripe-integration/SKILL.md
Normal 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
|
||||
Reference in New Issue
Block a user