Billing Webhooks

import hmac
import hashlib

def verify_webhook_signature(request, secret):
  timestamp = request.headers.get('X-Meshpay-Timestamp')
  signature = request.headers.get('X-Meshpay-Signature')
  body = request.get_data(as_text=True)  # Raw body
  
  # Compute expected signature
  payload = f"{timestamp}.{body}"
  expected_signature = hmac.new(
      secret.encode(),
      payload.encode(),
      hashlib.sha256
  ).hexdigest()
  
  # Compare signatures (timing-safe)
  return hmac.compare_digest(signature, expected_signature)

# Usage in Flask
@app.route('/webhooks/meshpay', methods=['POST'])
def handle_webhook():
  is_valid = verify_webhook_signature(
      request, 
      os.environ['MESHPAY_WEBHOOK_SECRET']
  )
  
  if not is_valid:
      return jsonify({'error': 'Invalid signature'}), 401
  
  event = request.json
  # Process event...
  
  return jsonify({'received': True}), 200

from flask import Flask, request, jsonify
import hmac
import hashlib
import os
from datetime import datetime

app = Flask(__name__)

@app.route('/webhooks/meshpay', methods=['POST'])
def handle_webhook():
  # 1. Verify signature
  timestamp = request.headers.get('X-Meshpay-Timestamp')
  signature = request.headers.get('X-Meshpay-Signature')
  event_id = request.headers.get('X-Meshpay-Event-Id')
  body = request.get_data(as_text=True)
  
  expected_signature = hmac.new(
      os.environ['MESHPAY_WEBHOOK_SECRET'].encode(),
      f"{timestamp}.{body}".encode(),
      hashlib.sha256
  ).hexdigest()
  
  if not hmac.compare_digest(signature, expected_signature):
      print('Invalid webhook signature')
      return jsonify({'error': 'Invalid signature'}), 401
  
  # 2. Parse event
  data = request.json
  event_type = data.get('event')
  event_data = data.get('data')
  
  # 3. Check idempotency
  existing_event = db.webhook_events.find_one({'event_id': event_id})
  if existing_event:
      print(f'Event {event_id} already processed')
      return jsonify({'status': 'already_processed'}), 200
  
  # 4. Process event based on type
  try:
      if event_type == 'billing.transaction.succeeded':
          handle_payment_succeeded(event_data)
      elif event_type == 'billing.transaction.failed':
          handle_payment_failed(event_data)
      else:
          print(f'Unhandled event type: {event_type}')
      
      # 5. Mark event as processed
      db.webhook_events.insert_one({
          'event_id': event_id,
          'event_type': event_type,
          'processed_at': datetime.utcnow(),
      })
      
      return jsonify({'status': 'processed'}), 200
  except Exception as e:
      print(f'Error processing webhook: {e}')
      return jsonify({'error': 'Processing failed'}), 500


def handle_payment_succeeded(data):
  print(f"Payment succeeded: {data['id']}")
  
  # Grant entitlement to user
  if data.get('customer_ref') and data.get('resource_ref'):
      db.entitlements.update_one(
          {
              'user_id': data['customer_ref'],
              'resource_id': data['resource_ref'],
          },
          {
              '$set': {
                  'transaction_id': data['id'],
                  'amount': data['amount'],
                  'currency': data['currency'],
                  'granted_at': datetime.utcnow(),
              }
          },
          upsert=True
      )
  
  # Send confirmation email
  if data.get('customer_ref'):
      send_payment_confirmation_email(data['customer_ref'], data)


def handle_payment_failed(data):
  print(f"Payment failed: {data['id']}")
  
  # Log the failure
  db.payment_failures.insert_one({
      'transaction_id': data['id'],
      'customer_ref': data.get('customer_ref'),
      'resource_ref': data.get('resource_ref'),
      'amount': data['amount'],
      'currency': data['currency'],
      'failed_at': datetime.utcnow(),
  })