git / moonbite-network / merchants.py · raw
"""MoonBite merchant adoption layer — non-custodial "Accept MBITE" rails.This module stores a merchant *directory* (businesses that voluntarily acceptMBITE) and the *invoices* they raise. It NEVER holds coins, keys, or balances.A payment is detected the way a watch-only wallet does it: we look at the chainfor coins that landed on the merchant's own address. The customer pays directly,wallet-to-wallet; the server only *observes* that it happened and marks theinvoice paid. Keep it that way — do not add custody, escrow, or fiat handling.Payment verification is deliberately injected (`received_lookup`) so this modulestays decoupled from any particular node. In this demo the web app wires it tothe local educational node; in production it points at the MoonBite node."""from __future__ import annotationsimport hmacimport osimport reimport secretsimport sqlite3import threadingimport timeimport uuidfrom decimal import ROUND_DOWN, Decimal, InvalidOperationfrom pathlib import Pathfrom typing import Callable, Optionalclass MerchantAuthError(Exception): """Raised when a merchant-scoped read is missing or presents a bad token. Distinct from ValueError (bad input) so the web layer can answer 403, not 400 — the request was well-formed but not authorised for that ledger. """# Categories a merchant can pick. Kept small and honest — no fake taxonomy.CATEGORIES = ("goods", "services", "food", "digital", "donation", "other")# App unit: base units per coin. The educational chain uses 100; the production# Litecoin-fork node uses 1e8 (sats). Configurable so a production deployment can# match its chain's precision (set MOONBITE_UNITS_PER_COIN=100000000) while the# in-process demo keeps 100. The invoiced amount and the injected received_lookup# MUST share this basis — that pairing is wired in web_app.py.UNITS_PER_COIN = int(os.environ.get("MOONBITE_UNITS_PER_COIN", "100"))# Upper bound on a single invoice, above the ~33M supply cap so no real payment# is blocked, but finite so an astronomically large (yet finite) Decimal cannot# overflow the base-unit conversion downstream._MAX_INVOICE_COINS = Decimal(os.environ.get("MOONBITE_MAX_INVOICE_COINS", "100000000"))# An invoice is payable for this long, then it expires (seconds).DEFAULT_INVOICE_TTL = 60 * 60 # 1 hour# Anti-spam: cap directory size churn per address.MAX_MERCHANTS_PER_ADDRESS = 3_URL_RE = re.compile(r"^https?://[^\s]{3,200}$", re.IGNORECASE)_DB_PATH = Path(__file__).with_name("merchants.db")_lock = threading.Lock()_conn: Optional[sqlite3.Connection] = None# A callable(address:str) -> int (received base units, monotonic over chain life).ReceivedLookup = Callable[[str], int]def _connect() -> sqlite3.Connection: global _conn if _conn is None: _conn = sqlite3.connect(str(_DB_PATH), check_same_thread=False) _conn.row_factory = sqlite3.Row _conn.execute( """ CREATE TABLE IF NOT EXISTS merchants ( id TEXT PRIMARY KEY, name TEXT NOT NULL, url TEXT NOT NULL DEFAULT '', category TEXT NOT NULL, mbite_address TEXT NOT NULL, blurb TEXT NOT NULL DEFAULT '', created_at INTEGER NOT NULL, manage_token TEXT NOT NULL DEFAULT '' ) """ ) # Migration for databases created before manage_token existed. A merchant # with an empty token cannot authenticate a ledger read until re-registered # (there is no bulk-read path that leaks other merchants' invoices). cols = {r["name"] for r in _conn.execute("PRAGMA table_info(merchants)")} if "manage_token" not in cols: _conn.execute( "ALTER TABLE merchants ADD COLUMN manage_token TEXT NOT NULL DEFAULT ''" ) _conn.execute( """ CREATE TABLE IF NOT EXISTS invoices ( id TEXT PRIMARY KEY, merchant_id TEXT, address TEXT NOT NULL, amount TEXT NOT NULL, -- MBITE, decimal string amount_units INTEGER NOT NULL,-- base units for on-chain check memo TEXT NOT NULL DEFAULT '', status TEXT NOT NULL DEFAULT 'pending', baseline_units INTEGER NOT NULL,-- received at address when raised created_at INTEGER NOT NULL, expires_at INTEGER NOT NULL, paid_at INTEGER ) """ ) _conn.commit() return _conn# --------------------------------------------------------------------------- ## Validation helpers# --------------------------------------------------------------------------- #def _clean_text(value, field: str, lo: int, hi: int, required: bool = True) -> str: if value is None: value = "" if not isinstance(value, str): raise ValueError(f"{field} must be text") v = value.strip() if required and len(v) < lo: raise ValueError(f"{field} must be at least {lo} characters") if len(v) > hi: raise ValueError(f"{field} must be at most {hi} characters") return vdef _clean_address(value, validate: Optional[Callable[[str], bool]]) -> str: if not isinstance(value, str): raise ValueError("MBITE address is required") v = value.strip() if not (6 <= len(v) <= 120) or any(c.isspace() for c in v): raise ValueError("MBITE address does not look valid") # When the caller supplies an address validator (the app's own format), use # it so invoices raised against this address can actually be auto-verified. if validate is not None and not validate(v): raise ValueError("MBITE address failed checksum validation") return vdef _pos_amount(value) -> Decimal: try: d = Decimal(str(value)) except (InvalidOperation, ValueError, TypeError): raise ValueError("amount must be a number") # Decimal("inf")/"nan" parse cleanly and slip past the <= 0 check (NaN # compares false, Infinity compares greater), so reject non-finite first. if not d.is_finite(): raise ValueError("amount must be a finite number") if d <= 0: raise ValueError("amount must be greater than zero") if d > _MAX_INVOICE_COINS: raise ValueError("amount is too large") return ddef _amount_units(d: Decimal) -> int: # Convert a MBITE amount to the chain's integer base units WITHOUT silently # discarding value: an amount finer than one base unit used to round to 0, # creating a zero-value invoice (goods for free). Reject it instead. scaled = d * UNITS_PER_COIN integral = scaled.to_integral_value(rounding=ROUND_DOWN) if scaled != integral: raise ValueError("amount is finer than the smallest MBITE unit") units = int(integral) if units <= 0: raise ValueError("amount is below the smallest MBITE unit") return units# --------------------------------------------------------------------------- ## Merchant directory# --------------------------------------------------------------------------- #def add_merchant( name, category, mbite_address, url="", blurb="", address_validator: Optional[Callable[[str], bool]] = None,) -> dict: """Register a merchant that voluntarily accepts MBITE. Returns the record.""" name_c = _clean_text(name, "name", 2, 60) if category not in CATEGORIES: raise ValueError(f"category must be one of: {', '.join(CATEGORIES)}") addr = _clean_address(mbite_address, address_validator) url_c = _clean_text(url, "url", 0, 200, required=False) if url_c and not _URL_RE.match(url_c): raise ValueError("url must start with http:// or https://") blurb_c = _clean_text(blurb, "blurb", 0, 160, required=False) with _lock: conn = _connect() n = conn.execute( "SELECT COUNT(*) AS n FROM merchants WHERE mbite_address = ?", (addr,) ).fetchone()["n"] if n >= MAX_MERCHANTS_PER_ADDRESS: raise ValueError("too many listings for this address") mid = uuid.uuid4().hex # The manage token is the only credential that authorises reading this # merchant's invoice ledger (list/poll). It is returned ONCE here and is # never echoed by any public directory read — store it now. token = secrets.token_hex(24) conn.execute( """INSERT INTO merchants (id, name, url, category, mbite_address, blurb, created_at, manage_token) VALUES (?, ?, ?, ?, ?, ?, ?, ?)""", (mid, name_c, url_c, category, addr, blurb_c, int(time.time()), token), ) conn.commit() record = get_merchant(mid) or {} # Surface the secret exactly once, on creation. record["manage_token"] = token return recorddef _public_merchant(row) -> dict: """A merchant dict safe to return to anyone — the manage token is stripped.""" d = dict(row) d.pop("manage_token", None) return ddef get_merchant(merchant_id: str) -> Optional[dict]: with _lock: conn = _connect() row = conn.execute( "SELECT * FROM merchants WHERE id = ?", (merchant_id,) ).fetchone() return _public_merchant(row) if row else Nonedef _verify_merchant(merchant_id: Optional[str], manage_token: Optional[str]) -> None: """Authorise a merchant-scoped ledger read, or raise MerchantAuthError. Requires a concrete merchant_id AND a token that matches the stored one in constant time. The unscoped (merchant_id=None) case — which used to dump every merchant's ledger — is rejected outright. """ if not merchant_id or not isinstance(merchant_id, str): raise MerchantAuthError("merchant_id is required") if not manage_token or not isinstance(manage_token, str): raise MerchantAuthError("merchant manage token is required") with _lock: conn = _connect() row = conn.execute( "SELECT manage_token FROM merchants WHERE id = ?", (merchant_id,) ).fetchone() stored = row["manage_token"] if row else "" # Compare even on a miss (against a dummy) to keep timing uniform, and reject # the empty/legacy token so a tokenless merchant can never be read. if not stored or not hmac.compare_digest(str(stored), manage_token): raise MerchantAuthError("invalid merchant credentials")def list_merchants(category: Optional[str] = None) -> list[dict]: with _lock: conn = _connect() if category is not None and category not in CATEGORIES: raise ValueError("unknown category") if category: rows = conn.execute( "SELECT * FROM merchants WHERE category = ? ORDER BY created_at DESC", (category,), ).fetchall() else: rows = conn.execute( "SELECT * FROM merchants ORDER BY created_at DESC" ).fetchall() return [_public_merchant(r) for r in rows]# --------------------------------------------------------------------------- ## Invoices (non-custodial payment requests)# --------------------------------------------------------------------------- #def build_pay_uri(address: str, amount: Decimal, memo: str = "") -> str: """A BIP21-style MoonBite payment URI a wallet can parse into a prefilled send.""" uri = f"moonbite:{address}?amount={format(amount, 'f')}" if memo: # keep it URL-safe-ish; wallets treat label as display text label = re.sub(r"[^\w \-.,]", "", memo)[:60].replace(" ", "%20") if label: uri += f"&label={label}" return uridef create_invoice( address, amount, received_lookup: ReceivedLookup, merchant_id: Optional[str] = None, memo: str = "", ttl: int = DEFAULT_INVOICE_TTL, address_validator: Optional[Callable[[str], bool]] = None, manage_token: Optional[str] = None,) -> dict: """Raise a payment request against a merchant address. Non-custodial. We snapshot how much the address has *already* received so that only a NEW inbound payment of at least `amount` marks this specific invoice paid. Authenticated (audit H1): only the registered merchant, proving ownership with its manage token, may raise an invoice, and only against its own registered address. Anonymous invoices used to be accepted against ANY address, so an attacker could pre-seed a dust invoice on a shop's address and have every real payment settle the attacker's row first (oldest-first allocation), starving the real invoice forever. ``address`` may be omitted (None/"") to mean "the merchant's registered address". """ _verify_merchant(merchant_id, manage_token) with _lock: conn = _connect() mrow = conn.execute( "SELECT mbite_address FROM merchants WHERE id = ?", (merchant_id,) ).fetchone() if mrow is None: # verified above, but never trust a race raise MerchantAuthError("invalid merchant credentials") owned = mrow["mbite_address"] if address is None or (isinstance(address, str) and not address.strip()): address = owned addr = _clean_address(address, address_validator) if addr != owned: raise ValueError("address is not registered to this merchant") amt = _pos_amount(amount) memo_c = _clean_text(memo, "memo", 0, 140, required=False) baseline = int(received_lookup(addr)) now = int(time.time()) inv_id = uuid.uuid4().hex with _lock: conn = _connect() conn.execute( """INSERT INTO invoices (id, merchant_id, address, amount, amount_units, memo, status, baseline_units, created_at, expires_at, paid_at) VALUES (?, ?, ?, ?, ?, ?, 'pending', ?, ?, ?, NULL)""", ( inv_id, merchant_id, addr, format(amt, "f"), _amount_units(amt), memo_c, baseline, now, now + int(ttl), ), ) conn.commit() return invoice_status(inv_id, received_lookup)def _enrich(inv: dict, paid_units: int, received_now: int, now: int) -> dict: amt = Decimal(inv["amount"]) return { **inv, "paid_units": paid_units, "received_now": received_now, "seconds_left": max(0, inv["expires_at"] - now), "pay_uri": build_pay_uri(inv["address"], amt, inv["memo"]), }def _settle_address_group(group_rows, received_now: int, now: int) -> dict: """Allocate ONE address's received total across its invoices and persist. This is the fix for the double-settlement bug: the address's monotonic received total is a single pool, so a payment must back *one* invoice, not every invoice sharing the address. We walk the address's invoices oldest first with an allocation ``cursor`` — the base units already spoken for by invoices that are (or become) paid. Each invoice can only be satisfied by value above ``max(its baseline, cursor)``, and a paid invoice advances the cursor past the slice it consumed. So two 100-MBITE invoices on one address need 200 MBITE received, not 100 counted twice. ``group_rows`` MUST be every invoice on this (address, merchant_id) pair, or attribution would miss a claimant. The group is scoped to ONE merchant: rows raised by another merchant id, or legacy anonymous rows (merchant_id NULL), never join a merchant's allocation group, so a foreign/dust row cannot sit ahead of the real invoice in the oldest-first walk (audit H1). Returns {id: settled}. """ ordered = sorted(group_rows, key=lambda r: (r["created_at"], r["id"])) cursor = 0 # received units already allocated to paid invoices on this address results: dict = {} transitions: list = [] # (id, new_status, paid_at) for row in ordered: inv = dict(row) baseline = int(inv["baseline_units"]) amount = int(inv["amount_units"]) status = inv["status"] paid_at = inv["paid_at"] floor = max(baseline, cursor) available = max(0, received_now - floor) if status == "paid": # Already terminal: it still owns its slice so a later invoice can't # reuse the same coins. Advance the cursor; never re-touch it. cursor = floor + amount report_units = amount elif status == "pending": if available >= amount: status = "paid" paid_at = now cursor = floor + amount report_units = amount transitions.append((inv["id"], "paid", paid_at)) else: if now > int(inv["expires_at"]): status = "expired" transitions.append((inv["id"], "expired", paid_at)) report_units = available # unpaid/expired consume nothing else: # 'expired' already — consumes nothing, cursor unchanged report_units = 0 inv["status"] = status inv["paid_at"] = paid_at results[inv["id"]] = _enrich(inv, report_units, received_now, now) if transitions: with _lock: conn = _connect() for inv_id, st, pa in transitions: conn.execute( "UPDATE invoices SET status = ?, paid_at = ? WHERE id = ?", (st, pa, inv_id), ) conn.commit() return resultsdef _settle_addresses(groups, received_lookup: ReceivedLookup, now: int) -> dict: """Settle the invoice group for each (address, merchant_id) pair once. ``groups`` is an iterable of ``(address, merchant_id)``. A merchant_id of None selects only the legacy anonymous rows on that address — they settle among themselves and never share a pool with a registered merchant's invoices. Each distinct address is looked up on-chain once. Returns {id: settled}. """ out: dict = {} received_by_addr: dict = {} for addr, mid in set(groups): with _lock: conn = _connect() if mid is None: group = conn.execute( "SELECT * FROM invoices WHERE address = ? AND merchant_id IS NULL", (addr,), ).fetchall() else: group = conn.execute( "SELECT * FROM invoices WHERE address = ? AND merchant_id = ?", (addr, mid), ).fetchall() if addr not in received_by_addr: received_by_addr[addr] = int(received_lookup(addr)) out.update(_settle_address_group(group, received_by_addr[addr], now)) return outdef invoice_status(invoice_id: str, received_lookup: ReceivedLookup) -> dict: """Return an invoice with a *fresh* payment check against the chain. Reachable by knowing the (unguessable) invoice id, so it is not merchant gated — but it settles the whole address group so the per-invoice allocation is consistent no matter which invoice is polled. """ with _lock: conn = _connect() row = conn.execute( "SELECT * FROM invoices WHERE id = ?", (invoice_id,) ).fetchone() if row is None: raise ValueError("invoice not found") addr = row["address"] mid = row["merchant_id"] results = _settle_addresses([(addr, mid)], received_lookup, int(time.time())) return results[invoice_id]def list_invoices( received_lookup: ReceivedLookup, merchant_id: Optional[str] = None, manage_token: Optional[str] = None,) -> list[dict]: """Return ONE authorised merchant's invoices, each freshly checked on-chain. Requires the merchant's manage token (see add_merchant). The old unscoped read that returned every merchant's ledger to anyone is gone — callers must identify and prove ownership of exactly one merchant. The on-chain scan is therefore bounded to that merchant's own addresses. """ _verify_merchant(merchant_id, manage_token) with _lock: conn = _connect() rows = conn.execute( "SELECT * FROM invoices WHERE merchant_id = ? ORDER BY created_at DESC", (merchant_id,), ).fetchall() now = int(time.time()) addrs = {(r["address"], merchant_id) for r in rows} settled = _settle_addresses(addrs, received_lookup, now) # Newest first, this merchant's invoices only. return [settled[r["id"]] for r in rows]def poll_pending_invoices( received_lookup: ReceivedLookup, merchant_id: Optional[str] = None, manage_token: Optional[str] = None,) -> dict: """Sweep ONE authorised merchant's pending invoices, advancing paid/expired. Requires the merchant's manage token and scopes the sweep (and the on-chain scan fan-out) to that merchant's own addresses — so an anonymous caller can no longer force a system-wide ``scantxoutset`` per address against the node. """ _verify_merchant(merchant_id, manage_token) with _lock: conn = _connect() rows = conn.execute( "SELECT * FROM invoices WHERE merchant_id = ? AND status = 'pending'", (merchant_id,), ).fetchall() now = int(time.time()) addrs = {(r["address"], merchant_id) for r in rows} settled = _settle_addresses(addrs, received_lookup, now) paid = [r["id"] for r in rows if settled[r["id"]]["status"] == "paid"] expired = [r["id"] for r in rows if settled[r["id"]]["status"] == "expired"] return { "checked": len(rows), "paid": paid, "expired": expired, "addresses_scanned": len(addrs), }