Python store fulfillment reference

python3 -m venv /private/kdpika-fulfillment-venv
/private/kdpika-fulfillment-venv/bin/python -m pip install -r python/requirements.txt
/private/kdpika-fulfillment-venv/bin/python -m pip check

/private/kdpika-fulfillment-venv/bin/python python/server.py serve

/private/kdpika-fulfillment-venv/bin/python python/server.py work

/private/kdpika-fulfillment-venv/bin/python python/server.py inspect

/private/kdpika-fulfillment-venv/bin/python python/server.py export YOUR_ORDER_ID /private/key.json

#!/usr/bin/env python3
"""Durable trusted-store command receiver and licensing worker; not a payment gateway.
Python 3.11+, cryptography==50.0.2. POSIX file locking, private SQLite/AES-GCM outbox.
Set KDP_FULFILLMENT_CONFIG/STATE/SECRET/OUTBOX_KEY and KDP_API_KEY privately.
Run: python3 server.py serve | work | inspect | export YOUR_ORDER_ID /private/key.json
Expected: safe acceptance/work state JSON, never a complete key in stdout.
"""
import base64
from contextlib import contextmanager
import fcntl
import hashlib
import hmac
from http.server import BaseHTTPRequestHandler, HTTPServer
import json
import os
from pathlib import Path
import re
import sqlite3
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from cryptography.hazmat.primitives.ciphers.aead import AESGCM

LIMIT = 262144
ID = re.compile(r'[A-Za-z0-9_.:-]{1,190}')
ACTIONS = {'issue', 'renew', 'suspend', 'restore', 'revoke', 'record_cancellation'}
STATES = {'active', 'suspended', 'revoked', 'expired', 'pending', 'cancelled'}
FIELDS = {'version', 'source', 'seller_id', 'environment', 'event_id', 'effect_id',
          'order_id', 'sequence', 'action', 'offer_id', 'customer',
          'expected_status', 'expected_expires_at'}


def encoded(value):
    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(',', ':'), allow_nan=False)


def identity(*values):
    return hashlib.sha256('\0'.join(values).encode()).hexdigest()


class Refusal(Exception):
    pass


def decoded(raw):
    def unique(pairs):
        result = {}
        for key, value in pairs:
            if key in result:
                raise Refusal('DUPLICATE_JSON_MEMBER')
            result[key] = value
        return result
    def invalid_constant(value):
        raise Refusal('NONSTANDARD_JSON_NUMBER')
    return json.loads(raw, object_pairs_hook=unique, parse_constant=invalid_constant)


class NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, request, response, code, message, headers, new_url):
        return None


