diff --git a/docs/CONTRIBUTING.md b/docs/CONTRIBUTING.md index 7a32990..5c61ef3 100644 --- a/docs/CONTRIBUTING.md +++ b/docs/CONTRIBUTING.md @@ -44,6 +44,29 @@ request must hit a route documented on the page, and each fixture is compiled wi When an example calls an endpoint, add one fixture per request it makes. `no-check` is not allowed on that page. +### Python quickstart + +`guides/quickstarts/python.mdx` runs end to end in CI, offline, as part of +`check:snippets-extended`: + +```bash +npm run check:python-quickstart +``` + +`scripts/python-quickstart/run_quickstart.py` builds `wraith_crypto.py` from the +fence that starts with `# wraith_crypto.py` (plus any `# Add to wraith_crypto.py` +fences), then runs every other Python fence in page order in one session. The +Soroban RPC, Horizon, and Soroban transaction calls go to local fakes that serve +recorded announcer events and verify each transaction's signatures. Results are +compared with `scripts/python-quickstart/vectors.json`, which +`npm run generate:python-quickstart-vectors` computes from the pinned SDK. + +Dependencies are pinned in `scripts/python-quickstart/requirements.txt`. The +`--bootstrap` flag creates a virtualenv with those pins under +`node_modules/.cache`. Every Python fence in the guide must run. A fence that +cannot run needs both words after the language, `python Pseudocode no-check`, so +readers see it is pseudocode. + ## Navigation coverage Every `.mdx` page in the shipped taxonomy (root pages plus `architecture/`, diff --git a/guides/quickstarts/python.mdx b/guides/quickstarts/python.mdx index 9c5ad87..209769f 100644 --- a/guides/quickstarts/python.mdx +++ b/guides/quickstarts/python.mdx @@ -1,14 +1,14 @@ --- title: "Python Quickstart" description: "Derive stealth keys, fetch announcements, scan for owned payments, and send a stealth transfer — all from Python 3.11+ using stellar-sdk and a small crypto utility." -keywords: "Python, stellar-sdk, stealth payment, Soroban RPC, ed25519, X25519, futurenet, scan key" +keywords: "Python, stellar-sdk, stealth payment, Soroban RPC, ed25519, X25519, testnet, scan key" --- - This guide targets the Wraith contracts on the Stellar **futurenet**. The - derivation crypto and RPC calls are identical on testnet and mainnet — only the - `soroban_rpc_url`, `network_passphrase`, and `announcer_contract_id` differ. - See [Stellar Networks](/reference/stellar-networks) for the full address table. + This guide targets the Wraith contracts on Stellar **testnet**. The derivation + crypto and RPC calls are identical on mainnet — only the RPC URLs, + `NETWORK_PASS`, and `ANNOUNCER_CONTRACT_ID` differ. See + [Stellar Networks](/reference/stellar-networks) for the full address table. By the end of this guide you will have: @@ -17,20 +17,25 @@ By the end of this guide you will have: - Fetched announcement events from the Soroban RPC without the TypeScript SDK - Scanned those announcements to find payments addressed to you - Sent a stealth payment and published an on-chain announcement +- Spent from a stealth address with its derived private scalar -**Prerequisites:** Python 3.11+, `pip`, and a Stellar keypair you can fund on futurenet. +**Prerequisites:** Python 3.11+, `pip`, and a Stellar testnet keypair funded with Friendbot. + +The steps build on each other: run them in order in one Python session, or paste +them into one script. --- ## Step 1 — Install dependencies -```bash no-check -pip install stellar-sdk==11.* cryptography +```bash +pip install stellar-sdk==16.1.0 cryptography==50.0.2 httpx==0.28.1 ``` `stellar-sdk` handles Soroban RPC calls, XDR encoding, transaction building, and -`StrKey` address encoding. `cryptography` provides X25519 Diffie-Hellman needed for -the shared-secret step. +`StrKey` address encoding. `cryptography` provides the ed25519 and X25519 +primitives the crypto utility builds on, and `httpx` makes the JSON-RPC calls in +Step 4. These are the versions the docs CI runs this guide against. No TypeScript toolchain is required. @@ -42,7 +47,7 @@ This utility file implements the three primitives the rest of the guide relies o **key derivation**, **stealth address generation**, and **announcement scanning**. It is intentionally self-contained — copy it into your project and import from it. -```python no-check +```python # wraith_crypto.py """ Wraith Protocol — Stellar stealth-address crypto, pure Python. @@ -50,28 +55,25 @@ Wraith Protocol — Stellar stealth-address crypto, pure Python. Implements the same derivation as @wraith-protocol/sdk/chains/stellar: - SHA-256 domain separation for spending/viewing seeds - X25519 ECDH via Edwards-to-Montgomery conversion - - View-tag prefilter (1 byte, ~255/256 skip rate) - - ed25519 scalar addition for stealth pub-key derivation + - v2 view tag over the ephemeral and viewing public keys (1 byte, ~255/256 skip rate) + - ed25519 point arithmetic for stealth public keys + +The curve arithmetic uses Python integers and is not constant-time. It suits +scanning and tooling; keep production signing keys on hardened code. -Requires: cryptography>=41, stellar-sdk>=11 +Requires: stellar-sdk and cryptography (pinned versions in Step 1) """ from __future__ import annotations import hashlib -import hmac -import struct +import os from dataclasses import dataclass from typing import Optional from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey, X25519PublicKey -from cryptography.hazmat.primitives.serialization import ( - Encoding, - PublicFormat, - PrivateFormat, - NoEncryption, -) +from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat from stellar_sdk import Keypair # --------------------------------------------------------------------------- @@ -86,7 +88,7 @@ STEALTH_SIGNING_MESSAGE = ( ) META_ADDRESS_PREFIX = "st:xlm:" -SCHEME_ID = 1 +SCHEME_ID = 2 # the stealth-announcer v2 contract only accepts scheme 2 # ed25519 group order L L = 2**252 + 27742317777372353535851937790883648493 @@ -103,8 +105,11 @@ def _sha256(*parts: bytes) -> bytes: return h.digest() -def _sha512(data: bytes) -> bytes: - return hashlib.sha512(data).digest() +def _sha512(*parts: bytes) -> bytes: + h = hashlib.sha512() + for part in parts: + h.update(part) + return h.digest() def _clamp_scalar(scalar_bytes: bytearray) -> bytearray: @@ -124,8 +129,7 @@ def _seed_to_scalar(seed: bytes) -> int: 3. Clamp a 4. Interpret as little-endian bigint """ - h = _sha512(seed) - a = bytearray(h[:32]) + a = bytearray(_sha512(seed)[:32]) _clamp_scalar(a) return int.from_bytes(a, "little") @@ -137,35 +141,83 @@ def _ed25519_pub_from_seed(seed: bytes) -> bytes: # --------------------------------------------------------------------------- -# Edwards-to-Montgomery conversion (RFC 7748) +# ed25519 point arithmetic (affine coordinates on -x^2 + y^2 = 1 + d*x^2*y^2) # --------------------------------------------------------------------------- -# Ed25519 curve parameters _P = 2**255 - 19 _D = -121665 * pow(121666, _P - 2, _P) % _P +_SQRT_M1 = pow(2, (_P - 1) // 4, _P) + +Point = tuple[int, int] +_IDENTITY: Point = (0, 1) -def _decompress_ed25519(point_bytes: bytes) -> tuple[int, int]: - """Decompress a 32-byte compressed ed25519 point to (x, y).""" +def _decompress_ed25519(point_bytes: bytes) -> Point: + """Decompress a 32-byte ed25519 point to (x, y), rejecting invalid encodings.""" + if len(point_bytes) != 32: + raise ValueError("ed25519 points are 32 bytes") y_int = int.from_bytes(point_bytes, "little") - sign = (y_int >> 255) & 1 + sign = y_int >> 255 y = y_int & ((1 << 255) - 1) + if y >= _P: + raise ValueError("not a valid ed25519 point") - # Recover x from the curve equation: -x^2 + y^2 = 1 + d*x^2*y^2 - y2 = (y * y) % _P - x2 = (y2 - 1) * pow(1 + _D * y2, _P - 2, _P) % _P + # Recover x from the curve equation: x^2 = (y^2 - 1) / (d*y^2 + 1) + y2 = y * y % _P + x2 = (y2 - 1) * pow(_D * y2 + 1, _P - 2, _P) % _P x = pow(x2, (_P + 3) // 8, _P) - if (x * x - x2) % _P != 0: - _I = pow(2, (_P - 1) // 4, _P) - x = x * _I % _P - + x = x * _SQRT_M1 % _P + if (x * x - x2) % _P != 0 or (x == 0 and sign == 1): + raise ValueError("not a valid ed25519 point") if x % 2 != sign: x = _P - x - return x, y +def _compress_ed25519(point: Point) -> bytes: + """Encode (x, y) as 32 bytes: y little-endian, sign of x in the top bit.""" + x, y = point + encoded = bytearray(y.to_bytes(32, "little")) + if x & 1: + encoded[31] |= 0x80 + return bytes(encoded) + + +def _point_add(p1: Point, p2: Point) -> Point: + """ + Edwards point addition (complete on ed25519, so it also doubles): + x3 = (x1*y2 + y1*x2) / (1 + d*x1*x2*y1*y2) + y3 = (y1*y2 + x1*x2) / (1 - d*x1*x2*y1*y2) + """ + x1, y1 = p1 + x2, y2 = p2 + t = _D * x1 * x2 * y1 * y2 % _P + x3 = (x1 * y2 + y1 * x2) * pow(1 + t, _P - 2, _P) % _P + y3 = (y1 * y2 + x1 * x2) * pow(1 - t, _P - 2, _P) % _P + return x3, y3 + + +def _scalar_mult(scalar: int, point: Point) -> Point: + """scalar * point by double-and-add.""" + result, addend = _IDENTITY, point + while scalar: + if scalar & 1: + result = _point_add(result, addend) + addend = _point_add(addend, addend) + scalar >>= 1 + return result + + +# Base point G: y = 4/5, x even +_BASE = _decompress_ed25519((4 * pow(5, _P - 2, _P) % _P).to_bytes(32, "little")) + + +def _scalar_mult_base(scalar: int) -> bytes: + """Return scalar * G, compressed.""" + return _compress_ed25519(_scalar_mult(scalar, _BASE)) + + def _edwards_to_montgomery_pub(ed_pub: bytes) -> bytes: """ Convert an ed25519 public key (Edwards form) to an X25519 key (Montgomery form). @@ -180,10 +232,7 @@ def _edwards_to_montgomery_priv(ed_seed: bytes) -> bytes: """ Convert an ed25519 seed to the X25519 scalar (lower 32 bytes of SHA-512, clamped). """ - h = _sha512(ed_seed) - a = bytearray(h[:32]) - _clamp_scalar(a) - return bytes(a) + return bytes(_clamp_scalar(bytearray(_sha512(ed_seed)[:32]))) # --------------------------------------------------------------------------- @@ -230,6 +279,8 @@ def derive_stealth_keys(signature: bytes) -> StealthKeys: Domain-separated hashing is used (instead of splitting r/s) because ed25519 signature components lack the independence of secp256k1 components. """ + if len(signature) != 64: + raise ValueError("Expected a 64-byte ed25519 signature") spending_key = _sha256(b"wraith:spending:", signature) viewing_key = _sha256(b"wraith:viewing:", signature) @@ -256,13 +307,14 @@ def _x25519_shared_secret(priv_ed_seed: bytes, pub_ed: bytes) -> bytes: return x_priv.exchange(x_pub) -def _compute_view_tag(shared_secret: bytes) -> int: +def compute_view_tag(ephemeral_pub_key: bytes, viewing_pub_key: bytes) -> int: """ - Compute the 1-byte view tag (v2 domain). - view_tag = SHA-256("wraith:stellar:view-tag:v2:" || shared_secret)[0] + Compute the 1-byte v2 view tag (mirrors computeAnnouncementViewTag in the SDK). + view_tag = SHA-256("wraith:stellar:view-tag:v2:" || ephemeral_pub || viewing_pub)[0] + + It needs no ECDH, so scanners reject ~255/256 announcements with one hash. """ - digest = _sha256(b"wraith:stellar:view-tag:v2:", shared_secret) - return digest[0] + return _sha256(b"wraith:stellar:view-tag:v2:", ephemeral_pub_key, viewing_pub_key)[0] def _hash_to_scalar(shared_secret: bytes) -> int: @@ -274,45 +326,14 @@ def _hash_to_scalar(shared_secret: bytes) -> int: return int.from_bytes(digest, "little") % L -def _ed25519_point_add(pub_bytes: bytes, scalar: int) -> bytes: +def _derive_stealth_pub_key(spending_pub_key: bytes, h_scalar: int) -> bytes: """ - Compute P + scalar*G on ed25519 and return the compressed 32-byte result. - Uses the cryptography library's internal scalar multiplication via a trick: - we derive scalar*G from a raw scalar key, then add to P via field arithmetic. - - This mirrors deriveStealthPubKey in the SDK (scalar.ts point addition). + stealth_pub = spending_pub + h_scalar * G + (mirrors deriveStealthPubKey in SDK scalar.ts) """ - # scalar * G → get the public key for scalar as a raw scalar - # We represent scalar as a little-endian 32-byte value and manually construct - # the point using the standard ed25519 signing key structure. - # Note: we're doing point addition in the ed25519 group. - - # Decompress both points - x1, y1 = _decompress_ed25519(pub_bytes) - - # scalar * G: derive public key bytes from scalar - scalar_bytes = scalar.to_bytes(32, "little") - scalar_ed = Ed25519PrivateKey.from_private_bytes(scalar_bytes) - scalar_g_bytes = scalar_ed.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw) - x2, y2 = _decompress_ed25519(scalar_g_bytes) - - # Edwards point addition formula: - # x3 = (x1*y2 + y1*x2) / (1 + d*x1*x2*y1*y2) - # y3 = (y1*y2 + x1*x2) / (1 - d*x1*x2*y1*y2) - x1y2 = x1 * y2 % _P - y1x2 = y1 * x2 % _P - x1x2 = x1 * x2 % _P - y1y2 = y1 * y2 % _P - dx1x2y1y2 = _D * x1x2 * y1y2 % _P - - x3 = (x1y2 + y1x2) * pow(1 + dx1x2y1y2, _P - 2, _P) % _P - y3 = (y1y2 + x1x2) * pow(1 - dx1x2y1y2, _P - 2, _P) % _P - - # Compress: encode y with sign bit of x - result = bytearray(y3.to_bytes(32, "little")) - if x3 & 1: - result[31] |= 0x80 - return bytes(result) + spending_point = _decompress_ed25519(spending_pub_key) + h_point = _scalar_mult(h_scalar, _BASE) + return _compress_ed25519(_point_add(spending_point, h_point)) def _pub_key_to_stellar_address(pub_key: bytes) -> str: @@ -334,28 +355,24 @@ def generate_stealth_address( Algorithm (mirrors generateStealthAddress in SDK stealth.ts): 1. Ephemeral ed25519 key pair 2. X25519 ECDH shared secret with recipient's viewing key - 3. view_tag = SHA-256("wraith:stellar:view-tag:v2:" || shared)[0] + 3. view_tag = SHA-256("wraith:stellar:view-tag:v2:" || ephemeral_pub || viewing_pub)[0] 4. h_scalar = SHA-256("wraith:scalar:" || shared) mod L 5. stealth_point = spending_pub + h_scalar * G (ed25519 point addition) 6. Encode as Stellar G... via StrKey """ - import os if ephemeral_seed is None: ephemeral_seed = os.urandom(32) ephemeral_pub = _ed25519_pub_from_seed(ephemeral_seed) shared = _x25519_shared_secret(ephemeral_seed, viewing_pub_key) - - view_tag = _compute_view_tag(shared) h_scalar = _hash_to_scalar(shared) - stealth_pub = _ed25519_point_add(spending_pub_key, h_scalar) - stealth_address = _pub_key_to_stellar_address(stealth_pub) + stealth_pub = _derive_stealth_pub_key(spending_pub_key, h_scalar) return GeneratedStealthAddress( - stealth_address=stealth_address, + stealth_address=_pub_key_to_stellar_address(stealth_pub), ephemeral_pub_key=ephemeral_pub, - view_tag=view_tag, + view_tag=compute_view_tag(ephemeral_pub, viewing_pub_key), ) @@ -386,6 +403,7 @@ def scan_announcements( Scan a list of announcement dicts for payments addressed to these keys. Each announcement dict must have: + scheme_id : int (2 for the v2 announcer) ephemeral_pub_key : hex string (64 chars, 32 bytes) metadata : hex string (first byte = view tag) stealth_address : str (G... address) @@ -393,29 +411,36 @@ def scan_announcements( Returns only announcements that match — with the derived private scalar needed to spend the funds. - View-tag prefilter: only ~1/256 non-matching announcements reach the - expensive X25519 step (false-positive rate ≈ 0.39 %). + View-tag prefilter: the tag costs one SHA-256, so only ~1/256 of the + non-matching announcements reach the X25519 step (false-positive rate ≈ 0.39 %). """ + viewing_pub_key = _ed25519_pub_from_seed(viewing_key) results: list[MatchedAnnouncement] = [] for ann in announcements: - eph_bytes = bytes.fromhex(ann["ephemeral_pub_key"]) - meta_bytes = bytes.fromhex(ann["metadata"]) - announced_view_tag = meta_bytes[0] + if ann.get("scheme_id") != SCHEME_ID: + continue + try: + eph_bytes = bytes.fromhex(ann["ephemeral_pub_key"]) + announced_view_tag = bytes.fromhex(ann["metadata"])[0] + except (KeyError, ValueError, IndexError): + continue # malformed announcement # ------------------------------------------------------------------ - # Cheap prefilter: compute expected view tag and compare + # Cheap prefilter: compare the view tag before any curve operation # ------------------------------------------------------------------ - shared = _x25519_shared_secret(viewing_key, eph_bytes) - expected_tag = _compute_view_tag(shared) - if expected_tag != announced_view_tag: + if compute_view_tag(eph_bytes, viewing_pub_key) != announced_view_tag: continue # skip ~255/256 of non-matching entries # ------------------------------------------------------------------ # Full check: derive the expected stealth address and compare # ------------------------------------------------------------------ + try: + shared = _x25519_shared_secret(viewing_key, eph_bytes) + except ValueError: + continue # the ephemeral key is not a valid ed25519 point h_scalar = _hash_to_scalar(shared) - stealth_pub = _ed25519_point_add(spending_pub_key, h_scalar) + stealth_pub = _derive_stealth_pub_key(spending_pub_key, h_scalar) derived_address = _pub_key_to_stellar_address(stealth_pub) if derived_address != ann["stealth_address"]: @@ -426,6 +451,8 @@ def scan_announcements( # stealth_scalar = (spending_scalar + h_scalar) mod L # ------------------------------------------------------------------ stealth_scalar = (spending_scalar + h_scalar) % L + if stealth_scalar == 0: + continue results.append(MatchedAnnouncement( stealth_address=derived_address, @@ -438,8 +465,8 @@ def scan_announcements( ``` - The crypto needed for Wraith's Stellar scheme is deliberately small — roughly - 200 lines of hashing and curve arithmetic. Shipping it inline keeps this + The crypto needed for Wraith's Stellar scheme is deliberately small — a few + hundred lines of hashing and curve arithmetic. Shipping it inline keeps this quickstart dependency-free beyond `stellar-sdk` and `cryptography`, both of which are already in most Python fintech backends. A standalone `wraith-stellar-py` package is on the roadmap. @@ -449,7 +476,7 @@ def scan_announcements( ## Step 3 — Derive your stealth keys -```python no-check +```python import os from stellar_sdk import Keypair from wraith_crypto import STEALTH_SIGNING_MESSAGE, derive_stealth_keys, encode_meta_address @@ -462,10 +489,11 @@ keypair = Keypair.from_secret(os.environ["STELLAR_SECRET"]) signature = keypair.sign(STEALTH_SIGNING_MESSAGE.encode()) keys = derive_stealth_keys(signature) +meta_address = encode_meta_address(keys.spending_pub_key, keys.viewing_pub_key) print("Spending pub key:", keys.spending_pub_key.hex()) print("Viewing pub key:", keys.viewing_pub_key.hex()) -print("Meta-address: ", encode_meta_address(keys.spending_pub_key, keys.viewing_pub_key)) +print("Meta-address: ", meta_address) # → st:xlm:<64 hex chars><64 hex chars> ``` @@ -477,15 +505,18 @@ one in the TypeScript SDK, so keys derived here are usable by any SDK implementa ## Step 4 — Fetch announcements from the Soroban RPC -The Soroban RPC's `getEvents` method returns paginated contract events. The utility -below fetches all `WRAITH_ANNOUNCEMENT` events from the announcer contract and +The Soroban RPC's `getEvents` method returns contract events one page at a time. +The utility below pages through recent events from the announcer contract and normalises them into the dict shape `scan_announcements` expects. -```python no-check -import httpx # pip install httpx +```python +import httpx +from stellar_sdk import scval, xdr as stellar_xdr -SOROBAN_RPC_URL = "https://rpc-futurenet.stellar.org" -ANNOUNCER_CONTRACT_ID = "CCJLJ2QRBJAAKIG6ELNQVXLLWMKKWVN5O2FKWUETHZGMPAD4MHK7WVWL" # testnet value; futurenet has no live contracts +from wraith_crypto import SCHEME_ID + +SOROBAN_RPC_URL = "https://soroban-testnet.stellar.org" +ANNOUNCER_CONTRACT_ID = "CCJLJ2QRBJAAKIG6ELNQVXLLWMKKWVN5O2FKWUETHZGMPAD4MHK7WVWL" # stealth-announcer v2, testnet def _rpc(method: str, params: dict) -> dict: @@ -502,125 +533,87 @@ def _rpc(method: str, params: dict) -> dict: return body["result"] -def _current_ledger_range() -> tuple[int, int]: - """Return (oldest_ledger, latest_ledger) from getLatestLedger.""" - info = _rpc("getLatestLedger", {}) - latest = info["sequence"] - # Soroban RPC retains ~17,280 ledgers (~24 h at 5 s/ledger) - oldest = max(latest - 17_000, info.get("oldestLedgerSequence", 0)) - return oldest, latest +def _decode(scval_xdr: str): + """Decode a base64 XDR ScVal into plain Python values.""" + return scval.to_native(stellar_xdr.SCVal.from_xdr(scval_xdr)) -def fetch_announcements() -> list[dict]: +def fetch_announcements(lookback_ledgers: int = 17_000) -> list[dict]: """ - Fetch all WRAITH_ANNOUNCEMENT events from the announcer contract. + Fetch recent announcements from the stealth-announcer contract. + + The v2 announcer publishes each announcement as + topics = ("announce", scheme_id, view_tag_bucket, metadata_kind) + value = (stealth_address, ephemeral_pub_key, metadata) - Returns a list of announcement dicts with keys: - stealth_address, caller, ephemeral_pub_key, metadata, scheme_id + Returns a list of dicts with keys: + scheme_id, stealth_address, ephemeral_pub_key, metadata """ - oldest, latest = _current_ledger_range() - - result = _rpc("getEvents", { - "startLedger": oldest, - "filters": [{ - "type": "contract", - "contractIds": [ANNOUNCER_CONTRACT_ID], - "topics": [["*", "*"]], # accept all topics from this contract - }], + latest = _rpc("getLatestLedger", {})["sequence"] + filters = [{"type": "contract", "contractIds": [ANNOUNCER_CONTRACT_ID]}] + # The RPC keeps a limited event history (~17,280 ledgers is ~24 h at 5 s/ledger) + params = { + "startLedger": max(latest - lookback_ledgers, 1), + "filters": filters, "pagination": {"limit": 200}, - }) + } announcements = [] - for event in result.get("events", []): - # Each Soroban event has a `topic` array and a `value` (XDR ScVal) - # The announcer contract emits: - # topic[0] = Symbol("WRAITH_ANNOUNCEMENT") - # topic[1] = scheme_id (u32) - # value = Map { stealth_address, caller, ephemeral_pub_key, metadata } - topics = event.get("topic", []) - if not topics: - continue - - # Decode the XDR value using stellar-sdk - from stellar_sdk import xdr as stellar_xdr - import base64 - - raw_xdr = event.get("value", {}).get("xdr", "") - if not raw_xdr: - continue - - try: - sc_val = stellar_xdr.SCVal.from_xdr(raw_xdr) - except Exception: - continue # malformed event — skip - - # Parse the map entries - ann: dict = {} - if sc_val.type == stellar_xdr.SCValType.SCV_MAP and sc_val.map: - for entry in sc_val.map.sc_map: - key_sym = entry.key.sym.sc_symbol.decode() if entry.key.sym else None - if key_sym == "stealth_address": - ann["stealth_address"] = entry.val.address.account_id # G... address - elif key_sym == "caller": - ann["caller"] = entry.val.address.account_id - elif key_sym == "ephemeral_pub_key": - ann["ephemeral_pub_key"] = entry.val.bytes.sc_bytes.hex() - elif key_sym == "metadata": - ann["metadata"] = entry.val.bytes.sc_bytes.hex() - - if {"stealth_address", "ephemeral_pub_key", "metadata"} <= ann.keys(): - announcements.append(ann) + while True: + page = _rpc("getEvents", params) + events = page.get("events", []) + if not events: + break + + for event in events: + topics = [_decode(topic) for topic in event["topic"]] + if len(topics) != 4 or topics[0] != "announce" or topics[1] != SCHEME_ID: + continue # not a v2 announcement + stealth_address, ephemeral_pub_key, metadata = _decode(event["value"]) + announcements.append({ + "scheme_id": topics[1], + "stealth_address": stealth_address.address, # G... address + "ephemeral_pub_key": ephemeral_pub_key.hex(), + "metadata": metadata.hex(), + }) + + # The cursor continues after the last event; it replaces startLedger + params = {"filters": filters, "pagination": {"cursor": page["cursor"], "limit": 200}} return announcements ``` - - Futurenet resets on demand and carries no live Wraith contracts — use the - testnet `ANNOUNCER_CONTRACT_ID` when testing against testnet. - On futurenet you can still exercise the full code path by injecting synthetic - announcements (see Step 6 below). - - - For high-throughput backends, consider caching the announcement list and - polling only for events newer than the last seen ledger sequence, incrementing - `startLedger` on each poll instead of re-fetching from `oldest`. + For high-throughput backends, cache the announcement list and keep the last + cursor. Each poll then asks only for events after it instead of re-fetching the + whole window. --- ## Step 5 — Scan for owned announcements -```python no-check -import os -from stellar_sdk import Keypair -from wraith_crypto import ( - STEALTH_SIGNING_MESSAGE, - derive_stealth_keys, - scan_announcements, -) - -keypair = Keypair.from_secret(os.environ["STELLAR_SECRET"]) -keys = derive_stealth_keys(keypair.sign(STEALTH_SIGNING_MESSAGE.encode())) +```python +from wraith_crypto import scan_announcements announcements = fetch_announcements() # from Step 4 matches = scan_announcements( announcements, - viewing_key=keys.viewing_key, + viewing_key=keys.viewing_key, # `keys` from Step 3 spending_pub_key=keys.spending_pub_key, spending_scalar=keys.spending_scalar, ) for m in matches: print(f"Found payment at: {m.stealth_address}") - print(f" Private scalar: {hex(m.stealth_private_scalar)}") - # m.stealth_private_scalar lets you sign transactions from this stealth address + # m.stealth_private_scalar signs for this address (Step 7). + # Treat it like a secret key: never log or share it. ``` -The scanner applies the view-tag prefilter first, so the expensive X25519 operation -runs only for ~1 in 256 non-matching announcements. On a list of 10,000 -announcements you would expect ~39 false-positive ECDH calls before the full +The scanner checks the view tag first. It costs one SHA-256, so the expensive +X25519 operation runs only for ~1 in 256 non-matching announcements. On a list of +10,000 announcements you would expect ~39 false-positive ECDH calls before the full address check eliminates them. --- @@ -633,12 +626,13 @@ announcer contract. ### 6a — Resolve the recipient's meta-address -```python no-check +```python from wraith_crypto import decode_meta_address, generate_stealth_address -# The recipient shares their meta-address — either directly or via a .wraith name. -# For this quickstart we use a literal; in production resolve it from the names contract. -RECIPIENT_META_ADDRESS = "st:xlm:" +# In production the recipient shares their meta-address, directly or through a +# .wraith name. Here you pay your own meta-address from Step 3, so Steps 4 and 5 +# find the payment once the announcement lands. +RECIPIENT_META_ADDRESS = meta_address spending_pub, viewing_pub = decode_meta_address(RECIPIENT_META_ADDRESS) stealth = generate_stealth_address(spending_pub, viewing_pub) @@ -654,25 +648,21 @@ New Stellar accounts must be activated with `createAccount`. After activation, c the Soroban announcer contract to publish the ephemeral key and view tag so the recipient can detect their payment. -```python no-check +```python import os -import base64 -from stellar_sdk import ( - Keypair, Network, Server, TransactionBuilder, Asset, - Operation, -) -from stellar_sdk.soroban_rpc import SorobanServer -from stellar_sdk import xdr as stellar_xdr +from stellar_sdk import Keypair, Network, Server, SorobanServer, TransactionBuilder, scval +from stellar_sdk.soroban_rpc import SendTransactionStatus -HORIZON_URL = "https://horizon-testnet.stellar.org" -SOROBAN_URL = "https://soroban-testnet.stellar.org" -NETWORK_PASS = Network.TESTNET_NETWORK_PASSPHRASE -AMOUNT_XLM = "10" # minimum to activate the new stealth account +from wraith_crypto import SCHEME_ID + +HORIZON_URL = "https://horizon-testnet.stellar.org" +NETWORK_PASS = Network.TESTNET_NETWORK_PASSPHRASE +AMOUNT_XLM = "10" # starting balance for the stealth account (the minimum is 1 XLM) sender_keypair = Keypair.from_secret(os.environ["STELLAR_SECRET"]) -horizon = Server(HORIZON_URL) -soroban = SorobanServer(SOROBAN_URL) +horizon = Server(HORIZON_URL) +soroban = SorobanServer(SOROBAN_RPC_URL) # from Step 4 # ------------------------------------------------------------------ # Part 1: createAccount — activate the stealth address @@ -699,19 +689,14 @@ print("Stealth account activated:", stealth.stealth_address) # ------------------------------------------------------------------ # Part 2: invoke stealth-announcer::announce # ------------------------------------------------------------------ -# Build the announcement metadata: first byte = view_tag -metadata_bytes = bytes([stealth.view_tag]) - -# Construct the Soroban invocation arguments # announce(scheme_id: u32, stealth_address: Address, -# ephemeral_pub_key: Bytes, metadata: Bytes) -from stellar_sdk import scval - +# ephemeral_pub_key: BytesN<32>, metadata: Bytes) +# The first metadata byte is the view tag. invoke_args = [ - scval.to_uint32(1), # scheme_id - scval.to_address(stealth.stealth_address), # stealth address - scval.to_bytes(stealth.ephemeral_pub_key), # ephemeral pub key - scval.to_bytes(metadata_bytes), # metadata (view tag) + scval.to_uint32(SCHEME_ID), # scheme_id (2) + scval.to_address(stealth.stealth_address), # stealth address + scval.to_bytes(stealth.ephemeral_pub_key), # ephemeral pub key + scval.to_bytes(bytes([stealth.view_tag])), # metadata (view tag) ] sender_account = horizon.load_account(sender_keypair.public_key) # refresh sequence @@ -731,19 +716,22 @@ announce_tx = ( .build() ) -# Simulate to get resource footprint before signing -sim = soroban.simulate_transaction(announce_tx) -announce_tx = sim.transaction # updated with resource fee +# Simulate, then attach the resource footprint and fee before signing +announce_tx = soroban.prepare_transaction(announce_tx) announce_tx.sign(sender_keypair) -response = soroban.send_transaction(announce_tx) -print("Announcement tx hash:", response.hash) +sent = soroban.send_transaction(announce_tx) +if sent.status != SendTransactionStatus.PENDING: + raise RuntimeError(f"Announcement rejected: {sent.status}") +result = soroban.poll_transaction(sent.hash) # waits for the ledger to close +print("Announcement tx:", sent.hash, result.status.value) ``` - On futurenet the Wraith contracts are not deployed, so the announcer call will - fail. Run Step 6 against **testnet** using the testnet `ANNOUNCER_CONTRACT_ID` - and the testnet Horizon/Soroban URLs from [Stellar Networks](/reference/stellar-networks). + Fund the sender on testnet with Friendbot before running Step 6. On mainnet, + swap in the mainnet Horizon and RPC URLs, `Network.PUBLIC_NETWORK_PASSPHRASE`, + and the mainnet `ANNOUNCER_CONTRACT_ID` from + [Stellar Networks](/reference/stellar-networks). --- @@ -752,84 +740,75 @@ print("Announcement tx hash:", response.hash) To spend funds held at a stealth address you need the private scalar derived during scanning. Because stealth scalars are not raw seeds, the standard -`Keypair.from_raw_ed25519_seed()` path does not work. Use `signWithScalar` from -`wraith_crypto` below. +`Keypair.from_raw_ed25519_seed()` path does not work. Add `sign_with_scalar` to +`wraith_crypto.py`: -```python no-check +```python # Add to wraith_crypto.py -def sign_with_scalar( - message: bytes, - scalar: int, - pub_key_bytes: bytes, -) -> bytes: +def sign_with_scalar(message: bytes, scalar: int, pub_key_bytes: bytes) -> bytes: """ Sign `message` using a raw ed25519 scalar (not a seed). Required because stealth scalars are (spending_scalar + h_scalar) mod L, which is not necessarily clamped and cannot be fed to standard seed-based - signing APIs. + signing APIs. Mirrors signWithScalar in SDK scalar.ts, byte for byte. Compatible with RFC 8032 ed25519 — the signature will verify against `pub_key_bytes` using any standard ed25519 verifier. """ - import struct - - # Encode scalar as little-endian 32 bytes + if not 0 < scalar < L: + raise ValueError("Scalar must be in range (0, L)") s_bytes = scalar.to_bytes(32, "little") - # Deterministic nonce: SHA-512(scalar_bytes || message), lower half, clamped - nonce_hash = _sha512(s_bytes + message) - r_bytes = bytearray(nonce_hash[:32]) - _clamp_scalar(r_bytes) - r = int.from_bytes(r_bytes, "little") + # Deterministic nonce: r = SHA-512(SHA-256(scalar_bytes) || message) mod L + r = int.from_bytes(_sha512(_sha256(s_bytes), message), "little") % L # R = r * G - r_key = Ed25519PrivateKey.from_private_bytes(bytes(r_bytes)) - R_bytes = r_key.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw) + R_bytes = _scalar_mult_base(r) # S = (r + SHA-512(R || pub_key || message) * s) mod L - k_hash = _sha512(R_bytes + pub_key_bytes + message) - k = int.from_bytes(k_hash, "little") % L + k = int.from_bytes(_sha512(R_bytes, pub_key_bytes, message), "little") % L S = (r + k * scalar) % L return R_bytes + S.to_bytes(32, "little") ``` -Use it when building a withdrawal transaction from a matched stealth address: +Use it to withdraw from a matched stealth address. The example merges the +stealth account into your primary account, which sends its whole balance (base +reserve included) and closes the one-time account: + +```python +from stellar_sdk import DecoratedSignature -```python no-check from wraith_crypto import sign_with_scalar -# `match` is a MatchedAnnouncement from scan_announcements() +if not matches: + raise SystemExit("No payments yet. Re-run Steps 4 and 5 after the Step 6 announcement lands.") + +match = matches[0] # a MatchedAnnouncement from Step 5 stealth_keypair = Keypair.from_raw_ed25519_public_key(match.stealth_pub_key_bytes) stealth_account = horizon.load_account(stealth_keypair.public_key) -# Build a payment back to your primary account withdrawal_tx = ( TransactionBuilder( source_account=stealth_account, network_passphrase=NETWORK_PASS, base_fee=1000, ) - .append_payment_op( - destination=sender_keypair.public_key, - asset=Asset.native(), - amount="9", # leave ~1 XLM for fees / base reserve - ) + .append_account_merge_op(destination=sender_keypair.public_key) .set_timeout(30) .build() ) -# Sign with raw scalar — not with a Keypair seed -tx_hash = withdrawal_tx.hash() -sig_bytes = sign_with_scalar(tx_hash, match.stealth_private_scalar, match.stealth_pub_key_bytes) - -withdrawal_tx.transaction.add_signature( - network_id=Network.TESTNET_NETWORK_PASSPHRASE, - public_key=stealth_keypair.public_key, - signature=sig_bytes, +# Sign with the raw scalar — not with a Keypair seed +sig_bytes = sign_with_scalar( + withdrawal_tx.hash(), match.stealth_private_scalar, match.stealth_pub_key_bytes +) +withdrawal_tx.signatures.append( + DecoratedSignature(stealth_keypair.signature_hint(), sig_bytes) ) + response = horizon.submit_transaction(withdrawal_tx) print("Withdrawal tx hash:", response["hash"]) ``` @@ -861,7 +840,7 @@ concerns — it handles only the cryptographic layer so it stays testable in iso ## Environment variables -```bash no-check +```bash # .env — never commit this file STELLAR_SECRET=S...your-secret-key diff --git a/package.json b/package.json index 6e19237..64c95f2 100644 --- a/package.json +++ b/package.json @@ -6,7 +6,9 @@ "check:placeholders": "tsx scripts/check-placeholders.ts", "check:snippets": "tsx scripts/check-snippets.ts && tsx scripts/check-api-examples.ts", "check:api-examples": "tsx scripts/check-api-examples.ts", - "check:snippets-extended": "tsx scripts/check-snippets-extended.ts", + "check:snippets-extended": "tsx scripts/check-snippets-extended.ts && npm run check:python-quickstart", + "check:python-quickstart": "tsx scripts/python-quickstart/generate-vectors.ts --check && python3 scripts/python-quickstart/run_quickstart.py --bootstrap", + "generate:python-quickstart-vectors": "tsx scripts/python-quickstart/generate-vectors.ts", "check:nav-coverage": "node scripts/check-nav-coverage.mjs", "check:contract-registry": "node scripts/check-contract-registry.mjs", "check:compatibility-matrix": "node scripts/check-compatibility-matrix.mjs", diff --git a/scripts/python-quickstart/generate-vectors.ts b/scripts/python-quickstart/generate-vectors.ts new file mode 100644 index 0000000..7ab477f --- /dev/null +++ b/scripts/python-quickstart/generate-vectors.ts @@ -0,0 +1,229 @@ +/** + * generate-vectors.ts + * + * Writes scripts/python-quickstart/vectors.json: the expected values the Python + * quickstart (guides/quickstarts/python.mdx) must reproduce, computed with the + * pinned @wraith-protocol/sdk Stellar primitives. + * + * tsx scripts/python-quickstart/generate-vectors.ts # write vectors.json + * tsx scripts/python-quickstart/generate-vectors.ts --check # fail if it is stale + * + * The announcement events use the stealth-announcer v2 layout + * (wraith-protocol/contracts stellar/stealth-announcer): topics + * ("announce", scheme_id, view_tag_bucket, metadata_kind) and value + * (stealth_address, ephemeral_pub_key, metadata). View tags use the v2 + * derivation SHA-256("wraith:stellar:view-tag:v2:" || R || V)[0] + * (architecture/view-tag-scanner.mdx). + */ + +import { createHash } from "node:crypto"; +import { readFile, writeFile } from "node:fs/promises"; +import path from "node:path"; +import process from "node:process"; +import { fileURLToPath } from "node:url"; +import { Address, Keypair, xdr } from "@stellar/stellar-sdk"; +import { + STEALTH_SIGNING_MESSAGE, + bytesToHex, + deriveStealthKeys, + deriveStealthPrivateScalar, + encodeStealthMetaAddress, + generateStealthAddress, + hexToBytes, + signWithScalar, +} from "@wraith-protocol/sdk/chains/stellar"; + +const here = path.dirname(fileURLToPath(import.meta.url)); +const outFile = path.join(here, "vectors.json"); + +const ANNOUNCER_CONTRACT_ID = "CCJLJ2QRBJAAKIG6ELNQVXLLWMKKWVN5O2FKWUETHZGMPAD4MHK7WVWL"; +const SCHEME_ID = 2; +const METADATA_KIND_VIEW_TAG = 1; +const LATEST_LEDGER = 1_990_000; + +const sha256 = (...parts: Array) => { + const hash = createHash("sha256"); + for (const part of parts) hash.update(part); + return new Uint8Array(hash.digest()); +}; + +const seed = (label: string) => sha256(`wraith:docs:python-quickstart:${label}`); + +/** The v2 view tag the stealth-announcer v2 contract indexes by. */ +function viewTagV2(ephemeralPubKey: Uint8Array, viewingPubKey: Uint8Array) { + return sha256("wraith:stellar:view-tag:v2:", ephemeralPubKey, viewingPubKey)[0]; +} + +/** A Stellar account and the stealth keys derived from its signature. */ +function identity(label: string) { + const keypair = Keypair.fromRawEd25519Seed(Buffer.from(seed(label))); + const signature = new Uint8Array(keypair.sign(Buffer.from(STEALTH_SIGNING_MESSAGE))); + const keys = deriveStealthKeys(signature); + return { keypair, signature, keys }; +} + +/** Pays `recipient` through a fresh stealth address, as the SDK builders do. */ +function pay(recipient: ReturnType, ephemeralLabel: string) { + const ephemeralSeed = seed(ephemeralLabel); + const generated = generateStealthAddress( + recipient.keys.spendingPubKey, + recipient.keys.viewingPubKey, + ephemeralSeed, + ); + const viewTag = viewTagV2(generated.ephemeralPubKey, recipient.keys.viewingPubKey); + return { ephemeralSeed, ...generated, viewTag }; +} + +function announcementEvent( + index: number, + stealthAddress: string, + ephemeralPubKey: Uint8Array, + metadata: Uint8Array, +) { + const ledger = LATEST_LEDGER - 100 + index; + const topic = [ + xdr.ScVal.scvSymbol("announce"), + xdr.ScVal.scvU32(SCHEME_ID), + xdr.ScVal.scvU32(metadata[0]), + xdr.ScVal.scvU32(METADATA_KIND_VIEW_TAG), + ]; + const value = xdr.ScVal.scvVec([ + new Address(stealthAddress).toScVal(), + xdr.ScVal.scvBytes(Buffer.from(ephemeralPubKey)), + xdr.ScVal.scvBytes(Buffer.from(metadata)), + ]); + + return { + type: "contract", + ledger, + ledgerClosedAt: new Date(Date.UTC(2026, 9, 1, 12, 0, index * 5)).toISOString(), + contractId: ANNOUNCER_CONTRACT_ID, + id: `${(BigInt(ledger) << 32n).toString().padStart(19, "0")}-${String(index).padStart(10, "0")}`, + txHash: bytesToHex(sha256(`tx:${index}`)), + inSuccessfulContractCall: true, + topic: topic.map((entry) => entry.toXDR("base64")), + value: value.toXDR("base64"), + }; +} + +function buildVectors() { + const owner = identity("owner"); + const other = identity("other-recipient"); + + // Announcements 0 and 2 pay the owner. 1 and 4 pay someone else. 3 is a decoy: + // it carries the owner's view tag for its ephemeral key but another stealth + // address, so it passes the view-tag prefilter and must fail the address check. + const toOwner = [pay(owner, "ephemeral:0"), pay(owner, "ephemeral:2")]; + const toOther = [pay(other, "ephemeral:1"), pay(other, "ephemeral:4")]; + const decoy = pay(other, "ephemeral:3"); + const decoyTag = viewTagV2(decoy.ephemeralPubKey, owner.keys.viewingPubKey); + + const announcements = [ + { payment: toOwner[0], viewTag: toOwner[0].viewTag }, + { payment: toOther[0], viewTag: toOther[0].viewTag }, + { payment: toOwner[1], viewTag: toOwner[1].viewTag }, + { payment: decoy, viewTag: decoyTag }, + { payment: toOther[1], viewTag: toOther[1].viewTag }, + ]; + + const events = announcements.map(({ payment, viewTag }, index) => + announcementEvent( + index, + payment.stealthAddress, + payment.ephemeralPubKey, + Uint8Array.of(viewTag), + ), + ); + + const matches = [0, 2].map((index) => { + const payment = announcements[index].payment; + const scalar = deriveStealthPrivateScalar( + owner.keys.spendingScalar, + owner.keys.viewingKey, + payment.ephemeralPubKey, + ); + const stealthPubKey = Keypair.fromPublicKey(payment.stealthAddress).rawPublicKey(); + return { + index, + stealthAddress: payment.stealthAddress, + stealthPrivateScalar: scalar.toString(), + stealthPubKey: bytesToHex(new Uint8Array(stealthPubKey)), + }; + }); + + const message = sha256("message"); + const signature = signWithScalar( + message, + BigInt(matches[0].stealthPrivateScalar), + hexToBytes(matches[0].stealthPubKey), + ); + + return { + $comment: + "Generated by scripts/python-quickstart/generate-vectors.ts from the pinned " + + "@wraith-protocol/sdk Stellar primitives. Do not edit by hand; run `pnpm run " + + "generate:python-quickstart-vectors`.", + stealthSigningMessage: STEALTH_SIGNING_MESSAGE, + owner: { + secret: owner.keypair.secret(), + publicKey: owner.keypair.publicKey(), + signature: bytesToHex(owner.signature), + spendingKey: bytesToHex(owner.keys.spendingKey), + viewingKey: bytesToHex(owner.keys.viewingKey), + spendingScalar: owner.keys.spendingScalar.toString(), + viewingScalar: owner.keys.viewingScalar.toString(), + spendingPubKey: bytesToHex(owner.keys.spendingPubKey), + viewingPubKey: bytesToHex(owner.keys.viewingPubKey), + metaAddress: encodeStealthMetaAddress(owner.keys.spendingPubKey, owner.keys.viewingPubKey), + }, + stealthAddresses: toOwner.map((payment) => ({ + ephemeralSeed: bytesToHex(payment.ephemeralSeed), + ephemeralPubKey: bytesToHex(payment.ephemeralPubKey), + stealthAddress: payment.stealthAddress, + viewTag: payment.viewTag, + })), + rpc: { + announcerContractId: ANNOUNCER_CONTRACT_ID, + latestLedger: LATEST_LEDGER, + events, + }, + announcements: announcements.map(({ payment, viewTag }) => ({ + scheme_id: SCHEME_ID, + stealth_address: payment.stealthAddress, + ephemeral_pub_key: bytesToHex(payment.ephemeralPubKey), + metadata: bytesToHex(Uint8Array.of(viewTag)), + })), + matches, + scalarSignature: { + message: bytesToHex(message), + scalar: matches[0].stealthPrivateScalar, + publicKey: matches[0].stealthPubKey, + signature: bytesToHex(signature), + }, + }; +} + +async function main() { + const content = `${JSON.stringify(buildVectors(), null, 2)}\n`; + + if (process.argv.includes("--check")) { + const current = await readFile(outFile, "utf8").catch(() => null); + if (current !== content) { + console.error( + "scripts/python-quickstart/vectors.json is out of date. " + + "Run `pnpm run generate:python-quickstart-vectors` and commit the result.", + ); + process.exit(1); + } + console.log("Python quickstart vectors are in sync with the pinned SDK."); + return; + } + + await writeFile(outFile, content, "utf8"); + console.log(`Wrote ${path.relative(process.cwd(), outFile)}`); +} + +main().catch((error) => { + console.error(error); + process.exit(1); +}); diff --git a/scripts/python-quickstart/requirements.txt b/scripts/python-quickstart/requirements.txt new file mode 100644 index 0000000..74b0f7a --- /dev/null +++ b/scripts/python-quickstart/requirements.txt @@ -0,0 +1,25 @@ +# Pinned dependencies for guides/quickstarts/python.mdx, checked by run_quickstart.py. +# Direct: stellar-sdk, cryptography, httpx (Step 1 of the guide). The rest is their +# dependency tree, frozen from a clean Python 3.11 environment. +annotated-types==0.8.0 +anyio==4.15.1 +certifi==2026.7.22 +cffi==2.1.1 +charset-normalizer==3.5.2 +cryptography==50.0.2 +h11==0.16.0 +httpcore==1.0.9 +httpx==0.28.1 +idna==3.20 +mnemonic==0.21 +pycparser==3.11 +pydantic==2.14.1 +pydantic_core==2.50.1 +PyNaCl==1.6.2 +requests==2.34.2 +requests-sse==0.5.3 +stellar-sdk==16.1.0 +typing-inspection==0.4.4 +typing_extensions==4.16.0 +urllib3==2.8.0 +xdrlib3==0.1.1 diff --git a/scripts/python-quickstart/run_quickstart.py b/scripts/python-quickstart/run_quickstart.py new file mode 100644 index 0000000..942dd90 --- /dev/null +++ b/scripts/python-quickstart/run_quickstart.py @@ -0,0 +1,572 @@ +#!/usr/bin/env python3 +""" +run_quickstart.py: runs guides/quickstarts/python.mdx offline. + +The guide's Python fences are the fixture. The fence that starts with +`# wraith_crypto.py` becomes that module, fences that start with +`# Add to wraith_crypto.py` are appended to it, and every other Python fence runs +in page order in one namespace, the way a reader pastes the steps into one +session. A Python fence may only be skipped if its info string says both +`no-check` and `pseudocode`. + +Network calls go to fakes instead of testnet: + - httpx.post answers getLatestLedger and serves the announcer events from + vectors.json, two per getEvents page, so the cursor loop runs. + - stellar_sdk.Server (Horizon) and stellar_sdk.SorobanServer check each + transaction's network and signatures and record it. + +The results are compared with vectors.json, which generate-vectors.ts computes +with the pinned TypeScript SDK, and with an independent libsodium (PyNaCl) +derivation for the random stealth address in Step 6a. + +Installed packages must match requirements.txt exactly. With --bootstrap the +script creates a virtualenv under node_modules/.cache with those pins and +re-runs itself there. + +Usage + python3 scripts/python-quickstart/run_quickstart.py [--bootstrap] +""" + +from __future__ import annotations + +import contextlib +import hashlib +import io +import json +import os +import re +import subprocess +import sys +import tempfile +import traceback +from dataclasses import dataclass +from importlib import metadata +from pathlib import Path +from types import SimpleNamespace +from typing import Any + +HERE = Path(__file__).resolve().parent +REPO = HERE.parents[1] +GUIDE = REPO / "guides" / "quickstarts" / "python.mdx" +GUIDE_LABEL = "guides/quickstarts/python.mdx" +VECTORS = HERE / "vectors.json" +REQUIREMENTS = HERE / "requirements.txt" +MODULE = "wraith_crypto" +TESTNET_PASSPHRASE = "Test SDF Network ; September 2015" +EVENTS_PER_PAGE = 2 + + +class QuickstartError(Exception): + """A check failed; the message says where and why.""" + + +# --------------------------------------------------------------------------- +# Pinned dependencies +# --------------------------------------------------------------------------- + +def _normalize(name: str) -> str: + return re.sub(r"[-_.]+", "-", name).lower() + + +def pin_mismatches() -> list[str]: + problems = [] + for raw in REQUIREMENTS.read_text().splitlines(): + line = raw.split("#", 1)[0].strip() + if not line: + continue + name, _, version = line.partition("==") + try: + installed = metadata.version(name) + except metadata.PackageNotFoundError: + problems.append(f"{name}=={version} is not installed") + continue + if installed != version: + problems.append(f"{name}=={version} is pinned, {installed} is installed") + return problems + + +def ensure_pinned_environment(argv: list[str]) -> None: + problems = pin_mismatches() + if not problems: + return + if "--bootstrap" not in argv or os.environ.get("WRAITH_QUICKSTART_VENV"): + raise SystemExit( + "Python dependencies do not match scripts/python-quickstart/requirements.txt:\n " + + "\n ".join(problems) + + "\nInstall them with `python3 -m pip install -r scripts/python-quickstart/requirements.txt`" + " in a virtualenv, or run with --bootstrap." + ) + + digest = hashlib.sha256(REQUIREMENTS.read_bytes()).hexdigest()[:12] + venv = REPO / "node_modules" / ".cache" / "python-quickstart" / f"venv-{digest}" + python = venv / ("Scripts/python.exe" if os.name == "nt" else "bin/python") + if not python.exists(): + print(f"Creating {venv.relative_to(REPO)} with the pinned dependencies...", flush=True) + subprocess.run([sys.executable, "-m", "venv", str(venv)], check=True) + subprocess.run( + [str(python), "-m", "pip", "install", "--quiet", "--disable-pip-version-check", + "-r", str(REQUIREMENTS)], + check=True, + ) + env = {**os.environ, "WRAITH_QUICKSTART_VENV": "1"} + args = [arg for arg in argv if arg != "--bootstrap"] + completed = subprocess.run([str(python), str(Path(__file__).resolve()), *args], env=env) + raise SystemExit(completed.returncode) + + +# --------------------------------------------------------------------------- +# Extracting the fixture from the guide +# --------------------------------------------------------------------------- + +@dataclass +class Fence: + lang: str + info: str + code: str + line: int # 1-based MDX line of the first line inside the fence + + +def parse_fences(markdown: str) -> list[Fence]: + fences: list[Fence] = [] + open_fence: dict[str, Any] | None = None + for index, line in enumerate(markdown.split("\n")): + if open_fence is None: + match = re.match(r"^```([A-Za-z0-9_-]*)(.*)$", line) + if match: + open_fence = {"lang": match[1].lower(), "info": match[2].strip(), + "line": index + 2, "body": []} + continue + if line.strip() == "```": + fences.append(Fence(open_fence["lang"], open_fence["info"], + "\n".join(open_fence["body"]), open_fence["line"])) + open_fence = None + continue + open_fence["body"].append(line) + return fences + + +@dataclass +class Fixture: + module_source: str + module_parts: list[Fence] + steps: list[Fence] + pseudocode: list[Fence] + + +def extract_fixture(markdown: str) -> Fixture: + module_parts: list[Fence] = [] + steps: list[Fence] = [] + pseudocode: list[Fence] = [] + + for fence in parse_fences(markdown): + if fence.lang not in ("python", "py"): + continue + if re.search(r"\bno-check\b", fence.info): + if not re.search(r"\bpseudocode\b", fence.info, re.IGNORECASE): + raise QuickstartError( + f"{GUIDE_LABEL}:{fence.line - 1}: Python fences run as part of the fixture. " + "Remove `no-check`, or mark pseudocode as ```python Pseudocode no-check." + ) + pseudocode.append(fence) + continue + + first_line = fence.code.lstrip().split("\n", 1)[0].strip() + if first_line == f"# {MODULE}.py": + if module_parts: + raise QuickstartError(f"{GUIDE_LABEL}:{fence.line}: a second `# {MODULE}.py` fence") + module_parts.append(fence) + elif first_line == f"# Add to {MODULE}.py": + if not module_parts: + raise QuickstartError(f"{GUIDE_LABEL}:{fence.line}: `# Add to {MODULE}.py` " + f"comes before the `# {MODULE}.py` fence") + module_parts.append(fence) + else: + steps.append(fence) + + if not module_parts: + raise QuickstartError(f"{GUIDE_LABEL}: no fence starts with `# {MODULE}.py`") + if not steps: + raise QuickstartError(f"{GUIDE_LABEL}: no runnable steps found") + + module_source = "\n\n".join(part.code for part in module_parts) + "\n" + return Fixture(module_source, module_parts, steps, pseudocode) + + +# --------------------------------------------------------------------------- +# Network fakes +# --------------------------------------------------------------------------- + +def _require(condition: bool, message: str) -> None: + if not condition: + raise QuickstartError(message) + + +class FakeRpc: + """Stands in for httpx.post against the Soroban RPC.""" + + def __init__(self, vectors: dict) -> None: + self.latest = vectors["rpc"]["latestLedger"] + self.contract_id = vectors["rpc"]["announcerContractId"] + self.events = vectors["rpc"]["events"] + self.calls: list[str] = [] + + def post(self, url: str, json: dict | None = None, timeout: float | None = None, **_: Any): + _require(isinstance(json, dict) and json.get("jsonrpc") == "2.0", + f"RPC call to {url} is not JSON-RPC 2.0") + method, params = json.get("method"), json.get("params") or {} + self.calls.append(method) + if method == "getLatestLedger": + result = {"id": "fixture", "protocolVersion": 23, "sequence": self.latest} + elif method == "getEvents": + result = self._get_events(params) + else: + raise QuickstartError(f"Unexpected RPC method {method!r}") + return SimpleNamespace( + status_code=200, + raise_for_status=lambda: None, + json=lambda: {"jsonrpc": "2.0", "id": json.get("id"), "result": result}, + ) + + def _get_events(self, params: dict) -> dict: + filters = params.get("filters") or [] + _require( + len(filters) == 1 and filters[0].get("contractIds") == [self.contract_id], + f"getEvents must filter on the announcer {self.contract_id}, got {filters}", + ) + pagination = params.get("pagination") or {} + cursor = pagination.get("cursor") + if cursor is None: + start = params.get("startLedger") + _require(isinstance(start, int) and 1 <= start <= self.latest, + f"getEvents startLedger must be within 1..{self.latest}, got {start!r}") + offset = 0 + else: + _require("startLedger" not in params, + "getEvents takes either startLedger or a cursor, not both") + offset = int(cursor) + page = self.events[offset:offset + EVENTS_PER_PAGE] + return { + "events": page, + "cursor": str(offset + len(page)), + "latestLedger": self.latest, + "oldestLedger": self.latest - 17_280, + } + + +def _verify_signed(envelope: Any, signer: str, label: str) -> None: + from stellar_sdk import Keypair + + _require(envelope.network_passphrase == TESTNET_PASSPHRASE, + f"{label}: built for {envelope.network_passphrase!r}, not testnet") + keypair = Keypair.from_public_key(signer) + tx_hash = envelope.hash() + for decorated in envelope.signatures: + if decorated.signature_hint != keypair.signature_hint(): + continue + try: + keypair.verify(tx_hash, decorated.signature) + return + except Exception: + continue + raise QuickstartError(f"{label}: no valid signature from {signer}") + + +class FakeHorizon: + """Stands in for stellar_sdk.Server against Horizon.""" + + instances: list["FakeHorizon"] = [] + + def __init__(self, horizon_url: str = "", *_: Any, **__: Any) -> None: + self.url = horizon_url + self.sequences: dict[str, int] = {} + self.submitted: list[Any] = [] + FakeHorizon.instances.append(self) + + def load_account(self, account_id: Any): + from stellar_sdk import Account + + address = getattr(account_id, "public_key", None) or getattr(account_id, "account_id", None) or account_id + return Account(address, self.sequences.setdefault(address, 4_200_000_000)) + + def submit_transaction(self, envelope: Any, *_: Any, **__: Any) -> dict: + source = envelope.transaction.source.account_id + _verify_signed(envelope, source, "Horizon transaction") + _require(envelope.transaction.sequence == self.sequences.get(source, 4_200_000_000) + 1, + f"Horizon transaction from {source} reuses a stale sequence number") + self.sequences[source] = envelope.transaction.sequence + self.submitted.append(envelope) + return {"hash": envelope.hash_hex(), "successful": True} + + +class FakeSoroban: + """Stands in for stellar_sdk.SorobanServer.""" + + instances: list["FakeSoroban"] = [] + + def __init__(self, server_url: str = "", *_: Any, **__: Any) -> None: + self.url = server_url + self.prepared: list[Any] = [] + self.sent: list[Any] = [] + FakeSoroban.instances.append(self) + + def prepare_transaction(self, envelope: Any, *_: Any, **__: Any): + _require(not envelope.signatures, "prepare_transaction must run before signing") + self.prepared.append(envelope) + return envelope + + def send_transaction(self, envelope: Any): + from stellar_sdk.soroban_rpc import SendTransactionStatus + + _require(any(envelope is prepared for prepared in self.prepared), + "send_transaction got a transaction that never went through prepare_transaction") + _verify_signed(envelope, envelope.transaction.source.account_id, "Soroban transaction") + self.sent.append(envelope) + return SimpleNamespace(status=SendTransactionStatus.PENDING, hash=envelope.hash_hex()) + + def poll_transaction(self, transaction_hash: str, *_: Any, **__: Any): + from stellar_sdk.soroban_rpc import GetTransactionStatus + + _require(any(env.hash_hex() == transaction_hash for env in self.sent), + f"poll_transaction for unknown hash {transaction_hash}") + return SimpleNamespace(status=GetTransactionStatus.SUCCESS, transaction_hash=transaction_hash) + + +# --------------------------------------------------------------------------- +# Independent reference (libsodium via PyNaCl) +# --------------------------------------------------------------------------- + +L = 2**252 + 27742317777372353535851937790883648493 + + +def reference_stealth_address(ephemeral_pub: bytes, viewing_seed: bytes, spending_pub: bytes) -> str: + from nacl import bindings + from stellar_sdk import Keypair + + viewing_pub = Keypair.from_raw_ed25519_seed(viewing_seed).raw_public_key() + x_priv = bindings.crypto_sign_ed25519_sk_to_curve25519(viewing_seed + viewing_pub) + x_pub = bindings.crypto_sign_ed25519_pk_to_curve25519(ephemeral_pub) + shared = bindings.crypto_scalarmult(x_priv, x_pub) + h = int.from_bytes(hashlib.sha256(b"wraith:scalar:" + shared).digest(), "little") % L + h_point = bindings.crypto_scalarmult_ed25519_base_noclamp(h.to_bytes(32, "little")) + return Keypair.from_raw_ed25519_public_key( + bindings.crypto_core_ed25519_add(spending_pub, h_point) + ).public_key + + +def reference_view_tag(ephemeral_pub: bytes, viewing_pub: bytes) -> int: + return hashlib.sha256(b"wraith:stellar:view-tag:v2:" + ephemeral_pub + viewing_pub).digest()[0] + + +# --------------------------------------------------------------------------- +# Checks +# --------------------------------------------------------------------------- + +def check_module(module: Any, vectors: dict) -> int: + """Compares the guide's wraith_crypto.py with the TypeScript SDK vectors.""" + from stellar_sdk import Keypair + + owner = vectors["owner"] + checks = 0 + + def same(actual: Any, expected: Any, what: str) -> None: + nonlocal checks + _require(actual == expected, f"{MODULE}.{what}: got {actual!r}, the SDK gives {expected!r}") + checks += 1 + + same(module.STEALTH_SIGNING_MESSAGE, vectors["stealthSigningMessage"], "STEALTH_SIGNING_MESSAGE") + + signature = Keypair.from_secret(owner["secret"]).sign(module.STEALTH_SIGNING_MESSAGE.encode()) + same(signature.hex(), owner["signature"], "STEALTH_SIGNING_MESSAGE signature") + keys = module.derive_stealth_keys(signature) + same(keys.spending_key.hex(), owner["spendingKey"], "derive_stealth_keys spending_key") + same(keys.viewing_key.hex(), owner["viewingKey"], "derive_stealth_keys viewing_key") + same(str(keys.spending_scalar), owner["spendingScalar"], "derive_stealth_keys spending_scalar") + same(str(keys.viewing_scalar), owner["viewingScalar"], "derive_stealth_keys viewing_scalar") + same(keys.spending_pub_key.hex(), owner["spendingPubKey"], "derive_stealth_keys spending_pub_key") + same(keys.viewing_pub_key.hex(), owner["viewingPubKey"], "derive_stealth_keys viewing_pub_key") + same(module.encode_meta_address(keys.spending_pub_key, keys.viewing_pub_key), + owner["metaAddress"], "encode_meta_address") + same(module.decode_meta_address(owner["metaAddress"]), + (keys.spending_pub_key, keys.viewing_pub_key), "decode_meta_address") + + for expected in vectors["stealthAddresses"]: + generated = module.generate_stealth_address( + keys.spending_pub_key, keys.viewing_pub_key, bytes.fromhex(expected["ephemeralSeed"]) + ) + same(generated.stealth_address, expected["stealthAddress"], "generate_stealth_address address") + same(generated.ephemeral_pub_key.hex(), expected["ephemeralPubKey"], + "generate_stealth_address ephemeral_pub_key") + same(generated.view_tag, expected["viewTag"], "generate_stealth_address view_tag") + + found = module.scan_announcements( + [dict(item) for item in vectors["announcements"]], + keys.viewing_key, keys.spending_pub_key, keys.spending_scalar, + ) + same([(m.stealth_address, str(m.stealth_private_scalar), m.stealth_pub_key_bytes.hex()) + for m in found], + [(m["stealthAddress"], m["stealthPrivateScalar"], m["stealthPubKey"]) + for m in vectors["matches"]], + "scan_announcements matches") + + signed = vectors["scalarSignature"] + signature = module.sign_with_scalar( + bytes.fromhex(signed["message"]), int(signed["scalar"]), bytes.fromhex(signed["publicKey"]) + ) + same(signature.hex(), signed["signature"], "sign_with_scalar") + Keypair.from_raw_ed25519_public_key(bytes.fromhex(signed["publicKey"])).verify( + bytes.fromhex(signed["message"]), signature + ) + return checks + + +def check_flow(ns: dict, vectors: dict, rpc: FakeRpc) -> list[str]: + """Checks what the steps did, using the names the guide defines.""" + from stellar_sdk import Address, scval + from stellar_sdk.operation import AccountMerge, CreateAccount, InvokeHostFunction + + owner = vectors["owner"] + results = [] + + def need(name: str) -> Any: + _require(name in ns, f"the guide's steps never define `{name}`") + return ns[name] + + keys = need("keys") + _require(keys.spending_pub_key.hex() == owner["spendingPubKey"] + and keys.viewing_pub_key.hex() == owner["viewingPubKey"], + "Step 3 derived different keys than the SDK") + _require(need("meta_address") == owner["metaAddress"], "Step 3 meta_address differs from the SDK") + results.append("Step 3 keys and meta-address match the SDK") + + _require(need("ANNOUNCER_CONTRACT_ID") == vectors["rpc"]["announcerContractId"], + "ANNOUNCER_CONTRACT_ID differs from the announcer in vectors.json") + pages = rpc.calls.count("getEvents") + _require(pages >= 2, "fetch_announcements stopped before the second getEvents page") + _require(need("announcements") == vectors["announcements"], + "Step 4 parsed announcements differ from the events in vectors.json") + results.append(f"Step 4 parsed {len(vectors['announcements'])} v2 events over {pages} getEvents pages") + + matches = need("matches") + _require([m.stealth_address for m in matches] == [m["stealthAddress"] for m in vectors["matches"]], + "Step 5 found different payments than the SDK scanner") + results.append(f"Step 5 found the {len(matches)} payments the SDK finds, and skipped the decoy") + + stealth = need("stealth") + spending_pub, viewing_pub = bytes.fromhex(owner["spendingPubKey"]), bytes.fromhex(owner["viewingPubKey"]) + expected_address = reference_stealth_address( + stealth.ephemeral_pub_key, bytes.fromhex(owner["viewingKey"]), spending_pub + ) + _require(stealth.stealth_address == expected_address, + "Step 6a stealth address differs from the libsodium reference derivation") + _require(stealth.view_tag == reference_view_tag(stealth.ephemeral_pub_key, viewing_pub), + "Step 6a view tag differs from the v2 derivation") + results.append("Step 6a stealth address and view tag match an independent libsodium derivation") + + horizon_txs = [env for server in FakeHorizon.instances for env in server.submitted] + soroban_txs = [env for server in FakeSoroban.instances for env in server.sent] + sender = need("sender_keypair").public_key + _require(len(horizon_txs) == 2 and len(soroban_txs) == 1, + f"expected 2 Horizon and 1 Soroban transactions, got {len(horizon_txs)} and {len(soroban_txs)}") + + activate = horizon_txs[0].transaction + op = activate.operations[0] + _require(activate.source.account_id == sender and isinstance(op, CreateAccount) + and op.destination == stealth.stealth_address, + "Step 6b createAccount does not fund the Step 6a stealth address from the sender") + + announce = soroban_txs[0].transaction + op = announce.operations[0] + _require(isinstance(op, InvokeHostFunction), "Step 6b announcement is not a contract call") + call = op.host_function.invoke_contract + _require(Address.from_xdr_sc_address(call.contract_address).address == vectors["rpc"]["announcerContractId"] + and call.function_name.sc_symbol == b"announce", + "Step 6b does not call announce on the announcer contract") + args = [scval.to_native(arg) for arg in call.args] + _require(len(args) == 4 and args[0] == 2 and args[1].address == stealth.stealth_address + and args[2] == stealth.ephemeral_pub_key and args[3][:1] == bytes([stealth.view_tag]), + f"Step 6b announce arguments are wrong: {args!r}") + results.append("Step 6b createAccount and announce(2, stealth, R, [view tag]) are signed for testnet") + + withdrawal = horizon_txs[1].transaction + op = withdrawal.operations[0] + _require(withdrawal.source.account_id == matches[0].stealth_address and isinstance(op, AccountMerge) + and op.destination.account_id == sender, + "Step 7 does not merge the matched stealth account into the sender") + results.append("Step 7 merge is signed by sign_with_scalar and verifies against the stealth key") + return results + + +# --------------------------------------------------------------------------- +# Main +# --------------------------------------------------------------------------- + +def run(fixture: Fixture, vectors: dict) -> None: + import httpx + import stellar_sdk + + workdir = Path(tempfile.mkdtemp(prefix="wraith-python-quickstart-")) + (workdir / f"{MODULE}.py").write_text(fixture.module_source) + sys.path.insert(0, str(workdir)) + sys.modules.pop(MODULE, None) + + try: + module = __import__(MODULE) + except Exception as error: # noqa: BLE001 + raise QuickstartError(f"{MODULE}.py from {GUIDE_LABEL} does not import:\n" + + "".join(traceback.format_exception(error))) from error + module_checks = check_module(module, vectors) + + rpc = FakeRpc(vectors) + httpx.post = rpc.post + stellar_sdk.Server = FakeHorizon + stellar_sdk.SorobanServer = FakeSoroban + os.environ["STELLAR_SECRET"] = vectors["owner"]["secret"] + + ns: dict[str, Any] = {"__name__": "__quickstart__"} + print(f"Running {len(fixture.steps)} steps from {GUIDE_LABEL} offline") + for step in fixture.steps: + # Pad with blank lines so tracebacks point at the real MDX line numbers + code = compile("\n" * (step.line - 1) + step.code, GUIDE_LABEL, "exec") + output = io.StringIO() + try: + with contextlib.redirect_stdout(output): + exec(code, ns) # noqa: S102 - the guide is the code under test + except QuickstartError as error: + raise QuickstartError(f"{GUIDE_LABEL}:{step.line}: {error}") from error + except BaseException as error: # noqa: BLE001 + raise QuickstartError( + f"The step at {GUIDE_LABEL}:{step.line} failed:\n" + + "".join(traceback.format_exception(error)) + ) from error + printed = output.getvalue().strip() + print(f" ok {GUIDE_LABEL}:{step.line}") + for line in printed.splitlines(): + print(f" {line}") + + results = check_flow(ns, vectors, rpc) + print(f"\n{MODULE}.py matches the TypeScript SDK on {module_checks} checks") + for result in results: + print(f" {result}") + if fixture.pseudocode: + lines = ", ".join(str(fence.line) for fence in fixture.pseudocode) + print(f"Skipped pseudocode fences at lines {lines}") + print("\nPython quickstart ran offline.") + + +def main(argv: list[str]) -> int: + if sys.version_info < (3, 11): + print("The Python quickstart targets Python 3.11+.", file=sys.stderr) + return 1 + ensure_pinned_environment(argv) + try: + fixture = extract_fixture(GUIDE.read_text()) + vectors = json.loads(VECTORS.read_text()) + run(fixture, vectors) + except QuickstartError as error: + print(f"Python quickstart check failed:\n\n{error}", file=sys.stderr) + return 1 + return 0 + + +if __name__ == "__main__": + sys.exit(main(sys.argv[1:])) diff --git a/scripts/python-quickstart/vectors.json b/scripts/python-quickstart/vectors.json new file mode 100644 index 0000000..d07ba33 --- /dev/null +++ b/scripts/python-quickstart/vectors.json @@ -0,0 +1,168 @@ +{ + "$comment": "Generated by scripts/python-quickstart/generate-vectors.ts from the pinned @wraith-protocol/sdk Stellar primitives. Do not edit by hand; run `pnpm run generate:python-quickstart-vectors`.", + "stealthSigningMessage": "Sign this message to generate your Wraith stealth keys.\n\nChain: Stellar\nNote: This signature is used for key derivation only and does not authorize any transaction.", + "owner": { + "secret": "SDOPMOTKW6U6YUOUQF44C4ZG2D4GC46UJPMOJGK5MKBTDKEJHWIZBA6P", + "publicKey": "GCWSBJO4NRFHAXZISP4VT5SN5BLC7RIIBPVCYKK2TBBI5VFX7B47JFRF", + "signature": "edbab2c5d58fdbbd08995f649f2fdc5bb70d9061b59f40dafa5e049a916c569f6caf00358759aa4a2639e3debd2d4464c91e27a4cbc7b7655c11e17eaacc0d06", + "spendingKey": "49ce21c9fc403478fee764bcc25e343de47744fa3d7c0860bb376cee990523ec", + "viewingKey": "2872bccf9e4d56b0e9f2227495e632e168086a382aed8149a8673832d913dc62", + "spendingScalar": "36165898451058754716118145400569639394529583296247500707205305789287386495712", + "viewingScalar": "44469554120281652802293231179823107218516810350333748018472723505713148190392", + "spendingPubKey": "951028f9a75bc86c14ef130e66de85f5946d4e1699baf94c73aef5444d6ff65f", + "viewingPubKey": "b67a08148e7afdf0b3bf0eb90cbd981e83d6a37dbd24ee6b74b99b4783a0cd91", + "metaAddress": "st:xlm:951028f9a75bc86c14ef130e66de85f5946d4e1699baf94c73aef5444d6ff65fb67a08148e7afdf0b3bf0eb90cbd981e83d6a37dbd24ee6b74b99b4783a0cd91" + }, + "stealthAddresses": [ + { + "ephemeralSeed": "b0126b5f4b7e2a6752281e977813c71e9e98832b4aeb98ae5e91a7bb87c37994", + "ephemeralPubKey": "471c22b4ce3805ab50cbbd880487fdf49b80c62d1e2d4cbe320f01a932c22332", + "stealthAddress": "GBPRSH57SFOQ5FY3TBWAHHW2YD27J3CCQ2XUYSHOGKKMGO3W6Z4JEOM5", + "viewTag": 51 + }, + { + "ephemeralSeed": "635b5072507169bb6eb2eea5d5ec48ca25cffb56ba11a6a77c49cbb18cc1630d", + "ephemeralPubKey": "8929a439bc8464aa0ef67c1a2e564c3d894a921143027a6842bc080f5137b30c", + "stealthAddress": "GCQETDXZEMWXMR3QG2PD35O4Q3RTTWFMSQPLLI43SHRPC23EEEFJQKOF", + "viewTag": 63 + } + ], + "rpc": { + "announcerContractId": "CCJLJ2QRBJAAKIG6ELNQVXLLWMKKWVN5O2FKWUETHZGMPAD4MHK7WVWL", + "latestLedger": 1990000, + "events": [ + { + "type": "contract", + "ledger": 1989900, + "ledgerClosedAt": "2026-10-01T12:00:00.000Z", + "contractId": "CCJLJ2QRBJAAKIG6ELNQVXLLWMKKWVN5O2FKWUETHZGMPAD4MHK7WVWL", + "id": "0008546555422310400-0000000000", + "txHash": "56f73ce7225c86bb063e074a5b1a874454b953615c30bdb435b9f06537949bcc", + "inSuccessfulContractCall": true, + "topic": [ + "AAAADwAAAAhhbm5vdW5jZQ==", + "AAAAAwAAAAI=", + "AAAAAwAAADM=", + "AAAAAwAAAAE=" + ], + "value": "AAAAEAAAAAEAAAADAAAAEgAAAAAAAAAAXxkfv5FdDpcbmGwDntrA9fTsQoavTEjuMpTDO3b2eJIAAAANAAAAIEccIrTOOAWrUMu9iASH/fSbgMYtHi1MvjIPAakywiMyAAAADQAAAAEzAAAA" + }, + { + "type": "contract", + "ledger": 1989901, + "ledgerClosedAt": "2026-10-01T12:00:05.000Z", + "contractId": "CCJLJ2QRBJAAKIG6ELNQVXLLWMKKWVN5O2FKWUETHZGMPAD4MHK7WVWL", + "id": "0008546559717277696-0000000001", + "txHash": "a1979ac06076b6412c21f7882e00bcf0baf3f790ea4db800ac36b70aeeb5289e", + "inSuccessfulContractCall": true, + "topic": [ + "AAAADwAAAAhhbm5vdW5jZQ==", + "AAAAAwAAAAI=", + "AAAAAwAAAKE=", + "AAAAAwAAAAE=" + ], + "value": "AAAAEAAAAAEAAAADAAAAEgAAAAAAAAAAALwNY/Qc+cTzHT/GARTAVrJPELsF2MfkCSgmLYt2uTsAAAANAAAAINu5i+jRxSbQ7j1QhFPW6LzCjumnZvozxe2jqPLlgea0AAAADQAAAAGhAAAA" + }, + { + "type": "contract", + "ledger": 1989902, + "ledgerClosedAt": "2026-10-01T12:00:10.000Z", + "contractId": "CCJLJ2QRBJAAKIG6ELNQVXLLWMKKWVN5O2FKWUETHZGMPAD4MHK7WVWL", + "id": "0008546564012244992-0000000002", + "txHash": "500ad5cf9c25e8f2d3a5212a24fe796da4d5610c9faa49721be50de88a60e937", + "inSuccessfulContractCall": true, + "topic": [ + "AAAADwAAAAhhbm5vdW5jZQ==", + "AAAAAwAAAAI=", + "AAAAAwAAAD8=", + "AAAAAwAAAAE=" + ], + "value": "AAAAEAAAAAEAAAADAAAAEgAAAAAAAAAAoEmO+SMtdkdwNp499dyG4znYrJQetaObkeLxa2QhCpgAAAANAAAAIIkppDm8hGSqDvZ8Gi5WTD2JSpIRQwJ6aEK8CA9RN7MMAAAADQAAAAE/AAAA" + }, + { + "type": "contract", + "ledger": 1989903, + "ledgerClosedAt": "2026-10-01T12:00:15.000Z", + "contractId": "CCJLJ2QRBJAAKIG6ELNQVXLLWMKKWVN5O2FKWUETHZGMPAD4MHK7WVWL", + "id": "0008546568307212288-0000000003", + "txHash": "60eb90252b68103311097c64fabfee713a9ba4a36176904f8b6d353775a89656", + "inSuccessfulContractCall": true, + "topic": [ + "AAAADwAAAAhhbm5vdW5jZQ==", + "AAAAAwAAAAI=", + "AAAAAwAAABY=", + "AAAAAwAAAAE=" + ], + "value": "AAAAEAAAAAEAAAADAAAAEgAAAAAAAAAAYHDKPKdmnK+1dQhnexzoxAfOIgefOuuGc6bi1yj2W6UAAAANAAAAIJompQty7o9izn46i6WNP2Gopg+kHRlxQ0ljHBgZYsv6AAAADQAAAAEWAAAA" + }, + { + "type": "contract", + "ledger": 1989904, + "ledgerClosedAt": "2026-10-01T12:00:20.000Z", + "contractId": "CCJLJ2QRBJAAKIG6ELNQVXLLWMKKWVN5O2FKWUETHZGMPAD4MHK7WVWL", + "id": "0008546572602179584-0000000004", + "txHash": "c2933ca05d68ddac1aaf121db0103c99c65eeddfd87ce88d16baf0a591176018", + "inSuccessfulContractCall": true, + "topic": [ + "AAAADwAAAAhhbm5vdW5jZQ==", + "AAAAAwAAAAI=", + "AAAAAwAAAPg=", + "AAAAAwAAAAE=" + ], + "value": "AAAAEAAAAAEAAAADAAAAEgAAAAAAAAAACYurZzTsdHKtBvydB9DnjmcRPtFE3DDYVrT/TwIxuNMAAAANAAAAIFz0doKcEKtpMZNCNZoTtq/ndYvhDYpUk7eQ8lbi6Z1yAAAADQAAAAH4AAAA" + } + ] + }, + "announcements": [ + { + "scheme_id": 2, + "stealth_address": "GBPRSH57SFOQ5FY3TBWAHHW2YD27J3CCQ2XUYSHOGKKMGO3W6Z4JEOM5", + "ephemeral_pub_key": "471c22b4ce3805ab50cbbd880487fdf49b80c62d1e2d4cbe320f01a932c22332", + "metadata": "33" + }, + { + "scheme_id": 2, + "stealth_address": "GAALYDLD6QOPTRHTDU74MAIUYBLLETYQXMC5RR7EBEUCMLMLO24TX4YL", + "ephemeral_pub_key": "dbb98be8d1c526d0ee3d508453d6e8bcc28ee9a766fa33c5eda3a8f2e581e6b4", + "metadata": "a1" + }, + { + "scheme_id": 2, + "stealth_address": "GCQETDXZEMWXMR3QG2PD35O4Q3RTTWFMSQPLLI43SHRPC23EEEFJQKOF", + "ephemeral_pub_key": "8929a439bc8464aa0ef67c1a2e564c3d894a921143027a6842bc080f5137b30c", + "metadata": "3f" + }, + { + "scheme_id": 2, + "stealth_address": "GBQHBSR4U5TJZL5VOUEGO6Y45DCAPTRCA6PTV24GOOTOFVZI6ZN2LOBT", + "ephemeral_pub_key": "9a26a50b72ee8f62ce7e3a8ba58d3f61a8a60fa41d19714349631c181962cbfa", + "metadata": "16" + }, + { + "scheme_id": 2, + "stealth_address": "GAEYXK3HGTWHI4VNA36J2B6Q46HGOEJ62FCNYMGYK22P6TYCGG4NHZNG", + "ephemeral_pub_key": "5cf476829c10ab69319342359a13b6afe7758be10d8a5493b790f256e2e99d72", + "metadata": "f8" + } + ], + "matches": [ + { + "index": 0, + "stealthAddress": "GBPRSH57SFOQ5FY3TBWAHHW2YD27J3CCQ2XUYSHOGKKMGO3W6Z4JEOM5", + "stealthPrivateScalar": "3437456400779383884853090559485828520176343062883920412031112698606428378328", + "stealthPubKey": "5f191fbf915d0e971b986c039edac0f5f4ec4286af4c48ee3294c33b76f67892" + }, + { + "index": 2, + "stealthAddress": "GCQETDXZEMWXMR3QG2PD35O4Q3RTTWFMSQPLLI43SHRPC23EEEFJQKOF", + "stealthPrivateScalar": "6640971120132170134333823985662900435482568511840634446737891804661085823421", + "stealthPubKey": "a0498ef9232d764770369e3df5dc86e339d8ac941eb5a39b91e2f16b64210a98" + } + ], + "scalarSignature": { + "message": "ab530a13e45914982b79f9b7e3fba994cfd1f3fb22f71cea1afbf02b460c6d1d", + "scalar": "3437456400779383884853090559485828520176343062883920412031112698606428378328", + "publicKey": "5f191fbf915d0e971b986c039edac0f5f4ec4286af4c48ee3294c33b76f67892", + "signature": "027adf8e4bfeab964c999b9e466aac3ee2259b9c35f4cc4fde54445f37c97b7cdebb35fb8bf64f38729b3e2b1cbce4b104ad3e2d3a269226960c502fa957520a" + } +}