Specter-DIY: import xpubs without the 3s timeout, one confirmation per key set, wrong-network handling

src/cryptoadvance/specter/devices/hwi/specter_diy.py

@@ -31,7 +31,10 @@ class SpecterClient(HardwareWalletClient):
     that hardware wallet subclasses should implement.
     """
 
-    # timeout large enough to handle xpub derivations
+    # bounded timeout for non-interactive operations (e.g. fingerprint,
+    # used for device discovery/identification). xpub now requires an
+    # on-device user confirmation, so get_pubkey_at_path() does not use
+    # this - it waits indefinitely, like sign_tx() already does.
     TIMEOUT = 3
 
     def __init__(self, path: str, password: str = "", expert: bool = False) -> None:
@@ -87,8 +90,9 @@ def get_pubkey_at_path(self, bip32_path: str) -> ExtendedKey:
         :param bip32_path: The BIP 32 derivation path
         :return: The extended public key
         """
-        # this should be fast
-        xpub = self.query("xpub %s" % bip32_path, timeout=self.TIMEOUT)
+        # xpub now requires on-device user confirmation and may take
+        # arbitrarily long - wait indefinitely, same as sign_tx()
+        xpub = self.query("xpub %s" % bip32_path)
         hd = ExtendedKey.deserialize(xpub)
         # Specter returns xpub with a prefix
         # for a network currently selected on the device

tests/test_specter_diy_hwi.py

@@ -0,0 +1,44 @@
+"""
+Regression tests for cryptoadvance.specter.devices.hwi.specter_diy.SpecterClient.
+
+Specter DIY firmware now requires an on-device user confirmation before
+returning an xpub, so get_pubkey_at_path() must not impose the old 3-second
+response timeout (it should wait indefinitely, like sign_tx() already does).
+get_master_fingerprint() stays non-interactive and keeps its short timeout.
+
+These are pure unit tests against a mocked transport - no bitcoind, no real
+device, no network required.
+"""
+from unittest.mock import MagicMock
+
+from cryptoadvance.specter.devices.hwi.specter_diy import SpecterClient
+
+
+def _client_with_mocked_transport():
+    # ":" in the path selects the (non-connecting-on-init) simulator transport
+    client = SpecterClient("127.0.0.1:9999")
+    client.chain = "main"
+    client.dev.query = MagicMock(return_value="deadbeef")
+    return client
+
+
+def test_get_pubkey_at_path_does_not_pass_a_timeout():
+    client = _client_with_mocked_transport()
+    client.dev.query.return_value = (
+        "tpubD6NzVbkrYhZ4WZaiWHz59q5EQ61bd6dUYfU4ggRWAtNAyyYRNWT6ktJ7UHJEXURvSCVW"
+        "shSCLtQ4pnyNSSVUXQfP7yzzKcVXBEeejuSsn7q"
+    )
+    client.get_pubkey_at_path("m/84h/0h/0h")
+    args, kwargs = client.dev.query.call_args
+    # positional call: self.dev.query(data, timeout) - the timeout arg
+    # (positional or via kwarg) must be None, i.e. "wait indefinitely"
+    passed_timeout = kwargs.get("timeout", args[1] if len(args) > 1 else None)
+    assert passed_timeout is None
+
+
+def test_get_master_fingerprint_keeps_bounded_timeout():
+    client = _client_with_mocked_transport()
+    client.get_master_fingerprint()
+    args, kwargs = client.dev.query.call_args
+    passed_timeout = kwargs.get("timeout", args[1] if len(args) > 1 else None)
+    assert passed_timeout == SpecterClient.TIMEOUT