class Fulfillment:
    def __init__(self):
        os.umask(0o077)
        config_path = Path(os.environ['KDP_FULFILLMENT_CONFIG'])
        if not config_path.is_absolute():
            raise Refusal('PRIVATE_ABSOLUTE_CONFIG_REQUIRED')
        self.config = decoded(config_path.read_text())
        self.directory = Path(os.environ['KDP_FULFILLMENT_STATE'])
        if not self.directory.is_absolute() or self.directory.is_symlink():
            raise Refusal('PRIVATE_ABSOLUTE_STATE_REQUIRED')
        self.directory.mkdir(parents=True, mode=0o700, exist_ok=True)
        if self.directory.stat().st_mode & 0o077:
            raise Refusal('PRIVATE_STATE_PERMISSIONS_REQUIRED')
        self.secret = os.environ['KDP_FULFILLMENT_SECRET']
        self.outbox_key = base64.b64decode(os.environ['KDP_FULFILLMENT_OUTBOX_KEY'], validate=True)
        if len(self.secret) < 32 or len(self.outbox_key) != 32:
            raise Refusal('PRIVATE_SECRETS_REQUIRED')
        self.environment = self.config['environment']
        if self.environment not in ('sandbox', 'live'):
            raise Refusal('INVALID_ENVIRONMENT')
        if self.environment == 'live' and os.environ.get('KDP_CONFIRM_LIVE') != '1':
            raise Refusal('LIVE_CONFIRMATION_REQUIRED')
        for field in ('source', 'seller_id', 'product_id'):
            if not isinstance(self.config[field], str) or not ID.fullmatch(self.config[field]):
                raise Refusal('INVALID_CONFIGURATION')
        if not isinstance(self.config['offers'], dict) or not self.config['offers']:
            raise Refusal('OFFERS_REQUIRED')
        for offer_id, offer in self.config['offers'].items():
            if not ID.fullmatch(offer_id) or not ID.fullmatch(offer['plan_id']) or type(offer['renewal_days']) is not int or not 1 <= offer['renewal_days'] <= 3650:
                raise Refusal('INVALID_OFFER_MAPPING')
        self.base = self.config['base_url'].rstrip('/')
        url = urllib.parse.urlsplit(self.base)
        local = self.environment == 'sandbox' and os.environ.get('KDP_ALLOW_LOOPBACK') == '1' and url.scheme == 'http' and url.hostname == '127.0.0.1'
        if url.username or url.password or url.query or url.fragment or url.path or not url.hostname or (url.scheme != 'https' and not local):
            raise Refusal('INVALID_BASE_URL')
        self.api_key = os.environ['KDP_API_KEY']
        if not self.api_key.startswith('lmp_' + self.environment + '_'):
            raise Refusal('API_KEY_ENVIRONMENT_MISMATCH')
        self.database = self.directory / 'fulfillment.sqlite'
        if self.database.is_symlink():
            raise Refusal('INVALID_STATE_FILE')
        with self.connect() as db:
            db.executescript('''
                CREATE TABLE IF NOT EXISTS inbox(event_id TEXT PRIMARY KEY,body_hash TEXT NOT NULL,effect_key TEXT NOT NULL,received_at INTEGER NOT NULL);
                CREATE TABLE IF NOT EXISTS orders(order_key TEXT PRIMARY KEY,offer_id TEXT NOT NULL,email TEXT NOT NULL,license_id TEXT,last_sequence INTEGER NOT NULL DEFAULT 0,held INTEGER NOT NULL DEFAULT 0);
                CREATE TABLE IF NOT EXISTS effects(effect_key TEXT PRIMARY KEY,order_key TEXT NOT NULL,sequence INTEGER NOT NULL,command_hash TEXT NOT NULL,command_json TEXT NOT NULL,state TEXT NOT NULL,request_id TEXT,error_code TEXT,created_at INTEGER NOT NULL,UNIQUE(order_key,sequence));
                CREATE TABLE IF NOT EXISTS outbox(order_key TEXT PRIMARY KEY,license_id TEXT NOT NULL,cipher TEXT NOT NULL,created_at INTEGER NOT NULL);
                CREATE TABLE IF NOT EXISTS binding(id INTEGER PRIMARY KEY CHECK(id=1),config_hash TEXT NOT NULL,outbox_key_hash TEXT NOT NULL);
            ''')
            config_hash = hashlib.sha256(encoded(self.config).encode()).hexdigest()
            key_hash = hashlib.sha256(self.outbox_key).hexdigest()
            db.execute('INSERT OR IGNORE INTO binding(id,config_hash,outbox_key_hash) VALUES(1,?,?)', (config_hash, key_hash))
            binding = db.execute('SELECT * FROM binding WHERE id=1').fetchone()
            if binding['config_hash'] != config_hash or binding['outbox_key_hash'] != key_hash:
                raise Refusal('CONFIGURATION_CHANGED_RECONCILE_FIRST')
        self.database.chmod(0o600)

    @contextmanager
    def connect(self):
        db = sqlite3.connect(self.database, timeout=3)
        db.row_factory = sqlite3.Row
        try:
            with db:
                yield db
        finally:
            db.close()

    def validate(self, event):
        if not isinstance(event, dict) or set(event) != FIELDS or event['version'] != 1 or type(event['version']) is not int:
            raise Refusal('INVALID_COMMAND')
        for key in ('source', 'seller_id', 'environment'):
            if event[key] != self.config[key]:
                raise Refusal('WRONG_SELLER_OR_ENVIRONMENT')
        for key in ('event_id', 'effect_id', 'order_id', 'offer_id'):
            if not isinstance(event[key], str) or not ID.fullmatch(event[key]):
                raise Refusal('INVALID_IDENTIFIER')
        if event['offer_id'] not in self.config['offers'] or event['action'] not in ACTIONS or type(event['sequence']) is not int or not 1 <= event['sequence'] <= 2147483647:
            raise Refusal('UNSUPPORTED_COMMAND')
        customer = event['customer']
        if not isinstance(customer, dict) or set(customer) != {'email', 'name'} or not isinstance(customer['email'], str) or not re.fullmatch(r'[^\s@]{1,120}@[^\s@]{1,120}', customer['email']) or len(customer['email']) > 190 or not isinstance(customer['name'], str) or len(customer['name']) > 160:
            raise Refusal('INVALID_CUSTOMER')
        expiry = event['expected_expires_at']
        if expiry is not None and (not isinstance(expiry, str) or len(expiry) > 40):
            raise Refusal('INVALID_EXPECTED_EXPIRY')
        if event['action'] == 'issue':
            if event['sequence'] != 1 or event['expected_status'] is not None or expiry is not None:
                raise Refusal('INVALID_ISSUANCE_COMMAND')
        elif event['expected_status'] not in STATES:
            raise Refusal('CURRENT_STATE_REQUIRED')
        return event

    def accept(self, body, signature):
        match = re.fullmatch(r't=([1-9][0-9]{0,10}),v1=([0-9a-f]{64})', signature)
        if len(body) > LIMIT or not match or abs(int(time.time()) - int(match[1])) > 300:
            raise Refusal('INVALID_SIGNATURE')
        digest = hmac.new(self.secret.encode(), match[1].encode() + b'.' + body, hashlib.sha256).hexdigest()
        if not hmac.compare_digest(digest, match[2]):
            raise Refusal('INVALID_SIGNATURE')
        event = self.validate(decoded(body.decode('utf-8', errors='strict')))
        order_key = identity(event['source'], event['seller_id'], event['environment'], event['order_id'])
        effect_key = identity(event['source'], event['seller_id'], event['environment'], event['effect_id'])
        # Different provider deliveries may carry the same durable effect; event ID is not the effect key.
        semantic = {key: value for key, value in event.items() if key != 'event_id'}
        command_json = encoded(semantic)
        command_hash = hashlib.sha256(command_json.encode()).hexdigest()
        raw_hash = hashlib.sha256(body).hexdigest()
        with self.connect() as db:
            db.execute('BEGIN IMMEDIATE')
            previous = db.execute('SELECT body_hash FROM inbox WHERE event_id=?', (event['event_id'],)).fetchone()
            if previous:
                if previous['body_hash'] != raw_hash:
                    raise Refusal('EVENT_ID_CONFLICT')
                return {'accepted': True, 'duplicate': True}
            effect = db.execute('SELECT command_hash FROM effects WHERE effect_key=?', (effect_key,)).fetchone()
            if effect and effect['command_hash'] != command_hash:
                raise Refusal('EFFECT_CONFLICT')
            order = db.execute('SELECT * FROM orders WHERE order_key=?', (order_key,)).fetchone()
            if order and (order['offer_id'] != event['offer_id'] or order['email'] != event['customer']['email']):
                raise Refusal('ORDER_MAPPING_CONFLICT')
            db.execute('INSERT OR IGNORE INTO orders(order_key,offer_id,email) VALUES(?,?,?)', (order_key, event['offer_id'], event['customer']['email']))
            if not effect:
                try:
                    db.execute("INSERT INTO effects(effect_key,order_key,sequence,command_hash,command_json,state,created_at) VALUES(?,?,?,?,?,'WAITING',?)",
                               (effect_key, order_key, event['sequence'], command_hash, command_json, int(time.time())))
                except sqlite3.IntegrityError as error:
                    raise Refusal('SEQUENCE_CONFLICT') from error
            db.execute('INSERT INTO inbox(event_id,body_hash,effect_key,received_at) VALUES(?,?,?,?)', (event['event_id'], raw_hash, effect_key, int(time.time())))
            db.commit()
        return {'accepted': True, 'duplicate': effect is not None}

    def api(self, method, path, body=None, idempotency=None):
        headers = {'Authorization': 'Bearer ' + self.api_key, 'Accept': 'application/json', 'Content-Type': 'application/json', 'User-Agent': 'KDPika-Reference-Fulfillment/1'}
        if idempotency:
            headers['Idempotency-Key'] = idempotency
        request = urllib.request.Request(self.base + '/api/v1' + path, data=encoded(body).encode() if body is not None else None, headers=headers, method=method)
        opener = urllib.request.build_opener(NoRedirect())
        try:
            response = opener.open(request, timeout=15)
        except urllib.error.HTTPError as error:
            response = error
        with response:
            raw = response.read(1048577)
            if len(raw) > 1048576:
                raise Refusal('UNEXPECTED_API_BODY')
            data = decoded(raw)
            if not isinstance(data, dict):
                raise Refusal('UNEXPECTED_API_BODY')
            return response.status, data

    def perform(self, command, order, effect_key):
        action = command['action']
        if action == 'issue':
            if order['license_id']:
                raise Refusal('ORDER_ALREADY_HAS_LICENSE')
            offer = self.config['offers'][command['offer_id']]
            status, result = self.api('POST', '/licenses', {'product_id': self.config['product_id'], 'plan_id': offer['plan_id'], 'customer': command['customer']}, 'store_' + effect_key)
            item = result.get('license', {})
            if status != 201 or result.get('success') is not True or not isinstance(item.get('id'), str) or not re.fullmatch(r'lic_[A-Za-z0-9_-]{1,100}', item['id']) or not isinstance(item.get('license_key'), str) or not 12 <= len(item['license_key']) <= 100 or item.get('environment') != self.environment:
                raise Refusal('ISSUANCE_RESULT_REQUIRES_REVIEW')
            return item['id'], item['license_key'], result.get('request_id')
        if not order['license_id']:
            raise Refusal('ISSUE_FIRST')
        license_id = order['license_id']
        path = '/licenses/' + urllib.parse.quote(license_id, safe='')
        status, current = self.api('GET', path)
        item = current.get('license', {})
        if status != 200 or current.get('success') is not True or item.get('id') != license_id or item.get('environment') != self.environment or item.get('status') != command['expected_status'] or item.get('expires_at') != command['expected_expires_at']:
            raise Refusal('LICENSE_CHANGED_RECONCILE_FIRST')
        if action == 'record_cancellation':
            # This records an authenticated decision; it does not cancel/refund a store payment or erase paid access.
            return license_id, None, current.get('request_id')
        body = {'days': self.config['offers'][command['offer_id']]['renewal_days']} if action == 'renew' else {'reason': 'Verified store fulfillment decision'}
        status, changed = self.api('POST', path + '/' + action, body)
        expected = {'suspend': 'suspended', 'restore': 'active', 'revoke': 'revoked',
                    'renew': 'active' if item['status'] == 'expired' else item['status']}[action]
        changed_item = changed.get('license', {})
        if status != 200 or changed.get('success') is not True or changed_item.get('id') != license_id or changed_item.get('status') != expected or (action == 'renew' and not isinstance(changed_item.get('expires_at'), str)):
            raise Refusal('MUTATION_RESULT_REQUIRES_REVIEW')
        return license_id, None, changed.get('request_id')

    def work(self):
        lock_path = self.directory / 'worker.lock'
        if lock_path.is_symlink():
            raise Refusal('INVALID_LOCK_FILE')
        with lock_path.open('a+b') as lock:
            try:
                fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
            except BlockingIOError as error:
                raise Refusal('WORKER_BUSY') from error
            processed = 0
            # One bounded cron run. Per-order sequence gaps wait; other independent orders can progress.
            while processed < 25:
                with self.connect() as db:
                    db.execute('BEGIN IMMEDIATE')
                    effect = db.execute("SELECT e.* FROM effects e JOIN orders o ON o.order_key=e.order_key WHERE e.state='WAITING' AND o.held=0 AND e.sequence=o.last_sequence+1 ORDER BY e.created_at,e.effect_key LIMIT 1").fetchone()
                    if not effect:
                        break
                    order = db.execute('SELECT * FROM orders WHERE order_key=?', (effect['order_key'],)).fetchone()
                    db.execute("UPDATE effects SET state='IN_FLIGHT' WHERE effect_key=?", (effect['effect_key'],))
                    db.execute('UPDATE orders SET held=1 WHERE order_key=?', (effect['order_key'],))
                    db.commit()
                try:
                    license_id, key, request_id = self.perform(json.loads(effect['command_json']), order, effect['effect_key'])
                    if request_id is not None and (not isinstance(request_id, str) or len(request_id) > 190):
                        raise Refusal('UNEXPECTED_REQUEST_ID')
                    with self.connect() as db:
                        db.execute('BEGIN IMMEDIATE')
                        if key:
                            nonce = os.urandom(12)
                            cipher = nonce + AESGCM(self.outbox_key).encrypt(nonce, key.encode(), effect['order_key'].encode())
                            db.execute('INSERT INTO outbox(order_key,license_id,cipher,created_at) VALUES(?,?,?,?)', (effect['order_key'], license_id, base64.b64encode(cipher).decode(), int(time.time())))
                        db.execute("UPDATE effects SET state='DONE',request_id=? WHERE effect_key=?", (request_id, effect['effect_key']))
                        db.execute('UPDATE orders SET license_id=?,last_sequence=?,held=0 WHERE order_key=?', (license_id, effect['sequence'], effect['order_key']))
                        db.commit()
                except Exception:
                    with self.connect() as db:
                        db.execute("UPDATE effects SET state='REVIEW',error_code='RECONCILE_BEFORE_RETRY' WHERE effect_key=?", (effect['effect_key'],))
                    # held=1 remains. A crash before this write leaves IN_FLIGHT, also held and never auto-retried.
                processed += 1
                time.sleep(1.1)
            return {'processed': processed, 'states': self.inspect()}

    def inspect(self):
        with self.connect() as db:
            return {row['state']: row['count'] for row in db.execute('SELECT state,COUNT(*) AS count FROM effects GROUP BY state')}

    def export(self, order_id, path):
        if not ID.fullmatch(order_id):
            raise Refusal('INVALID_ORDER_ID')
        order_key = identity(self.config['source'], self.config['seller_id'], self.environment, order_id)
        with self.connect() as db:
            item = db.execute('SELECT b.*,o.email FROM outbox b JOIN orders o ON o.order_key=b.order_key WHERE b.order_key=?', (order_key,)).fetchone()
        if not item:
            raise Refusal('NO_SAVED_KEY')
        target = Path(path)
        if not target.is_absolute() or not target.parent.is_dir() or target.parent.stat().st_mode & 0o077:
            raise Refusal('PRIVATE_EXPORT_DIRECTORY_REQUIRED')
        raw = base64.b64decode(item['cipher'], validate=True)
        key = AESGCM(self.outbox_key).decrypt(raw[:12], raw[12:], order_key.encode()).decode()
        with target.open('x') as stream:
            target.chmod(0o600)
            json.dump({'license_id': item['license_id'], 'license_key': key, 'email': item['email']}, stream)
        return {'exported': True, 'contains_secret': True}


