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(): set_list = await sq.acat2id(SQUARE_LEGO_CAT_ID) # type: ignore # print(f"[@DEBUG]\t {set_list}") # type: ignore for l_set in set_list: # print(f"[@DEBUG]{l_set}") try: l_setinfo = await lego.throttled_get_set(l_set["number"]) print(f"[@DEBUG] Found {l_setinfo["pw_name"]}") except Exception as e: print(f"Error fetching set\t: {l_set}: {str(e)}") continue # print(f"[@DEBUG]{l_setinfo}") # await sq.update_itm(l_set, l_setinfo) """for l_set in set_list: print(f"[@DEBUG]{l_set}") try: l_setinfo = await lego.throttled_get_set(l_set["number"]) except Exception as e: print(f"Error fetching set\t: {l_set}: {str(e)}") continue print(f"[@DEBUG]{l_setinfo}")""" # await sq.update_itm(l_set, l_setinfo) 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)