interface: protocol 1.7: some response types changed: array->dict

This was motivated by wanting to include the chaintip in responses
but that part was reverted and postponed to later.

ref spesmilo/electrum-protocol#17

Author: SomberNight

electrum/interface.py

@@ -131,9 +131,13 @@ def assert_hex_str(val: Any, *, allow_odd_len: bool = False) -> None:
         raise RequestCorrupted(f'{val!r} should be a hex str')
 
 
-def assert_dict_contains_field(d: Any, *, field_name: str) -> Any:
+def assert_dict(d: Any) -> None:
     if not isinstance(d, dict):
         raise RequestCorrupted(f'{d!r} should be a dict')
+
+
+def assert_dict_contains_field(d: Any, *, field_name: str) -> Any:
+    assert_dict(d)
     if field_name not in d:
         raise RequestCorrupted(f'required field {field_name!r} missing from dict')
     return d[field_name]
@@ -235,7 +239,7 @@ def set_default_timeout(self, timeout):
         assert hasattr(self, "max_send_delay")        # in base class
         self.max_send_delay = timeout
 
-    async def subscribe(self, method: str, params: List, queue: asyncio.Queue):
+    async def subscribe(self, method: str, params: List, queue: asyncio.Queue) -> None:
         # note: until the cache is written for the first time,
         # each 'subscribe' call might make a request on the network.
         notif_params = params[:]
@@ -1471,9 +1475,10 @@ async def get_history_for_spk(self, spk: str) -> List[dict]:
         # do request
         res = await self.session.send_request('blockchain.scriptpubkey.get_history', [spk])
         # check response
-        assert_list_or_tuple(res)
+        hist = assert_dict_contains_field(res, field_name='history')
+        assert_list_or_tuple(hist)
         prev_height = 1
-        for tx_item in res:
+        for tx_item in hist:
             height = assert_dict_contains_field(tx_item, field_name='height')
             assert_dict_contains_field(tx_item, field_name='tx_hash')
             assert_integer(height)
@@ -1491,23 +1496,24 @@ async def get_history_for_spk(self, spk: str) -> List[dict]:
                 prev_height = height
         if self.active_protocol_tuple >= (1, 6):
             # enforce order of mempool txs
-            mempool_txs = [tx_item for tx_item in res if tx_item['height'] <= 0]
+            mempool_txs = [tx_item for tx_item in hist if tx_item['height'] <= 0]
             if mempool_txs != sorted(mempool_txs, key=lambda x: (-x['height'], bytes.fromhex(x['tx_hash']))):
                 raise RequestCorrupted(f'mempool txs not in canonical order')
-        hashes = set(map(lambda item: item['tx_hash'], res))
-        if len(hashes) != len(res):
+        hashes = set(map(lambda item: item['tx_hash'], hist))
+        if len(hashes) != len(hist):
             # Either server is sending garbage... or maybe if server is race-prone
             # a recently mined tx could be included in both last block and mempool?
             # Still, it's simplest to just disregard the response.
             raise RequestCorrupted(f"server history has non-unique txids for spk={spk}")
-        return res
+        return hist
 
     async def listunspent_for_spk(self, spk: str) -> List[dict]:
         # do request
         res = await self.session.send_request('blockchain.scriptpubkey.listunspent', [spk])
         # check response
-        assert_list_or_tuple(res)
-        for utxo_item in res:
+        utxos = assert_dict_contains_field(res, field_name='utxos')
+        assert_list_or_tuple(utxos)
+        for utxo_item in utxos:
             assert_dict_contains_field(utxo_item, field_name='tx_pos')
             assert_dict_contains_field(utxo_item, field_name='value')
             assert_dict_contains_field(utxo_item, field_name='tx_hash')
@@ -1516,7 +1522,7 @@ async def listunspent_for_spk(self, spk: str) -> List[dict]:
             assert_non_negative_integer(utxo_item['value'])
             assert_non_negative_integer(utxo_item['height'])
             assert_hash256_str(utxo_item['tx_hash'])
-        return res
+        return utxos
 
     async def get_balance_for_spk(self, spk: str) -> dict:
         # do request

electrum/synchronizer.py

@@ -35,7 +35,10 @@
 from .util import make_aiohttp_session, NetworkJobOnDefaultServer, random_shuffled_copy, OldTaskGroup, log_exceptions
 from .bitcoin import address_to_script, script_to_scripthash, is_address, neuter_bitcoin_address
 from .logging import Logger
-from .interface import GracefulDisconnect, NetworkTimeout, HistoryTooLong, assert_hash256_str
+from .interface import (
+    GracefulDisconnect, NetworkTimeout, HistoryTooLong, assert_hash256_str, assert_non_negative_integer,
+    assert_dict_contains_field, assert_dict, assert_integer,
+)
 
 if TYPE_CHECKING:
     from .network import Network
