diff --git a/.env.example b/.env.example index 4a76963..aa58bac 100644 --- a/.env.example +++ b/.env.example @@ -30,34 +30,33 @@ BRICKSET_API_KEY=your_brickset_api_key # SSL Certificate Configuration (Optional) # ============================================================================ -# NOTE: Only needed if running HTTPS directly (not recommended with nginx) -# For nginx reverse proxy setup, leave these empty +# NOTE: Only needed if running HTTPS directly in Python (not recommended) +# For production, use nginx reverse proxy with Let's Encrypt instead +# nginx handles SSL/TLS, your Python app runs HTTP internally -# Path to SSL certificate in PEM format +# Path to SSL certificate in PEM format (if running HTTPS directly) # SSL_CERT_PATH=/path/to/cert.pem -# Path to SSL private key in PEM format +# Path to SSL private key in PEM format (if running HTTPS directly) # SSL_KEY_PATH=/path/to/key.pem -# ============================================================================ -# Square Webhook Certificate (P7B Format) -# ============================================================================ - -# Path to P7B certificate file for Square webhook signature verification -# This file is UPLOADED to Square Dashboard, not used by your server -# Generate with: python convert_to_p7b.py -SQUARE_P7B_PATH=./cert.p7b - # ============================================================================ # Square Webhook Signature Verification # ============================================================================ # CRITICAL: Get these from Square Developer Dashboard → Webhooks -# Signature key used to verify webhook authenticity (prevents spoofing) +# 1. Go to https://developer.squareup.com/apps +# 2. Select your application +# 3. Click "Webhooks" section +# 4. Create webhook subscription (Notification URL + Events) +# 5. Copy the Signature Key that Square displays + +# Signature key from Square Dashboard (a string like "asdf1234") SQUARE_WEBHOOK_SIGNATURE_KEY=your_webhook_signature_key_from_square -# Your webhook notification URL (must match exactly what's in Square Dashboard) -# Example: https://studspider.sh/hook +# Your webhook notification URL (must match EXACTLY what's in Square Dashboard) +# This should be your public HTTPS URL (through nginx) +# Example: https://yourdomain.com/hook SQUARE_WEBHOOK_URL=https://yourdomain.com/hook # ============================================================================ diff --git a/src/get_sq_env.py b/src/get_sq_env.py index 25a8d9f..053b666 100644 --- a/src/get_sq_env.py +++ b/src/get_sq_env.py @@ -10,9 +10,10 @@ LEGO_CAT_ID = os.getenv("SQUARE_LEGO_CAT_ID") SSL_CERT_PATH = os.getenv("SSL_CERT_PATH") SSL_KEY_PATH = os.getenv("SSL_KEY_PATH") -# P7B certificate for Square webhook signature verification -# This is uploaded to Square Dashboard, not used by the server -SQUARE_P7B_PATH = os.getenv("SQUARE_P7B_PATH") +# Square Webhook Configuration +# Get signature key from Square Developer Dashboard → Webhooks +SQUARE_WEBHOOK_SIGNATURE_KEY = os.getenv("SQUARE_WEBHOOK_SIGNATURE_KEY") +SQUARE_WEBHOOK_URL = os.getenv("SQUARE_WEBHOOK_URL") if os.getenv("SQUARE_ENV") == "production": SQUARE_APP_ID = os.getenv("PROD_SQUARE_APP_ID") diff --git a/src/test_server.py b/src/test_server.py index cf8cb08..3835901 100644 --- a/src/test_server.py +++ b/src/test_server.py @@ -21,8 +21,9 @@ from time import time time = int(time()) import asquare as sq -import hmac -import hashlib + +# Square SDK webhook validation +from square.utilities.webhooks_helper import is_valid_webhook_event_signature # from asquareclass import AsyncSquareClient # type: ignore from abricksetapi import AsyncBricksetAPI @@ -91,46 +92,6 @@ processed_event_ids = set() MAX_STORED_EVENT_IDS = 10000 # Prevent memory growth -def verify_square_webhook_signature( - signature_header: str, body: bytes, signature_key: str, webhook_url: str -) -> bool: - """ - Verify Square webhook signature using HMAC-SHA-256 - - Args: - signature_header: Value from x-square-hmacsha256-signature header - body: Raw request body bytes - signature_key: Your webhook signature key from Square Dashboard - webhook_url: Your webhook notification URL - - Returns: - bool: True if signature is valid - """ - if not signature_key or not webhook_url: - logger.error("Square webhook signature key or URL not configured in .env") - return False - - # Concatenate webhook URL + request body - payload = webhook_url.encode("utf-8") + body - - # Generate HMAC-SHA-256 signature - computed_signature = hmac.new( - signature_key.encode("utf-8"), payload, hashlib.sha256 - ).digest() - - # Decode the signature from the header (base64) - import base64 - - try: - expected_signature = base64.b64decode(signature_header) - except Exception as e: - logger.error(f"Failed to decode signature header: {e}") - return False - - # Compare signatures using constant-time comparison - return hmac.compare_digest(computed_signature, expected_signature) - - async def hook_handler(request: web.Request) -> web.Response: """Handle incoming JSON webhook POSTs on /hook endpoint""" try: @@ -146,9 +107,11 @@ async def hook_handler(request: web.Request) -> web.Response: # Read raw body for signature verification body = await request.read() + body_str = body.decode("utf-8") + + # Verify Square webhook signature using Square SDK + signature_header = request.headers.get("x-square-hmacsha256-signature", "") - # Verify Square webhook signature - signature_header = request.headers.get("x-square-hmacsha256-signature") if SQUARE_WEBHOOK_SIGNATURE_KEY and SQUARE_WEBHOOK_URL: if not signature_header: logger.error( @@ -157,9 +120,15 @@ async def hook_handler(request: web.Request) -> web.Response: print(f"[@DEBUG] Missing x-square-hmacsha256-signature header") return web.Response(status=401, text="Unauthorized: Missing signature") - if not verify_square_webhook_signature( - signature_header, body, SQUARE_WEBHOOK_SIGNATURE_KEY, SQUARE_WEBHOOK_URL - ): + # Use Square SDK's official validation function + is_valid = is_valid_webhook_event_signature( + body_str, + signature_header, + SQUARE_WEBHOOK_SIGNATURE_KEY, + SQUARE_WEBHOOK_URL, + ) + + if not is_valid: logger.error(f"{current_time}\tInvalid signature from {request.remote}") print(f"[@DEBUG] Invalid webhook signature!") return web.Response(status=401, text="Unauthorized: Invalid signature") @@ -172,7 +141,7 @@ async def hook_handler(request: web.Request) -> web.Response: # Parse JSON data try: - data = json.loads(body.decode("utf-8")) + data = json.loads(body_str) except json.JSONDecodeError as e: print(f"[@DEBUG] Invalid JSON to /hook from {request.remote}: {str(e)}") logger.error(f"Invalid JSON to /hook from {request.remote}: {str(e)}") @@ -203,44 +172,53 @@ async def hook_handler(request: web.Request) -> web.Response: if len(processed_event_ids) > MAX_STORED_EVENT_IDS: processed_event_ids.pop() - # Log the received webhook with detailed output - print(f"\n{'='*60}") - print(f"[@DEBUG] WEBHOOK RECEIVED at {current_time}") - print(f"[@DEBUG] From: {request.remote}") - print(f"[@DEBUG] Headers: {dict(request.headers)}") - print(f"[@DEBUG] Data type: {data.get('type', 'unknown')}") - print(f"[@DEBUG] Event ID: {data.get('event_id', 'none')}") - print(f"[@DEBUG] Full payload:") - print(json.dumps(data, indent=2)) - print(f"{'='*60}\n") - - logger.info( - f"{current_time}\tWebhook received from {request.remote}: {json.dumps(data)}" - ) - - # Process the webhook data here - # TODO: Add your webhook processing logic + # Get event info for quick response webhook_type = data.get("type", "unknown") + event_id_val = data.get("event_id", "none") - if webhook_type == "inventory.count.updated": - print(f"[@DEBUG] Processing inventory count update...") - # Extract inventory data - inventory_counts = ( - data.get("data", {}).get("object", {}).get("inventory_counts", []) - ) - for count in inventory_counts: - print(f"[@DEBUG] Catalog Object: {count.get('catalog_object_id')}") - print(f"[@DEBUG] Quantity: {count.get('quantity')}") - print(f"[@DEBUG] State: {count.get('state')}") - + # RESPOND IMMEDIATELY - Square requires fast response (< 2 seconds) + # Don't do any processing before responding! response_data = { "status": "success", - "message": f"Webhook received: {webhook_type}", - "timestamp": current_time, - "event_id": data.get("event_id"), - "data_received": data, + "message": "Webhook received", + "event_id": event_id_val, } + # Log and process AFTER sending response (async background tasks) + async def log_and_process(): + # Log the received webhook with detailed output + print(f"\n{'='*60}") + print(f"[@DEBUG] WEBHOOK RECEIVED at {current_time}") + print(f"[@DEBUG] From: {request.remote}") + print(f"[@DEBUG] Headers: {dict(request.headers)}") + print(f"[@DEBUG] Data type: {webhook_type}") + print(f"[@DEBUG] Event ID: {event_id_val}") + print(f"[@DEBUG] Full payload:") + print(json.dumps(data, indent=2)) + print(f"{'='*60}\n") + + logger.info( + f"{current_time}\tWebhook received from {request.remote}: {json.dumps(data)}" + ) + + # Process the webhook data + if webhook_type == "inventory.count.updated": + print(f"[@DEBUG] Processing inventory count update...") + # Extract inventory data + inventory_counts = ( + data.get("data", {}).get("object", {}).get("inventory_counts", []) + ) + for count in inventory_counts: + print( + f"[@DEBUG] Catalog Object: {count.get('catalog_object_id')}" + ) + print(f"[@DEBUG] Quantity: {count.get('quantity')}") + print(f"[@DEBUG] State: {count.get('state')}") + + # Start background processing (don't wait for it) + asyncio.create_task(log_and_process()) + + # Return response immediately return web.json_response(response_data) except json.JSONDecodeError as e: @@ -346,8 +324,7 @@ Examples: Environment Variables (.env): SSL_CERT_PATH - Default SSL certificate path - SSL_KEY_PATH - Default SSL key path - SQUARE_P7B_PATH - P7B file to upload to Square Dashboard + SSL_KEY_PATH - Default SSL key path For more information, see doc/ """, @@ -417,23 +394,12 @@ if __name__ == "__main__": ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH) ssl_context.load_cert_chain(args.ssl_cert, args.ssl_key) print(f"✓ SSL enabled with cert: {args.ssl_cert}") - - # Show P7B info if configured - if get_sq_env.SQUARE_P7B_PATH: - print( - f"ℹ️ Square P7B certificate for upload: {get_sq_env.SQUARE_P7B_PATH}" - ) - if os.path.exists(get_sq_env.SQUARE_P7B_PATH): - print(f" ✓ P7B file exists") - else: - print(f" ⚠️ P7B file not found - run: python convert_to_p7b.py") except FileNotFoundError as e: print(f"ERROR: Certificate file not found: {e}") exit(1) except ssl.SSLError as e: print(f"ERROR: Invalid certificate or key file: {e}") print("\nMake sure you're using PEM format files.") - print("Generate certificates with: python generate_cert.py") exit(1) elif args.ssl_cert or args.ssl_key: print("ERROR: Both --ssl-cert and --ssl-key must be provided together") @@ -443,20 +409,6 @@ if __name__ == "__main__": print(" For production with nginx reverse proxy, this is recommended.") print(" Nginx will handle HTTPS, forward HTTP to this server.") - # Show P7B info even without SSL (for nginx setup) - if get_sq_env.SQUARE_P7B_PATH: - print( - f"\nℹ️ Square P7B certificate configured: {get_sq_env.SQUARE_P7B_PATH}" - ) - if os.path.exists(get_sq_env.SQUARE_P7B_PATH): - print(f" ✓ P7B file exists - upload this to Square Dashboard") - print(f" Square Dashboard → Webhooks → Manage Notification Settings") - else: - print(f" ⚠️ P7B file not found") - print( - f" Generate with: python convert_to_p7b.py --cert /path/to/nginx/cert.pem" - ) - # If hook-port is specified, run a separate server for webhooks if args.hook_port: protocol_main = "https" if (ssl_context and not args.hook_ssl_only) else "http" diff --git a/test_square_webhook.py b/test_square_webhook.py new file mode 100644 index 0000000..575ef11 --- /dev/null +++ b/test_square_webhook.py @@ -0,0 +1,48 @@ +""" +Simple test script to verify Square webhook signature verification works correctly. + +This uses the same test data from Square's documentation to ensure our implementation +matches their expected behavior. +""" + +import sys + +sys.path.insert(0, "src") + +from square.utilities.webhooks_helper import is_valid_webhook_event_signature + +# Test data from Square's documentation +TEST_BODY = '{"hello":"world"}' +TEST_SIGNATURE = "2kRE5qRU2tR+tBGlDwMEw2avJ7QM4ikPYD/PJ3bd9Og=" +TEST_SIGNATURE_KEY = "asdf1234" +TEST_NOTIFICATION_URL = "https://example.com/webhook" + + +def test_signature_validation(): + """Test that signature validation works with Square SDK""" + + print("Testing Square webhook signature validation...") + print(f"Body: {TEST_BODY}") + print(f"Signature: {TEST_SIGNATURE}") + print(f"Key: {TEST_SIGNATURE_KEY}") + print(f"URL: {TEST_NOTIFICATION_URL}") + print() + + # This should return True with Square's test data + is_valid = is_valid_webhook_event_signature( + TEST_BODY, TEST_SIGNATURE, TEST_SIGNATURE_KEY, TEST_NOTIFICATION_URL + ) + + if is_valid: + print("✅ SUCCESS: Signature validation passed!") + print("The webhook implementation is using Square SDK correctly.") + return True + else: + print("❌ FAILURE: Signature validation failed!") + print("Something is wrong with the signature verification.") + return False + + +if __name__ == "__main__": + success = test_signature_validation() + sys.exit(0 if success else 1)