#!/usr/bin/env python3 """Python 3.10+, standard library only. Synthetic HTTP sandbox; never sends a webhook.""" import hashlib import hmac import json import os import re import sqlite3 import tempfile import time import urllib.error import urllib.parse import urllib.request def validate_base(value): url = urllib.parse.urlsplit(value) main = url.scheme == 'https' and url.netloc == 'ludovex.org' local = url.scheme == 'http' and url.hostname in ('localhost', '127.0.0.1', '::1') if (not (main or local) or url.username or url.password or url.query or url.fragment or url.path != '/api/sandbox/v1'): raise ValueError('Only the main-domain sandbox or a loopback HTTP sandbox is allowed.') return value def verify_delivery(delivery, secret, now=None): now = int(time.time()) if now is None else now headers = delivery.get('headers', {}) timestamp = headers.get('X-Ludovex-Timestamp', '') signature = headers.get('X-Ludovex-Signature', '') delivery_id = headers.get('X-Ludovex-Delivery-Id', '') if not isinstance(timestamp, str) or not re.fullmatch(r'[0-9]{10}', timestamp): raise ValueError('Invalid timestamp') if not 0 <= now - int(timestamp) <= 300: raise ValueError('Stale or future timestamp') if not isinstance(signature, str) or not re.fullmatch(r'sha256=[a-f0-9]{64}', signature): raise ValueError('Invalid signature header') if not isinstance(delivery_id, str) or not re.fullmatch(r'[0-9]{18}', delivery_id): raise ValueError('Invalid delivery identifier') raw = delivery.get('raw_body') if not isinstance(raw, str) or len(raw.encode()) > 16384: raise ValueError('Invalid body') expected = hmac.new(secret.encode(), (timestamp + '.' + raw).encode(), hashlib.sha256).hexdigest() if not hmac.compare_digest(expected, signature[7:]): raise ValueError('Invalid signature') # Parse only after verifying exact bytes. Never HMAC a reserialized JSON value. event = json.loads(raw) if (event.get('sandbox') is not True or event.get('event') not in ('order.completed', 'order.failed', 'order.refunded') or not re.fullmatch(r'sb_ord_[a-f0-9]{24}', event.get('data', {}).get('id', ''))): raise ValueError('Invalid test event') return delivery_id, event def record_verified(db, delivery, secret): delivery_id, event = verify_delivery(delivery, secret) db.execute('BEGIN IMMEDIATE') try: result = db.execute('INSERT OR IGNORE INTO deliveries (id) VALUES (?)', (delivery_id,)) if result.rowcount: # Receipt and synthetic effect commit together. A real integration needs # its own persistent database and a transaction/outbox for real effects. db.execute('INSERT INTO effects (delivery_id, event, order_id) VALUES (?, ?, ?)', (delivery_id, event['event'], event['data']['id'])) db.commit() return result.rowcount == 1 except Exception: db.rollback() raise class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): raise ValueError('Redirects are not allowed') def run_checks(value=None): base = validate_base(value or os.getenv('SANDBOX_BASE_URL', 'https://ludovex.org/api/sandbox/v1')) opener = urllib.request.build_opener(NoRedirect()) key = None def request(path, method='GET', body=None, expected=200): headers = {'Accept': 'application/json', 'User-Agent': 'Ludovex-Sandbox-Sample/1.0'} if key: headers['X-Sandbox-Key'] = key if body is not None: headers['Content-Type'] = 'application/json' req = urllib.request.Request(base + path, data=None if body is None else json.dumps(body).encode(), headers=headers, method=method) try: response = opener.open(req, timeout=12) except urllib.error.HTTPError as error: response = error with response: status = response.status raw = response.read(65537) if len(raw) > 65536: raise ValueError('Response too large') result = json.loads(raw) assert status == expected, 'Unexpected sandbox HTTP status' assert result.get('sandbox') is True and result.get('ok') == (expected < 400) return result session = request('/sessions', 'POST', {}, 201)['data'] key = session['api_key'] try: assert re.fullmatch(r'lvx_test_[a-f0-9]{64}', key) assert session['currency'] == 'TEST_USDT' catalog = request('/catalog')['data'][0] def body(scenario): return dict(game_key=catalog['game_key'], variation_id=catalog['variation_id'], scenario=scenario, idempotency_key='sample-' + scenario) first = request('/order', 'POST', body('success'), 201)['data'] assert first['status'] == 'COMPLETED' duplicate = request('/order', 'POST', body('success'), 201)['data'] assert duplicate['id'] == first['id'] and duplicate['idempotent_replay'] is True conflict = dict(body('success'), scenario='delayed') assert request('/order', 'POST', conflict, 409)['error']['code'] == 'IDEMPOTENCY_CONFLICT' delayed = request('/order', 'POST', body('delayed'), 201)['data'] refund = request('/order', 'POST', body('failed_refund'), 201)['data'] assert delayed['status'] == refund['status'] == 'PROCESSING' assert request('/order', 'POST', body('low_balance'), 402)['error']['code'] == 'INSUFFICIENT_BALANCE' time.sleep(5.2) assert request('/order/' + delayed['id'])['data']['status'] == 'COMPLETED' assert request('/order/' + refund['id'])['data']['status'] == 'REFUNDED' assert request('/order/' + refund['id'])['data']['refunded'] is True assert request('/balance')['data']['balance'] == '93.50' assert len(request('/orders')['data']) == 3 events = request('/events')['data'] assert len(events) == 4 with tempfile.TemporaryDirectory(prefix='ludovex-sandbox-') as directory: db = sqlite3.connect(os.path.join(directory, 'inbox.sqlite')) try: db.executescript('CREATE TABLE deliveries (id TEXT PRIMARY KEY); CREATE TABLE effects (delivery_id TEXT PRIMARY KEY, event TEXT NOT NULL, order_id TEXT NOT NULL);') for event in events: first_delivery = request('/events/' + event['id'] + '/replay', 'POST', {})['data'] assert first_delivery['external_delivery'] is False assert record_verified(db, first_delivery, session['webhook_secret']) second_delivery = request('/events/' + event['id'] + '/replay', 'POST', {})['data'] assert second_delivery['headers']['X-Ludovex-Delivery-Id'] == first_delivery['headers']['X-Ludovex-Delivery-Id'] assert second_delivery['raw_body'] == first_delivery['raw_body'] assert not record_verified(db, second_delivery, session['webhook_secret']) for tampered, at in [(dict(first_delivery, raw_body=first_delivery['raw_body'] + ' '), None), (first_delivery, int(first_delivery['headers']['X-Ludovex-Timestamp']) + 301)]: try: verify_delivery(tampered, session['webhook_secret'], at) except ValueError: pass else: raise AssertionError('Invalid delivery accepted') assert db.execute('SELECT count(*) FROM effects').fetchone()[0] == 4 finally: db.close() print('PASS: success, delayed completion, one refund, low balance, idempotency, HMAC, timestamp and atomic delivery deduplication.') return True finally: try: request('/session', 'DELETE') except Exception: pass # Absolute TTL also expires temporary sessions. key = None if __name__ == '__main__': try: run_checks() except Exception: print('Sandbox checks failed. Check the sandbox URL, availability and rate limits; no credentials were logged.') raise SystemExit(1)