"""Python >=3.11. Standard library only. Use one worker per local database.""" import hashlib import hmac import json import os import re import sqlite3 import sys import time from urllib.error import HTTPError from urllib.parse import parse_qs, quote, urlparse from urllib.request import Request, urlopen API = 'https://merchant-api.tradeon.market' def open_store(path='merchant.sqlite'): db = sqlite3.connect(path) db.execute('PRAGMA journal_mode=WAL') db.execute('PRAGMA synchronous=FULL') db.executescript(''' CREATE TABLE IF NOT EXISTS intents(custom_id TEXT PRIMARY KEY, method TEXT, body TEXT); CREATE TABLE IF NOT EXISTS inbox(id TEXT PRIMARY KEY, body TEXT, done INTEGER DEFAULT 0); CREATE TABLE IF NOT EXISTS effects(business_key TEXT PRIMARY KEY, body TEXT); CREATE TABLE IF NOT EXISTS reconcile(purchase_id TEXT PRIMARY KEY); CREATE TABLE IF NOT EXISTS catalog(id TEXT PRIMARY KEY, body TEXT); CREATE TABLE IF NOT EXISTS metadata(id INTEGER PRIMARY KEY CHECK(id=1), body TEXT); ''') return db def request(method, path, body=None): key = os.environ['TRADEON_API_KEY'] req = Request(API + '/api/v1/merchant' + path, method=method, data=json.dumps(body).encode() if body else None, headers={'Authorization': f'Bearer {key}', 'Content-Type': 'application/json'}) try: response = urlopen(req, timeout=30) except HTTPError as error: response = error with response: return {'status': response.status, 'body': json.load(response)} def validate_recipient(body, method='buy'): url = urlparse(body['trade_url' if method == 'buy-by-name' else 'target_trade_url']) query = parse_qs(url.query) partner = query.get('partner', [''])[0] if (url.scheme != 'https' or url.hostname != 'steamcommunity.com' or url.path != '/tradeoffer/new/' or not partner.isdigit() or not query.get('token')): raise ValueError('Invalid Steam trade URL') if body.get('target_steam_id') and int(body['target_steam_id']) != 76561197960265728 + int(partner): raise ValueError('target_steam_id does not match trade URL partner') def remember_intent(db, method, body): if method not in ('buy', 'buy-sync', 'buy-by-name'): raise ValueError('Unknown purchase method') if not isinstance(body.get('custom_id'), str) or not body['custom_id'].strip(): raise ValueError('Supply the stable business custom_id') validate_recipient(body, method) encoded = json.dumps(body, sort_keys=True, separators=(',', ':')) with db: db.execute('INSERT OR IGNORE INTO intents VALUES (?, ?, ?)', (body['custom_id'], method, encoded)) old = db.execute('SELECT method, body FROM intents WHERE custom_id=?', (body['custom_id'],)).fetchone() if old != (method, encoded): raise ValueError('Do not reuse custom_id for a different intent') return body['custom_id'] def submit_intent(db, custom_id, transport=request): intent = db.execute('SELECT method, body FROM intents WHERE custom_id=?', (custom_id,)).fetchone() if not intent: raise ValueError('Persist intent before POST') try: result = transport('POST', '/' + intent[0], json.loads(intent[1])) except (OSError, ValueError): return {'kind': 'unknown', 'custom_id': custom_id, 'next': 'Reconcile, then retry this saved intent; never create a new ID'} status = result['status'] kind = 'accepted' if 200 <= status < 300 else 'unknown' if status >= 500 or status in (408, 429) else 'rejected' return {'kind': kind, **result} def reconcile_intent(custom_id, transport=request): purchases, cursor = [], None while True: path = '/purchases?custom_id=' + quote(custom_id, safe='') + '&limit=200' if cursor: path += '&cursor=' + quote(cursor, safe='') result = transport('GET', path) if result['status'] != 200: raise ValueError(f"Reconciliation returned {result['status']}") purchases.extend(result['body']['purchases']) cursor = result['body'].get('next_cursor') if not cursor: break # Absence does not prove an in-flight POST failed. By-name attempts may keep changing. return {'purchases': purchases, 'next_cursor': None} def sign(raw, timestamp, secret): return hmac.new(secret.encode(), timestamp.encode() + b'.' + raw, hashlib.sha256).hexdigest() def verify(raw, headers, secret, now=None): now = time.time() if now is None else now timestamp, signature = headers.get('x-timestamp', ''), headers.get('x-signature', '') if (not re.fullmatch(r'[0-9]+', timestamp) or abs(now - int(timestamp)) > 300 or not re.fullmatch(r'[0-9a-fA-F]{64}', signature) or not hmac.compare_digest(sign(raw, timestamp, secret), signature.lower())): raise ValueError('Invalid signature or timestamp') event = json.loads(raw) if (not isinstance(event, dict) or not event.get('event_id') or event['event_id'] != headers.get('x-event-id') or not event.get('event') or not isinstance(event.get('data'), dict) or headers.get('x-event') != event['event']): raise ValueError('Header/payload mismatch or invalid envelope') return event def accept_webhook(db, raw, headers, secret, now=None): event = verify(raw, headers, secret, now) with db: old = db.execute('SELECT body FROM inbox WHERE id=?', (event['event_id'],)).fetchone() if old and old[0] != raw.decode(): raise ValueError('Event ID reused with a different body') db.execute('INSERT OR IGNORE INTO inbox(id, body) VALUES (?, ?)', (event['event_id'], raw.decode())) return 204 # The commit above must succeed before acknowledging an accepted duplicate too. def process_inbox(db): count = 0 for event_id, body in db.execute('SELECT id, body FROM inbox WHERE done=0').fetchall(): with db: event = json.loads(body) data, key = event['data'], None if event['event'] == 'balance.deposited': if data['provider'] not in ('heleket', 'whitepay') or not data['provider_uuid']: raise ValueError('Missing deposit identity') # Keep previously saved Heleket keys valid across an example upgrade. prefix = 'deposit:' if data['provider'] == 'heleket' else 'deposit:whitepay:' key = prefix + data['provider_uuid'] elif event['event'] in ('purchase.created', 'purchase.sent_to_steam', 'purchase.completed', 'purchase.failed', 'purchase.refunded', 'purchase.reverted_by_buyer', 'purchase.reverted_by_seller'): key = data['purchase_id'] + ':' + event['event'] if key: inserted = db.execute('INSERT OR IGNORE INTO effects VALUES (?, ?)', (key, body)).rowcount if inserted and data.get('purchase_id'): db.execute('INSERT OR IGNORE INTO reconcile VALUES (?)', (data['purchase_id'],)) # Unknown types stay in the durable inbox without creating business effects. db.execute('UPDATE inbox SET done=1 WHERE id=?', (event_id,)) count += 1 return count def main(): command, *arguments = sys.argv[1:] db = open_store() try: if command == 'first': print(request('GET', '/items?limit=1')) elif command == 'prepare': with open(arguments[0]) as source: intent = json.load(source) print({'saved': remember_intent(db, intent.get('method', 'buy'), intent['body'])}) elif command == 'submit': print(submit_intent(db, arguments[0])) elif command == 'recover': print(reconcile_intent(arguments[0])) elif command == 'status': print(request('GET', '/purchases/' + quote(arguments[0], safe=''))) elif command == 'work': print({'processed': process_inbox(db), 'reconcile': db.execute('SELECT * FROM reconcile').fetchall()}) else: raise ValueError('Use first|prepare intent.json|submit custom_id|recover custom_id|status purchase_id|work') finally: db.close() if __name__ == '__main__': main()