""" Production webhook server using FastAPI + Gunicorn/Uvicorn Receives Square webhooks with signature verification """ import os import json import asyncio from typing import Dict, Any from fastapi import FastAPI, Request, Response, status from fastapi.responses import JSONResponse import dotenv # Load environment variables dotenv.load_dotenv() # Square webhook validation from square.utils.webhooks_helper import verify_signature # Configuration SQUARE_WEBHOOK_SIGNATURE_KEY = os.getenv("SANDBOX_SQUARE_WEBHOOK_INVENTORY_ID") SQUARE_WEBHOOK_URL = os.getenv("SQUARE_WEBHOOK_URL") # Debug: Print configuration (redact key) if SQUARE_WEBHOOK_SIGNATURE_KEY: print(f"[CONFIG] Signature key loaded: {SQUARE_WEBHOOK_SIGNATURE_KEY[:10]}...") else: print("[CONFIG] WARNING: SQUARE_WEBHOOK_SIGNATURE_KEY not set!") if SQUARE_WEBHOOK_URL: print(f"[CONFIG] Webhook URL: {SQUARE_WEBHOOK_URL}") else: print("[CONFIG] WARNING: SQUARE_WEBHOOK_URL not set!") # In-memory store for processed event IDs (idempotency) processed_event_ids = set() MAX_STORED_EVENT_IDS = 10000 # Create FastAPI app app = FastAPI(title="Studspider") @app.get("/") async def root(): """Landing page""" return { "service": "Studspider", "status": "running", "endpoints": {"webhook": "/hook (POST)", "health": "/health (GET)"}, } @app.get("/health") async def health_check(): """Health check endpoint for monitoring""" return { "status": "healthy", "signature_verification": ( "enabled" if SQUARE_WEBHOOK_SIGNATURE_KEY else "disabled" ), } @app.post("/hook") async def webhook_handler(request: Request): """ Handle Square webhook events - MUST respond within 2 seconds! Square sends POST requests with: - x-square-hmacsha256-signature header - JSON body with event data - event_id for idempotency """ try: # Log incoming request print(f"\n[WEBHOOK] Incoming request from {request.client.host}") print(f"[WEBHOOK] Method: {request.method}") print(f"[WEBHOOK] URL: {request.url}") print(f"[WEBHOOK] Headers: {dict(request.headers)}") # Read raw body for signature verification body = await request.body() body_str = body.decode("utf-8") print(f"[WEBHOOK] Body length: {len(body_str)} bytes") # Get signature header signature_header = request.headers.get("x-square-hmacsha256-signature", "") # Verify Square webhook signature (if configured) if SQUARE_WEBHOOK_SIGNATURE_KEY and SQUARE_WEBHOOK_URL: if not signature_header: print( f"[ERROR] Missing x-square-hmacsha256-signature header from {request.client.host}" ) return JSONResponse( status_code=status.HTTP_401_UNAUTHORIZED, content={"error": "Missing signature header"}, ) # Debug signature verification inputs print(f"[DEBUG] Signature header: {signature_header[:20]}...") print(f"[DEBUG] Signature key (first 10 chars): {SQUARE_WEBHOOK_SIGNATURE_KEY[:10]}...") print(f"[DEBUG] Notification URL: {SQUARE_WEBHOOK_URL}") print(f"[DEBUG] Body (first 100 chars): {body_str[:100]}...") try: is_valid = verify_signature( request_body=body_str, signature_header=signature_header, signature_key=SQUARE_WEBHOOK_SIGNATURE_KEY, notification_url=SQUARE_WEBHOOK_URL, ) if not is_valid: print( f"[ERROR] Invalid webhook signature from {request.client.host}" ) print(f"[ERROR] This usually means:") print(f"[ERROR] 1. SQUARE_WEBHOOK_URL doesn't match Square Dashboard exactly") print(f"[ERROR] 2. SQUARE_WEBHOOK_SIGNATURE_KEY is incorrect") print(f"[ERROR] 3. Request body was modified in transit") return JSONResponse( status_code=status.HTTP_401_UNAUTHORIZED, content={"error": "Invalid signature"}, ) print(f"[INFO] ✓ Webhook signature verified from {request.client.host}") except Exception as e: print(f"[ERROR] Signature verification exception: {e}") return JSONResponse( status_code=status.HTTP_401_UNAUTHORIZED, content={"error": f"Signature verification failed: {str(e)}"}, ) else: print( f"[WARN] Webhook signature verification DISABLED - configure SQUARE_WEBHOOK_SIGNATURE_KEY" ) # Parse JSON data try: data = json.loads(body_str) except json.JSONDecodeError as e: print(f"[ERROR] Invalid JSON from {request.client.host}: {e}") return JSONResponse( status_code=status.HTTP_400_BAD_REQUEST, content={"error": "Invalid JSON"}, ) # Check idempotency (event_id) event_id = data.get("event_id") if event_id: if event_id in processed_event_ids: print( f"[INFO] Duplicate event {event_id} - already processed (idempotency)" ) return JSONResponse( status_code=status.HTTP_200_OK, content={ "status": "success", "message": "Event already processed", "event_id": event_id, }, ) # Add to processed set processed_event_ids.add(event_id) # Prevent memory growth if len(processed_event_ids) > MAX_STORED_EVENT_IDS: processed_event_ids.pop() # Get event info for quick response webhook_type = data.get("type", "unknown") event_id_val = data.get("event_id", "none") merchant_id = data.get("merchant_id", "unknown") # RESPOND IMMEDIATELY - Square requires fast response (< 2 seconds) response_data = { "status": "success", "message": "Webhook received", "event_id": event_id_val, } # Process webhook in background (async task) asyncio.create_task( process_webhook( webhook_type, event_id_val, merchant_id, data, request.client.host ) ) # Return immediate response print(f"[INFO] 200 OK response sent for event {event_id_val}") return JSONResponse(status_code=status.HTTP_200_OK, content=response_data) except Exception as e: # Catch-all for any unexpected errors - still respond quickly print(f"[ERROR] Unexpected error in webhook handler: {e}") import traceback traceback.print_exc() return JSONResponse( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, content={"error": "Internal server error"}, ) # Return immediate response return JSONResponse(status_code=status.HTTP_200_OK, content=response_data) async def process_webhook( webhook_type: str, event_id: str, merchant_id: str, data: Dict[str, Any], client_ip: str, ): """ Process webhook data asynchronously after responding to Square This runs in the background and doesn't block the HTTP response """ try: print("\n" + "=" * 60) print(f"[WEBHOOK] Received at {asyncio.get_event_loop().time()}") print(f"[WEBHOOK] From: {client_ip}") print(f"[WEBHOOK] Type: {webhook_type}") print(f"[WEBHOOK] Event ID: {event_id}") print(f"[WEBHOOK] Merchant: {merchant_id}") print(f"[WEBHOOK] Full payload:") print(json.dumps(data, indent=2)) print("=" * 60 + "\n") # Process based on webhook type if webhook_type == "inventory.count.updated": print(f"[PROCESS] Processing inventory count update...") inventory_counts = ( data.get("data", {}).get("object", {}).get("inventory_counts", []) ) for count in inventory_counts: catalog_id = count.get("catalog_object_id") quantity = count.get("quantity") state = count.get("state") print( f"[PROCESS] Catalog: {catalog_id}, Qty: {quantity}, State: {state}" ) elif webhook_type == "catalog.version.updated": print(f"[PROCESS] Processing catalog version update...") # Add your catalog processing logic here else: print(f"[PROCESS] Webhook type '{webhook_type}' - no handler configured") print(f"[PROCESS] ✓ Webhook processing complete for {event_id}") except Exception as e: print(f"[ERROR] Failed to process webhook {event_id}: {e}") if __name__ == "__main__": # For development/testing - use uvicorn directly import uvicorn print("\n" + "=" * 60) print("StudSpider Webhook Server - Development Mode") print("=" * 60) print( f"Signature verification: {'ENABLED' if SQUARE_WEBHOOK_SIGNATURE_KEY else 'DISABLED'}" ) print(f"Webhook URL: {SQUARE_WEBHOOK_URL or 'NOT CONFIGURED'}") print("\nFor production, use Gunicorn:") print( " gunicorn webhook_server:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:7433" ) print("=" * 60 + "\n") uvicorn.run(app, host="0.0.0.0", port=7433, log_level="info")