@@ -120,10 +123,10 @@ async def _on_address_status(self, addr: str, status: Optional[str]):
         """
         raise NotImplementedError()  # implemented by subclasses
 
-    async def _on_outpoint_status(self, txid: str, index:int, status: dict):
+    async def _on_outpoint_status(self, txid: str, index: int, status: dict):
         raise NotImplementedError()  # implemented by subclasses
 
-    async def _subscribe_to_address(self, addr):
+    async def _subscribe_to_address(self, addr: str):
         spk = address_to_script(addr)
         h = script_to_scripthash(spk)
         self.scripthash_to_address[h] = addr
@@ -153,8 +156,13 @@ async def _subscribe_to_outpoint(self, outpoint: str, *, spk_hint: str) -> None:
     @log_exceptions
     async def handle_address_status(self):
         while True:
-            h, status = await self.address_status_queue.get()
-            addr = self.scripthash_to_address[h]
+            sh, status = await self.address_status_queue.get()
+            # basic checks for response
+            assert_hash256_str(sh)
+            if status is not None:
+                assert_hash256_str(status)
+            # process status
+            addr = self.scripthash_to_address[sh]
             self._handling_addr_statuses.add(addr)
             self.requested_addrs.discard(addr)  # ok for addr not to be present
             await self.taskgroup.spawn(self._on_address_status, addr, status)
@@ -164,7 +172,12 @@ async def handle_address_status(self):
     async def handle_outpoint_status(self):
         while True:
             txid, index, status = await self.outpoint_status_queue.get()
-            outpoint = txid + ':%d'%index
+            # basic checks for response. more checks done in _on_outpoint_status
+            assert_hash256_str(txid)
+            assert_non_negative_integer(index)
+            assert_dict(status)
+            # process status
+            outpoint = f"{txid}:{index}"
             self._handling_outpoint_statuses.add(outpoint)
             self.requested_outpoints.discard(outpoint)  # ok for addr not to be present
             await self.taskgroup.spawn(self._on_outpoint_status, txid, index, status)
@@ -284,26 +297,28 @@ async def disconnect_if_still_stale():
 
     async def _on_outpoint_status(self, txid: str, index: int, status: dict):
         try:
-            assert isinstance(status, dict)
+            # check inputs (coming from server)
             assert_hash256_str(txid)
+            assert_non_negative_integer(index)
+            #
             txs = set()
             txs.add(txid)
-            height = status.get('funder_height')
-            if height is not None:
-                assert isinstance(height, int)
+            funder_height = status.get('funder_height')
+            if funder_height is not None:
+                assert_integer(funder_height)
                 # spv the input
-                self.adb.add_unverified_or_unconfirmed_tx(txid, height)
+                self.adb.add_unverified_or_unconfirmed_tx(txid, funder_height)
             # fetch the output
             spender_txid = status.get('spender_txhash')
             if spender_txid is not None:
                 assert_hash256_str(spender_txid)
-                spender_height = status.get('spender_height')
-                assert isinstance(spender_height, int)
+                spender_height = assert_dict_contains_field(status, field_name='spender_height')
+                assert_integer(spender_height)
                 self.adb.add_unverified_or_unconfirmed_tx(spender_txid, spender_height)
                 txs.add(spender_txid)
             await self._request_missing_txs(txs, allow_server_not_finding_tx=False)
         finally:
-            outpoint = txid + ':%d'%index
+            outpoint = f"{txid}:{index}"
             self._handling_outpoint_statuses.discard(outpoint)
 
 

tests/toyserver/toyserver.py

@@ -6,7 +6,7 @@
 import collections
 from dataclasses import dataclass
 from functools import partial
-from typing import Optional, Sequence, Iterable, List, Set, Callable, TypeVar
+from typing import Optional, Sequence, Iterable, List, Set, Callable, TypeVar, Any
 
 import aiorpcx
 from aiorpcx import RPCError
@@ -490,12 +490,11 @@ async def _handle_server_version(self, client_name='', protocol_version=None, *a
     async def _handle_server_features(self) -> dict:
         return {
             'genesis_hash': constants.net.GENESIS,
-            'hosts': {"14.3.140.101": {"tcp_port": 51001, "ssl_port": 51002}},
+            'hosts': {"127.0.0.1": {"tcp_port": 51001, "ssl_port": 51002}},
             'protocol_max': '1.7',
             'protocol_min': '1.7',
             'pruning': None,
-            'server_version': 'ElectrumX 1.19.0',
-            'hash_function': 'sha256',
+            'server_version': 'toyserver 0.1',
         }
 
     async def _handle_estimatefee(self, number, mode=None):
@@ -574,13 +573,15 @@ async def _handle_spk_subscribe(self, spk: str) -> Optional[str]:
         hist = self.svr.calc_spk_history(spk)
         return history_status(hist)
 
-    async def _handle_spk_get_history(self, spk: str) -> Sequence[dict]:
+    async def _handle_spk_get_history(self, spk: str) -> dict[str, Any]:
+        res = dict()
         hist_tuples = self.svr.calc_spk_history(spk)
         hist_dicts = [{"height": height, "tx_hash": txid} for (txid, height) in hist_tuples]
         for hist_dict in hist_dicts:  # add "fee" key for mempool txs
             if hist_dict["height"] in (0, -1,):
                 hist_dict["fee"] = 0
-        return hist_dicts
+        res["history"] = hist_dicts
+        return res
 
     async def server_send_notifications(self, *, touched_spks: Iterable[str], height_changed: bool = False) -> None:
         if height_changed and self.subbed_headers and self.notified_height != self.svr.cur_height: