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(