#!/usr/bin/env python3 """Check before you book: batch, verify offline, read the fields, watch for change. Usage and context: https://lodestarstamp.com/check-before-you-book Standard library only; `cryptography` is optional and only needed to verify signatures. """ from __future__ import annotations import argparse import base64 import hashlib import json import os import re import sys import urllib.error import urllib.parse import urllib.request from pathlib import Path STATEMENT_SCHEMA = "https://lodestarstamp.com/schema/receipt-signature-v2" FIELDS = ("real", "at_that_address", "licensed") KEY_ENV = "LODESTAR_KEY" KEY_SHAPE = re.compile(r"^[A-Za-z0-9._~-]{8,256}$") LOOPBACK = {"127.0.0.1", "localhost", "::1"} class ApiError(RuntimeError): """The Trust API gave no 2xx answer: an error status, or no answer at all.""" def held_key(api: str) -> str: """LODESTAR_KEY, if set. It goes to the Trust API only, over HTTPS (or plain HTTP to this machine, for a local Worker), and never to the discovery card's host.""" key = os.environ.get(KEY_ENV, "").strip() if not key: return "" if not KEY_SHAPE.match(key): raise ValueError(f"{KEY_ENV} is set but is not an X-Lodestar-Key") parts = urllib.parse.urlsplit(api) if parts.scheme != "https" and parts.hostname not in LOOPBACK: raise ValueError(f"{KEY_ENV} is sent over HTTPS only, and {api} is not") return key def canonical_json(value) -> str: """Sorted keys, no whitespace: the serialisation the Stamp hashes and signs.""" return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False) def sha256_hex(text: str) -> str: return hashlib.sha256(text.encode("utf-8")).hexdigest() def unb64url(text: str) -> bytes: return base64.urlsafe_b64decode(text + "=" * (-len(text) % 4)) def get_json(url: str, consumer: str, key: str = "") -> dict: req = urllib.request.Request(url, headers={"Accept": "application/json", "X-Lodestar-Consumer": consumer, "User-Agent": f"{consumer} (lodestar reference client)"}) if key: # urllib copies ordinary headers onto a redirect, to any host; the key never follows one. req.add_unredirected_header("X-Lodestar-Key", key) try: with urllib.request.urlopen(req, timeout=30) as res: return json.loads(res.read().decode("utf-8")) except urllib.error.HTTPError as exc: text = exc.read().decode("utf-8", "replace") try: body = json.loads(text) except ValueError: body = {} said = (body.get("message") or body.get("error")) if isinstance(body, dict) else None raise ApiError(f"{exc.code} from {urllib.parse.urlsplit(url).path}: {said or text[:200]}") from None except (urllib.error.URLError, OSError) as exc: raise ApiError(f"no answer from {urllib.parse.urlsplit(url).path}: {getattr(exc, 'reason', exc)}") from None def get_changes(api: str, since: str, consumer: str, key: str = "", max_pages: int = 1000) -> dict: """/v1/changes?since=, every page joined: an answer holds at most `limit` events, and its next_cursor, sent back as cursor, reads the next until it is null. A page marked data_source "unavailable" could not be read, so it is not the end.""" url = f"{api}/v1/changes?since={urllib.parse.quote(since)}" def read(page_url: str) -> dict: page = get_json(page_url, consumer, key) if page.get("data_source") == "unavailable": raise RuntimeError("the change feed could not be read (data_source: unavailable); try again later") return page page = read(url) events = list(page.get("events") or []) for _ in range(max_pages - 1): if not page.get("next_cursor"): return {**page, "events": events} page = read(f"{url}&cursor={urllib.parse.quote(page['next_cursor'])}") events += page.get("events") or [] if page.get("next_cursor"): raise RuntimeError(f"the change feed did not end within {max_pages} pages") return {**page, "events": events} def statement_subject(business: dict) -> tuple[str, str]: """What a statement names the business by: the licence number where business.id is license: (a licence-keyed market), the domain otherwise.""" ident = str(business.get("id") or "") if ident.lower().startswith("license:"): return "license", ident[len("license:"):] return "domain", str(business.get("domain") or "") def verify(result: dict, keys: dict[str, str]) -> tuple[str, str]: """('valid' | 'INVALID' | 'unsigned' | 'unchecked', reason).""" sig = result.get("signature") if not isinstance(sig, dict) or not sig.get("signed"): return "unsigned", (sig or {}).get("reason", "no signature on the answer") served = sha256_hex(canonical_json(result.get("receipt") or {})) statement = sig.get("statement") or {} if statement.get("schema") != STATEMENT_SCHEMA: return "INVALID", "unexpected statement schema" if served != statement.get("receipt_hash") or served != result.get("receipt_hash"): return "INVALID", "receipt bytes do not hash to the signed receipt_hash" business = result.get("business") or {} key, value = statement_subject(business) if (not value or statement.get(key) != value or statement.get("market") != business.get("market") or statement.get("edition") != result.get("edition")): return "INVALID", "statement names a different business, market or edition" x = keys.get(sig.get("kid")) if not x: return "INVALID", f"no published key with kid {sig.get('kid')}" try: from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey except ImportError: return "unchecked", "hashes match; install `cryptography` to verify the Ed25519 signature" try: Ed25519PublicKey.from_public_bytes(unb64url(x)).verify(unb64url(sig["sig"]), canonical_json(statement).encode("utf-8")) except Exception: # noqa: BLE001 - any failure is an invalid signature to the caller return "INVALID", "Ed25519 signature does not verify" return "valid", f"kid {sig['kid']}" def describe(result: dict, status: str, reason: str) -> str: b = result.get("business") or {} lines = [f"{b.get('name') or result.get('query')} ({b.get('domain') or result.get('query')})", f" signature: {status} ({reason})", f" edition: {result.get('edition')} as of {result.get('edition_as_of')}"] fields = ((result.get("receipt") or {}).get("fields") or {}) for f in FIELDS: env = fields.get(f) or {} state = env.get("state") or "not_published" when = f", checked {env.get('verified_on')}, expires {env.get('expires_on')}" if state == "verified" else "" src = f" · {env.get('source_url')}" if env.get("source_url") else "" lines.append(f" {f}: {state}{when}{src}") return "\n".join(lines) def main(argv: list[str] | None = None) -> int: ap = argparse.ArgumentParser(description=__doc__) ap.add_argument("domains", help="comma-separated, up to 20") ap.add_argument("--api", help="API origin override; defaults to the public discovery card") ap.add_argument("--consumer", default="reference-client") ap.add_argument("--since", help="YYYY-MM-DD for the change feed") ap.add_argument("--fixture-dir", type=Path, help="offline: batch.json, keys.json, changes.json") args = ap.parse_args(argv) asked = [d.strip() for d in args.domains.split(",") if d.strip()] if args.fixture_dir: batch = json.loads((args.fixture_dir / "batch.json").read_text(encoding="utf-8")) keys_body = json.loads((args.fixture_dir / "keys.json").read_text(encoding="utf-8")) changes = json.loads((args.fixture_dir / "changes.json").read_text(encoding="utf-8")) if (args.fixture_dir / "changes.json").exists() else None else: try: if not args.api: card = get_json("https://lodestarstamp.com/.well-known/mcp.json", args.consumer) endpoint = urllib.parse.urlsplit(card["endpoints"]["trust"]) if endpoint.scheme != "https" or not endpoint.netloc: raise ValueError("Discovery must name an HTTPS receipt endpoint") args.api = f"{endpoint.scheme}://{endpoint.netloc}" key = held_key(args.api) q = urllib.parse.quote(",".join(asked), safe=",") batch = get_json(f"{args.api}/v1/trust/batch?domains={q}", args.consumer, key) keys_body = get_json(f"{args.api}/v1/keys", args.consumer, key) changes = get_changes(args.api, args.since, args.consumer, key) if args.since else None except (RuntimeError, ValueError) as error: # ApiError, an unreadable change feed, a bad key print(f"error: {error}", file=sys.stderr) return 2 keys = {k["kid"]: k["x"] for k in keys_body.get("keys", []) if k.get("kid") and k.get("x")} print(f"Lodestar Stamp · {batch.get('count')} asked · order: {batch.get('order')}") print("We report source-backed facts as of the date shown. We do not approve the booking. The assistant decides.\n") invalid = 0 for r in batch.get("results", []): if r.get("status") != "ok": print(f"{r.get('query')}\n {r.get('status')}: {r.get('message')} (not adverse: absence is coverage)\n") continue status, reason = verify(r, keys) invalid += status == "INVALID" print(describe(r, status, reason) + "\n") if changes: wanted = set(asked) moved = [e for e in changes.get("events", []) if e.get("domain") in wanted] print(f"Changes since {args.since}: {len(moved)} event(s) on these domains" + ("" if moved else ".")) for e in moved: fields = ", ".join(f"{c['field']} {c['from']}→{c['to']}" for c in e.get("fields_changed") or []) or "re-dated only" print(f" {e['as_of']} {e['type']} {e['domain']}: {fields}") return 1 if invalid else 0 if __name__ == "__main__": sys.exit(main())