From a51c19ea53e327692b5ba7a0bf138bae1ec80461 Mon Sep 17 00:00:00 2001 From: C <0xicl33n@gmail.com> Date: Sat, 3 Jan 2026 16:08:48 -0800 Subject: [PATCH] complete project restructure --- README.md | 2 +- {src/webhook => config}/gunicorn.conf.py | 5 - edderkop.code-workspace | 11 - src/{ => old}/ebricksetapi.py | 232 ++--- src/{ => old}/server.py | 226 ++--- src/webhook/server.py | 278 ------ studspider.py | 285 +++++- {src => tests}/test_client.py | 78 +- {src => tests}/test_server.py | 904 +++++++++--------- {src/webhook => tests}/test_square_webhook.py | 0 {src => tests}/test_webhook.py | 0 11 files changed, 995 insertions(+), 1026 deletions(-) rename {src/webhook => config}/gunicorn.conf.py (94%) delete mode 100644 edderkop.code-workspace rename src/{ => old}/ebricksetapi.py (97%) rename src/{ => old}/server.py (96%) delete mode 100644 src/webhook/server.py rename {src => tests}/test_client.py (96%) rename {src => tests}/test_server.py (97%) rename {src/webhook => tests}/test_square_webhook.py (100%) rename {src => tests}/test_webhook.py (100%) diff --git a/README.md b/README.md index 7569660..88537d0 100644 --- a/README.md +++ b/README.md @@ -12,7 +12,7 @@ Clone the project ```bash git clone http://git.traps.moe/ph0sgene/studspider.git - git checkout uv + git checkout branch ``` Go to the project directory diff --git a/src/webhook/gunicorn.conf.py b/config/gunicorn.conf.py similarity index 94% rename from src/webhook/gunicorn.conf.py rename to config/gunicorn.conf.py index ef781e9..7a70997 100644 --- a/src/webhook/gunicorn.conf.py +++ b/config/gunicorn.conf.py @@ -1,8 +1,3 @@ -""" -Gunicorn configuration for StudSpider webhook server -Production-ready settings for running with nginx reverse proxy -""" - import multiprocessing import os diff --git a/edderkop.code-workspace b/edderkop.code-workspace deleted file mode 100644 index e6d0824..0000000 --- a/edderkop.code-workspace +++ /dev/null @@ -1,11 +0,0 @@ -{ - "folders": [ - { - "path": "." - }, - { - "path": "../kopling" - } - ], - "settings": {} -} \ No newline at end of file diff --git a/src/ebricksetapi.py b/src/old/ebricksetapi.py similarity index 97% rename from src/ebricksetapi.py rename to src/old/ebricksetapi.py index f291363..26b48fd 100644 --- a/src/ebricksetapi.py +++ b/src/old/ebricksetapi.py @@ -1,116 +1,116 @@ -import os -import requests -import json -import dotenv -from typing import Dict, Any - -dotenv.load_dotenv() - - -class BricksetAPI: - """Class for interacting with the Brickset API""" - - def __init__(self, debug: bool = False): - self.username = os.getenv("BRICKSET_USERNAME") - self.password = os.getenv("BRICKSET_PASS") - self.api_key = os.getenv("BRICKSET_API_KEY") - self.debug = debug - self.headers = {"UserAgent": "studspider/0.1"} - self.user_hash = None - - def login(self) -> Dict[str, Any]: - """Login to Brickset API and get user hash""" - url = "https://brickset.com/api/v3.asmx/login" - params = { - "username": self.username, - "password": self.password, - "apiKey": self.api_key, - } - - response = requests.get(url, params=params, headers=self.headers) - if self.debug: - print(f"[@DEBUG] Login response: {response.status_code}") - print(f"[@DEBUG] Response content: {response.text}") - - try: - data = response.json() - self.user_hash = data.get("hash") - if not self.user_hash: - raise ValueError("No hash received in login response") - return data - except json.JSONDecodeError as e: - print(f"[@ERROR] Failed to decode JSON response: {e}") - print(f"[@DEBUG] Raw response: {response.text}") - raise - - def get_set(self, set_number: str) -> Dict[str, Any]: - """Get LEGO set information by set number""" - if not self.user_hash: - self.login() - - url = "https://brickset.com/api/v3.asmx/getSets" - params = { - "apiKey": self.api_key, - "userHash": self.user_hash, - "params": json.dumps({"setNumber": f"{set_number}-1"}), - } - - response = requests.get(url, params=params, headers=self.headers) - - if self.debug: - print(f"[@DEBUG] Get set response: {response.status_code}") - print(f"[@DEBUG] Response content: {response.text}") - - if response.status_code == 200: - try: - data = response.json() - if not data.get("sets"): - raise ValueError(f"No set data found for {set_number}") - - set_data = data["sets"][0] - our_set = { - "set_num": set_data["number"], - "set_name": set_data["name"], - "set_image": set_data["image"]["thumbnailURL"], - "set_barcode": set_data["barcode"]["UPC"], - "set_retailprice": round(set_data["LEGOCom"]["US"]["retailPrice"]), - } - - # Calculate adjusted price (10% markup) - retail_price = our_set["set_retailprice"] - our_set["set_adjustredprice"] = retail_price + ( - (10 / 100) * retail_price - ) - # Set Playwell inventory name - if our_set["theme"] == "Duplo": - our_set["pw_name"] = ( - f"{our_set['set_num']} - DUPLO {our_set['set_name']}" - ) - else: - our_set["pw_name"] = f"{our_set['set_num']} - {our_set['set_name']}" - - return our_set - - except json.JSONDecodeError as e: - # print(f"[@ERROR] Failed to decode JSON response: {e}") - # print(f"[@DEBUG] Raw response: {response.text}") - raise - except KeyError as e: - # print(f"[@ERROR] Missing expected field in response: {e}") - # print(f"[@DEBUG] Response data: {data}") - raise - elif response.status_code == 500: - raise Exception( - "Internal Server Error: Please check API key and parameters." - ) - else: - raise Exception(f"Unexpected response: {response.status_code}") - - -def example(): - client = BricksetAPI(debug=True) - try: - set_info = client.get_set("10931") - print(set_info) - except Exception as e: - print(f"Error: {e}") +import os +import requests +import json +import dotenv +from typing import Dict, Any + +dotenv.load_dotenv() + + +class BricksetAPI: + """Class for interacting with the Brickset API""" + + def __init__(self, debug: bool = False): + self.username = os.getenv("BRICKSET_USERNAME") + self.password = os.getenv("BRICKSET_PASS") + self.api_key = os.getenv("BRICKSET_API_KEY") + self.debug = debug + self.headers = {"UserAgent": "studspider/0.1"} + self.user_hash = None + + def login(self) -> Dict[str, Any]: + """Login to Brickset API and get user hash""" + url = "https://brickset.com/api/v3.asmx/login" + params = { + "username": self.username, + "password": self.password, + "apiKey": self.api_key, + } + + response = requests.get(url, params=params, headers=self.headers) + if self.debug: + print(f"[@DEBUG] Login response: {response.status_code}") + print(f"[@DEBUG] Response content: {response.text}") + + try: + data = response.json() + self.user_hash = data.get("hash") + if not self.user_hash: + raise ValueError("No hash received in login response") + return data + except json.JSONDecodeError as e: + print(f"[@ERROR] Failed to decode JSON response: {e}") + print(f"[@DEBUG] Raw response: {response.text}") + raise + + def get_set(self, set_number: str) -> Dict[str, Any]: + """Get LEGO set information by set number""" + if not self.user_hash: + self.login() + + url = "https://brickset.com/api/v3.asmx/getSets" + params = { + "apiKey": self.api_key, + "userHash": self.user_hash, + "params": json.dumps({"setNumber": f"{set_number}-1"}), + } + + response = requests.get(url, params=params, headers=self.headers) + + if self.debug: + print(f"[@DEBUG] Get set response: {response.status_code}") + print(f"[@DEBUG] Response content: {response.text}") + + if response.status_code == 200: + try: + data = response.json() + if not data.get("sets"): + raise ValueError(f"No set data found for {set_number}") + + set_data = data["sets"][0] + our_set = { + "set_num": set_data["number"], + "set_name": set_data["name"], + "set_image": set_data["image"]["thumbnailURL"], + "set_barcode": set_data["barcode"]["UPC"], + "set_retailprice": round(set_data["LEGOCom"]["US"]["retailPrice"]), + } + + # Calculate adjusted price (10% markup) + retail_price = our_set["set_retailprice"] + our_set["set_adjustredprice"] = retail_price + ( + (10 / 100) * retail_price + ) + # Set Playwell inventory name + if our_set["theme"] == "Duplo": + our_set["pw_name"] = ( + f"{our_set['set_num']} - DUPLO {our_set['set_name']}" + ) + else: + our_set["pw_name"] = f"{our_set['set_num']} - {our_set['set_name']}" + + return our_set + + except json.JSONDecodeError as e: + # print(f"[@ERROR] Failed to decode JSON response: {e}") + # print(f"[@DEBUG] Raw response: {response.text}") + raise + except KeyError as e: + # print(f"[@ERROR] Missing expected field in response: {e}") + # print(f"[@DEBUG] Response data: {data}") + raise + elif response.status_code == 500: + raise Exception( + "Internal Server Error: Please check API key and parameters." + ) + else: + raise Exception(f"Unexpected response: {response.status_code}") + + +def example(): + client = BricksetAPI(debug=True) + try: + set_info = client.get_set("10931") + print(set_info) + except Exception as e: + print(f"Error: {e}") diff --git a/src/server.py b/src/old/server.py similarity index 96% rename from src/server.py rename to src/old/server.py index ba4eaa0..3b9f552 100644 --- a/src/server.py +++ b/src/old/server.py @@ -1,113 +1,113 @@ -import asyncio -import aiohttp.web -import uuid -import os -from base64 import b64decode -import dotenv - -dotenv.load_dotenv() -SQUARE_LEGO_CAT_ID = os.environ.get("SQUARE_LEGO_CAT_ID") - -from asyncio_throttle.throttler import Throttler - -import asquare as sq -import abricksetapi as alego - - -def valid_uuid(val): - try: - uuid.UUID(str(val)) - return True - except ValueError: - return False - - -async def fullsync(): - """' - full_db = await sq.acat2id(SQUARE_LEGO_CAT_ID) - for l_set in full_db: - l_setinfo = await alego.throttled_get_set(l_set) - await sq.update_itm(l_set, l_setinfo)""" - - -async def stop(): - pass - - -async def priceonly(): - pass - - -async def handle_post(request): - - valid_cmd = ["RUNFULLSYNC", "STOP", "RUNPRICEONLY"] - - async def handle_cmd(cmd: str): - if cmd not in valid_cmd: - return False - else: - return True - - idp_key = request.headers.get("Idempotency-Key") - user_agent = request.headers.get("User-Agent") - - if not idp_key and not valid_uuid(idp_key) and user_agent != "Kopling/0.1": - # 412 Precondition Failed - # The server does not meet one of the preconditions that the requester put on the request header fields. - return aiohttp.web.Response(status=412, text="412 Headers rejected!") - - else: - data = await request.post() - reqdata = data.get(b64decode("CMD")) - - if await handle_cmd(reqdata): - match reqdata: - case "RUNFULLSYNC": - print("Running full sync...") - await fullsync() - return True - case "STOP": - await stop() - return True - case "RUNPRICEONLY": - await priceonly() - return True - else: - return aiohttp.web.Response(status=400, text="400 Bad Request.") - result = {"message": f"studspider/0.1 received command: {reqdata}"} - return aiohttp.web.json_response(status=200, data=result) - - -app = aiohttp.web.Application() -app.router.add_post("/", handle_post) - -import subprocess - - -def generate_toilet_banner(text, font="smblock", filter_effect=None, colors=None): - command = ["toilet", "-f", font] - if filter_effect: - command.extend(["-F", filter_effect]) - if colors: - # Toilet uses a specific format for color arguments, e.g., --gay or --metal - # For general colors, you might need to use specific filters or direct color codes. - # This example focuses on filters that inherently use colors. - pass # More complex color handling would be needed for arbitrary colors - - command.append(text) - - try: - result = subprocess.run(command, capture_output=True, text=True, check=True) - return result.stdout - except subprocess.CalledProcessError as e: - return f"Error executing toilet: {e}\n{e.stderr}" - - -async def main(): - throttler = Throttler(rate_limit=500, period=60) # 500 requests per minute - - -if __name__ == "__main__": - print(generate_toilet_banner("studspider v0.1")) - print("Built for PLAY-WELL TEKNOLOGIES by contact@traps.moe\n\n") - aiohttp.web.run_app(app, port=7432) +import asyncio +import aiohttp.web +import uuid +import os +from base64 import b64decode +import dotenv + +dotenv.load_dotenv() +SQUARE_LEGO_CAT_ID = os.environ.get("SQUARE_LEGO_CAT_ID") + +from asyncio_throttle.throttler import Throttler + +import asquare as sq +import abricksetapi as alego + + +def valid_uuid(val): + try: + uuid.UUID(str(val)) + return True + except ValueError: + return False + + +async def fullsync(): + """' + full_db = await sq.acat2id(SQUARE_LEGO_CAT_ID) + for l_set in full_db: + l_setinfo = await alego.throttled_get_set(l_set) + await sq.update_itm(l_set, l_setinfo)""" + + +async def stop(): + pass + + +async def priceonly(): + pass + + +async def handle_post(request): + + valid_cmd = ["RUNFULLSYNC", "STOP", "RUNPRICEONLY"] + + async def handle_cmd(cmd: str): + if cmd not in valid_cmd: + return False + else: + return True + + idp_key = request.headers.get("Idempotency-Key") + user_agent = request.headers.get("User-Agent") + + if not idp_key and not valid_uuid(idp_key) and user_agent != "Kopling/0.1": + # 412 Precondition Failed + # The server does not meet one of the preconditions that the requester put on the request header fields. + return aiohttp.web.Response(status=412, text="412 Headers rejected!") + + else: + data = await request.post() + reqdata = data.get(b64decode("CMD")) + + if await handle_cmd(reqdata): + match reqdata: + case "RUNFULLSYNC": + print("Running full sync...") + await fullsync() + return True + case "STOP": + await stop() + return True + case "RUNPRICEONLY": + await priceonly() + return True + else: + return aiohttp.web.Response(status=400, text="400 Bad Request.") + result = {"message": f"studspider/0.1 received command: {reqdata}"} + return aiohttp.web.json_response(status=200, data=result) + + +app = aiohttp.web.Application() +app.router.add_post("/", handle_post) + +import subprocess + + +def generate_toilet_banner(text, font="smblock", filter_effect=None, colors=None): + command = ["toilet", "-f", font] + if filter_effect: + command.extend(["-F", filter_effect]) + if colors: + # Toilet uses a specific format for color arguments, e.g., --gay or --metal + # For general colors, you might need to use specific filters or direct color codes. + # This example focuses on filters that inherently use colors. + pass # More complex color handling would be needed for arbitrary colors + + command.append(text) + + try: + result = subprocess.run(command, capture_output=True, text=True, check=True) + return result.stdout + except subprocess.CalledProcessError as e: + return f"Error executing toilet: {e}\n{e.stderr}" + + +async def main(): + throttler = Throttler(rate_limit=500, period=60) # 500 requests per minute + + +if __name__ == "__main__": + print(generate_toilet_banner("studspider v0.1")) + print("Built for PLAY-WELL TEKNOLOGIES by contact@traps.moe\n\n") + aiohttp.web.run_app(app, port=7432) diff --git a/src/webhook/server.py b/src/webhook/server.py deleted file mode 100644 index dceda46..0000000 --- a/src/webhook/server.py +++ /dev/null @@ -1,278 +0,0 @@ - """ - 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 boot dev") - 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") diff --git a/studspider.py b/studspider.py index 61efcbd..6255678 100644 --- a/studspider.py +++ b/studspider.py @@ -1,14 +1,277 @@ -import src.test_server +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") -class Studspider: - def __init__(self): - self.name = "Studspider" - self.version = "0.0.1" - self.description = """ - studspider is an application designed to interact with the Brickset API and directly with Square. - It allows users to search for items, manage collections, and perform various operations related to bricksets. - """ +@app.get("/") +async def root(): + """Landing page""" + return { + "service": "Studspider", + "status": "running", + "endpoints": {"webhook": "/hook (POST)", "health": "/health (GET)"}, + } - def run(self): - print(f"{self.name} v{self.version} is running.") + +@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 boot dev") + 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 conf/studspider: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") diff --git a/src/test_client.py b/tests/test_client.py similarity index 96% rename from src/test_client.py rename to tests/test_client.py index f5725e3..6c59234 100644 --- a/src/test_client.py +++ b/tests/test_client.py @@ -1,39 +1,39 @@ -import asyncio -import aiohttp -from uuid import uuid4 -from base64 import b64encode -import os - - -async def idp_request(cmd: str): - headers = { - "Idempotency-Key": f"{uuid4()}", - "Content-Type": "application/json", - "User-Agent": "Kopling/0.12", - } - print(f"[@DEBUG] Idempotency-Key: {headers['Idempotency-Key']}") - - enc_cmd = b64encode(f"{cmd}".encode("utf-8")).decode("utf-8") - print("[@DEBUG] Encoded command:", enc_cmd) - print("[@DEBUG] Sending request..") - try: - async with aiohttp.ClientSession() as session: - async with session.post( - "http://localhost:7432", # Added http:// protocol - headers=headers, - json={"CMD": enc_cmd}, # Changed to json instead of data - ) as resp: - print(f"[@DEBUG] Response status: {resp.status}") - print(f"[@DEBUG] Response body: {await resp.text()}") - except Exception as e: - print(f"\nRequest failed: {e}") - os._exit(-1) - - -async def main(): - # Await the task instead of just creating it - await idp_request("RUNUPDATE") - - -if __name__ == "__main__": - asyncio.run(main()) +import asyncio +import aiohttp +from uuid import uuid4 +from base64 import b64encode +import os + + +async def idp_request(cmd: str): + headers = { + "Idempotency-Key": f"{uuid4()}", + "Content-Type": "application/json", + "User-Agent": "Kopling/0.12", + } + print(f"[@DEBUG] Idempotency-Key: {headers['Idempotency-Key']}") + + enc_cmd = b64encode(f"{cmd}".encode("utf-8")).decode("utf-8") + print("[@DEBUG] Encoded command:", enc_cmd) + print("[@DEBUG] Sending request..") + try: + async with aiohttp.ClientSession() as session: + async with session.post( + "http://localhost:7432", # Added http:// protocol + headers=headers, + json={"CMD": enc_cmd}, # Changed to json instead of data + ) as resp: + print(f"[@DEBUG] Response status: {resp.status}") + print(f"[@DEBUG] Response body: {await resp.text()}") + except Exception as e: + print(f"\nRequest failed: {e}") + os._exit(-1) + + +async def main(): + # Await the task instead of just creating it + await idp_request("RUNUPDATE") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/src/test_server.py b/tests/test_server.py similarity index 97% rename from src/test_server.py rename to tests/test_server.py index beb4292..4eec006 100644 --- a/src/test_server.py +++ b/tests/test_server.py @@ -1,452 +1,452 @@ -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) +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) diff --git a/src/webhook/test_square_webhook.py b/tests/test_square_webhook.py similarity index 100% rename from src/webhook/test_square_webhook.py rename to tests/test_square_webhook.py diff --git a/src/test_webhook.py b/tests/test_webhook.py similarity index 100% rename from src/test_webhook.py rename to tests/test_webhook.py