From a21e058cdd0d53f348fc4552ac3ee1471c0c1830 Mon Sep 17 00:00:00 2001 From: Davide Grilli Date: Tue, 21 Jul 2026 10:25:38 +0200 Subject: [PATCH] Add Electrum SPV client and address listener Minimal Electrum protocol client (client.py) plus a scripthash subscription listener (listener.py) that watches user deposit addresses for confirmed UTXOs, with the address->scripthash conversion helper and a manual smoke-test script against the dev bootstrap server. Co-Authored-By: Claude Sonnet 5 --- app/electrum/__init__.py | 0 app/electrum/client.py | 114 +++++++++++++++++++++++++++++ app/electrum/listener.py | 111 ++++++++++++++++++++++++++++ app/electrum/scripthash.py | 15 ++++ scripts/electrum_smoke_test.py | 27 +++++++ tests/unit/test_electrum_client.py | 77 +++++++++++++++++++ tests/unit/test_scripthash.py | 9 +++ 7 files changed, 353 insertions(+) create mode 100644 app/electrum/__init__.py create mode 100644 app/electrum/client.py create mode 100644 app/electrum/listener.py create mode 100644 app/electrum/scripthash.py create mode 100644 scripts/electrum_smoke_test.py create mode 100644 tests/unit/test_electrum_client.py create mode 100644 tests/unit/test_scripthash.py diff --git a/app/electrum/__init__.py b/app/electrum/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/electrum/client.py b/app/electrum/client.py new file mode 100644 index 0000000..7e4b5e1 --- /dev/null +++ b/app/electrum/client.py @@ -0,0 +1,114 @@ +import asyncio +import itertools +import json +import ssl + + +class ElectrumError(Exception): + pass + + +class ElectrumClient: + """Minimal asyncio Electrum protocol client: line-delimited JSON-RPC over TLS. + + Push notifications (blockchain.headers.subscribe, blockchain.scripthash.subscribe) + arrive under the *same* method name as the subscribe call, multiplexed for every + scripthash subscribed — callers read `notifications(method)` and, for scripthash + pushes, dispatch on `params[0]` (the scripthash) themselves. + """ + + def __init__(self, host: str, port: int, use_ssl: bool = True): + self.host = host + self.port = port + self.use_ssl = use_ssl + self._reader: asyncio.StreamReader | None = None + self._writer: asyncio.StreamWriter | None = None + self._id_counter = itertools.count(1) + self._pending: dict[int, asyncio.Future] = {} + self._subscriptions: dict[str, asyncio.Queue] = {} + self._read_task: asyncio.Task | None = None + + async def connect(self) -> None: + # Electrum servers commonly present self-signed certs; the protocol's trust + # model is server consensus, not TLS PKI, so we only use SSL for transport + # encryption and don't verify the certificate chain/hostname. + ssl_context = None + if self.use_ssl: + ssl_context = ssl.create_default_context() + ssl_context.check_hostname = False + ssl_context.verify_mode = ssl.CERT_NONE + self._reader, self._writer = await asyncio.open_connection(self.host, self.port, ssl=ssl_context) + self._read_task = asyncio.create_task(self._read_loop()) + await self.request("server.version", ["plm-lottery", "1.4"]) + + async def close(self) -> None: + if self._read_task is not None: + self._read_task.cancel() + if self._writer is not None: + self._writer.close() + try: + await asyncio.wait_for(self._writer.wait_closed(), timeout=2) + except (ssl.SSLError, TimeoutError, asyncio.TimeoutError): + pass # some Electrum servers don't send a clean TLS close_notify + + async def request(self, method: str, params: list | None = None) -> object: + if self._writer is None: + raise ElectrumError("not connected") + request_id = next(self._id_counter) + future: asyncio.Future = asyncio.get_event_loop().create_future() + self._pending[request_id] = future + payload = json.dumps({"id": request_id, "method": method, "params": params or []}) + "\n" + self._writer.write(payload.encode()) + await self._writer.drain() + return await future + + def notifications(self, method: str) -> asyncio.Queue: + return self._subscriptions.setdefault(method, asyncio.Queue()) + + async def subscribe_headers(self) -> dict: + self.notifications("blockchain.headers.subscribe") + return await self.request("blockchain.headers.subscribe") + + async def subscribe_scripthash(self, scripthash: str) -> str | None: + self.notifications("blockchain.scripthash.subscribe") + return await self.request("blockchain.scripthash.subscribe", [scripthash]) + + async def listunspent(self, scripthash: str) -> list[dict]: + return await self.request("blockchain.scripthash.listunspent", [scripthash]) + + async def broadcast(self, raw_tx_hex: str) -> str: + return await self.request("blockchain.transaction.broadcast", [raw_tx_hex]) + + async def get_transaction(self, txid: str, verbose: bool = False) -> object: + return await self.request("blockchain.transaction.get", [txid, verbose]) + + async def _read_loop(self) -> None: + assert self._reader is not None + try: + while True: + line = await self._reader.readline() + if not line: + break + message = json.loads(line) + self._dispatch(message) + finally: + error = ElectrumError("connection closed") + for future in self._pending.values(): + if not future.done(): + future.set_exception(error) + self._pending.clear() + + def _dispatch(self, message: dict) -> None: + message_id = message.get("id") + if message_id is not None and message_id in self._pending: + future = self._pending.pop(message_id) + if future.done(): + return + if message.get("error"): + future.set_exception(ElectrumError(message["error"])) + else: + future.set_result(message.get("result")) + elif "method" in message: + queue = self._subscriptions.get(message["method"]) + if queue is not None: + queue.put_nowait(message.get("params")) diff --git a/app/electrum/listener.py b/app/electrum/listener.py new file mode 100644 index 0000000..48a7d7a --- /dev/null +++ b/app/electrum/listener.py @@ -0,0 +1,111 @@ +import asyncio +import logging +from collections.abc import Callable + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import async_sessionmaker + +from app.db.models import User +from app.deposits.service import credit_confirmed_utxos +from app.electrum.client import ElectrumClient +from app.electrum.scripthash import address_to_scripthash + +logger = logging.getLogger(__name__) + + +class ElectrumListener: + """Long-lived background task: keeps one Electrum connection open, subscribes + every user's address (plus any address added later via add_address), and + credits confirmed deposits as scripthash-change notifications arrive. + + Reconnects with backoff on any failure; a fresh connection re-subscribes to + every user pulled straight from the DB, so no in-memory subscription state is + ever a stale source of truth. + """ + + def __init__(self, client_factory: Callable[[], ElectrumClient], session_factory: async_sessionmaker): + self._client_factory = client_factory + self._session_factory = session_factory + self._scripthash_to_user: dict[str, int] = {} + self.tip_height: int = 0 + self.tip_header_hex: str | None = None + self.client: ElectrumClient | None = None + + def address_for_new_user(self, user_id: int, address: str) -> None: + """Called right after a user registers so their deposit address starts + being watched immediately, without waiting for the next reconnect cycle.""" + scripthash = address_to_scripthash(address) + self._scripthash_to_user[scripthash] = user_id + if self.client is not None: + asyncio.create_task(self._subscribe_and_refresh(scripthash, user_id)) + + async def run(self) -> None: + backoff = 1 + while True: + try: + await self._run_once() + except asyncio.CancelledError: + raise + except Exception: + logger.exception("Electrum listener error, reconnecting in %ss", backoff) + self.client = None + await asyncio.sleep(backoff) + backoff = min(backoff * 2, 30) + continue + backoff = 1 + + async def _run_once(self) -> None: + client = self._client_factory() + await client.connect() + self.client = client + + header = await client.subscribe_headers() + self.tip_height = header["height"] + self.tip_header_hex = header.get("hex") + + await self._subscribe_all_users() + + headers_queue = client.notifications("blockchain.headers.subscribe") + scripthash_queue = client.notifications("blockchain.scripthash.subscribe") + try: + await asyncio.gather( + self._consume_headers(headers_queue), + self._consume_scripthash(scripthash_queue), + ) + finally: + await client.close() + + async def _subscribe_all_users(self) -> None: + async with self._session_factory() as session: + users = (await session.scalars(select(User))).all() + for user in users: + scripthash = address_to_scripthash(user.address) + self._scripthash_to_user[scripthash] = user.id + await self._subscribe_and_refresh(scripthash, user.id) + + async def _subscribe_and_refresh(self, scripthash: str, user_id: int) -> None: + assert self.client is not None + await self.client.subscribe_scripthash(scripthash) + await self._refresh_user(user_id, scripthash) + + async def _consume_headers(self, queue: asyncio.Queue) -> None: + while True: + params = await queue.get() + for header in params: + self.tip_height = header["height"] + self.tip_header_hex = header.get("hex") + + async def _consume_scripthash(self, queue: asyncio.Queue) -> None: + while True: + scripthash, _status = await queue.get() + user_id = self._scripthash_to_user.get(scripthash) + if user_id is not None: + await self._refresh_user(user_id, scripthash) + + async def _refresh_user(self, user_id: int, scripthash: str) -> None: + assert self.client is not None + entries = await self.client.listunspent(scripthash) + async with self._session_factory() as session: + credited = await credit_confirmed_utxos(session, user_id, entries) + if credited: + logger.info("credited %s new UTXO(s) for user_id=%s", credited, user_id) diff --git a/app/electrum/scripthash.py b/app/electrum/scripthash.py new file mode 100644 index 0000000..dff74ac --- /dev/null +++ b/app/electrum/scripthash.py @@ -0,0 +1,15 @@ +import hashlib + +from embit.script import Script + + +def address_to_scripthash(address: str) -> str: + """Electrum protocol scripthash: sha256(scriptPubKey), byte-reversed, hex. + + Uses `.data` (the raw scriptPubKey bytes), not `.serialize()` — the latter + prefixes a compact-size length byte meant for embedding the script as pushdata + elsewhere (e.g. a P2SH redeemScript), which is not part of the actual on-chain + output script and produces a wrong (unmatchable) scripthash if used here. + """ + script_pubkey = Script.from_address(address).data + return hashlib.sha256(script_pubkey).digest()[::-1].hex() diff --git a/scripts/electrum_smoke_test.py b/scripts/electrum_smoke_test.py new file mode 100644 index 0000000..585c961 --- /dev/null +++ b/scripts/electrum_smoke_test.py @@ -0,0 +1,27 @@ +"""Manual smoke test: connect to the configured Electrum server, do the +server.version handshake, subscribe to headers and confirm a sane tip height. + +Usage: PYTHONPATH=. python scripts/electrum_smoke_test.py +""" + +import asyncio + +from app.config import settings +from app.electrum.client import ElectrumClient + + +async def main() -> None: + client = ElectrumClient(settings.electrum_host, settings.electrum_port, settings.electrum_use_ssl) + await client.connect() + print("connected + handshake OK") + + header = await client.subscribe_headers() + print("tip:", header) + assert header["height"] > 0 + + await client.close() + print("OK") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/tests/unit/test_electrum_client.py b/tests/unit/test_electrum_client.py new file mode 100644 index 0000000..83a9653 --- /dev/null +++ b/tests/unit/test_electrum_client.py @@ -0,0 +1,77 @@ +import asyncio +import json + +from app.electrum.client import ElectrumClient, ElectrumError + + +class FakeWriter: + def __init__(self): + self.written = b"" + + def write(self, data: bytes) -> None: + self.written += data + + async def drain(self) -> None: + pass + + def close(self) -> None: + pass + + async def wait_closed(self) -> None: + pass + + +async def _client_with_fake_transport() -> tuple[ElectrumClient, asyncio.StreamReader, FakeWriter]: + client = ElectrumClient("localhost", 1234) + reader = asyncio.StreamReader() + writer = FakeWriter() + client._reader = reader + client._writer = writer + client._read_task = asyncio.create_task(client._read_loop()) + return client, reader, writer + + +async def test_request_resolves_on_matching_response(): + client, reader, writer = await _client_with_fake_transport() + + task = asyncio.create_task(client.request("blockchain.headers.subscribe")) + await asyncio.sleep(0) # let request() write the payload + sent = json.loads(writer.written.decode()) + assert sent["method"] == "blockchain.headers.subscribe" + + reader.feed_data((json.dumps({"id": sent["id"], "result": {"height": 100}}) + "\n").encode()) + result = await task + assert result == {"height": 100} + + client._read_task.cancel() + + +async def test_error_response_raises_electrum_error(): + client, reader, writer = await _client_with_fake_transport() + + task = asyncio.create_task(client.request("blockchain.transaction.broadcast", ["deadbeef"])) + await asyncio.sleep(0) + sent = json.loads(writer.written.decode()) + + reader.feed_data((json.dumps({"id": sent["id"], "error": "bad tx"}) + "\n").encode()) + try: + await task + assert False, "expected ElectrumError" + except ElectrumError: + pass + + client._read_task.cancel() + + +async def test_notification_delivered_to_subscription_queue(): + client, reader, writer = await _client_with_fake_transport() + queue = client.notifications("blockchain.scripthash.subscribe") + + push = {"method": "blockchain.scripthash.subscribe", "params": ["abcd", "newstatus"]} + reader.feed_data((json.dumps(push) + "\n").encode()) + await asyncio.sleep(0) + + params = await asyncio.wait_for(queue.get(), timeout=1) + assert params == ["abcd", "newstatus"] + + client._read_task.cancel() diff --git a/tests/unit/test_scripthash.py b/tests/unit/test_scripthash.py new file mode 100644 index 0000000..d09d3bd --- /dev/null +++ b/tests/unit/test_scripthash.py @@ -0,0 +1,9 @@ +from app.electrum.scripthash import address_to_scripthash + + +def test_matches_known_mainnet_vector(): + # Ground truth: scriptPubKey "0014a195473740aea3b4df1690fbdcb51243fe4e7a20" of a real + # confirmed mainnet output to this address (txid 1ebd0219...b0d48a, vout 1), cross-checked + # against the Electrum server's own listunspent for this scripthash. + address = "plm1q5x25wd6q463mfhckjraaedgjg0lyu73qfcj43n" + assert address_to_scripthash(address) == "5b7744e34b1d6ee3ae6eef7e1ce82aa4c28ff13f10d4aa905640c722c9249111"