(WIP) Add stream isolation for name commands

electrum_nmc/electrum/commands.py

@@ -303,12 +303,12 @@ async def make_seed(self, nbits=132, language=None, seed_type=None):
         return s
 
     @command('n')
-    async def getaddresshistory(self, address):
+    async def getaddresshistory(self, address, stream_id=None):
         """Return the transaction history of any address. Note: This is a
         walletless server query, results are not checked by SPV.
         """
         sh = bitcoin.address_to_scripthash(address)
-        return await self.network.get_history_for_scripthash(sh)
+        return await self.network.get_history_for_scripthash(sh, stream_id=stream_id)
 
     @command('w')
     async def listunspent(self, wallet: Abstract_Wallet = None):
@@ -372,12 +372,12 @@ async def name_list(self, identifier=None):
         return result
 
     @command('n')
-    async def getaddressunspent(self, address):
+    async def getaddressunspent(self, address, stream_id=None):
         """Returns the UTXO list of any address. Note: This
         is a walletless server query, results are not checked by SPV.
         """
         sh = bitcoin.address_to_scripthash(address)
-        return await self.network.listunspent_for_scripthash(sh)
+        return await self.network.listunspent_for_scripthash(sh, stream_id=stream_id)
 
     @command('')
     async def serialize(self, jsontx):
@@ -428,10 +428,10 @@ async def deserialize(self, tx):
         return tx.deserialize(force_full_parse=True)
 
     @command('n')
-    async def broadcast(self, tx):
+    async def broadcast(self, tx, stream_id=None):
         """Broadcast a transaction to the network. """
         tx = Transaction(tx)
-        await self.network.broadcast_transaction(tx)
+        await self.network.broadcast_transaction(tx, stream_id=stream_id)
         return tx.txid()
 
     @command('')
@@ -497,21 +497,21 @@ async def getbalance(self, wallet: Abstract_Wallet = None):
         return out
 
     @command('n')
-    async def getaddressbalance(self, address):
+    async def getaddressbalance(self, address, stream_id=None):
         """Return the balance of any address. Note: This is a walletless
         server query, results are not checked by SPV.
         """
         sh = bitcoin.address_to_scripthash(address)
-        out = await self.network.get_balance_for_scripthash(sh)
+        out = await self.network.get_balance_for_scripthash(sh, stream_id=stream_id)
         out["confirmed"] =  str(Decimal(out["confirmed"])/COIN)
         out["unconfirmed"] =  str(Decimal(out["unconfirmed"])/COIN)
         return out
 
     @command('n')
-    async def getmerkle(self, txid, height):
+    async def getmerkle(self, txid, height, stream_id=None):
         """Get Merkle branch of a transaction included in a block. Electrum
         uses this to verify transactions (Simple Payment Verification)."""
-        return await self.network.get_merkle_for_transaction(txid, int(height))
+        return await self.network.get_merkle_for_transaction(txid, int(height), stream_id=stream_id)
 
     @command('n')
     async def getservers(self):
