439 lines
16 KiB
Python
439 lines
16 KiB
Python
import argparse
|
||
import asyncio
|
||
import json
|
||
import ssl
|
||
from aiohttp import web
|
||
from base64 import b64decode
|
||
from typing import Dict, Any
|
||
import uuid
|
||
|
||
import dotenv
|
||
import os
|
||
|
||
import re
|
||
import logging
|
||
from logging.handlers import RotatingFileHandler
|
||
from pathlib import Path
|
||
|
||
import banner
|
||
import get_sq_env
|
||
from time import time
|
||
|
||
time = int(time())
|
||
import asquare as sq
|
||
|
||
# Square SDK webhook validation
|
||
from square.utils.webhooks_helper import verify_signature
|
||
|
||
# from asquareclass import AsyncSquareClient # type: ignore
|
||
from abricksetapi import AsyncBricksetAPI
|
||
|
||
print("Loading environment variables...")
|
||
dotenv.load_dotenv()
|
||
SQUARE_LEGO_CAT_ID = os.environ.get("SQUARE_LEGO_CAT_ID")
|
||
SQUARE_WEBHOOK_SIGNATURE_KEY = os.environ.get("SQUARE_WEBHOOK_SIGNATURE_KEY")
|
||
SQUARE_WEBHOOK_URL = os.environ.get("SQUARE_WEBHOOK_URL")
|
||
|
||
from asyncio_throttle.throttler import Throttler
|
||
|
||
# set up logger
|
||
print("Setting up logger...")
|
||
logger = logging.getLogger("studspider")
|
||
logger.setLevel(logging.DEBUG)
|
||
handler = RotatingFileHandler("./logs/studspider.log", maxBytes=125000, backupCount=5)
|
||
logger.addHandler(handler)
|
||
|
||
lego = AsyncBricksetAPI(debug=True, logger=logger)
|
||
# sq = AsyncSquareClient(logger=logger)
|
||
# logging.basicConfig(
|
||
# filename="./logs/studspider.log", encoding="utf-8", level=logging.DEBUG
|
||
# )
|
||
print("Setting up rate limiter...")
|
||
print("Doing RE things...")
|
||
UA = re.compile(r"^kopling\/?", re.IGNORECASE)
|
||
|
||
|
||
async def run_update():
|
||
"""
|
||
DEPRECATED: Use glue.enrich_bulk() instead
|
||
|
||
Legacy RUNUPDATE command handler - kept for backwards compatibility
|
||
but now delegates to the new glue.enrich_bulk() function
|
||
"""
|
||
from src import glue
|
||
|
||
await glue.enrich_bulk()
|
||
|
||
|
||
def valid_uuid(val):
|
||
try:
|
||
uuid.UUID(str(val))
|
||
return True
|
||
except ValueError:
|
||
return False
|
||
|
||
|
||
# In-memory store for processed event IDs (idempotency)
|
||
processed_event_ids = set()
|
||
MAX_STORED_EVENT_IDS = 10000 # Prevent memory growth
|
||
|
||
|
||
async def hook_handler(request: web.Request) -> web.Response:
|
||
"""Handle incoming JSON webhook POSTs on /hook endpoint"""
|
||
try:
|
||
current_time = int(asyncio.get_event_loop().time())
|
||
|
||
# Verify request method
|
||
if request.method != "POST":
|
||
logger.error(
|
||
f"{current_time}\tInvalid request method to /hook from %s",
|
||
request.remote,
|
||
)
|
||
return web.Response(status=405, text="Method not allowed. Use POST.")
|
||
|
||
# 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", "")
|
||
|
||
if SQUARE_WEBHOOK_SIGNATURE_KEY and SQUARE_WEBHOOK_URL:
|
||
if not signature_header:
|
||
logger.error(
|
||
f"{current_time}\tMissing signature header from {request.remote}"
|
||
)
|
||
print(f"[@DEBUG] Missing x-square-hmacsha256-signature header")
|
||
return web.Response(status=401, text="Unauthorized: Missing signature")
|
||
|
||
# Use Square SDK's official validation function
|
||
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:
|
||
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")
|
||
|
||
print(f"[@DEBUG] ✓ Webhook signature verified")
|
||
else:
|
||
print(
|
||
f"[@DEBUG] ⚠️ Webhook signature verification DISABLED (set SQUARE_WEBHOOK_SIGNATURE_KEY and SQUARE_WEBHOOK_URL in .env)"
|
||
)
|
||
|
||
# Parse JSON data
|
||
try:
|
||
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)}")
|
||
return web.Response(status=400, text="Invalid JSON in request body")
|
||
|
||
# Check for idempotency (event_id)
|
||
event_id = data.get("event_id")
|
||
if event_id:
|
||
if event_id in processed_event_ids:
|
||
print(
|
||
f"[@DEBUG] Duplicate event {event_id} - already processed (idempotency)"
|
||
)
|
||
logger.info(
|
||
f"{current_time}\tDuplicate event {event_id} from {request.remote}"
|
||
)
|
||
return web.json_response(
|
||
{
|
||
"status": "success",
|
||
"message": "Event already processed (idempotency)",
|
||
"event_id": event_id,
|
||
}
|
||
)
|
||
|
||
# Add to processed set
|
||
processed_event_ids.add(event_id)
|
||
|
||
# Prevent memory growth - remove oldest if too many
|
||
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: {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:
|
||
error_msg = f"JSON decode error in webhook: {str(e)}"
|
||
print(f"[@DEBUG] {error_msg}")
|
||
logger.error(error_msg)
|
||
return web.Response(status=400, text=error_msg)
|
||
except Exception as e:
|
||
error_msg = f"Webhook handler error: {str(e)}"
|
||
print(f"[@DEBUG] {error_msg}")
|
||
print(f"[@DEBUG] Exception type: {type(e).__name__}")
|
||
import traceback
|
||
|
||
print(f"[@DEBUG] Traceback:\n{traceback.format_exc()}")
|
||
logger.error(f"{error_msg}\n{traceback.format_exc()}")
|
||
return web.Response(status=500, text="Internal server error")
|
||
|
||
|
||
async def request_handler(request: web.Request) -> web.Response:
|
||
"""Handle incoming POST requests with command processing"""
|
||
try:
|
||
# Verify request method
|
||
if request.method != "POST":
|
||
logger.error(f"{time}\tInvalid request method from %s at ", request.remote)
|
||
return web.Response(status=405, text="Method not allowed. Use POST.")
|
||
|
||
# Get JSON data from request
|
||
try:
|
||
data = await request.json()
|
||
except json.JSONDecodeError:
|
||
logger.error("Invalid JSON from %s", request.remote)
|
||
return web.Response(
|
||
status=400, text="{time()}\tInvalid JSON in request body"
|
||
)
|
||
# Validate Idempotency-Key header
|
||
if valid_uuid(request.headers.get("Idempotency-Key")) is False:
|
||
logger.error("{time()}\tInvalid Idempotency-Key from %s", request.remote)
|
||
return web.Response(status=412, text="412 Headers rejected!")
|
||
# Validate User-Agent header
|
||
if not re.match(UA, str(request.headers.get("User-Agent"))):
|
||
logger.error("{time()}\tInvalid User-Agent from %s", request.remote)
|
||
return web.Response(status=412, text="412 Headers rejected!")
|
||
# Validate command exists
|
||
if "CMD" not in data:
|
||
logger.error("{time()}\tMissing CMD field from %s", request.remote)
|
||
return web.Response(status=400, text="Missing CMD field in request")
|
||
|
||
# Decode and process command
|
||
try:
|
||
decoded_cmd = b64decode(data["CMD"]).decode("utf-8")
|
||
print(
|
||
f"""At: {time}\nReceived command: {decoded_cmd}\nUser-Agent: {request.headers.get('User-Agent')}\nIdempotency-Key: {request.headers.get('Idempotency-Key')}"""
|
||
)
|
||
logging.info(
|
||
f"{time}\tCommand: {decoded_cmd}\tFrom: {request.remote}\tUA: {request.headers.get('User-Agent')}\tIdempotency-Key: {request.headers.get('Idempotency-Key')}"
|
||
)
|
||
|
||
# Process the command
|
||
if decoded_cmd == "RUNUPDATE":
|
||
response_data = {"status": "success", "message": "Update started"}
|
||
await run_update()
|
||
logging.info(f"{time}\tUpdate command processed successfully")
|
||
return web.json_response(response_data)
|
||
else:
|
||
return web.Response(status=400, text=f"Unknown command: {decoded_cmd}")
|
||
|
||
except Exception as e:
|
||
print(f"Error processing command: {str(e)}")
|
||
return web.Response(status=500, text=f"Server error: {str(e)}")
|
||
|
||
except Exception as e:
|
||
print(f"Request handler error: {str(e)}")
|
||
return web.Response(status=500, text="Internal server error")
|
||
|
||
|
||
print("Starting web server...")
|
||
os.system("cls" if os.name == "nt" else "clear")
|
||
app = web.Application()
|
||
app.router.add_post("/api", request_handler)
|
||
app.router.add_post("/hook", hook_handler)
|
||
|
||
parser = argparse.ArgumentParser(
|
||
description="studspider - LEGO Inventory Sync Service for Square POS and Brickset",
|
||
epilog="""
|
||
Examples:
|
||
# Basic HTTP server (recommended with nginx reverse proxy)
|
||
python src/test_server.py --host 0.0.0.0 --port 7432
|
||
|
||
# With separate webhook port
|
||
python src/test_server.py --port 7432 --hook-port 8080
|
||
|
||
# HTTPS using certificates from .env (set SSL_CERT_PATH and SSL_KEY_PATH)
|
||
python src/test_server.py --hook-port 8443 --hook-ssl-only
|
||
|
||
# HTTPS with command line override (self-signed cert for testing)
|
||
python src/test_server.py --ssl-cert cert.pem --ssl-key key.pem --hook-port 8443 --hook-ssl-only
|
||
|
||
# Full HTTPS (both endpoints)
|
||
python src/test_server.py --ssl-cert cert.pem --ssl-key key.pem --hook-port 8443
|
||
|
||
# Production behind nginx (no SSL needed in app)
|
||
python src/test_server.py --host 127.0.0.1 --port 7432
|
||
|
||
Environment Variables (.env):
|
||
SSL_CERT_PATH - Default SSL certificate path
|
||
SSL_KEY_PATH - Default SSL key path
|
||
|
||
For more information, see doc/
|
||
""",
|
||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||
)
|
||
parser.add_argument(
|
||
"--host",
|
||
type=str,
|
||
default="localhost",
|
||
help="IP address to listen on (default: localhost, use 0.0.0.0 for all interfaces)",
|
||
)
|
||
parser.add_argument(
|
||
"--port", type=int, default=7432, help="Port for main API endpoint (default: 7432)"
|
||
)
|
||
parser.add_argument(
|
||
"--hook-port",
|
||
type=int,
|
||
default=None,
|
||
help="Separate port for webhook endpoint (default: None - uses same port as --port)",
|
||
)
|
||
parser.add_argument(
|
||
"--ssl-cert",
|
||
type=str,
|
||
default=get_sq_env.SSL_CERT_PATH,
|
||
help="Path to SSL certificate file in PEM format (default: from .env SSL_CERT_PATH)",
|
||
)
|
||
parser.add_argument(
|
||
"--ssl-key",
|
||
type=str,
|
||
default=get_sq_env.SSL_KEY_PATH,
|
||
help="Path to SSL private key file in PEM format (default: from .env SSL_KEY_PATH)",
|
||
)
|
||
parser.add_argument(
|
||
"--hook-ssl-only",
|
||
action="store_true",
|
||
help="Enable HTTPS only for webhook endpoint, main API stays HTTP (requires --ssl-cert and --ssl-key)",
|
||
)
|
||
args = parser.parse_args()
|
||
|
||
if __name__ == "__main__":
|
||
print("Creating log directory if it doesn't exist...")
|
||
os.makedirs("./logs/", exist_ok=True)
|
||
print(banner.generate("studspider v0.1"))
|
||
print("Built for PLAY-WELL TEKNOLOGIES by contact@studspider.sh\n\n")
|
||
|
||
# Setup SSL context if certificates provided
|
||
ssl_context = None
|
||
if args.ssl_cert and args.ssl_key:
|
||
# Check if trying to use P7B file
|
||
if args.ssl_cert.endswith(".p7b") or args.ssl_key.endswith(".p7b"):
|
||
print("ERROR: Cannot use P7B files directly for HTTPS server!")
|
||
print("\nP7B files are for Square's webhook signature verification only.")
|
||
print("Your server needs PEM format files (.pem)")
|
||
print("\nWorkflow:")
|
||
print("1. Generate PEM files: python generate_cert.py")
|
||
print("2. Convert to P7B for Square: python convert_to_p7b.py")
|
||
print("3. Upload P7B to Square Dashboard")
|
||
print("4. Run server with PEM files:")
|
||
print(
|
||
" python src/test_server.py --ssl-cert cert.pem --ssl-key key.pem --hook-port 8443"
|
||
)
|
||
print("\nIf you have a P7B and need PEM:")
|
||
print(" python convert_from_p7b.py your_cert.p7b")
|
||
exit(1)
|
||
|
||
try:
|
||
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}")
|
||
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.")
|
||
exit(1)
|
||
elif args.ssl_cert or args.ssl_key:
|
||
print("ERROR: Both --ssl-cert and --ssl-key must be provided together")
|
||
exit(1)
|
||
else:
|
||
print("ℹ️ Running without SSL (HTTP only)")
|
||
print(" For production with nginx reverse proxy, this is recommended.")
|
||
print(" Nginx will handle HTTPS, forward HTTP to this server.")
|
||
|
||
# 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"
|
||
protocol_hook = "https" if ssl_context else "http"
|
||
|
||
# Create separate app for webhook
|
||
hook_app = web.Application()
|
||
hook_app.router.add_post("/hook", hook_handler)
|
||
|
||
# Run both servers concurrently
|
||
async def run_servers():
|
||
runner1 = web.AppRunner(app)
|
||
await runner1.setup()
|
||
# Main server uses SSL only if not hook-ssl-only mode
|
||
main_ssl = None if args.hook_ssl_only else ssl_context
|
||
site1 = web.TCPSite(runner1, args.host, args.port, ssl_context=main_ssl)
|
||
|
||
runner2 = web.AppRunner(hook_app)
|
||
await runner2.setup()
|
||
# Webhook server always uses SSL if certificates provided
|
||
site2 = web.TCPSite(
|
||
runner2, args.host, args.hook_port, ssl_context=ssl_context
|
||
)
|
||
|
||
await site1.start()
|
||
await site2.start()
|
||
|
||
print(f"Starting main server on {args.host}:{args.port} ({protocol_main})")
|
||
print(
|
||
f"Starting webhook server on {args.host}:{args.hook_port} ({protocol_hook})"
|
||
)
|
||
|
||
# Keep running
|
||
await asyncio.Event().wait()
|
||
|
||
asyncio.run(run_servers())
|
||
else:
|
||
protocol = "https" if ssl_context else "http"
|
||
print(f"Server running on {args.host}:{args.port} ({protocol})")
|
||
print(f"Webhook endpoint: {protocol}://{args.host}:{args.port}/hook")
|
||
web.run_app(app, host=args.host, port=args.port, ssl_context=ssl_context)
|