4 Commits
Author SHA1 Message Date
C F d25e5aac60 fix type error 2025-12-17 14:56:57 -08:00
C F c88695f1e7 fix sq path so test works 2025-12-17 14:56:46 -08:00
C F 8413a89a47 test works and tests port 2025-12-17 14:56:34 -08:00
C F 6bb11a206b docker it up 2025-12-17 14:56:21 -08:00
10 changed files with 69 additions and 785 deletions

No files matched your search

+21
View File
@@ -0,0 +1,21 @@
FROM python:3.13-alpine
# Install uv
RUN pip install uv
# Set working directory
WORKDIR /app
# Copy project files
COPY pyproject.toml ./
COPY README.md ./
# Install dependencies using uv without venv
RUN uv pip compile pyproject.toml -o requirements.txt && \
uv pip install --system -r requirements.txt
# Copy the rest of the application
COPY . .
# Default command
CMD ["python", "tests/test.py"]
-125
View File
@@ -1,125 +0,0 @@
# HTTPS Webhook Setup Guide
## Quick Start (Self-Signed Certificate for Testing)
### 1. Generate Certificate
```powershell
python generate_cert.py
```
This creates `cert.pem` and `key.pem` files.
### 2. Run Server with HTTPS
#### Option A: HTTPS for webhook only (recommended for development)
```powershell
python src/test_server.py --ssl-cert cert.pem --ssl-key key.pem --hook-port 8443 --hook-ssl-only
```
- Main server: `http://localhost:7432/` (HTTP)
- Webhook: `https://localhost:8443/hook` (HTTPS)
#### Option B: HTTPS for everything
```powershell
python src/test_server.py --ssl-cert cert.pem --ssl-key key.pem --hook-port 8443
```
- Main server: `https://localhost:7432/` (HTTPS)
- Webhook: `https://localhost:8443/hook` (HTTPS)
#### Option C: Single port with HTTPS
```powershell
python src/test_server.py --ssl-cert cert.pem --ssl-key key.pem
```
- Everything on: `https://localhost:7432/`
### 3. Test the Webhook
Update `src/test_webhook.py` to use HTTPS:
```python
url = "https://localhost:8443/hook"
```
Then run:
```powershell
python src/test_webhook.py
```
## Production Setup (with Real Certificate)
### Option 1: Let's Encrypt (Free, Recommended)
On Linux/WSL:
```bash
# Install certbot
sudo apt-get install certbot
# Generate certificate (replace yourdomain.com)
sudo certbot certonly --standalone -d yourdomain.com
# Certificates will be at:
# /etc/letsencrypt/live/yourdomain.com/fullchain.pem
# /etc/letsencrypt/live/yourdomain.com/privkey.pem
# Run server
python src/test_server.py \
--ssl-cert /etc/letsencrypt/live/yourdomain.com/fullchain.pem \
--ssl-key /etc/letsencrypt/live/yourdomain.com/privkey.pem \
--host 0.0.0.0 \
--hook-port 443 \
--hook-ssl-only
```
### Option 2: Reverse Proxy (nginx/Caddy)
Run studspider on HTTP internally, use nginx/Caddy for HTTPS:
```nginx
# nginx config
server {
listen 443 ssl;
server_name yourdomain.com;
ssl_certificate /path/to/cert.pem;
ssl_certificate_key /path/to/key.pem;
location /hook {
proxy_pass http://localhost:7432/hook;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
}
}
```
Then run studspider normally:
```powershell
python src/test_server.py
```
## Troubleshooting
### "OpenSSL not found" Error
- **Windows**: Install from https://slproweb.com/products/Win32OpenSSL.html
- **Alternative**: Use Git Bash or WSL which include OpenSSL
### Self-Signed Certificate Warnings
- Normal for development
- Use `--insecure` flag with curl/tools for testing
- For production, get a proper certificate from Let's Encrypt
### Port 443 Already in Use
- Use a different port like 8443
- Or stop the service using port 443
### Certificate File Not Found
- Use absolute paths: `--ssl-cert C:\path\to\cert.pem`
- Or run from the directory containing the certificates
## Square Webhook Configuration
When setting up webhooks in Square Dashboard:
1. Use your HTTPS URL: `https://yourdomain.com:8443/hook`
2. Square requires valid SSL (not self-signed)
3. Consider using ngrok for local testing:
```powershell
ngrok http 8443
```
Then use the ngrok HTTPS URL in Square Dashboard
-191
View File
@@ -1,191 +0,0 @@
# Square Webhook Setup - Quick Reference
## Complete Setup Process
### Step 1: Generate SSL Certificate
```powershell
# Generate PEM certificate
python generate_cert.py
# Creates: cert.pem, key.pem
```
### Step 2: Convert to P7B for Square
```powershell
# Convert PEM to P7B (PKCS#7) format
python convert_to_p7b.py
# Creates: cert.p7b
```
### Step 3: Start HTTPS Server
```powershell
# Start webhook server with HTTPS
python src/test_server.py --ssl-cert cert.pem --ssl-key key.pem --hook-port 8443 --hook-ssl-only
# Server URLs:
# Main: http://localhost:7432/
# Webhook: https://localhost:8443/hook (or https://yourdomain.com:8443/hook)
```
### Step 4: Configure Square Webhook
1. **Upload P7B Certificate to Square:**
- Go to: https://developer.squareup.com/apps
- Select your application
- Navigate to: **Webhooks** → **Manage Notification Settings**
- Upload `cert.p7b` file
- This allows Square to verify webhook signatures
2. **Add Webhook Subscription:**
- Still in Webhooks section
- Click **"Add Subscription"**
- Enter URL: `https://yourdomain.com:8443/hook`
- Select events to subscribe to (e.g., `inventory.count.updated`)
- Click **Save**
### Step 5: Test the Webhook
```powershell
# Test from command line
python src/test_webhook.py
# Or manually with curl
curl -X POST https://localhost:8443/hook \
-H "Content-Type: application/json" \
-d '{"type":"inventory.count.updated","merchant_id":"test"}'
```
## Production Deployment
### Using Let's Encrypt (Recommended)
```bash
# 1. Get Let's Encrypt certificate (Linux/WSL)
sudo certbot certonly --standalone -d yourdomain.com
# 2. Convert to P7B for Square
python convert_to_p7b.py \
--cert /etc/letsencrypt/live/yourdomain.com/fullchain.pem \
--output square.p7b
# 3. Upload square.p7b to Square Dashboard
# 4. Run server
python src/test_server.py \
--ssl-cert /etc/letsencrypt/live/yourdomain.com/fullchain.pem \
--ssl-key /etc/letsencrypt/live/yourdomain.com/privkey.pem \
--host 0.0.0.0 \
--hook-port 443 \
--hook-ssl-only
```
### Using ngrok (Easy Testing)
```powershell
# Terminal 1: Start your server (HTTP is fine)
python src/test_server.py --hook-port 8080
# Terminal 2: Expose with ngrok (provides HTTPS automatically)
ngrok http 8080
# Use the ngrok HTTPS URL in Square:
# Example: https://abc123.ngrok.io/hook
# For P7B certificate with ngrok:
# ngrok provides its own certificate, so you don't need to upload one to Square
```
## Webhook Event Types
Common Square webhook events you can subscribe to:
- `inventory.count.updated` - Inventory quantity changes
- `catalog.version.updated` - Catalog item updates
- `payment.created` - New payment received
- `payment.updated` - Payment status changes
- `order.created` - New order created
- `order.updated` - Order status changes
## Troubleshooting
### Certificate Upload Fails
- Ensure file is in P7B format (not PEM)
- File must be valid PKCS#7 structure
- Check file isn't corrupted: `openssl pkcs7 -print_certs -in cert.p7b`
### Webhook Not Receiving Data
1. Check server is running and accessible from internet
2. Verify URL in Square Dashboard is correct
3. Check firewall allows incoming traffic on webhook port
4. Review server logs: `./logs/studspider.log`
### SSL Certificate Errors
- Self-signed certificates won't work with Square in production
- Use Let's Encrypt or commercial CA for production
- Use ngrok for quick testing (provides valid HTTPS)
### Testing Locally
```powershell
# Square can't reach localhost - use one of these:
# 1. ngrok (easiest)
ngrok http 8443
# 2. Cloudflare Tunnel
cloudflared tunnel --url http://localhost:8443
# 3. Deploy to cloud (AWS, Azure, DigitalOcean, etc.)
```
## Verifying Webhook Signatures
Square signs webhook requests. Your server should verify signatures:
```python
# TODO: Add to hook_handler in test_server.py
import hmac
import hashlib
def verify_square_signature(payload: bytes, signature: str, webhook_signature_key: str) -> bool:
"""Verify Square webhook signature"""
computed_signature = hmac.new(
webhook_signature_key.encode('utf-8'),
payload,
hashlib.sha256
).digest()
expected_signature = base64.b64decode(signature)
return hmac.compare_digest(computed_signature, expected_signature)
```
Get your webhook signature key from Square Dashboard → Webhooks → Signature Key.
## Environment Variables
Add to your `.env` file:
```env
# Square Webhook Configuration
SQUARE_WEBHOOK_SIGNATURE_KEY=your_signature_key_from_square_dashboard
# SSL Certificate Paths (production)
SSL_CERT_PATH=/path/to/fullchain.pem
SSL_KEY_PATH=/path/to/privkey.pem
```
## Commands Quick Reference
```powershell
# Generate certificate
python generate_cert.py
# Convert to P7B
python convert_to_p7b.py
# Start HTTPS server
python src/test_server.py --ssl-cert cert.pem --ssl-key key.pem --hook-port 8443 --hook-ssl-only
# Test webhook
python src/test_webhook.py
# Start with ngrok
ngrok http 8443
```
-166
View File
@@ -1,166 +0,0 @@
"""
Convert PEM certificate to P7B (PKCS#7) format for Square webhooks
Square requires P7B format for webhook signature verification
"""
import subprocess
import sys
import os
def convert_pem_to_p7b(cert_file="cert.pem", output_file="cert.p7b"):
"""
Convert PEM certificate to P7B (PKCS#7) format
Args:
cert_file: Input PEM certificate file
output_file: Output P7B file
"""
# Check if OpenSSL is available
try:
subprocess.run(["openssl", "version"], capture_output=True, check=True)
except (subprocess.CalledProcessError, FileNotFoundError):
print("ERROR: OpenSSL not found!")
print("\nFor Windows, you can:")
print(
"1. Install OpenSSL from: https://slproweb.com/products/Win32OpenSSL.html"
)
print("2. Or use Git Bash which includes OpenSSL")
print("3. Or use WSL (Windows Subsystem for Linux)")
return False
# Check if input file exists
if not os.path.exists(cert_file):
print(f"ERROR: Certificate file not found: {cert_file}")
print(f"Current directory: {os.getcwd()}")
print("\nMake sure you've generated the certificate first:")
print(" python generate_cert.py")
return False
print(f"Converting {cert_file} to P7B format...")
# Convert PEM to P7B (PKCS#7)
# -certfile can include intermediate certificates if needed
cmd = [
"openssl",
"crl2pkcs7",
"-nocrl",
"-certfile",
cert_file,
"-out",
output_file,
]
try:
result = subprocess.run(cmd, check=True, capture_output=True, text=True)
if os.path.exists(output_file):
print(f"\n✓ P7B certificate created: {os.path.abspath(output_file)}")
print(f" File size: {os.path.getsize(output_file)} bytes")
print("\n" + "=" * 60)
print("Square Webhook Configuration:")
print("=" * 60)
print(f"1. Upload this file to Square Developer Dashboard:")
print(f" {os.path.abspath(output_file)}")
print("\n2. In Square Dashboard:")
print(" - Go to Webhooks settings")
print(" - Click 'Manage Notification Settings'")
print(" - Upload the P7B file for signature verification")
print("\n3. Square will use this to verify webhook signatures")
print("=" * 60)
# Show file contents preview
print("\nP7B file preview (first 500 chars):")
with open(output_file, "r") as f:
content = f.read(500)
print(content)
if len(content) == 500:
print("...")
return True
else:
print("ERROR: Output file was not created")
return False
except subprocess.CalledProcessError as e:
print(f"ERROR: Conversion failed: {e}")
if e.stderr:
print(f"OpenSSL error: {e.stderr}")
return False
def convert_with_chain(cert_file="cert.pem", chain_file=None, output_file="cert.p7b"):
"""
Convert PEM certificate with certificate chain to P7B format
Use this if you have intermediate certificates
Args:
cert_file: Your certificate file
chain_file: Intermediate/chain certificate file (optional)
output_file: Output P7B file
"""
if not chain_file:
return convert_pem_to_p7b(cert_file, output_file)
# Check files exist
for f in [cert_file, chain_file]:
if not os.path.exists(f):
print(f"ERROR: File not found: {f}")
return False
print(f"Converting {cert_file} with chain {chain_file} to P7B format...")
# Create combined cert file
combined = "combined_cert.pem"
with open(combined, "w") as out:
with open(cert_file, "r") as f:
out.write(f.read())
with open(chain_file, "r") as f:
out.write(f.read())
result = convert_pem_to_p7b(combined, output_file)
# Clean up combined file
if os.path.exists(combined):
os.remove(combined)
return result
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser(
description="Convert PEM certificate to P7B (PKCS#7) for Square webhooks"
)
parser.add_argument(
"--cert",
default="cert.pem",
help="Input PEM certificate file (default: cert.pem)",
)
parser.add_argument(
"--chain", default=None, help="Optional: Intermediate certificate chain file"
)
parser.add_argument(
"--output", default="cert.p7b", help="Output P7B file (default: cert.p7b)"
)
args = parser.parse_args()
print("=" * 60)
print("PEM to P7B Certificate Converter for Square Webhooks")
print("=" * 60)
print()
if args.chain:
success = convert_with_chain(args.cert, args.chain, args.output)
else:
success = convert_pem_to_p7b(args.cert, args.output)
if not success:
sys.exit(1)
print("\n✓ Conversion complete!")
-76
View File
@@ -1,76 +0,0 @@
"""
Generate self-signed SSL certificate for testing HTTPS webhooks
Run this script to create cert.pem and key.pem files
"""
import subprocess
import sys
import os
def generate_certificate():
"""Generate self-signed certificate using OpenSSL"""
cert_file = "cert.pem"
key_file = "key.pem"
# Check if OpenSSL is available
try:
subprocess.run(["openssl", "version"], capture_output=True, check=True)
except (subprocess.CalledProcessError, FileNotFoundError):
print("ERROR: OpenSSL not found!")
print("\nFor Windows, you can:")
print(
"1. Install OpenSSL from: https://slproweb.com/products/Win32OpenSSL.html"
)
print("2. Or use Git Bash which includes OpenSSL")
print("3. Or use WSL (Windows Subsystem for Linux)")
return False
print("Generating self-signed certificate...")
print("This will create cert.pem and key.pem files")
print()
# Generate certificate valid for 365 days
cmd = [
"openssl",
"req",
"-x509",
"-newkey",
"rsa:4096",
"-keyout",
key_file,
"-out",
cert_file,
"-days",
"365",
"-nodes", # No passphrase
"-subj",
"/C=US/ST=State/L=City/O=Organization/CN=localhost",
]
try:
subprocess.run(cmd, check=True)
print(f"\n✓ Certificate created: {os.path.abspath(cert_file)}")
print(f"✓ Private key created: {os.path.abspath(key_file)}")
print("\nYou can now run the server with HTTPS:")
print(
f" python src/test_server.py --ssl-cert {cert_file} --ssl-key {key_file} --hook-port 8443"
)
print("\nOr webhook-only HTTPS:")
print(
f" python src/test_server.py --ssl-cert {cert_file} --ssl-key {key_file} --hook-port 8443 --hook-ssl-only"
)
print(
"\nNote: Self-signed certificates will show security warnings in browsers."
)
print("For production, use certificates from a trusted CA like Let's Encrypt.")
return True
except subprocess.CalledProcessError as e:
print(f"ERROR: Failed to generate certificate: {e}")
return False
if __name__ == "__main__":
if not generate_certificate():
sys.exit(1)
+1 -1
View File
@@ -48,7 +48,7 @@ def get_set(set_number: int) -> dict:
}
# add PW 10% b4 tax
our_set.update(
{"set_adjustredprice": our_set + ((10 / 100) * our_set["set_retailprice"])}
{"set_adjustredprice": our_set["set_retailprice"] + ((10 / 100) * our_set["set_retailprice"])}
)
our_set["pw_name"] = (
f"{our_set['set_num']} - {our_set["set_theme"].upper()} {our_set['set_name']}"
+1 -1
View File
@@ -1,6 +1,6 @@
import json
import get_sq_env
from . import get_sq_env
SQUARE_TOKEN = get_sq_env.SQUARE_TOKEN
LEGO_CAT_ID = get_sq_env.LEGO_CAT_ID
+20 -122
View File
@@ -1,7 +1,6 @@
import argparse
import asyncio
import json
import ssl
from aiohttp import web
from base64 import b64decode
from typing import Dict, Any
@@ -88,77 +87,41 @@ def valid_uuid(val):
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,
f"{time}\tInvalid request method to /hook from %s", 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 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)}")
except json.JSONDecodeError:
logger.error("Invalid JSON to /hook from %s", request.remote)
return web.Response(status=400, text="Invalid JSON in request body")
# 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: {data.get('type', 'unknown')}")
print(f"[@DEBUG] Event ID: {data.get('event_id', 'none')}")
print(f"[@DEBUG] Full payload:")
print(json.dumps(data, indent=2))
print(f"{'='*60}\n")
# Log the received webhook
print(f"[@DEBUG] Webhook received at {time} from {request.remote}")
logger.info(
f"{current_time}\tWebhook received from {request.remote}: {json.dumps(data)}"
f"{time}\tWebhook received from {request.remote}: {json.dumps(data)}"
)
# Process the webhook data here
# TODO: Add your webhook processing logic
webhook_type = data.get("type", "unknown")
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')}")
response_data = {
"status": "success",
"message": f"Webhook received: {webhook_type}",
"timestamp": current_time,
"event_id": data.get("event_id"),
"message": "Webhook received",
"timestamp": time,
"data_received": data,
}
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()}")
logger.error(f"Webhook handler error: {str(e)}")
print(f"Webhook handler error: {str(e)}")
return web.Response(status=500, text="Internal server error")
@@ -205,7 +168,7 @@ async def request_handler(request: web.Request) -> web.Response:
if decoded_cmd == "RUNUPDATE":
response_data = {"status": "success", "message": "Update started"}
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)
else:
return web.Response(status=400, text=f"Unknown command: {decoded_cmd}")
@@ -241,73 +204,16 @@ parser.add_argument(
default=8080,
help="Optional separate port for webhook endpoint. default: 8080",
)
parser.add_argument(
"--ssl-cert",
type=str,
default=None,
help="Path to SSL certificate file for HTTPS (PEM format)",
)
parser.add_argument(
"--ssl-key",
type=str,
default=None,
help="Path to SSL private key file for HTTPS (PEM format)",
)
parser.add_argument(
"--hook-ssl-only",
action="store_true",
help="Enable HTTPS only for webhook endpoint (main server stays HTTP)",
)
args = parser.parse_args()
if __name__ == "__main__":
print(banner.generate("studspider v0.1"))
print("Built for PLAY-WELL TEKNOLOGIES by contact@traps.moe\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.")
print("Generate certificates with: python generate_cert.py")
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"
print(f"Starting main server on {args.host}:{args.port}")
print(f"Starting webhook server on {args.host}:{args.hook_port}")
# Create separate app for webhook
hook_app = web.Application()
@@ -317,31 +223,23 @@ if __name__ == "__main__":
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)
site1 = web.TCPSite(runner1, args.host, args.port)
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
)
site2 = web.TCPSite(runner2, args.host, args.hook_port)
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})"
)
print(f"Main server running on http://{args.host}:{args.port}")
print(f"Webhook server running on http://{args.host}:{args.hook_port}/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)
print(f"Server running on {args.host}:{args.port}")
print(f"Webhook endpoint: http://{args.host}:{args.port}/hook")
web.run_app(app, host=args.host, port=args.port)
-102
View File
@@ -1,102 +0,0 @@
"""
Test script for webhook endpoint
Sends the Square inventory.count.updated webhook payload
"""
import asyncio
import aiohttp
import json
import ssl
import sys
async def test_webhook(verify_ssl=True):
url = "https://studspider.sh:8443/hook"
# Your Square webhook payload
payload = {
"merchant_id": "6SSW7HV8K2ST5",
"type": "inventory.count.updated",
"event_id": "df5f3813-a913-45a1-94e9-fdc3f7d5e3b6",
"created_at": "2019-10-29T18:38:45.455006797Z",
"data": {
"type": "inventory_counts",
"id": "84e4ac73-d605-4dbd-a9e5-ffff794ddb9d",
"object": {
"inventory_counts": [
{
"calculated_at": "2019-10-29T18:38:45.10296Z",
"catalog_object_id": "FGQ5JJWT2PYTHF35CKZ2DSKP",
"catalog_object_type": "ITEM_VARIATION",
"location_id": "YYQR03DGCTXA4",
"quantity": "10",
"state": "IN_STOCK",
}
]
},
},
}
print("Sending webhook test payload...")
print(f"URL: {url}")
print(
f"SSL Verification: {'Enabled' if verify_ssl else 'Disabled (accepting self-signed certs)'}"
)
print(json.dumps(payload, indent=2))
print("\n" + "=" * 60 + "\n")
# Create SSL context that allows self-signed certificates if verify_ssl is False
ssl_context = None
if not verify_ssl:
ssl_context = ssl.create_default_context()
ssl_context.check_hostname = False
ssl_context.verify_mode = ssl.CERT_NONE
try:
connector = aiohttp.TCPConnector(ssl=ssl_context)
async with aiohttp.ClientSession(connector=connector) as session:
async with session.post(url, json=payload) as resp:
print(f"Status: {resp.status}")
print(f"Response headers: {dict(resp.headers)}")
response_text = await resp.text()
print(f"\nResponse body:")
try:
response_json = json.loads(response_text)
print(json.dumps(response_json, indent=2))
except json.JSONDecodeError:
print(response_text)
except aiohttp.ClientConnectorError as e:
print(f"Connection error: {e}")
print("\nTroubleshooting:")
print("1. Make sure the server is running")
print("2. Check if the hostname and port are correct")
print("3. For self-signed certificates, the script now accepts them by default")
print("\nServer start command:")
print(
" python src/test_server.py --ssl-cert cert.pem --ssl-key key.pem --hook-port 8443 --hook-ssl-only"
)
except aiohttp.ServerTimeoutError as e:
print(f"Timeout error: {e}")
print("The server took too long to respond")
except Exception as e:
print(f"Error: {type(e).__name__}: {e}")
if __name__ == "__main__":
# Check command line arguments
verify_ssl = True
if len(sys.argv) > 1 and sys.argv[1] in ["--no-verify", "--insecure", "-k"]:
verify_ssl = False
print("⚠️ SSL verification disabled - accepting self-signed certificates\n")
elif len(sys.argv) > 1:
print("Usage: python src/test_webhook.py [--no-verify]")
print(
" --no-verify: Disable SSL certificate verification (for self-signed certs)"
)
sys.exit(1)
# Default to no verification since we're likely using self-signed certs
asyncio.run(test_webhook(verify_ssl=False))
+26 -1
View File
@@ -1,6 +1,31 @@
import sys
from pathlib import Path
from pprint import pprint
sys.path.insert(0, str(Path(__file__).parent.parent))
import socket
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
import src.libsquare as sq
import src.bricksetapi as lego
print("=" * 80)
result = sock.connect_ex(('127.0.0.1', 7432))
if result == 0:
print("PORT 7432 IS IN USE")
else:
print("PORT 7432 IS AVAILABLE")
sock.close()
print("=" * 80)
print("SQUARE OBJECT FOR SET 71417:")
print("=" * 80)
oursq = sq.get_itm("71417")
pprint(oursq, width=120, depth=None)
print("\n" + "=" * 80)
print("LEGO SET INFORMATION FOR SET 71417:")
print("=" * 80)
ourlego = lego.get_set(71417)
print(f"{oursq}\n\n{ourlego}\n\n")
print("=" * 80)
pprint(ourlego, width=120, depth=None)