@@ -689,12 +689,12 @@ async def paytomany(self, outputs, fee=None, feerate=None, from_addr=None, from_
         return tx.as_dict()
 
     @command('wp')
-    async def name_new(self, identifier, destination=None, amount=0.0, fee=None, from_addr=None, change_addr=None, nocheck=False, unsigned=False, rbf=None, password=None, locktime=None, allow_existing=False):
+    async def name_new(self, identifier, destination=None, amount=0.0, fee=None, from_addr=None, change_addr=None, nocheck=False, unsigned=False, rbf=None, password=None, locktime=None, allow_existing=False, stream_id=None):
         """Create a name_new transaction. """
         if not allow_existing:
             name_exists = True
             try:
-                show = self.name_show(identifier)
+                show = self.name_show(identifier, stream_id=stream_id)
             except NameNotFoundError:
                 name_exists = False
             if name_exists:
@@ -772,7 +772,7 @@ async def name_update(self, identifier, value=None, destination=None, amount=0.0
         return tx.as_dict()
 
     @command('wpn')
-    async def name_autoregister(self, identifier, value, destination=None, amount=0.0, fee=None, from_addr=None, change_addr=None, nocheck=False, rbf=None, password=None, locktime=None, allow_existing=False):
+    async def name_autoregister(self, identifier, value, destination=None, amount=0.0, fee=None, from_addr=None, change_addr=None, nocheck=False, rbf=None, password=None, locktime=None, allow_existing=False, stream_id=None):
         """Creates a name_new transaction, broadcasts it, creates a corresponding name_firstupdate transaction, and queues it. """
 
         # Validate the value before we try to pre-register the name.  That way,
@@ -781,12 +781,12 @@ async def name_autoregister(self, identifier, value, destination=None, amount=0.
         validate_value_length(value)
 
         # TODO: Don't hardcode the 0.005 name_firstupdate fee
-        new_result = self.name_new(identifier, amount=amount+0.005, fee=fee, from_addr=from_addr, change_addr=change_addr, nocheck=nocheck, rbf=rbf, password=password, locktime=locktime, allow_existing=allow_existing)
+        new_result = self.name_new(identifier, amount=amount+0.005, fee=fee, from_addr=from_addr, change_addr=change_addr, nocheck=nocheck, rbf=rbf, password=password, locktime=locktime, allow_existing=allow_existing, stream_id=stream_id)
         new_txid = new_result["txid"]
         new_rand = new_result["rand"]
         new_tx = new_result["tx"]["hex"]
 
-        self.broadcast(new_tx)
+        self.broadcast(new_tx, stream_id=stream_id)
 
         # We add the name_new transaction to the wallet explicitly because
         # otherwise, the wallet will only learn about the name_new once the
@@ -882,13 +882,13 @@ async def listaddresses(self, receiving=False, change=False, labels=False, froze
         return out
 
     @command('n')
-    async def gettransaction(self, txid, wallet: Abstract_Wallet = None):
+    async def gettransaction(self, txid, stream_id=None, wallet: Abstract_Wallet = None):
         """Retrieve a transaction. """
         tx = None
         if wallet:
             tx = wallet.db.get_transaction(txid)
         if tx is None:
-            raw = await self.network.get_transaction(txid)
+            raw = await self.network.get_transaction(txid, stream_id=stream_id)
             if raw:
                 tx = Transaction(raw)
             else:
@@ -1043,7 +1043,9 @@ async def updatequeuedtransactions(self):
             if trigger_name is not None:
                 # TODO: handle non-ASCII trigger_name
                 try:
-                    current_height = self.name_show(trigger_name)["height"]
+                    # TODO: Store a stream ID in the queue, so that we can be
+                    # more intelligent than using the txid.
+                    current_height = self.name_show(trigger_name, stream_id="txid: " + txid)["height"]
                     current_depth = chain_height - current_height + 1
                 except NameNotFoundError:
                     current_depth = 36000
@@ -1056,7 +1058,9 @@ async def updatequeuedtransactions(self):
             if current_depth >= trigger_depth:
                 tx = queue_item["tx"]
                 try:
-                    self.broadcast(tx)
+                    # TODO: Store a stream ID in the queue, so that we can be
+                    # more intelligent than using the txid.
+                    self.broadcast(tx, stream_id="txid: " + txid)
                 except Exception as e:
                     errors[txid] = str(e)
 
@@ -1128,12 +1132,12 @@ async def getfeerate(self, fee_method=None, fee_level=None):
         return self.config.fee_per_kb(dyn=dyn, mempool=mempool, fee_level=fee_level)
 
     @command('n')
-    async def name_show(self, identifier):
+    async def name_show(self, identifier, stream_id=None):
         # TODO: support non-ASCII encodings
         identifier_bytes = identifier.encode("ascii")
         sh = name_identifier_to_scripthash(identifier_bytes)
 
-        txs = self.network.run_from_another_thread(self.network.get_history_for_scripthash(sh))
+        txs = self.network.run_from_another_thread(self.network.get_history_for_scripthash(sh, stream_id=stream_id))
 
         # Pick the most recent name op that's [12, 36000) confirmations.
         chain_height = self.network.blockchain().height()
@@ -1156,10 +1160,10 @@ async def name_show(self, identifier):
         header = self.network.blockchain().read_header(height)
         if header is None:
             if height < constants.net.max_checkpoint():
-                self.network.run_from_another_thread(self.network.request_chunk(height, None))
+                self.network.run_from_another_thread(self.network.request_chunk(height, None, stream_id=stream_id))
 
         # (from verifier._request_and_verify_single_proof)
-        merkle = self.network.run_from_another_thread(self.network.get_merkle_for_transaction(txid, height))
+        merkle = self.network.run_from_another_thread(self.network.get_merkle_for_transaction(txid, height, stream_id=stream_id))
         if height != merkle.get('block_height'):
             raise Exception('requested height {} differs from received height {} for txid {}'
                             .format(height, merkle.get('block_height'), txid))
@@ -1177,7 +1181,7 @@ async def wait_for_header():
         if self.wallet and txid in self.wallet.db.transactions:
             tx = self.wallet.db.transactions[txid]
         else:
-            raw = self.network.run_from_another_thread(self.network.get_transaction(txid))
+            raw = self.network.run_from_another_thread(self.network.get_transaction(txid, stream_id=stream_id))
             if raw:
                 tx = Transaction(raw)
             else:
@@ -1390,6 +1394,7 @@ def eval_bool(x: str) -> bool:
     'fee_level':   (None, "Float between 0.0 and 1.0, representing fee slider position"),
     'from_height': (None, "Only show transactions that confirmed after given block height"),
     'to_height':   (None, "Only show transactions that confirmed before given block height"),
+    'stream_id':   (None, "Stream-isolate the network connection using this stream ID (only used with Tor)"),
     'destination': (None, "Namecoin address, contact or alias"),
     'amount':      (None, "Amount to be sent (in NMC). Type \'!\' to send the maximum available."),
     'allow_existing': (None, "Allow pre-registering a name that already is registered.  Your registration fee will be forfeited until you can register the name after it expires."),

electrum_nmc/electrum/interface.py

@@ -709,6 +709,31 @@ def do_bucket():
         return self._ipaddr_bucket
 
 
+class InterfaceSecondary(Interface):
+    """An Interface that doesn't try to fetch blocks, and instead stays idle
+    until it's explicitly used for something."""
+
+    async def ping(self):
+        # Since InterfaceSecondary doesn't ping periodically once it becomes
+        # dirty, it will time out if the user stops using it.  That's good,
+        # since otherwise we'd accumulate a giant pile of secondary interfaces
+        # for stream ID's that aren't in use anymore.
+        while True:
+            await asyncio.sleep(300)
+            if self not in self.network.interfaces_clean.values():
+                break
+            await self.session.send_request('server.ping')
+
+    async def run_fetch_blocks(self):
+        if self.ready.cancelled():
+            raise GracefulDisconnect('conn establishment was too slow; *ready* future was cancelled')
+        if self.ready.done():
+            return
+
+        # Without this, the Interface will think the connection timed out.
+        self.ready.set_result(1)
+
+
 def _assert_header_does_not_check_against_any_chain(header: dict) -> None:
     chain_bad = blockchain.check_header(header) if 'mock' not in header else header['mock']['check'](header)
     if chain_bad: