Validate Bitcoin Core state types

Require exact nonnegative integer types for wallet counts and unlock state,
including final relock verification. This prevents booleans and malformed RPC
values from passing numeric equality checks.

Security: fail closed on untrusted Bitcoin Core state while preserving valid
encrypted and unencrypted wallet flows.

Validation: python -m pytest -q; python -O -m pytest -q; Ruff check and format;
strict mypy; differential_wallet.py --verify.

Author: BenWestgate

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(