wallet: Validate Bitcoin Core state types

src/codex32/_bitcoin_core.py

@@ -304,19 +304,27 @@ def _target(self, name: str) -> tuple[bool, bool] | None:
         listing, info = self._rpc("listdescriptors", wallet=name), self._rpc("getwalletinfo", wallet=name)
         if not isinstance(info, dict) or not isinstance(listing, dict):
             raise BitcoinCoreError("Unexpected Bitcoin Core wallet information.")
+        txcount = info.get("txcount")
+        keypoolsize = info.get("keypoolsize")
+        internal_keypool = info.get("keypoolsize_hd_internal", 0)
+        has_unlock_state = "unlocked_until" in info
+        unlocked = info.get("unlocked_until")
         eligible = (
             info.get("descriptors") is True
             and info.get("private_keys_enabled") is True
             and info.get("external_signer", False) is False
-            and info.get("txcount") == 0
-            and info.get("keypoolsize") == 0
-            and info.get("keypoolsize_hd_internal", 0) == 0
+            and type(txcount) is int
+            and txcount == 0
+            and type(keypoolsize) is int
+            and keypoolsize == 0
+            and type(internal_keypool) is int
+            and internal_keypool == 0
             and info.get("scanning") is False
             and listing.get("descriptors") == []
             and name.isprintable()
+            and (not has_unlock_state or type(unlocked) is int and unlocked >= 0)
         )
-        unlocked = info.get("unlocked_until")
-        return (unlocked is not None, unlocked == 0) if eligible else None
+        return (has_unlock_state, unlocked == 0) if eligible else None
 
     def _create_account_zero(self, secret: MasterSeed, wallet: str) -> None:
         root = _master_xprv_from_seed(secret.seed_bytes, testnet=self.chain != "main")
@@ -510,7 +518,8 @@ def initialize(
                         continue
                     except BitcoinCoreError as error:
                         raise BitcoinCoreError(warning) from error
-                    if not isinstance(info, dict) or info.get("unlocked_until") != 0:
+                    unlocked_until = info.get("unlocked_until") if isinstance(info, dict) else None
+                    if type(unlocked_until) is not int or unlocked_until != 0:
                         raise BitcoinCoreError(warning)
                     relock = False
             if waited:

tests/test_bitcoin_core.py

@@ -214,6 +214,11 @@ def test_candidate_filter_rejects_every_unsafe_wallet_property(
         "transactions": (_empty_info(txcount=1), []),
         "keys": (_empty_info(keypoolsize=1), []),
         "change": (_empty_info(keypoolsize_hd_internal=1), []),
+        "boolean transactions": (_empty_info(txcount=False), []),
+        "boolean keys": (_empty_info(keypoolsize=False), []),
+        "boolean change": (_empty_info(keypoolsize_hd_internal=False), []),
+        "invalid unlock": (_empty_info(unlocked_until="unlocked"), []),
+        "null unlock": (_empty_info(unlocked_until=None), []),
         "scanning": (_empty_info(scanning={"duration": 1}), []),
         "descriptors": (_empty_info(), [{"desc": "public"}]),
         "bad\x1bname": (_empty_info(), []),
@@ -721,6 +726,60 @@ def interrupt_once(
     assert rpc.locked and lock_calls == 2
 
 
+def test_walletlock_failure_requires_manual_lock_confirmation(monkeypatch: pytest.MonkeyPatch) -> None:
+    rpc = _ImportRPC(locked=False)
+    original = rpc.__call__
+
+    def fail_lock(
+        client: BitcoinCore, *arguments: str, wallet: str | None = None, stdin: str | None = None
+    ) -> object:
+        if arguments == ("walletlock",):
+            raise BitcoinCoreError("suppressed lock failure")
+        return original(client, *arguments, wallet=wallet, stdin=stdin)
+
+    monkeypatch.setattr(BitcoinCore, "_rpc", fail_lock)
+    with pytest.raises(
+        BitcoinCoreError, match="Confirm immediately in Bitcoin Core that the wallet is locked"
+    ):
+        BitcoinCore("bitcoin-cli", "main", 300000).initialize(
+            _SEED,
+            lambda _prompt: "yes",
+            lambda _message: None,
+            expected_fingerprint=_FINGERPRINT,
+        )
+
+
+@pytest.mark.parametrize("unlocked_until", (100, False, "locked"))
+def test_failed_lock_verification_requires_manual_confirmation(
+    monkeypatch: pytest.MonkeyPatch, unlocked_until: object
+) -> None:
+    rpc = _ImportRPC(locked=False)
+    original = rpc.__call__
+    lock_requested = False
+
+    def remain_unlocked(
+        client: BitcoinCore, *arguments: str, wallet: str | None = None, stdin: str | None = None
+    ) -> object:
+        nonlocal lock_requested
+        if arguments == ("walletlock",):
+            lock_requested = True
+            return None
+        if lock_requested and arguments == ("getwalletinfo",):
+            return _empty_info(unlocked_until=unlocked_until)
+        return original(client, *arguments, wallet=wallet, stdin=stdin)
+
+    monkeypatch.setattr(BitcoinCore, "_rpc", remain_unlocked)
+    with pytest.raises(
+        BitcoinCoreError, match="Confirm immediately in Bitcoin Core that the wallet is locked"
+    ):
+        BitcoinCore("bitcoin-cli", "main", 300000).initialize(
+            _SEED,
+            lambda _prompt: "yes",
+            lambda _message: None,
+            expected_fingerprint=_FINGERPRINT,
+        )
+
+
 def test_unencrypted_wallet_imports_without_a_lock_call(monkeypatch: pytest.MonkeyPatch) -> None:
     rpc = _ImportRPC(encrypted=False, locked=False)
     monkeypatch.setattr(