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

Compare commits

base: ph0sgene/studspider:webhook
ph0sgene/studspider:rewrite ph0sgene/studspider:webhook ph0sgene/studspider:old ph0sgene/studspider:docker ph0sgene/studspider:uv
..
compare: ph0sgene/studspider:48835cd85737e0dfc4cd3cd9c6e2d4ee0c1fd297
ph0sgene/studspider:rewrite ph0sgene/studspider:webhook ph0sgene/studspider:old ph0sgene/studspider:docker ph0sgene/studspider:uv
4 Commits
webhook .. 48835cd857
Author SHA1 Message Date
C 48835cd857 im actually gonna crash out rn dude 2026-01-03 17:24:49 -08:00
C c4bca1e4d0 bro my grammar was so ass. was i drunk when i wrote this? 2026-01-03 16:45:00 -08:00
C 42b4575cc4 update env 2026-01-03 16:34:19 -08:00
C a51c19ea53 complete project restructure 2026-01-03 16:08:48 -08:00
12 changed files with 1019 additions and 1108 deletions
Expand all files Collapse all files

No files matched your search

.env.example
+17 -78
View File
@@ -1,83 +1,22 @@
# studspider Environment Configuration # Brickset.com credentials - LEGO database for pricing, images, UPC codes
BRICKSET_USERNAME = "your_brickset_username"
BRICKSET_PASS = "your_brickset_password"
BRICKSET_API_KEY = "your_brickset_api_key"
# ============================================================================ # Square environment - "sandbox" for testing, "production" for live store
# Square API Configuration SQUARE_ENV = "sandbox" # or "production"
# ============================================================================
# Environment: "production" or "sandbox" SQUARE_WEBHOOK_URL = "https://yourdomain.com/hook"
SQUARE_ENV=sandbox
# Sandbox credentials # Sandbox credentials - test environment (fake data, safe for development)
SANDBOX_SQUARE_APP_ID=your_sandbox_app_id SANDBOX_SQUARE_APP_ID = "your_sandbox_app_id"
SANDBOX_SQUARE_TOKEN=your_sandbox_access_token SANDBOX_SQUARE_TOKEN = "your_sandbox_access_token"
SANDBOX_SQUARE_WEBHOOK_INVENTORY_ID = "your_sandbox_webhook_signature_key"
# Production credentials # Production credentials - live store (real inventory and sales)
PROD_SQUARE_APP_ID=your_production_app_id PROD_SQUARE_APP_ID = "your_production_app_id"
PROD_SQUARE_TOKEN=your_production_access_token PROD_SQUARE_TOKEN = "your_production_access_token"
PROD_SQUARE_WEBHOOK_INVENTORY_ID = "your_production_webhook_signature_key"
# Square LEGO category ID (find this in Square Dashboard) # Category ID for LEGO items in Square catalog (find in Dashboard → Items → Categories)
SQUARE_LEGO_CAT_ID=your_lego_category_id SQUARE_LEGO_CAT_ID = "your_lego_category_id"
# ============================================================================
# Brickset API Configuration
# ============================================================================
BRICKSET_USERNAME=your_brickset_username
BRICKSET_PASS=your_brickset_password
BRICKSET_API_KEY=your_brickset_api_key
# ============================================================================
# SSL Certificate Configuration (Optional)
# ============================================================================
# NOTE: Only needed if running HTTPS directly in Python (not recommended)
# For production, use nginx reverse proxy with Let's Encrypt instead
# nginx handles SSL/TLS, your Python app runs HTTP internally
# Path to SSL certificate in PEM format (if running HTTPS directly)
# SSL_CERT_PATH=/path/to/cert.pem
# Path to SSL private key in PEM format (if running HTTPS directly)
# SSL_KEY_PATH=/path/to/key.pem
# ============================================================================
# Square Webhook Signature Verification
# ============================================================================
# CRITICAL: Get these from Square Developer Dashboard → Webhooks
# 1. Go to https://developer.squareup.com/apps
# 2. Select your application
# 3. Click "Webhooks" section
# 4. Create webhook subscription (Notification URL + Events)
# 5. Copy the Signature Key that Square displays
# Signature key from Square Dashboard (a string like "asdf1234")
SQUARE_WEBHOOK_SIGNATURE_KEY=your_webhook_signature_key_from_square
# Your webhook notification URL (must match EXACTLY what's in Square Dashboard)
# This should be your public HTTPS URL (through nginx)
# Example: https://yourdomain.com/hook
SQUARE_WEBHOOK_URL=https://yourdomain.com/hook
# ============================================================================
# Server Configuration (Optional)
# ============================================================================
# Default host and port (can override with command line args)
# SERVER_HOST=0.0.0.0
# SERVER_PORT=7432
# SERVER_HOOK_PORT=8080
# ============================================================================
# Usage Examples
# ============================================================================
# Basic HTTP server (recommended with nginx):
# python src/test_server.py
#
# With SSL from .env:
# Set SSL_CERT_PATH and SSL_KEY_PATH above, then:
# python src/test_server.py --hook-port 8443 --hook-ssl-only
#
# Override SSL from command line:
# python src/test_server.py --ssl-cert /custom/cert.pem --ssl-key /custom/key.pem
README.md
+6 -5
View File
@@ -5,14 +5,14 @@
is a tool used to talk to brickset.com and a square inventory to pull lego set information for entering in the square db is a tool used to talk to brickset.com, pull lego set information and automatically enter it into the square inventory!
## Run me ## Run me
Clone the project Clone the project
```bash ```bash
git clone http://git.traps.moe/ph0sgene/studspider.git git clone http://git.traps.moe/ph0sgene/studspider.git
git checkout uv git checkout branch
``` ```
Go to the project directory Go to the project directory
@@ -30,14 +30,15 @@ Install dependencies
Start the server Start the server
```bash ```bash
uvx run studspider.py uv run studspider.py
``` ```
## Tech ## Tech
- gunicorn
- uvicorn
- aiohttp - aiohttp
- squareup - squareup
- apscheduler - apscheduler
- aiosqlite - aiosqlite
src/webhook/gunicorn.conf.py → config/gunicorn.conf.py
-5
View File
@@ -1,8 +1,3 @@
"""
Gunicorn configuration for StudSpider webhook server
Production-ready settings for running with nginx reverse proxy
"""
import multiprocessing import multiprocessing
import os import os
edderkop.code-workspace
-11
View File
@@ -1,11 +0,0 @@
{
"folders": [
{
"path": "."
},
{
"path": "../kopling"
}
],
"settings": {}
}
src/ebricksetapi.py → src/old/ebricksetapi.py
+116 -116
View File
@@ -1,116 +1,116 @@
import os import os
import requests import requests
import json import json
import dotenv import dotenv
from typing import Dict, Any from typing import Dict, Any
dotenv.load_dotenv() dotenv.load_dotenv()
class BricksetAPI: class BricksetAPI:
"""Class for interacting with the Brickset API""" """Class for interacting with the Brickset API"""
def __init__(self, debug: bool = False): def __init__(self, debug: bool = False):
self.username = os.getenv("BRICKSET_USERNAME") self.username = os.getenv("BRICKSET_USERNAME")
self.password = os.getenv("BRICKSET_PASS") self.password = os.getenv("BRICKSET_PASS")
self.api_key = os.getenv("BRICKSET_API_KEY") self.api_key = os.getenv("BRICKSET_API_KEY")
self.debug = debug self.debug = debug
self.headers = {"UserAgent": "studspider/0.1"} self.headers = {"UserAgent": "studspider/0.1"}
self.user_hash = None self.user_hash = None
def login(self) -> Dict[str, Any]: def login(self) -> Dict[str, Any]:
"""Login to Brickset API and get user hash""" """Login to Brickset API and get user hash"""
url = "https://brickset.com/api/v3.asmx/login" url = "https://brickset.com/api/v3.asmx/login"
params = { params = {
"username": self.username, "username": self.username,
"password": self.password, "password": self.password,
"apiKey": self.api_key, "apiKey": self.api_key,
} }
response = requests.get(url, params=params, headers=self.headers) response = requests.get(url, params=params, headers=self.headers)
if self.debug: if self.debug:
print(f"[@DEBUG] Login response: {response.status_code}") print(f"[@DEBUG] Login response: {response.status_code}")
print(f"[@DEBUG] Response content: {response.text}") print(f"[@DEBUG] Response content: {response.text}")
try: try:
data = response.json() data = response.json()
self.user_hash = data.get("hash") self.user_hash = data.get("hash")
if not self.user_hash: if not self.user_hash:
raise ValueError("No hash received in login response") raise ValueError("No hash received in login response")
return data return data
except json.JSONDecodeError as e: except json.JSONDecodeError as e:
print(f"[@ERROR] Failed to decode JSON response: {e}") print(f"[@ERROR] Failed to decode JSON response: {e}")
print(f"[@DEBUG] Raw response: {response.text}") print(f"[@DEBUG] Raw response: {response.text}")
raise raise
def get_set(self, set_number: str) -> Dict[str, Any]: def get_set(self, set_number: str) -> Dict[str, Any]:
"""Get LEGO set information by set number""" """Get LEGO set information by set number"""
if not self.user_hash: if not self.user_hash:
self.login() self.login()
url = "https://brickset.com/api/v3.asmx/getSets" url = "https://brickset.com/api/v3.asmx/getSets"
params = { params = {
"apiKey": self.api_key, "apiKey": self.api_key,
"userHash": self.user_hash, "userHash": self.user_hash,
"params": json.dumps({"setNumber": f"{set_number}-1"}), "params": json.dumps({"setNumber": f"{set_number}-1"}),
} }
response = requests.get(url, params=params, headers=self.headers) response = requests.get(url, params=params, headers=self.headers)
if self.debug: if self.debug:
print(f"[@DEBUG] Get set response: {response.status_code}") print(f"[@DEBUG] Get set response: {response.status_code}")
print(f"[@DEBUG] Response content: {response.text}") print(f"[@DEBUG] Response content: {response.text}")
if response.status_code == 200: if response.status_code == 200:
try: try:
data = response.json() data = response.json()
if not data.get("sets"): if not data.get("sets"):
raise ValueError(f"No set data found for {set_number}") raise ValueError(f"No set data found for {set_number}")
set_data = data["sets"][0] set_data = data["sets"][0]
our_set = { our_set = {
"set_num": set_data["number"], "set_num": set_data["number"],
"set_name": set_data["name"], "set_name": set_data["name"],
"set_image": set_data["image"]["thumbnailURL"], "set_image": set_data["image"]["thumbnailURL"],
"set_barcode": set_data["barcode"]["UPC"], "set_barcode": set_data["barcode"]["UPC"],
"set_retailprice": round(set_data["LEGOCom"]["US"]["retailPrice"]), "set_retailprice": round(set_data["LEGOCom"]["US"]["retailPrice"]),
} }
# Calculate adjusted price (10% markup) # Calculate adjusted price (10% markup)
retail_price = our_set["set_retailprice"] retail_price = our_set["set_retailprice"]
our_set["set_adjustredprice"] = retail_price + ( our_set["set_adjustredprice"] = retail_price + (
(10 / 100) * retail_price (10 / 100) * retail_price
) )
# Set Playwell inventory name # Set Playwell inventory name
if our_set["theme"] == "Duplo": if our_set["theme"] == "Duplo":
our_set["pw_name"] = ( our_set["pw_name"] = (
f"{our_set['set_num']} - DUPLO {our_set['set_name']}" f"{our_set['set_num']} - DUPLO {our_set['set_name']}"
) )
else: else:
our_set["pw_name"] = f"{our_set['set_num']} - {our_set['set_name']}" our_set["pw_name"] = f"{our_set['set_num']} - {our_set['set_name']}"
return our_set return our_set
except json.JSONDecodeError as e: except json.JSONDecodeError as e:
# print(f"[@ERROR] Failed to decode JSON response: {e}") # print(f"[@ERROR] Failed to decode JSON response: {e}")
# print(f"[@DEBUG] Raw response: {response.text}") # print(f"[@DEBUG] Raw response: {response.text}")
raise raise
except KeyError as e: except KeyError as e:
# print(f"[@ERROR] Missing expected field in response: {e}") # print(f"[@ERROR] Missing expected field in response: {e}")
# print(f"[@DEBUG] Response data: {data}") # print(f"[@DEBUG] Response data: {data}")
raise raise
elif response.status_code == 500: elif response.status_code == 500:
raise Exception( raise Exception(
"Internal Server Error: Please check API key and parameters." "Internal Server Error: Please check API key and parameters."
) )
else: else:
raise Exception(f"Unexpected response: {response.status_code}") raise Exception(f"Unexpected response: {response.status_code}")
def example(): def example():
client = BricksetAPI(debug=True) client = BricksetAPI(debug=True)
try: try:
set_info = client.get_set("10931") set_info = client.get_set("10931")
print(set_info) print(set_info)
except Exception as e: except Exception as e:
print(f"Error: {e}") print(f"Error: {e}")
src/server.py → src/old/server.py
+113 -113
View File
@@ -1,113 +1,113 @@
import asyncio import asyncio
import aiohttp.web import aiohttp.web
import uuid import uuid
import os import os
from base64 import b64decode from base64 import b64decode
import dotenv import dotenv
dotenv.load_dotenv() dotenv.load_dotenv()
SQUARE_LEGO_CAT_ID = os.environ.get("SQUARE_LEGO_CAT_ID") SQUARE_LEGO_CAT_ID = os.environ.get("SQUARE_LEGO_CAT_ID")
from asyncio_throttle.throttler import Throttler from asyncio_throttle.throttler import Throttler
import asquare as sq import asquare as sq
import abricksetapi as alego import abricksetapi as alego
def valid_uuid(val): def valid_uuid(val):
try: try:
uuid.UUID(str(val)) uuid.UUID(str(val))
return True return True
except ValueError: except ValueError:
return False return False
async def fullsync(): async def fullsync():
"""' """'
full_db = await sq.acat2id(SQUARE_LEGO_CAT_ID) full_db = await sq.acat2id(SQUARE_LEGO_CAT_ID)
for l_set in full_db: for l_set in full_db:
l_setinfo = await alego.throttled_get_set(l_set) l_setinfo = await alego.throttled_get_set(l_set)
await sq.update_itm(l_set, l_setinfo)""" await sq.update_itm(l_set, l_setinfo)"""
async def stop(): async def stop():
pass pass
async def priceonly(): async def priceonly():
pass pass
async def handle_post(request): async def handle_post(request):
valid_cmd = ["RUNFULLSYNC", "STOP", "RUNPRICEONLY"] valid_cmd = ["RUNFULLSYNC", "STOP", "RUNPRICEONLY"]
async def handle_cmd(cmd: str): async def handle_cmd(cmd: str):
if cmd not in valid_cmd: if cmd not in valid_cmd:
return False return False
else: else:
return True return True
idp_key = request.headers.get("Idempotency-Key") idp_key = request.headers.get("Idempotency-Key")
user_agent = request.headers.get("User-Agent") user_agent = request.headers.get("User-Agent")
if not idp_key and not valid_uuid(idp_key) and user_agent != "Kopling/0.1": if not idp_key and not valid_uuid(idp_key) and user_agent != "Kopling/0.1":
# 412 Precondition Failed # 412 Precondition Failed
# The server does not meet one of the preconditions that the requester put on the request header fields. # 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!") return aiohttp.web.Response(status=412, text="412 Headers rejected!")
else: else:
data = await request.post() data = await request.post()
reqdata = data.get(b64decode("CMD")) reqdata = data.get(b64decode("CMD"))
if await handle_cmd(reqdata): if await handle_cmd(reqdata):
match reqdata: match reqdata:
case "RUNFULLSYNC": case "RUNFULLSYNC":
print("Running full sync...") print("Running full sync...")
await fullsync() await fullsync()
return True return True
case "STOP": case "STOP":
await stop() await stop()
return True return True
case "RUNPRICEONLY": case "RUNPRICEONLY":
await priceonly() await priceonly()
return True return True
else: else:
return aiohttp.web.Response(status=400, text="400 Bad Request.") return aiohttp.web.Response(status=400, text="400 Bad Request.")
result = {"message": f"studspider/0.1 received command: {reqdata}"} result = {"message": f"studspider/0.1 received command: {reqdata}"}
return aiohttp.web.json_response(status=200, data=result) return aiohttp.web.json_response(status=200, data=result)
app = aiohttp.web.Application() app = aiohttp.web.Application()
app.router.add_post("/", handle_post) app.router.add_post("/", handle_post)
import subprocess import subprocess
def generate_toilet_banner(text, font="smblock", filter_effect=None, colors=None): def generate_toilet_banner(text, font="smblock", filter_effect=None, colors=None):
command = ["toilet", "-f", font] command = ["toilet", "-f", font]
if filter_effect: if filter_effect:
command.extend(["-F", filter_effect]) command.extend(["-F", filter_effect])
if colors: if colors:
# Toilet uses a specific format for color arguments, e.g., --gay or --metal # 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. # For general colors, you might need to use specific filters or direct color codes.
# This example focuses on filters that inherently use colors. # This example focuses on filters that inherently use colors.
pass # More complex color handling would be needed for arbitrary colors pass # More complex color handling would be needed for arbitrary colors
command.append(text) command.append(text)
try: try:
result = subprocess.run(command, capture_output=True, text=True, check=True) result = subprocess.run(command, capture_output=True, text=True, check=True)
return result.stdout return result.stdout
except subprocess.CalledProcessError as e: except subprocess.CalledProcessError as e:
return f"Error executing toilet: {e}\n{e.stderr}" return f"Error executing toilet: {e}\n{e.stderr}"
async def main(): async def main():
throttler = Throttler(rate_limit=500, period=60) # 500 requests per minute throttler = Throttler(rate_limit=500, period=60) # 500 requests per minute
if __name__ == "__main__": if __name__ == "__main__":
print(generate_toilet_banner("studspider v0.1")) print(generate_toilet_banner("studspider v0.1"))
print("Built for PLAY-WELL TEKNOLOGIES by contact@traps.moe\n\n") print("Built for PLAY-WELL TEKNOLOGIES by contact@traps.moe\n\n")
aiohttp.web.run_app(app, port=7432) aiohttp.web.run_app(app, port=7432)
src/webhook/server.py
-278
View File
@@ -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")
studspider.py
+276 -11
View File
@@ -1,14 +1,279 @@
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_SIGNATURE_KEY") or os.getenv(
"PROD_SQUARE_SIGNATURE_KEY"
)
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: @app.get("/")
def __init__(self): async def root():
self.name = "Studspider" """Landing page"""
self.version = "0.0.1" return {
self.description = """ "service": "Studspider",
studspider is an application designed to interact with the Brickset API and directly with Square. "status": "running",
It allows users to search for items, manage collections, and perform various operations related to bricksets. "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")
src/test_client.py → tests/test_client.py
+39 -39
View File
@@ -1,39 +1,39 @@
import asyncio import asyncio
import aiohttp import aiohttp
from uuid import uuid4 from uuid import uuid4
from base64 import b64encode from base64 import b64encode
import os import os
async def idp_request(cmd: str): async def idp_request(cmd: str):
headers = { headers = {
"Idempotency-Key": f"{uuid4()}", "Idempotency-Key": f"{uuid4()}",
"Content-Type": "application/json", "Content-Type": "application/json",
"User-Agent": "Kopling/0.12", "User-Agent": "Kopling/0.12",
} }
print(f"[@DEBUG] Idempotency-Key: {headers['Idempotency-Key']}") print(f"[@DEBUG] Idempotency-Key: {headers['Idempotency-Key']}")
enc_cmd = b64encode(f"{cmd}".encode("utf-8")).decode("utf-8") enc_cmd = b64encode(f"{cmd}".encode("utf-8")).decode("utf-8")
print("[@DEBUG] Encoded command:", enc_cmd) print("[@DEBUG] Encoded command:", enc_cmd)
print("[@DEBUG] Sending request..") print("[@DEBUG] Sending request..")
try: try:
async with aiohttp.ClientSession() as session: async with aiohttp.ClientSession() as session:
async with session.post( async with session.post(
"http://localhost:7432", # Added http:// protocol "http://localhost:7432", # Added http:// protocol
headers=headers, headers=headers,
json={"CMD": enc_cmd}, # Changed to json instead of data json={"CMD": enc_cmd}, # Changed to json instead of data
) as resp: ) as resp:
print(f"[@DEBUG] Response status: {resp.status}") print(f"[@DEBUG] Response status: {resp.status}")
print(f"[@DEBUG] Response body: {await resp.text()}") print(f"[@DEBUG] Response body: {await resp.text()}")
except Exception as e: except Exception as e:
print(f"\nRequest failed: {e}") print(f"\nRequest failed: {e}")
os._exit(-1) os._exit(-1)
async def main(): async def main():
# Await the task instead of just creating it # Await the task instead of just creating it
await idp_request("RUNUPDATE") await idp_request("RUNUPDATE")
if __name__ == "__main__": if __name__ == "__main__":
asyncio.run(main()) asyncio.run(main())
src/test_server.py → tests/test_server.py
+452 -452
View File
@@ -1,452 +1,452 @@
import argparse import argparse
import asyncio import asyncio
import json import json
import ssl import ssl
from aiohttp import web from aiohttp import web
from base64 import b64decode from base64 import b64decode
from typing import Dict, Any from typing import Dict, Any
import uuid import uuid
import dotenv import dotenv
import os import os
import re import re
import logging import logging
from logging.handlers import RotatingFileHandler from logging.handlers import RotatingFileHandler
from pathlib import Path from pathlib import Path
import banner import banner
import get_sq_env import get_sq_env
from time import time from time import time
time = int(time()) time = int(time())
import asquare as sq import asquare as sq
# Square SDK webhook validation # Square SDK webhook validation
from square.utils.webhooks_helper import verify_signature from square.utils.webhooks_helper import verify_signature
# from asquareclass import AsyncSquareClient # type: ignore # from asquareclass import AsyncSquareClient # type: ignore
from abricksetapi import AsyncBricksetAPI from abricksetapi import AsyncBricksetAPI
print("Loading environment variables...") print("Loading environment variables...")
dotenv.load_dotenv() dotenv.load_dotenv()
SQUARE_LEGO_CAT_ID = os.environ.get("SQUARE_LEGO_CAT_ID") SQUARE_LEGO_CAT_ID = os.environ.get("SQUARE_LEGO_CAT_ID")
SQUARE_WEBHOOK_SIGNATURE_KEY = os.environ.get("SQUARE_WEBHOOK_SIGNATURE_KEY") SQUARE_WEBHOOK_SIGNATURE_KEY = os.environ.get("SQUARE_WEBHOOK_SIGNATURE_KEY")
SQUARE_WEBHOOK_URL = os.environ.get("SQUARE_WEBHOOK_URL") SQUARE_WEBHOOK_URL = os.environ.get("SQUARE_WEBHOOK_URL")
from asyncio_throttle.throttler import Throttler from asyncio_throttle.throttler import Throttler
# set up logger # set up logger
print("Setting up logger...") print("Setting up logger...")
logger = logging.getLogger("studspider") logger = logging.getLogger("studspider")
logger.setLevel(logging.DEBUG) logger.setLevel(logging.DEBUG)
handler = RotatingFileHandler("./logs/studspider.log", maxBytes=125000, backupCount=5) handler = RotatingFileHandler("./logs/studspider.log", maxBytes=125000, backupCount=5)
logger.addHandler(handler) logger.addHandler(handler)
lego = AsyncBricksetAPI(debug=True, logger=logger) lego = AsyncBricksetAPI(debug=True, logger=logger)
# sq = AsyncSquareClient(logger=logger) # sq = AsyncSquareClient(logger=logger)
# logging.basicConfig( # logging.basicConfig(
# filename="./logs/studspider.log", encoding="utf-8", level=logging.DEBUG # filename="./logs/studspider.log", encoding="utf-8", level=logging.DEBUG
# ) # )
print("Setting up rate limiter...") print("Setting up rate limiter...")
print("Doing RE things...") print("Doing RE things...")
UA = re.compile(r"^kopling\/?", re.IGNORECASE) UA = re.compile(r"^kopling\/?", re.IGNORECASE)
async def run_update(): async def run_update():
set_list = await sq.acat2id(SQUARE_LEGO_CAT_ID) # type: ignore set_list = await sq.acat2id(SQUARE_LEGO_CAT_ID) # type: ignore
# print(f"[@DEBUG]\t {set_list}") # type: ignore # print(f"[@DEBUG]\t {set_list}") # type: ignore
for l_set in set_list: for l_set in set_list:
# print(f"[@DEBUG]{l_set}") # print(f"[@DEBUG]{l_set}")
try: try:
l_setinfo = await lego.throttled_get_set(l_set["number"]) l_setinfo = await lego.throttled_get_set(l_set["number"])
print(f"[@DEBUG] Found {l_setinfo["pw_name"]}") print(f"[@DEBUG] Found {l_setinfo["pw_name"]}")
except Exception as e: except Exception as e:
print(f"Error fetching set\t: {l_set}: {str(e)}") print(f"Error fetching set\t: {l_set}: {str(e)}")
continue continue
# print(f"[@DEBUG]{l_setinfo}") # print(f"[@DEBUG]{l_setinfo}")
# await sq.update_itm(l_set, l_setinfo) # await sq.update_itm(l_set, l_setinfo)
"""for l_set in set_list: """for l_set in set_list:
print(f"[@DEBUG]{l_set}") print(f"[@DEBUG]{l_set}")
try: try:
l_setinfo = await lego.throttled_get_set(l_set["number"]) l_setinfo = await lego.throttled_get_set(l_set["number"])
except Exception as e: except Exception as e:
print(f"Error fetching set\t: {l_set}: {str(e)}") print(f"Error fetching set\t: {l_set}: {str(e)}")
continue continue
print(f"[@DEBUG]{l_setinfo}")""" print(f"[@DEBUG]{l_setinfo}")"""
# await sq.update_itm(l_set, l_setinfo) # await sq.update_itm(l_set, l_setinfo)
def valid_uuid(val): def valid_uuid(val):
try: try:
uuid.UUID(str(val)) uuid.UUID(str(val))
return True return True
except ValueError: except ValueError:
return False return False
# In-memory store for processed event IDs (idempotency) # In-memory store for processed event IDs (idempotency)
processed_event_ids = set() processed_event_ids = set()
MAX_STORED_EVENT_IDS = 10000 # Prevent memory growth MAX_STORED_EVENT_IDS = 10000 # Prevent memory growth
async def hook_handler(request: web.Request) -> web.Response: async def hook_handler(request: web.Request) -> web.Response:
"""Handle incoming JSON webhook POSTs on /hook endpoint""" """Handle incoming JSON webhook POSTs on /hook endpoint"""
try: try:
current_time = int(asyncio.get_event_loop().time()) current_time = int(asyncio.get_event_loop().time())
# Verify request method # Verify request method
if request.method != "POST": if request.method != "POST":
logger.error( logger.error(
f"{current_time}\tInvalid request method to /hook from %s", f"{current_time}\tInvalid request method to /hook from %s",
request.remote, request.remote,
) )
return web.Response(status=405, text="Method not allowed. Use POST.") return web.Response(status=405, text="Method not allowed. Use POST.")
# Read raw body for signature verification # Read raw body for signature verification
body = await request.read() body = await request.read()
body_str = body.decode("utf-8") body_str = body.decode("utf-8")
# Verify Square webhook signature using Square SDK # Verify Square webhook signature using Square SDK
signature_header = request.headers.get("x-square-hmacsha256-signature", "") signature_header = request.headers.get("x-square-hmacsha256-signature", "")
if SQUARE_WEBHOOK_SIGNATURE_KEY and SQUARE_WEBHOOK_URL: if SQUARE_WEBHOOK_SIGNATURE_KEY and SQUARE_WEBHOOK_URL:
if not signature_header: if not signature_header:
logger.error( logger.error(
f"{current_time}\tMissing signature header from {request.remote}" f"{current_time}\tMissing signature header from {request.remote}"
) )
print(f"[@DEBUG] Missing x-square-hmacsha256-signature header") print(f"[@DEBUG] Missing x-square-hmacsha256-signature header")
return web.Response(status=401, text="Unauthorized: Missing signature") return web.Response(status=401, text="Unauthorized: Missing signature")
# Use Square SDK's official validation function # Use Square SDK's official validation function
is_valid = verify_signature( is_valid = verify_signature(
request_body=body_str, request_body=body_str,
signature_header=signature_header, signature_header=signature_header,
signature_key=SQUARE_WEBHOOK_SIGNATURE_KEY, signature_key=SQUARE_WEBHOOK_SIGNATURE_KEY,
notification_url=SQUARE_WEBHOOK_URL, notification_url=SQUARE_WEBHOOK_URL,
) )
if not is_valid: if not is_valid:
logger.error(f"{current_time}\tInvalid signature from {request.remote}") logger.error(f"{current_time}\tInvalid signature from {request.remote}")
print(f"[@DEBUG] Invalid webhook signature!") print(f"[@DEBUG] Invalid webhook signature!")
return web.Response(status=401, text="Unauthorized: Invalid signature") return web.Response(status=401, text="Unauthorized: Invalid signature")
print(f"[@DEBUG] ✓ Webhook signature verified") print(f"[@DEBUG] ✓ Webhook signature verified")
else: else:
print( print(
f"[@DEBUG] ⚠️ Webhook signature verification DISABLED (set SQUARE_WEBHOOK_SIGNATURE_KEY and SQUARE_WEBHOOK_URL in .env)" f"[@DEBUG] ⚠️ Webhook signature verification DISABLED (set SQUARE_WEBHOOK_SIGNATURE_KEY and SQUARE_WEBHOOK_URL in .env)"
) )
# Parse JSON data # Parse JSON data
try: try:
data = json.loads(body_str) data = json.loads(body_str)
except json.JSONDecodeError as e: except json.JSONDecodeError as e:
print(f"[@DEBUG] Invalid JSON to /hook from {request.remote}: {str(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)}") logger.error(f"Invalid JSON to /hook from {request.remote}: {str(e)}")
return web.Response(status=400, text="Invalid JSON in request body") return web.Response(status=400, text="Invalid JSON in request body")
# Check for idempotency (event_id) # Check for idempotency (event_id)
event_id = data.get("event_id") event_id = data.get("event_id")
if event_id: if event_id:
if event_id in processed_event_ids: if event_id in processed_event_ids:
print( print(
f"[@DEBUG] Duplicate event {event_id} - already processed (idempotency)" f"[@DEBUG] Duplicate event {event_id} - already processed (idempotency)"
) )
logger.info( logger.info(
f"{current_time}\tDuplicate event {event_id} from {request.remote}" f"{current_time}\tDuplicate event {event_id} from {request.remote}"
) )
return web.json_response( return web.json_response(
{ {
"status": "success", "status": "success",
"message": "Event already processed (idempotency)", "message": "Event already processed (idempotency)",
"event_id": event_id, "event_id": event_id,
} }
) )
# Add to processed set # Add to processed set
processed_event_ids.add(event_id) processed_event_ids.add(event_id)
# Prevent memory growth - remove oldest if too many # Prevent memory growth - remove oldest if too many
if len(processed_event_ids) > MAX_STORED_EVENT_IDS: if len(processed_event_ids) > MAX_STORED_EVENT_IDS:
processed_event_ids.pop() processed_event_ids.pop()
# Get event info for quick response # Get event info for quick response
webhook_type = data.get("type", "unknown") webhook_type = data.get("type", "unknown")
event_id_val = data.get("event_id", "none") event_id_val = data.get("event_id", "none")
# RESPOND IMMEDIATELY - Square requires fast response (< 2 seconds) # RESPOND IMMEDIATELY - Square requires fast response (< 2 seconds)
# Don't do any processing before responding! # Don't do any processing before responding!
response_data = { response_data = {
"status": "success", "status": "success",
"message": "Webhook received", "message": "Webhook received",
"event_id": event_id_val, "event_id": event_id_val,
} }
# Log and process AFTER sending response (async background tasks) # Log and process AFTER sending response (async background tasks)
async def log_and_process(): async def log_and_process():
# Log the received webhook with detailed output # Log the received webhook with detailed output
print(f"\n{'='*60}") print(f"\n{'='*60}")
print(f"[@DEBUG] WEBHOOK RECEIVED at {current_time}") print(f"[@DEBUG] WEBHOOK RECEIVED at {current_time}")
print(f"[@DEBUG] From: {request.remote}") print(f"[@DEBUG] From: {request.remote}")
print(f"[@DEBUG] Headers: {dict(request.headers)}") print(f"[@DEBUG] Headers: {dict(request.headers)}")
print(f"[@DEBUG] Data type: {webhook_type}") print(f"[@DEBUG] Data type: {webhook_type}")
print(f"[@DEBUG] Event ID: {event_id_val}") print(f"[@DEBUG] Event ID: {event_id_val}")
print(f"[@DEBUG] Full payload:") print(f"[@DEBUG] Full payload:")
print(json.dumps(data, indent=2)) print(json.dumps(data, indent=2))
print(f"{'='*60}\n") print(f"{'='*60}\n")
logger.info( logger.info(
f"{current_time}\tWebhook received from {request.remote}: {json.dumps(data)}" f"{current_time}\tWebhook received from {request.remote}: {json.dumps(data)}"
) )
# Process the webhook data # Process the webhook data
if webhook_type == "inventory.count.updated": if webhook_type == "inventory.count.updated":
print(f"[@DEBUG] Processing inventory count update...") print(f"[@DEBUG] Processing inventory count update...")
# Extract inventory data # Extract inventory data
inventory_counts = ( inventory_counts = (
data.get("data", {}).get("object", {}).get("inventory_counts", []) data.get("data", {}).get("object", {}).get("inventory_counts", [])
) )
for count in inventory_counts: for count in inventory_counts:
print( print(
f"[@DEBUG] Catalog Object: {count.get('catalog_object_id')}" f"[@DEBUG] Catalog Object: {count.get('catalog_object_id')}"
) )
print(f"[@DEBUG] Quantity: {count.get('quantity')}") print(f"[@DEBUG] Quantity: {count.get('quantity')}")
print(f"[@DEBUG] State: {count.get('state')}") print(f"[@DEBUG] State: {count.get('state')}")
# Start background processing (don't wait for it) # Start background processing (don't wait for it)
asyncio.create_task(log_and_process()) asyncio.create_task(log_and_process())
# Return response immediately # Return response immediately
return web.json_response(response_data) return web.json_response(response_data)
except json.JSONDecodeError as e: except json.JSONDecodeError as e:
error_msg = f"JSON decode error in webhook: {str(e)}" error_msg = f"JSON decode error in webhook: {str(e)}"
print(f"[@DEBUG] {error_msg}") print(f"[@DEBUG] {error_msg}")
logger.error(error_msg) logger.error(error_msg)
return web.Response(status=400, text=error_msg) return web.Response(status=400, text=error_msg)
except Exception as e: except Exception as e:
error_msg = f"Webhook handler error: {str(e)}" error_msg = f"Webhook handler error: {str(e)}"
print(f"[@DEBUG] {error_msg}") print(f"[@DEBUG] {error_msg}")
print(f"[@DEBUG] Exception type: {type(e).__name__}") print(f"[@DEBUG] Exception type: {type(e).__name__}")
import traceback import traceback
print(f"[@DEBUG] Traceback:\n{traceback.format_exc()}") print(f"[@DEBUG] Traceback:\n{traceback.format_exc()}")
logger.error(f"{error_msg}\n{traceback.format_exc()}") logger.error(f"{error_msg}\n{traceback.format_exc()}")
return web.Response(status=500, text="Internal server error") return web.Response(status=500, text="Internal server error")
async def request_handler(request: web.Request) -> web.Response: async def request_handler(request: web.Request) -> web.Response:
"""Handle incoming POST requests with command processing""" """Handle incoming POST requests with command processing"""
try: try:
# Verify request method # Verify request method
if request.method != "POST": if request.method != "POST":
logger.error(f"{time}\tInvalid request method from %s at ", request.remote) logger.error(f"{time}\tInvalid request method from %s at ", request.remote)
return web.Response(status=405, text="Method not allowed. Use POST.") return web.Response(status=405, text="Method not allowed. Use POST.")
# Get JSON data from request # Get JSON data from request
try: try:
data = await request.json() data = await request.json()
except json.JSONDecodeError: except json.JSONDecodeError:
logger.error("Invalid JSON from %s", request.remote) logger.error("Invalid JSON from %s", request.remote)
return web.Response( return web.Response(
status=400, text="{time()}\tInvalid JSON in request body" status=400, text="{time()}\tInvalid JSON in request body"
) )
# Validate Idempotency-Key header # Validate Idempotency-Key header
if valid_uuid(request.headers.get("Idempotency-Key")) is False: if valid_uuid(request.headers.get("Idempotency-Key")) is False:
logger.error("{time()}\tInvalid Idempotency-Key from %s", request.remote) logger.error("{time()}\tInvalid Idempotency-Key from %s", request.remote)
return web.Response(status=412, text="412 Headers rejected!") return web.Response(status=412, text="412 Headers rejected!")
# Validate User-Agent header # Validate User-Agent header
if not re.match(UA, str(request.headers.get("User-Agent"))): if not re.match(UA, str(request.headers.get("User-Agent"))):
logger.error("{time()}\tInvalid User-Agent from %s", request.remote) logger.error("{time()}\tInvalid User-Agent from %s", request.remote)
return web.Response(status=412, text="412 Headers rejected!") return web.Response(status=412, text="412 Headers rejected!")
# Validate command exists # Validate command exists
if "CMD" not in data: if "CMD" not in data:
logger.error("{time()}\tMissing CMD field from %s", request.remote) logger.error("{time()}\tMissing CMD field from %s", request.remote)
return web.Response(status=400, text="Missing CMD field in request") return web.Response(status=400, text="Missing CMD field in request")
# Decode and process command # Decode and process command
try: try:
decoded_cmd = b64decode(data["CMD"]).decode("utf-8") decoded_cmd = b64decode(data["CMD"]).decode("utf-8")
print( print(
f"""At: {time}\nReceived command: {decoded_cmd}\nUser-Agent: {request.headers.get('User-Agent')}\nIdempotency-Key: {request.headers.get('Idempotency-Key')}""" f"""At: {time}\nReceived command: {decoded_cmd}\nUser-Agent: {request.headers.get('User-Agent')}\nIdempotency-Key: {request.headers.get('Idempotency-Key')}"""
) )
logging.info( logging.info(
f"{time}\tCommand: {decoded_cmd}\tFrom: {request.remote}\tUA: {request.headers.get('User-Agent')}\tIdempotency-Key: {request.headers.get('Idempotency-Key')}" f"{time}\tCommand: {decoded_cmd}\tFrom: {request.remote}\tUA: {request.headers.get('User-Agent')}\tIdempotency-Key: {request.headers.get('Idempotency-Key')}"
) )
# Process the command # Process the command
if decoded_cmd == "RUNUPDATE": if decoded_cmd == "RUNUPDATE":
response_data = {"status": "success", "message": "Update started"} response_data = {"status": "success", "message": "Update started"}
await run_update() await run_update()
logging.info(f"{time}\tUpdate command processed successfully") logging.info(f"{time}\tUpdate command processed successfully")
return web.json_response(response_data) return web.json_response(response_data)
else: else:
return web.Response(status=400, text=f"Unknown command: {decoded_cmd}") return web.Response(status=400, text=f"Unknown command: {decoded_cmd}")
except Exception as e: except Exception as e:
print(f"Error processing command: {str(e)}") print(f"Error processing command: {str(e)}")
return web.Response(status=500, text=f"Server error: {str(e)}") return web.Response(status=500, text=f"Server error: {str(e)}")
except Exception as e: except Exception as e:
print(f"Request handler error: {str(e)}") print(f"Request handler error: {str(e)}")
return web.Response(status=500, text="Internal server error") return web.Response(status=500, text="Internal server error")
print("Starting web server...") print("Starting web server...")
os.system("cls" if os.name == "nt" else "clear") os.system("cls" if os.name == "nt" else "clear")
app = web.Application() app = web.Application()
app.router.add_post("/api", request_handler) app.router.add_post("/api", request_handler)
app.router.add_post("/hook", hook_handler) app.router.add_post("/hook", hook_handler)
parser = argparse.ArgumentParser( parser = argparse.ArgumentParser(
description="studspider - LEGO Inventory Sync Service for Square POS and Brickset", description="studspider - LEGO Inventory Sync Service for Square POS and Brickset",
epilog=""" epilog="""
Examples: Examples:
# Basic HTTP server (recommended with nginx reverse proxy) # Basic HTTP server (recommended with nginx reverse proxy)
python src/test_server.py --host 0.0.0.0 --port 7432 python src/test_server.py --host 0.0.0.0 --port 7432
# With separate webhook port # With separate webhook port
python src/test_server.py --port 7432 --hook-port 8080 python src/test_server.py --port 7432 --hook-port 8080
# HTTPS using certificates from .env (set SSL_CERT_PATH and SSL_KEY_PATH) # HTTPS using certificates from .env (set SSL_CERT_PATH and SSL_KEY_PATH)
python src/test_server.py --hook-port 8443 --hook-ssl-only python src/test_server.py --hook-port 8443 --hook-ssl-only
# HTTPS with command line override (self-signed cert for testing) # 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 python src/test_server.py --ssl-cert cert.pem --ssl-key key.pem --hook-port 8443 --hook-ssl-only
# Full HTTPS (both endpoints) # Full HTTPS (both endpoints)
python src/test_server.py --ssl-cert cert.pem --ssl-key key.pem --hook-port 8443 python src/test_server.py --ssl-cert cert.pem --ssl-key key.pem --hook-port 8443
# Production behind nginx (no SSL needed in app) # Production behind nginx (no SSL needed in app)
python src/test_server.py --host 127.0.0.1 --port 7432 python src/test_server.py --host 127.0.0.1 --port 7432
Environment Variables (.env): Environment Variables (.env):
SSL_CERT_PATH - Default SSL certificate path SSL_CERT_PATH - Default SSL certificate path
SSL_KEY_PATH - Default SSL key path SSL_KEY_PATH - Default SSL key path
For more information, see doc/ For more information, see doc/
""", """,
formatter_class=argparse.RawDescriptionHelpFormatter, formatter_class=argparse.RawDescriptionHelpFormatter,
) )
parser.add_argument( parser.add_argument(
"--host", "--host",
type=str, type=str,
default="localhost", default="localhost",
help="IP address to listen on (default: localhost, use 0.0.0.0 for all interfaces)", help="IP address to listen on (default: localhost, use 0.0.0.0 for all interfaces)",
) )
parser.add_argument( parser.add_argument(
"--port", type=int, default=7432, help="Port for main API endpoint (default: 7432)" "--port", type=int, default=7432, help="Port for main API endpoint (default: 7432)"
) )
parser.add_argument( parser.add_argument(
"--hook-port", "--hook-port",
type=int, type=int,
default=None, default=None,
help="Separate port for webhook endpoint (default: None - uses same port as --port)", help="Separate port for webhook endpoint (default: None - uses same port as --port)",
) )
parser.add_argument( parser.add_argument(
"--ssl-cert", "--ssl-cert",
type=str, type=str,
default=get_sq_env.SSL_CERT_PATH, default=get_sq_env.SSL_CERT_PATH,
help="Path to SSL certificate file in PEM format (default: from .env SSL_CERT_PATH)", help="Path to SSL certificate file in PEM format (default: from .env SSL_CERT_PATH)",
) )
parser.add_argument( parser.add_argument(
"--ssl-key", "--ssl-key",
type=str, type=str,
default=get_sq_env.SSL_KEY_PATH, default=get_sq_env.SSL_KEY_PATH,
help="Path to SSL private key file in PEM format (default: from .env SSL_KEY_PATH)", help="Path to SSL private key file in PEM format (default: from .env SSL_KEY_PATH)",
) )
parser.add_argument( parser.add_argument(
"--hook-ssl-only", "--hook-ssl-only",
action="store_true", action="store_true",
help="Enable HTTPS only for webhook endpoint, main API stays HTTP (requires --ssl-cert and --ssl-key)", help="Enable HTTPS only for webhook endpoint, main API stays HTTP (requires --ssl-cert and --ssl-key)",
) )
args = parser.parse_args() args = parser.parse_args()
if __name__ == "__main__": if __name__ == "__main__":
print("Creating log directory if it doesn't exist...") print("Creating log directory if it doesn't exist...")
os.makedirs("./logs/", exist_ok=True) os.makedirs("./logs/", exist_ok=True)
print(banner.generate("studspider v0.1")) print(banner.generate("studspider v0.1"))
print("Built for PLAY-WELL TEKNOLOGIES by contact@studspider.sh\n\n") print("Built for PLAY-WELL TEKNOLOGIES by contact@studspider.sh\n\n")
# Setup SSL context if certificates provided # Setup SSL context if certificates provided
ssl_context = None ssl_context = None
if args.ssl_cert and args.ssl_key: if args.ssl_cert and args.ssl_key:
# Check if trying to use P7B file # Check if trying to use P7B file
if args.ssl_cert.endswith(".p7b") or args.ssl_key.endswith(".p7b"): if args.ssl_cert.endswith(".p7b") or args.ssl_key.endswith(".p7b"):
print("ERROR: Cannot use P7B files directly for HTTPS server!") print("ERROR: Cannot use P7B files directly for HTTPS server!")
print("\nP7B files are for Square's webhook signature verification only.") print("\nP7B files are for Square's webhook signature verification only.")
print("Your server needs PEM format files (.pem)") print("Your server needs PEM format files (.pem)")
print("\nWorkflow:") print("\nWorkflow:")
print("1. Generate PEM files: python generate_cert.py") print("1. Generate PEM files: python generate_cert.py")
print("2. Convert to P7B for Square: python convert_to_p7b.py") print("2. Convert to P7B for Square: python convert_to_p7b.py")
print("3. Upload P7B to Square Dashboard") print("3. Upload P7B to Square Dashboard")
print("4. Run server with PEM files:") print("4. Run server with PEM files:")
print( print(
" python src/test_server.py --ssl-cert cert.pem --ssl-key key.pem --hook-port 8443" " 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("\nIf you have a P7B and need PEM:")
print(" python convert_from_p7b.py your_cert.p7b") print(" python convert_from_p7b.py your_cert.p7b")
exit(1) exit(1)
try: try:
ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH) ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
ssl_context.load_cert_chain(args.ssl_cert, args.ssl_key) ssl_context.load_cert_chain(args.ssl_cert, args.ssl_key)
print(f"✓ SSL enabled with cert: {args.ssl_cert}") print(f"✓ SSL enabled with cert: {args.ssl_cert}")
except FileNotFoundError as e: except FileNotFoundError as e:
print(f"ERROR: Certificate file not found: {e}") print(f"ERROR: Certificate file not found: {e}")
exit(1) exit(1)
except ssl.SSLError as e: except ssl.SSLError as e:
print(f"ERROR: Invalid certificate or key file: {e}") print(f"ERROR: Invalid certificate or key file: {e}")
print("\nMake sure you're using PEM format files.") print("\nMake sure you're using PEM format files.")
exit(1) exit(1)
elif args.ssl_cert or args.ssl_key: elif args.ssl_cert or args.ssl_key:
print("ERROR: Both --ssl-cert and --ssl-key must be provided together") print("ERROR: Both --ssl-cert and --ssl-key must be provided together")
exit(1) exit(1)
else: else:
print("ℹ️ Running without SSL (HTTP only)") print("ℹ️ Running without SSL (HTTP only)")
print(" For production with nginx reverse proxy, this is recommended.") print(" For production with nginx reverse proxy, this is recommended.")
print(" Nginx will handle HTTPS, forward HTTP to this server.") print(" Nginx will handle HTTPS, forward HTTP to this server.")
# If hook-port is specified, run a separate server for webhooks # If hook-port is specified, run a separate server for webhooks
if args.hook_port: if args.hook_port:
protocol_main = "https" if (ssl_context and not args.hook_ssl_only) else "http" protocol_main = "https" if (ssl_context and not args.hook_ssl_only) else "http"
protocol_hook = "https" if ssl_context else "http" protocol_hook = "https" if ssl_context else "http"
# Create separate app for webhook # Create separate app for webhook
hook_app = web.Application() hook_app = web.Application()
hook_app.router.add_post("/hook", hook_handler) hook_app.router.add_post("/hook", hook_handler)
# Run both servers concurrently # Run both servers concurrently
async def run_servers(): async def run_servers():
runner1 = web.AppRunner(app) runner1 = web.AppRunner(app)
await runner1.setup() await runner1.setup()
# Main server uses SSL only if not hook-ssl-only mode # Main server uses SSL only if not hook-ssl-only mode
main_ssl = None if args.hook_ssl_only else ssl_context main_ssl = None if args.hook_ssl_only else ssl_context
site1 = web.TCPSite(runner1, args.host, args.port, ssl_context=main_ssl) site1 = web.TCPSite(runner1, args.host, args.port, ssl_context=main_ssl)
runner2 = web.AppRunner(hook_app) runner2 = web.AppRunner(hook_app)
await runner2.setup() await runner2.setup()
# Webhook server always uses SSL if certificates provided # Webhook server always uses SSL if certificates provided
site2 = web.TCPSite( site2 = web.TCPSite(
runner2, args.host, args.hook_port, ssl_context=ssl_context runner2, args.host, args.hook_port, ssl_context=ssl_context
) )
await site1.start() await site1.start()
await site2.start() await site2.start()
print(f"Starting main server on {args.host}:{args.port} ({protocol_main})") print(f"Starting main server on {args.host}:{args.port} ({protocol_main})")
print( print(
f"Starting webhook server on {args.host}:{args.hook_port} ({protocol_hook})" f"Starting webhook server on {args.host}:{args.hook_port} ({protocol_hook})"
) )
# Keep running # Keep running
await asyncio.Event().wait() await asyncio.Event().wait()
asyncio.run(run_servers()) asyncio.run(run_servers())
else: else:
protocol = "https" if ssl_context else "http" protocol = "https" if ssl_context else "http"
print(f"Server running on {args.host}:{args.port} ({protocol})") print(f"Server running on {args.host}:{args.port} ({protocol})")
print(f"Webhook endpoint: {protocol}://{args.host}:{args.port}/hook") print(f"Webhook endpoint: {protocol}://{args.host}:{args.port}/hook")
web.run_app(app, host=args.host, port=args.port, ssl_context=ssl_context) web.run_app(app, host=args.host, port=args.port, ssl_context=ssl_context)
src/webhook/test_square_webhook.py → tests/test_square_webhook.py
View File
File renamed without changes.
src/test_webhook.py → tests/test_webhook.py
View File
File renamed without changes.