From bbcfb56176bb27e5d762daa0f69b48c6f9e5d9dd Mon Sep 17 00:00:00 2001 From: Ben Westgate Date: Mon, 21 Sep 2026 02:03:33 -0500 Subject: [PATCH] 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. --- src/codex32/_bitcoin_core.py | 21 +++++++++---- tests/test_bitcoin_core.py | 59 ++++++++++++++++++++++++++++++++++++ 2 files changed, 74 insertions(+), 6 deletions(-) diff --git a/src/codex32/_bitcoin_core.py b/src/codex32/_bitcoin_core.py index 51712e4..4fc85e5 100644 --- a/src/codex32/_bitcoin_core.py +++ b/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: diff --git a/tests/test_bitcoin_core.py b/tests/test_bitcoin_core.py index 9737203..3eaf322 100644 --- a/tests/test_bitcoin_core.py +++ b/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(