ph0sgene/studspider
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix webhook code we hope

Browse Source
This commit is contained in:
C committed 2026-01-02 14:28:37 -08:00
1 parent a8ce535985
commit 4320b4cba9
4 files changed
+107 -107
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

.env.example
+15 -16
View File
@@ -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
# ============================================================================
src/get_sq_env.py
+4 -3
View File
@@ -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")
src/test_server.py
+40 -88
View File
@@ -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,13 +172,27 @@ async def hook_handler(request: web.Request) -> web.Response:
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")
# RESPOND IMMEDIATELY - Square requires fast response (< 2 seconds)
# Don't do any processing before responding!
response_data = {
"status": "success",
"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: {data.get('type', 'unknown')}")
print(f"[@DEBUG] Event ID: {data.get('event_id', 'none')}")
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")
@@ -218,10 +201,7 @@ async def hook_handler(request: web.Request) -> web.Response:
f"{current_time}\tWebhook received from {request.remote}: {json.dumps(data)}"
)
# Process the webhook data here
# TODO: Add your webhook processing logic
webhook_type = data.get("type", "unknown")
# Process the webhook data
if webhook_type == "inventory.count.updated":
print(f"[@DEBUG] Processing inventory count update...")
# Extract inventory data
@@ -229,18 +209,16 @@ async def hook_handler(request: web.Request) -> web.Response:
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] Catalog Object: {count.get('catalog_object_id')}"
)
print(f"[@DEBUG] Quantity: {count.get('quantity')}")
print(f"[@DEBUG] State: {count.get('state')}")
response_data = {
"status": "success",
"message": f"Webhook received: {webhook_type}",
"timestamp": current_time,
"event_id": data.get("event_id"),
"data_received": data,
}
# 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:
@@ -347,7 +325,6 @@ 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
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"
test_square_webhook.py
+48
View File
@@ -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)