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 <noreply@anthropic.com>

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"))

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)

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()

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())

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()

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"