class Receiver(BaseHTTPRequestHandler):
    def log_message(self, *args):
        return

    def reply(self, status, value):
        raw = encoded(value).encode()
        self.send_response(status)
        self.send_header('Content-Type', 'application/json')
        self.send_header('Content-Length', str(len(raw)))
        self.send_header('Cache-Control', 'no-store')
        self.send_header('X-Content-Type-Options', 'nosniff')
        self.end_headers()
        self.wfile.write(raw)

    def do_POST(self):
        self.connection.settimeout(5)
        if self.path != '/store-event':
            return self.reply(404, {'accepted': False})
        length = self.headers.get_all('Content-Length', [])
        signatures = self.headers.get_all('X-Fulfillment-Signature', [])
        if self.headers.get('Transfer-Encoding') or len(length) != 1 or not length[0].isdigit():
            return self.reply(400, {'accepted': False})
        if int(length[0]) > LIMIT:
            return self.reply(413, {'accepted': False})
        if len(signatures) != 1:
            return self.reply(401, {'accepted': False})
        if self.headers.get('Content-Type', '').split(';')[0].strip().lower() != 'application/json':
            return self.reply(415, {'accepted': False})
        try:
            raw = self.rfile.read(int(length[0]))
            if len(raw) != int(length[0]):
                return self.reply(400, {'accepted': False})
            accepted = self.server.application.accept(raw, signatures[0])
            return self.reply(202, accepted)
        except Refusal as error:
            return self.reply(401 if str(error) == 'INVALID_SIGNATURE' else 409, {'accepted': False})
        except (ValueError, TypeError, KeyError):
            return self.reply(400, {'accepted': False})
        except (OSError, sqlite3.Error):
            return self.reply(503, {'accepted': False})


def main():
    try:
        app = Fulfillment()
        mode = sys.argv[1] if len(sys.argv) > 1 else ''
        if mode == 'serve' and len(sys.argv) == 2:
            port = int(os.environ.get('KDP_FULFILLMENT_PORT', '8112'))
            with HTTPServer(('127.0.0.1', port), Receiver) as server:
                server.application = app
                print('{"listening":true,"loopback_only":true}', flush=True)
                server.serve_forever()
        elif mode == 'work' and len(sys.argv) == 2:
            print(encoded(app.work()))
        elif mode == 'inspect' and len(sys.argv) == 2:
            print(encoded({'states': app.inspect()}))
        elif mode == 'export' and len(sys.argv) == 4:
            print(encoded(app.export(sys.argv[2], sys.argv[3])))
        else:
            raise Refusal('USE_SERVE_WORK_INSPECT_OR_EXPORT')
    except Exception:
        print('{"completed":false,"error":"REVIEW_PRIVATE_CONFIGURATION_OR_STATE"}')
        return 1
    return 0


if __name__ == '__main__':
    raise SystemExit(main())