4 Commits
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

No files matched your search

+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 API Configuration
# ============================================================================
# Square environment - "sandbox" for testing, "production" for live store
SQUARE_ENV = "sandbox" # or "production"
# Environment: "production" or "sandbox"
SQUARE_ENV=sandbox
SQUARE_WEBHOOK_URL = "https://yourdomain.com/hook"
# Sandbox credentials
SANDBOX_SQUARE_APP_ID=your_sandbox_app_id
SANDBOX_SQUARE_TOKEN=your_sandbox_access_token
# Sandbox credentials - test environment (fake data, safe for development)
SANDBOX_SQUARE_APP_ID = "your_sandbox_app_id"
SANDBOX_SQUARE_TOKEN = "your_sandbox_access_token"
SANDBOX_SQUARE_WEBHOOK_INVENTORY_ID = "your_sandbox_webhook_signature_key"
# Production credentials
PROD_SQUARE_APP_ID=your_production_app_id
PROD_SQUARE_TOKEN=your_production_access_token
# Production credentials - live store (real inventory and sales)
PROD_SQUARE_APP_ID = "your_production_app_id"
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)
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
# Category ID for LEGO items in Square catalog (find in Dashboard → Items → Categories)
SQUARE_LEGO_CAT_ID = "your_lego_category_id"
+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
Clone the project
```bash
git clone http://git.traps.moe/ph0sgene/studspider.git
git checkout uv
git checkout branch
```
Go to the project directory
@@ -30,14 +30,15 @@ Install dependencies
Start the server
```bash
uvx run studspider.py
uv run studspider.py
```
## Tech
- gunicorn
- uvicorn
- aiohttp
- squareup
- apscheduler
- aiosqlite
- aiosqlite
@@ -1,8 +1,3 @@
"""
Gunicorn configuration for StudSpider webhook server
Production-ready settings for running with nginx reverse proxy
"""
import multiprocessing
import os
-11
View File
@@ -1,11 +0,0 @@
{
"folders": [
{
"path": "."
},
{
"path": "../kopling"
}
],
"settings": {}
}
+116 -116
View File
@@ -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}")
+113 -113
View File
@@ -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)
-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")
+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:
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")
+39 -39
View File
@@ -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())
+452 -452
View File
@@ -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)
File renamed without changes.
File renamed without changes.