diff --git a/AGENTS.md b/AGENTS.md index fd47d95..cf7df8c 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -61,7 +61,7 @@ avoid comments or tests that restate the implementation. Add or update concise docstrings when changing public behavior. Write codex32 in lowercase except when referring to the Codex32 Book. -Keep the installed package below 5,200 logical review lines, as enforced by the +Keep the installed package below 5,250 logical review lines, as enforced by the existing test. New dependencies, public API signature or return-shape changes, and lint suppressions require user authorization; an explicit request can already provide that authorization. diff --git a/docs/developer/api.md b/docs/developer/api.md index 4a290d8..dadf99f 100644 --- a/docs/developer/api.md +++ b/docs/developer/api.md @@ -105,7 +105,7 @@ unsupported but remains in the review scope. ### Size budget -V1 keeps the installed package below 5,200 logical review lines, excluding +V1 keeps the installed package below 5,250 logical review lines, excluding blank and comment-only lines while counting subpackages recursively. Changing the budget requires explicit review and authorization together with the matching documentation and enforcement update. @@ -195,6 +195,8 @@ requires a complete explicit `ms1` string; it never infers or corrects a missing HRP or separator. No entropy is drawn for this path; raw hexadecimal seeds retain the generation path. Existing imports use timestamp zero to include prior history. Changing a supplied secret's identifier requires a sharing threshold. +Existing-seed creation uses the same recorded-fingerprint or explicit no-record +confirmation as wallet restoration before import, including after re-sharing. Shared creation uses an explicit threshold or full backup header. Without an explicit share count or indices, thresholds 2 and 3 produce the reviewed 2-of-3 and 3-of-5 @@ -624,6 +626,13 @@ supplies the root xprv. Confirmation text is never reparsed into this source. The key is sent only through `bitcoin-cli -stdin`; raw Core errors are suppressed, and no passphrase interface exists. +For wallet restoration and `ms32 create --existing`, callers make the wallet-record +decision before initialization. `BitcoinCore.initialize()` calls `verify_identity()` +before `_select()` or any wallet mutation. A supplied fingerprint must match the +recovered master seed; `None` is reserved for fresh creation or the operator's +explicit no-record fallback. A mismatch stops before a destination wallet is +selected or changed. + Core v32 accepts the key with `addhdkey` and creates external and internal account-0 descriptors for BIP44/49/84/86 with `createwalletdescriptor`. Python checks each call's result but trusts Core to derive and store the wallet policy. diff --git a/docs/security/invariants.md b/docs/security/invariants.md index 98a35ef..be5ed48 100644 --- a/docs/security/invariants.md +++ b/docs/security/invariants.md @@ -11,6 +11,13 @@ and evidence. 3. Shared creation uses a separate OS-CSPRNG call for each random initial share, gated by confirmation. Input cannot replace entropy or the original secret. 4. Wallet setup uses the original ceremony result or a validated recovered seed. + Restore authenticates the recovered seed before any wallet is listed, + unlocked, or imported into: normally with the master fingerprint typed from + the wallet record, or by an explicit no-record choice made after seeing the + recovered fingerprint and whether the backup identifier matched a + seed-derived rule or its standard Bails check was unavailable. Fresh + `ms32 create` ceremonies do not authenticate against a pre-existing wallet; + they require the operator to record the new fingerprint. 5. Correction shares one mass bound and deadline across target lengths. The public API fails closed on incomplete required work; CLI searches may return one primary-best-so-far eligible candidate at the deadline. Incomplete diff --git a/docs/security/model.md b/docs/security/model.md index 2cd177f..e025fb3 100644 --- a/docs/security/model.md +++ b/docs/security/model.md @@ -45,8 +45,9 @@ The operator must: balances or history; - protect recovery cards and store shared cards in different trusted places; - confirm every newly recorded secret or share; -- keep wallet records separate from shares and compare recovered fingerprints, - addresses, account, policy, and history with those records; +- keep wallet records separate from shares, type the master fingerprint from + the record before a restore import, and compare addresses, account, policy, + and history with those records; - compare every correction suggestion with the original codex32 string and stop when recovered information and wallet records disagree; and - never put recovery text in command arguments or transfer a master seed, @@ -230,6 +231,7 @@ signing setup belong to Bitcoin Core's maintained v32 workflow. | Control | Required behavior | |---|---| | Preflight | Before entropy or recovery input, explicit chain arguments probe the five standard local networks for Bitcoin Core 32 or newer. One response is selected automatically; multiple responses require operator selection. | +| Recovery identity | `ms32 wallet` and `ms32 create --existing` authenticate a recovered seed before any wallet is listed. For `create --existing`, the wallet-record decision also precedes generation or display of any new card. Core derives the recovered fingerprint statelessly, and a mismatch raises `FingerprintMismatch` before any wallet RPC. The restore prompt does not show the recovered value, so the operator compares by typing the fingerprint from the wallet record. Without a record, the operator is shown the recovered fingerprint, whether the backup identifier was derived from the seed (the codex32 fingerprint rule, Bails' RIPEMD-160 rule, or its mid-2023 alpha's SHA-256 rule), and a warning, and then chooses. If RIPEMD-160 is unavailable, the standard Bails identifier check is reported as inconclusive rather than a mismatch; the Bails-alpha SHA-256 rule remains checkable. Fresh `ms32 create` has no pre-existing wallet to authenticate: it shows the newly created seed's fingerprint and requires the operator to acknowledge recording it. These checks catch mistakes such as wrong or mixed cards; anyone able to replace a threshold of cards could already read them. | | Process boundary | codex32 invokes the reviewed `bitcoin-cli` from `PATH` as a child without a shell, direct RPC socket, wallet database, or wallet-creation operation. Every call uses loopback and the selected chain. | | Destination | Only an empty descriptor wallet with private keys enabled, no external signer, transactions, descriptors, keypool entries, or active scan is eligible. One eligible wallet is offered directly; multiple wallets are selected by number. New wallets are detected by polling, and rejection returns to every eligible wallet. The escaped name is confirmed exactly. | | Seed source | The original ceremony result or validated recovered master seed supplies a root xprv for Core's reported chain. Core v32 creates BIP44, BIP49, BIP84, and BIP86 account-0 descriptors from that key. | diff --git a/docs/user/guide.md b/docs/user/guide.md index b2d78bb..05225cb 100644 --- a/docs/user/guide.md +++ b/docs/user/guide.md @@ -122,8 +122,12 @@ easier. Already have a complete codex32 `ms` secret? Run `ms32 create --existing` to write and confirm its recovery card and initialize a Bitcoin Core wallet. The existing secret is preserved unchanged. To split it into three cards -requiring any two, use `ms32 create 2 --existing` instead. Enter the secret -only when prompted. Bitcoin Core also scans for prior transactions. +requiring any two, use `ms32 create 2 --existing` instead. First type the +master fingerprint from the separate wallet record, then enter the secret +when prompted; a mismatch must be resolved before any new card is shown. If +you have no record, press Enter; the explicit recordless-restore choice and +visual fingerprint check happen right after the secret. Bitcoin Core also +scans for prior transactions. ### 3. Make a Bitcoin Core wallet @@ -171,9 +175,12 @@ wallet should be trusted until initialization completes. ### 4. Complete the record and store the cards -Copy the displayed backup identifier, wallet name, Bitcoin Core version, -master fingerprint, derivation standards, and account number to the wallet -record. Add the approximate +Before a freshly created wallet is filled, write the displayed master fingerprint on the +wallet record and confirm that you wrote it down. Fresh creation has no pre-existing +fingerprint or descriptor to authenticate; `ms32 create --existing` instead uses the +restore identity gate. Then copy the displayed +backup identifier, wallet name, Bitcoin Core version, derivation standards, and +account number to the wallet record. Add the approximate creation / earliest-use date. Do not put a descriptor timestamp on a recovery card; Core's public descriptor export preserves its stored timestamps. @@ -234,16 +241,23 @@ its public wallet data with the separate wallet record. If you know when the wallet was first used, an earlier Unix timestamp can shorten the rescan; `0` remains the safest choice when unsure. -5. Select and confirm that wallet. If it is locked, follow the displayed +5. Type the master fingerprint from the wallet record, then enter the cards. + A suggested correction says whether it matches the record without showing + the fingerprint, and the record picks between equally likely corrections. + A mismatch stops before Bitcoin Core is changed. Press Enter with nothing + typed only if there is no record; after the cards, codex32 then shows the + recovered fingerprint and what the backup identifier says, and asks before + restoring. +6. Select and confirm that wallet. If it is locked, follow the displayed Bitcoin-Qt Console instructions; codex32 waits and continues automatically. It gives Core the master private key, asks Core to create the standard account-0 descriptors, scans history, and relocks an encrypted wallet. -6. If you need an online watch-only counterpart, keep the restored signer +7. If you need an online watch-only counterpart, keep the restored signer offline and follow Bitcoin Core v32's [offline-signing tutorial](https://github.com/bitcoin/bitcoin/blob/v32.0rc1/doc/offline-signing-tutorial.md) to export and restore the watch-only wallet. Let the online node synchronize, - then compare the recovered fingerprint, account, policy, addresses, balance, - and transaction history with the wallet record. + then compare the account, policy, addresses, balance, and transaction + history with the wallet record. A timestamp of zero safely scans all history and may take time; it belongs in the recovery command, not on a paper card. During an emergency recovery, move diff --git a/src/codex32/_bitcoin_core.py b/src/codex32/_bitcoin_core.py index cfec702..51712e4 100644 --- a/src/codex32/_bitcoin_core.py +++ b/src/codex32/_bitcoin_core.py @@ -1,8 +1,10 @@ from __future__ import annotations +import hashlib import json import re import shutil +import string import subprocess from collections.abc import Callable from dataclasses import dataclass @@ -10,6 +12,8 @@ from typing import Literal from codex32._bip32 import _master_xprv_from_seed +from codex32.bech32 import _u5_to_chars, convertbits +from codex32.generation import _fingerprint_identifier from codex32.profiles.ms32 import MasterSeed from codex32.wallet import _descriptor_records @@ -18,6 +22,10 @@ class BitcoinCoreError(Exception): pass +class FingerprintMismatch(BitcoinCoreError): + """The recovered seed is not the wallet the operator's record describes.""" + + _CHAINS = ( ("main", "mainnet"), ("test", "testnet3"), @@ -32,6 +40,63 @@ class BitcoinCoreError(Exception): _OUTPUT_TYPES = ("legacy", "p2sh-segwit", "bech32", "bech32m") +def parse_fingerprint(text: str) -> bytes: + """Read a master fingerprint as written on a wallet record: 8 hex digits, any case or spacing.""" + compact = "".join(text.split()) + if len(compact) != 8 or not all(character in string.hexdigits for character in compact): + raise ValueError("A master fingerprint is 8 characters, each 0-9 or A-F.") + return bytes.fromhex(compact) + + +NO_RECORD_WARNING = ( + "Without the wallet record, nothing can prove these cards are the wallet you expect. Compare the " + "fingerprint with any other copy, such as another wallet app, a hardware wallet or a descriptor backup. " + "After restoring, let Bitcoin Core finish scanning and check that the balance, past payments and " + "addresses are ones you recognise before sending money here. Replaced cards can come with a history " + "too: if you do not know what this wallet should hold, have someone you trust check it. Once you are " + "sure, write the fingerprint on a new wallet record." +) + + +def identifier_note(origin: str | None) -> str: + # Say what `identifier_origin` found, for an operator restoring without a record. + if origin == "Bails check unavailable": + return ( + "The standard Bails identifier could not be checked because RIPEMD-160 is unavailable. " + "This does not prove the cards are wrong or mixed up. Compare the fingerprint and any " + "other wallet record you have before restoring." + ) + if origin is None: + return ( + "The backup identifier was not made from this seed. That can be normal for codex32 backups " + "made from split shares, supplied seed bytes or an explicit identifier. Bails made every " + "identifier from its seed, so for a Bails backup these are the wrong or mixed-up cards." + ) + return ( + f"The backup identifier matches this seed ({origin} rule). That rules out most mixed-up cards, " + "but not cards replaced on purpose." + ) + + +def identifier_origin(secret: MasterSeed, fingerprint: bytes) -> str | None: + """Check codex32's fingerprint or Bails' three-character seed-digest identifier.""" + identifier = secret.header.identifier + if identifier == _fingerprint_identifier(fingerprint): + return "codex32" + ripemd_unavailable = False + for name, digest in (("Bails", "ripemd160"), ("Bails alpha", "sha256")): + try: + hashed = hashlib.new(digest, secret.seed_bytes).digest() + except ValueError: + if name == "Bails": + ripemd_unavailable = True + continue + derived = convertbits(hashed, 8, 5, pad=True) + if identifier[:3] == _u5_to_chars(tuple(derived[:3])): + return name + return "Bails check unavailable" if ripemd_unavailable else None + + @dataclass(frozen=True) class BitcoinCore: executable: str @@ -154,6 +219,14 @@ def fingerprint(self, secret: MasterSeed) -> bytes: raise TypeError("wallet operations accept only MasterSeed") return self.fingerprint_seed(secret.seed_bytes) + def verify_identity(self, secret: MasterSeed, expected_fingerprint: bytes | None) -> None: + """Refuse a recovered seed that is not the recorded wallet before any wallet is touched.""" + if expected_fingerprint is not None and self.fingerprint(secret) != expected_fingerprint: + raise FingerprintMismatch( + "The recovered master fingerprint does not match the one from the wallet record. " + "Bitcoin Core was not changed." + ) + def _root_xpub(self, wallet: str) -> str: result = self._rpc("gethdkeys", wallet=wallet) if not isinstance(result, list) or len(result) != 1 or not isinstance(result[0], dict): @@ -375,15 +448,18 @@ def initialize( ask: Callable[[str], str], tell: Callable[[str], None], *, + expected_fingerprint: bytes | None, account: int = 0, timestamp: int | Literal["now"] = "now", ) -> str: + """Validate input and identity before selecting or changing a wallet.""" if not isinstance(secret, MasterSeed): raise TypeError("wallet operations accept only MasterSeed") if type(account) is not int or account != 0: raise ValueError("Bitcoin Core wallet initialization currently supports only account 0") if timestamp != "now" and (type(timestamp) is not int or timestamp < 0): raise ValueError("timestamp must be a nonnegative integer or 'now'") + self.verify_identity(secret, expected_fingerprint) while True: name = self._select(ask, tell) state = self._target(name) diff --git a/src/codex32/_cli_input.py b/src/codex32/_cli_input.py index cf5d25d..d72fbd4 100644 --- a/src/codex32/_cli_input.py +++ b/src/codex32/_cli_input.py @@ -200,28 +200,36 @@ def _card_text(text: str, highlight: bool = True, observed: str = "") -> str: return rendered if changed else rendered.replace("\x1b[0m ", " ") +def _completed(artifact: Artifact, accepted: Sequence[Artifact]) -> Artifact: + # Provisional recovery is exclusively for fingerprint previews and record checks. + if ( + isinstance(artifact, Share) + and artifact.profile is Profile.MS + and len(accepted) + 1 == artifact.header.threshold + ): + return recover_secret(cast(list[Share], [*accepted, artifact])) + return artifact + + def _confirm_correction( candidate: CorrectionCandidate, accepted: list[Artifact], basis: bool, fingerprint: Callable[[MasterSeed], bytes] | None = None, + record: bytes | None = None, ) -> bool | None: _require_correction_confirmation(candidate.low_checksum_discrimination) artifact = candidate.artifact - # Provisional recovery is exclusively for this fingerprint preview. - preview = artifact try: - if ( - isinstance(artifact, Share) - and artifact.profile is Profile.MS - and not basis - and (len(accepted) + 1 == artifact.header.threshold) - ): - preview = recover_secret(cast(list[Share], [*accepted, artifact])) + preview = artifact if basis else _completed(artifact, accepted) + # A typed wallet record is checked without showing the recovered value. fingerprint_text = ( - f"Master fingerprint: {fingerprint(preview).hex().upper()}\n\n" - if isinstance(preview, MasterSeed) and fingerprint is not None - else "" + "" + if not isinstance(preview, MasterSeed) or fingerprint is None + else f"Master fingerprint: {fingerprint(preview).hex().upper()}\n\n" + if record is None + else f"Master fingerprint {'matches' if fingerprint(preview) == record else 'does not match'} " + "your wallet record.\n\n" ) except CodexError: _stderr("Rejected: Could not recover a valid Bitcoin master seed using this correction.") @@ -410,36 +418,24 @@ def _correction_plan( byte_length: int | Literal["?"] | None, count: int, target: int | None, -) -> tuple[tuple[int, ...], frozenset[int], frozenset[int], bool]: +) -> tuple[int, ...]: if target is not None: - return ( - (target,), - frozenset((target,)), - frozenset(), - True, - ) + return (target,) normalized_hrp = hrp.value if isinstance(hrp, Profile) else hrp.lower() if normalized_hrp == Profile.CL.value: - return (74,), frozenset((74,)), frozenset(), True + return (74,) if isinstance(byte_length, int): - return ( - ((length := _text_length(byte_length)),), - frozenset((length,)), - frozenset(), - True, - ) + return (_text_length(byte_length),) if byte_length == "?": - return TEXT_LENGTHS, frozenset(TEXT_LENGTHS), frozenset(), True + return TEXT_LENGTHS if normalized_hrp == Profile.MS.value: nearest = min(_PRIMARY_MS, key=lambda length: abs(count - length)) targets = (nearest, *(length for length in TEXT_LENGTHS if length != nearest)) - return targets, frozenset(targets), frozenset(), True + return targets rules = _optional_profile_rules(normalized_hrp) if rules is not None and hasattr(rules, "text_length"): - targets = (rules.text_length,) - return targets, frozenset(targets), frozenset(), True - targets = tuple(sorted({count + delta for delta in (*range(-4, 5), -8, 8)})) - return targets, frozenset(targets), frozenset(), True + return (rules.text_length,) + return tuple(sorted({count + delta for delta in (*range(-4, 5), -8, 8)})) def _correction_candidates( @@ -459,7 +455,7 @@ def _correction_candidates( optional_only: bool = False, ) -> tuple[tuple[CorrectionCandidate, ...], bool, float]: count = len(value.replace(" ", "")) - targets, primary, reduced, _timed = _correction_plan(profile, byte_length, count, target) + targets = _correction_plan(profile, byte_length, count, target) deadline = monotonic() + 10 if deadline is None else deadline contexts = tuple(CorrectionContext(profile, length, immutable, excluded) for length in targets) from codex32.indel import _search_many @@ -467,8 +463,7 @@ def _correction_candidates( candidates, complete = _search_many( contexts, value, - primary=primary, - reduced=reduced, + primary=frozenset(targets), deadline=deadline, competitors=True, allowed=allowed, @@ -550,6 +545,8 @@ def _scheduled_candidates( def _fingerprint_matcher( fingerprint: Callable[[MasterSeed], bytes] | None, + record: bytes | None = None, + accepted: Sequence[Artifact] = (), ) -> Callable[[CorrectionCandidate], bool | None] | None: if fingerprint is None: return None @@ -557,9 +554,13 @@ def _fingerprint_matcher( def matches(candidate: CorrectionCandidate) -> bool | None: artifact = candidate.artifact - if not isinstance(artifact, MasterSeed) or artifact.header.threshold: - return None try: + if record is not None: + # Prefer corrections whose secret, or completed share set, matches the record. + seed = _completed(artifact, accepted) + return fingerprint(seed) == record if isinstance(seed, MasterSeed) else None + if not isinstance(artifact, MasterSeed) or artifact.header.threshold: + return None return _fingerprint_identifier(fingerprint(artifact)) == artifact.header.identifier except CodexError: return None @@ -575,8 +576,9 @@ def _suggestions( *, allowed: Callable[[CorrectionCandidate], bool] | None = None, fingerprint: Callable[[MasterSeed], bytes] | None = None, + record: bytes | None = None, ) -> tuple[CorrectionCandidate, ...]: - fingerprint_match = _fingerprint_matcher(fingerprint) + fingerprint_match = _fingerprint_matcher(fingerprint, record, accepted) erased = value if interpretation := _case_interpretation(value, prefix, profiles, allowed): candidate, value, erased, prefix = interpretation @@ -739,6 +741,7 @@ def _interactive( profiles: tuple[Profile, ...] | None, initial_prefix: str, fingerprint: Callable[[MasterSeed], bytes] | None, + record: bytes | None, ) -> list[Artifact]: accepted: list[Artifact] = [] prefix = initial_prefix @@ -783,10 +786,11 @@ def allowed(candidate: CorrectionCandidate) -> bool: accepted, allowed=allowed, fingerprint=fingerprint, + record=record, ) ) confirmation = ( - _confirm_correction(candidates[0], accepted, basis, fingerprint) + _confirm_correction(candidates[0], accepted, basis, fingerprint, record) if len(candidates) == 1 else None ) @@ -837,6 +841,7 @@ def read_artifacts( profiles: tuple[Profile, ...] | None = None, initial_prefix: str = "", fingerprint: Callable[[MasterSeed], bytes] | None = None, + record: bytes | None = None, ) -> list[Artifact]: if not sys.stdin.isatty(): return _redirected( @@ -853,6 +858,7 @@ def read_artifacts( profiles=profiles, initial_prefix=initial_prefix, fingerprint=fingerprint, + record=record, ) _stderr("") return result diff --git a/src/codex32/cli.py b/src/codex32/cli.py index b20bddd..2b653b2 100644 --- a/src/codex32/cli.py +++ b/src/codex32/cli.py @@ -8,7 +8,15 @@ from collections.abc import Callable, Sequence from typing import Literal, NamedTuple, cast -from codex32._bitcoin_core import BitcoinCore, BitcoinCoreError +from codex32._bitcoin_core import ( + NO_RECORD_WARNING, + BitcoinCore, + BitcoinCoreError, + FingerprintMismatch, + identifier_note, + identifier_origin, + parse_fingerprint, +) from codex32._cli_input import ( CorrectionDeclined, InteractiveConfirmationRequired, @@ -94,16 +102,18 @@ def _secret(artifacts: list[Artifact]) -> Secret: if not all(isinstance(artifact, Share) for artifact in artifacts): raise _UsageError("Recovery accepts ordinary shares or one complete secret.") try: - return recover_secret([artifact for artifact in artifacts if isinstance(artifact, Share)]) + return recover_secret(cast(list[Share], artifacts)) except CodexError as error: raise _UsageError(str(error)) from error -def _master_seed(fingerprint: Callable[[MasterSeed], bytes] | None = None) -> MasterSeed: - if isinstance( - value := _secret(_artifacts(profiles=(Profile.MS,), initial_prefix="MS1", fingerprint=fingerprint)), - MasterSeed, - ): +def _master_seed( + fingerprint: Callable[[MasterSeed], bytes] | None = None, record: bytes | None = None +) -> MasterSeed: + artifacts = _artifacts( + profiles=(Profile.MS,), initial_prefix="MS1", fingerprint=fingerprint, record=record + ) + if isinstance(value := _secret(artifacts), MasterSeed): return value raise _UsageError("Wallet commands accept only Bitcoin master-seed secrets.") @@ -223,6 +233,7 @@ def _creation_header(value: str | None) -> tuple[Profile, int | None, str | None def _creation_source( profile: Profile, fingerprint: Callable[[MasterSeed], bytes] | None = None, + record: bytes | None = None, ) -> bytes | Artifact: prefill = "" while True: @@ -250,13 +261,14 @@ def _creation_source( [], allowed=lambda item: isinstance(item.artifact, Secret) and item.artifact.profile is profile, fingerprint=fingerprint, + record=record, ) if ( len(candidates) == 1 and isinstance(candidate := candidates[0].artifact, Secret) and candidate.profile is profile ): - confirmation = _confirm_correction(candidates[0], [], False, fingerprint) + confirmation = _confirm_correction(candidates[0], [], False, fingerprint, record) if confirmation is True: return candidate if confirmation is False: @@ -339,6 +351,46 @@ def _generated_secret( ) +def _show_fingerprint(core: BitcoinCore, secret: MasterSeed, action: str) -> None: + _print(f"\nMaster fingerprint: {core.fingerprint(secret).hex().upper()}", err=True) + _text(f"{action}, then press Enter", optional=True, prompt_end=". ") + if sys.stderr.isatty(): + _print("\x1b[3J\x1b[2J\x1b[H", err=True) + + +def _without_record(core: BitcoinCore, secret: MasterSeed) -> bool: + fingerprint = core.fingerprint(secret) + _print(f"\nMaster fingerprint: {fingerprint.hex().upper()}", err=True) + _print(f"Backup identifier: {secret.header.identifier.upper()}", err=True) + _print(identifier_note(identifier_origin(secret, fingerprint)), err=True) + _print(NO_RECORD_WARNING, err=True) + return _text("Restore without a wallet record? [y/N]", optional=True).lower() in ("y", "yes") + + +def _record() -> bytes | None: + # Asked before the cards, so corrections can be checked without showing the fingerprint. + while text := _text("Type the master fingerprint from your wallet record (Enter if none)", optional=True): + try: + return parse_fingerprint(text) + except ValueError as error: + _print(str(error), err=True) + return None + + +def _recorded_fingerprint(core: BitcoinCore, secret: MasterSeed, expected: bytes | None) -> bytes | None: + # Re-ask for the record until the library accepts it, or the operator has none. + while expected is not None: + try: + core.verify_identity(secret, expected) + return expected + except FingerprintMismatch as error: + _print(str(error), err=True) + expected = _record() + if _without_record(core, secret): + return None + raise _WalletSetupInterrupted + + def _initialize_wallet( core: BitcoinCore, secret: MasterSeed, @@ -346,16 +398,24 @@ def _initialize_wallet( account: int = 0, timestamp: int | Literal["now"] = "now", fresh: bool = True, + restore: bool = False, confirmed: bool = True, + identity_checked: bool = False, + expected_fingerprint: bytes | None = None, ) -> int: assert isinstance(secret, MasterSeed) try: if confirmed: _print("Master-seed backup confirmed.\n", err=True) + if restore and not identity_checked: + expected_fingerprint = _recorded_fingerprint(core, secret, expected_fingerprint) + if not restore: + _show_fingerprint(core, secret, "Write it on the wallet record") name = core.initialize( secret, lambda prompt: _text(prompt, optional=True), lambda message: _print(message, err=True), + expected_fingerprint=expected_fingerprint, account=account, timestamp=timestamp, ) @@ -432,41 +492,53 @@ def _create( else: raise _UsageError("For thresholds 4 through 9, choose --shares or --indices.") core = _connected_core() - source = _creation_source(profile, core.fingerprint) if existing else None - if not existing and not sys.stdin.isatty() and _text("", optional=True): - raise _UsageError("Use --existing when supplying a seed or secret.") + try: + record = _record() if existing else None + except KeyboardInterrupt as error: + raise _WalletSetupInterrupted from error + source = _creation_source(profile, core.fingerprint if record else None, record) if existing else None if isinstance(source, (Share, Secret)) and not isinstance(source, MasterSeed): raise _UsageError(f"Enter one {_profile_rules(profile).label}, not a share or another backup type.") try: - if threshold == 0: - if isinstance(source, MasterSeed): - if identifier is not None and identifier != source.header.identifier: - raise _UsageError( - "To change the existing secret's identifier, choose a sharing threshold from 2 through 9." - ) - secret = source - else: - secret = _generated_secret(source, byte_length, identifier, core.fingerprint_seed) - _emit(secret, False, fingerprint=core.fingerprint) - if sys.stdin.isatty(): - _confirm_card(secret) - return ( - _initialize_wallet(core, secret, timestamp=0 if existing else "now", fresh=not existing) - if core is not None - else 0 - ) + existing_secret: MasterSeed | None = None if isinstance(source, MasterSeed): - ceremony = CreationCeremony.from_secret( - source, - threshold=threshold, - identifier=identifier, - share_count=shares, - indices=indices, - ) + if threshold == 0 and identifier is not None and identifier != source.header.identifier: + raise _UsageError( + "To change the existing secret's identifier, choose a sharing threshold from 2 through 9." + ) + existing_secret = source elif source is not None: - source_secret = _generated_secret(source, None, identifier, core.fingerprint_seed) + existing_secret = _generated_secret(source, None, identifier, core.fingerprint_seed) + if threshold and identifier is None: + identifier = existing_secret.header.identifier + try: + expected = ( + _recorded_fingerprint(core, existing_secret, record) if existing_secret is not None else None + ) + except (EOFError, KeyboardInterrupt) as error: + raise _WalletSetupInterrupted from error + + def finish_wallet(seed: MasterSeed) -> int: + return _initialize_wallet( + core, + seed, + timestamp=0 if existing else "now", + fresh=not existing, + restore=existing, + identity_checked=existing, + expected_fingerprint=expected, + ) + + if threshold == 0: + secret = existing_secret or _generated_secret( + None, byte_length, identifier, core.fingerprint_seed + ) + _emit(secret, False, fingerprint=None if existing else core.fingerprint) + _confirm_card(secret) + return finish_wallet(secret) + if existing_secret is not None: ceremony = CreationCeremony.from_secret( - source_secret, + existing_secret, threshold=threshold, identifier=identifier, share_count=shares, @@ -492,10 +564,7 @@ def _create( _print(f"Recovery card {position + 1} of {output_count} confirmed.", err=True) finished = ceremony.finish() assert isinstance(finished, MasterSeed) - if core is not None: - return _initialize_wallet(core, finished, timestamp=0 if existing else "now", fresh=not existing) - _print("\nEvery recovery card was confirmed from its re-entered text.", err=True) - return 0 + return finish_wallet(finished) def _correct( @@ -609,14 +678,19 @@ def _bitcoin_core(account: int, timestamp: int | Literal["now"]) -> int: danger=True, ) core = _connected_core() - secret = _master_seed(core.fingerprint) + # Keep the recovered fingerprint hidden until the operator has supplied + # independent wallet-record evidence or explicitly chosen recordless restore. + record = _record() + secret = _master_seed(core.fingerprint if record else None, record) return _initialize_wallet( core, secret, account=account, timestamp=timestamp, fresh=False, + restore=True, confirmed=False, + expected_fingerprint=record, ) diff --git a/src/codex32/indel.py b/src/codex32/indel.py index 26c0071..98a6369 100644 --- a/src/codex32/indel.py +++ b/src/codex32/indel.py @@ -552,7 +552,6 @@ def _search_many( damaged_text: str, *, primary: frozenset[int], - reduced: frozenset[int] = frozenset(), deadline: float | None = None, max_character_depth: int = 4, competitors: bool = False, diff --git a/tests/test_bitcoin_core.py b/tests/test_bitcoin_core.py index 72d6676..9737203 100644 --- a/tests/test_bitcoin_core.py +++ b/tests/test_bitcoin_core.py @@ -2,6 +2,7 @@ from __future__ import annotations +import hashlib import json import re import subprocess @@ -9,8 +10,16 @@ import pytest -from codex32._bitcoin_core import BitcoinCore, BitcoinCoreError +from codex32._bitcoin_core import ( + BitcoinCore, + BitcoinCoreError, + FingerprintMismatch, + identifier_note, + identifier_origin, + parse_fingerprint, +) from codex32.bip93 import parse_codex32 +from codex32.generation import _fingerprint_identifier from codex32.profiles.ms32 import MasterSeed from codex32.wallet import _with_checksum @@ -37,9 +46,16 @@ ), } _ROOT_XPUB = "xpub-root-fixture" +_FINGERPRINT = bytes.fromhex("3f3521a6") _PRIVATE_ACCOUNT = re.compile(r"/(?P44|49|84|86)h/0h/0h/<0;1>/\*") +@pytest.fixture(autouse=True) +def _recorded_fingerprint(monkeypatch: pytest.MonkeyPatch) -> None: + """Answer the pre-import identity check without the address-derivation RPCs tested separately.""" + monkeypatch.setattr(BitcoinCore, "fingerprint", lambda _client, _secret: _FINGERPRINT) + + def _descriptor_info(descriptor: str) -> dict[str, object]: """Return frozen Core-like normalization for the synthetic seed fixture.""" raw = descriptor.strip().split("#", 1)[0] @@ -386,7 +402,10 @@ def unlock(seconds: int) -> None: monkeypatch.setattr("codex32._bitcoin_core.sleep", unlock) - assert client.initialize(_SEED, lambda _prompt: "yes", messages.append) == "signer" + assert ( + client.initialize(_SEED, lambda _prompt: "yes", messages.append, expected_fingerprint=_FINGERPRINT) + == "signer" + ) private_calls = [call for call in rpc.calls if "xprv" in (call[2] or "")] assert len(private_calls) == 1 arguments, wallet, private_stdin = private_calls[0] @@ -418,7 +437,11 @@ def test_nonzero_or_noninteger_account_is_rejected_before_wallet_selection( monkeypatch.setattr(BitcoinCore, "_select", lambda *_args: pytest.fail("selected a wallet")) with pytest.raises(ValueError, match="only account 0"): BitcoinCore("bitcoin-cli", "main", 320000).initialize( - _SEED, lambda _prompt: "yes", lambda _message: None, account=account + _SEED, + lambda _prompt: "yes", + lambda _message: None, + expected_fingerprint=_FINGERPRINT, + account=account, ) @@ -439,7 +462,9 @@ def fail( monkeypatch.setattr(BitcoinCore, "_rpc", fail) client = BitcoinCore("bitcoin-cli", "main", 300000) with pytest.raises(BitcoinCoreError, match="did not create both wallet descriptors"): - client.initialize(_SEED, lambda _prompt: "yes", lambda _message: None) + client.initialize( + _SEED, lambda _prompt: "yes", lambda _message: None, expected_fingerprint=_FINGERPRINT + ) assert rpc.locked assert any(arguments == ("walletlock",) for arguments, _wallet, _stdin in rpc.calls) @@ -465,7 +490,7 @@ def rpc( monkeypatch.setattr(BitcoinCore, "_rpc", rpc) - assert BitcoinCore("bitcoin-cli", "main", 320000).fingerprint(_SEED) == bytes.fromhex("3f3521a6") + assert BitcoinCore("bitcoin-cli", "main", 320000).fingerprint_seed(_SEED.seed_bytes) == _FINGERPRINT assert calls == [ (("getdescriptorinfo",), None), (("deriveaddresses",), None), @@ -485,7 +510,13 @@ def test_numeric_timestamp_rescans_history(monkeypatch: pytest.MonkeyPatch, time ) client = BitcoinCore("bitcoin-cli", "main", 300000) assert ( - client.initialize(_SEED, lambda _prompt: "yes", lambda _message: None, timestamp=timestamp) + client.initialize( + _SEED, + lambda _prompt: "yes", + lambda _message: None, + expected_fingerprint=_FINGERPRINT, + timestamp=timestamp, + ) == "signer" ) calls = [(args, data) for args, _wallet, data in rpc.calls if args == ("importdescriptors",)] @@ -519,7 +550,11 @@ def test_failed_timestamped_rescan_relocks(monkeypatch: pytest.MonkeyPatch) -> N ) with pytest.raises(BitcoinCoreError, match="did not complete the timestamped wallet rescan"): BitcoinCore("bitcoin-cli", "main", 300000).initialize( - _SEED, lambda _prompt: "yes", lambda _message: None, timestamp=123 + _SEED, + lambda _prompt: "yes", + lambda _message: None, + expected_fingerprint=_FINGERPRINT, + timestamp=123, ) assert rpc.locked @@ -544,7 +579,9 @@ def fail( monkeypatch.setattr(BitcoinCore, "_rpc", fail) client = BitcoinCore("bitcoin-cli", "main", 300000) with pytest.raises(BitcoinCoreError, match="suppressed failure"): - client.initialize(_SEED, lambda _prompt: "yes", lambda _message: None) + client.initialize( + _SEED, lambda _prompt: "yes", lambda _message: None, expected_fingerprint=_FINGERPRINT + ) assert rpc.locked assert any(arguments == ("walletlock",) for arguments, _wallet, _stdin in rpc.calls) @@ -563,7 +600,9 @@ def interrupt( monkeypatch.setattr(BitcoinCore, "_rpc", interrupt) client = BitcoinCore("bitcoin-cli", "main", 300000) with pytest.raises(KeyboardInterrupt): - client.initialize(_SEED, lambda _prompt: "yes", lambda _message: None) + client.initialize( + _SEED, lambda _prompt: "yes", lambda _message: None, expected_fingerprint=_FINGERPRINT + ) assert rpc.locked @@ -590,7 +629,9 @@ def select(_client: BitcoinCore, _ask: object, _tell: object) -> str: ) with pytest.raises(KeyboardInterrupt): - BitcoinCore("bitcoin-cli", "main", 300000).initialize(_SEED, lambda _prompt: "yes", messages.append) + BitcoinCore("bitcoin-cli", "main", 300000).initialize( + _SEED, lambda _prompt: "yes", messages.append, expected_fingerprint=_FINGERPRINT + ) assert rpc.locked assert "That wallet is no longer eligible. Choose again." in messages assert any(arguments == ("walletlock",) for arguments, _wallet, _stdin in rpc.calls) @@ -613,7 +654,7 @@ def interrupt(_seconds: int) -> None: with pytest.raises(KeyboardInterrupt): BitcoinCore("bitcoin-cli", "main", 300000).initialize( - _SEED, lambda _prompt: "", lambda _message: None + _SEED, lambda _prompt: "", lambda _message: None, expected_fingerprint=_FINGERPRINT ) assert rpc.locked @@ -642,7 +683,7 @@ def target(*_args: object, **_options: object) -> tuple[bool, bool]: assert ( BitcoinCore("bitcoin-cli", "main", 300000).initialize( - _SEED, lambda _prompt: "", lambda _message: None + _SEED, lambda _prompt: "", lambda _message: None, expected_fingerprint=_FINGERPRINT ) == "signer" ) @@ -671,7 +712,12 @@ def interrupt_once( monkeypatch.setattr(BitcoinCore, "_rpc", interrupt_once) client = BitcoinCore("bitcoin-cli", "main", 300000) - assert client.initialize(_SEED, lambda _prompt: "yes", lambda _message: None) == "signer" + assert ( + client.initialize( + _SEED, lambda _prompt: "yes", lambda _message: None, expected_fingerprint=_FINGERPRINT + ) + == "signer" + ) assert rpc.locked and lock_calls == 2 @@ -686,7 +732,10 @@ def test_unencrypted_wallet_imports_without_a_lock_call(monkeypatch: pytest.Monk ) client = BitcoinCore("bitcoin-cli", "main", 300000) messages: list[str] = [] - assert client.initialize(_SEED, lambda _prompt: "yes", messages.append) == "signer" + assert ( + client.initialize(_SEED, lambda _prompt: "yes", messages.append, expected_fingerprint=_FINGERPRINT) + == "signer" + ) assert messages == [] assert not any(arguments == ("walletlock",) for arguments, _wallet, _stdin in rpc.calls) @@ -711,7 +760,7 @@ def test_immediate_revalidation_stops_before_private_import_and_relocks( ) with pytest.raises(BitcoinCoreError, match="changed before import"): - client.initialize(_SEED, lambda _prompt: "", lambda _message: None) + client.initialize(_SEED, lambda _prompt: "", lambda _message: None, expected_fingerprint=_FINGERPRINT) assert rpc.locked assert not any(arguments == ("addhdkey",) for arguments, _wallet, _stdin in rpc.calls) @@ -734,3 +783,90 @@ def run(command: list[str], **options: object) -> subprocess.CompletedProcess[st with pytest.raises(BitcoinCoreError) as failure: client._rpc("addhdkey", wallet="wallet", stdin=marker + "\n") assert marker not in str(failure.value) + + +@pytest.mark.parametrize("text", ("3f3521a6", "3F35 21A6", " 3f35\t21a6 ")) +def test_parse_fingerprint_accepts_record_spellings(text: str) -> None: + assert parse_fingerprint(text) == _FINGERPRINT + + +@pytest.mark.parametrize("text", ("", "3f3521a", "3f3521a6ff", "3f3521ag", "0x3f3521")) +def test_parse_fingerprint_rejects_other_text(text: str) -> None: + with pytest.raises(ValueError, match="8 characters"): + parse_fingerprint(text) + + +def test_identity_mismatch_stops_before_any_wallet_call(monkeypatch: pytest.MonkeyPatch) -> None: + rpc = _ImportRPC(locked=False) + monkeypatch.setattr( + BitcoinCore, + "_rpc", + lambda client, *args, wallet=None, stdin=None: rpc(client, *args, wallet=wallet, stdin=stdin), + ) + + with pytest.raises(FingerprintMismatch, match="Bitcoin Core was not changed"): + BitcoinCore("bitcoin-cli", "main", 300000).initialize( + _SEED, + lambda _prompt: "yes", + lambda _message: None, + expected_fingerprint=bytes.fromhex("3f3521a7"), + ) + assert rpc.calls == [] + + +def test_no_record_is_the_operators_choice_and_checks_nothing(monkeypatch: pytest.MonkeyPatch) -> None: + def unused(_client: BitcoinCore, _secret: MasterSeed) -> bytes: + raise AssertionError("no fingerprint is compared without a record") + + monkeypatch.setattr(BitcoinCore, "fingerprint", unused) + BitcoinCore("bitcoin-cli", "main", 300000).verify_identity(_SEED, None) + + +# Frozen from Bails' own ms32.seed_identifier for this seed: master (RIPEMD-160) and the +# June 2023 alpha (SHA-256). Bails checked three characters and kept the fourth for re-sharing. +_BAILS_SEED = bytes(range(16)) + + +@pytest.mark.parametrize( + ("identifier", "origin"), + ( + (_fingerprint_identifier(_FINGERPRINT), "codex32"), + ("d9k8", "Bails"), + ("d9kq", "Bails"), + ("hezu", "Bails alpha"), + ("test", None), + ), +) +def test_identifier_origin_names_the_rule_that_made_it(identifier: str, origin: str | None) -> None: + secret = MasterSeed.from_seed(_BAILS_SEED, identifier=identifier) + assert identifier_origin(secret, _FINGERPRINT) == origin + assert ("matches this seed" in identifier_note(origin)) is (origin is not None) + + +@pytest.mark.parametrize( + ("identifier", "expected"), + (("hezu", "Bails alpha"), ("d9k8", "Bails check unavailable")), +) +def test_identifier_origin_without_ripemd160( + monkeypatch: pytest.MonkeyPatch, identifier: str, expected: str +) -> None: + original_new = hashlib.new + + def without_ripemd160(name: str, data: bytes = b"") -> object: + if name == "ripemd160": + raise ValueError("unsupported hash type ripemd160") + return original_new(name, data) + + monkeypatch.setattr(hashlib, "new", without_ripemd160) + secret = MasterSeed.from_seed(_BAILS_SEED, identifier=identifier) + assert identifier_origin(secret, _FINGERPRINT) == expected + if expected == "Bails check unavailable": + assert "could not be checked" in identifier_note(expected) + assert "does not prove" in identifier_note(expected) + + +def test_identifier_note_allows_supported_nonderived_codex32_identifiers() -> None: + note = identifier_note(None) + assert "split shares" in note + assert "supplied seed bytes" in note + assert "explicit identifier" in note diff --git a/tests/test_cli.py b/tests/test_cli.py index 02e74c3..a008e91 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -32,6 +32,7 @@ parse_codex32, recover_secret, ) +from codex32._bitcoin_core import BitcoinCore from codex32.bech32 import _chars_to_u5, bech32_encode from codex32.checksums import _CODEX32, _CODEX32_LONG from codex32.cli import main, ms_main @@ -93,6 +94,7 @@ class _FakeBitcoinCore: imported: MasterSeed | None = None account: int | None = None timestamp: int | str | None = None + expected: bytes | None = None def fingerprint_seed(self, seed: bytes) -> bytes: return stub_fingerprint(seed) @@ -100,16 +102,22 @@ def fingerprint_seed(self, seed: bytes) -> bytes: def fingerprint(self, secret: MasterSeed) -> bytes: return self.fingerprint_seed(secret.seed_bytes) + def verify_identity(self, secret: MasterSeed, expected_fingerprint: bytes | None) -> None: + BitcoinCore.verify_identity(self, secret, expected_fingerprint) # type: ignore[arg-type] + def initialize( self, secret: MasterSeed, _ask: Callable[[str], str], _tell: Callable[[str], None], *, + expected_fingerprint: bytes | None, private: bool = True, account: int = 0, timestamp: int | str = "now", ) -> str: + self.verify_identity(secret, expected_fingerprint) + self.expected = expected_fingerprint self.imported = secret self.account, self.timestamp = account, timestamp return "test-wallet" @@ -120,6 +128,21 @@ def _offline_core(monkeypatch): monkeypatch.setattr("codex32.cli.BitcoinCore.connect", lambda *args, **kwargs: _FakeBitcoinCore()) +_RECORDED_FINGERPRINT = importlib.import_module("codex32.cli")._recorded_fingerprint +_RECORD = importlib.import_module("codex32.cli")._record +_SHOW_FINGERPRINT = importlib.import_module("codex32.cli")._show_fingerprint + + +@pytest.fixture(autouse=True) +def _matching_record(monkeypatch): + """Keep unrelated CLI tests independent of wallet-record interaction.""" + monkeypatch.setattr("codex32.cli._record", lambda: None) + monkeypatch.setattr( + "codex32.cli._recorded_fingerprint", lambda core, secret, _expected: core.fingerprint(secret) + ) + monkeypatch.setattr("codex32.cli._show_fingerprint", lambda _core, _secret, _action: None) + + def _invoke(args: list[str], *lines: str) -> _Result: stdin = io.StringIO("\n".join(lines) + "\n") stdout = io.StringIO() @@ -494,7 +517,8 @@ def answer(prompt: str, prefill: str = "") -> str: captured = capsys.readouterr() assert captured.out.strip().startswith(expected) if command[0] == "wallet": - assert "Possible correction:\n\nMaster fingerprint: 3F3521A6\n\n" in captured.err + assert "Possible correction:\n\nMaster fingerprint:" not in captured.err + assert "Master fingerprint: 3F3521A6" in captured.err else: assert "Master fingerprint:" not in captured.err assert input_module._card_text(original, False) in captured.err @@ -1815,14 +1839,13 @@ def test_correction_options_control_lengths_deadline_and_search_envelope( assert observed == damaged assert tuple(context.expected_length for context in contexts) == lengths assert (search.call_args.kwargs["deadline"] is not None) is bounded - assert search.call_args.kwargs["reduced"] == frozenset() def test_automatic_target_selection_covers_midpoints_and_supported_lengths() -> None: from codex32._cli_input import _correction_plan for observed in range(40, 136): - targets = _correction_plan(Profile.MS, None, observed, None)[0] + targets = _correction_plan(Profile.MS, None, observed, None) expected = 48 if observed <= 61 else 74 if observed <= 100 else 127 assert targets[0] == expected @@ -2021,6 +2044,7 @@ def test_wallet_commands_initialize_selected_master_seed_destinations() -> None: assert xprv.stderr.endswith("Keep it secret.\n\n") assert private.stdout == "" assert private_core.imported == parse_codex32(VECTOR_1["secret_s"]) + assert private_core.expected == private_core.fingerprint(private_core.imported) assert "Warning: This gives Bitcoin Core the master private key, which can spend funds." in private.stderr assert "Use only the intended encrypted wallet" not in private.stderr assert "\x1b[" not in private.stderr + private.stdout @@ -2094,7 +2118,7 @@ def test_wallet_private_warning_precedes_recovery_input( ) -> None: cli_module = importlib.import_module("codex32.cli") - def stop_before_input(_fingerprint=None) -> MasterSeed: + def stop_before_input(_fingerprint=None, _record=None) -> MasterSeed: assert ( "Warning: This gives Bitcoin Core the master private key, which can spend funds." in capsys.readouterr().err @@ -2172,7 +2196,7 @@ def test_production_size_budgets_are_enforced() -> None: for path in package.rglob("*.py") } - assert sum(counts.values()) < 5200, counts + assert sum(counts.values()) < 5250, counts @pytest.mark.parametrize( @@ -2544,6 +2568,45 @@ def test_final_share_preview_is_isolated_and_basis_has_no_preview(monkeypatch, c assert "Master fingerprint" not in capsys.readouterr().err +@pytest.mark.parametrize("matches", (True, False)) +def test_record_preview_and_ranking_never_show_the_fingerprint(monkeypatch, capsys, matches): + module = importlib.import_module("codex32._cli_input") + fingerprint = _FakeBitcoinCore().fingerprint + first = parse_codex32(VECTOR_2["share_A"]) + candidate = CorrectionCandidate( + parse_codex32(VECTOR_2["share_C"]), (), 1, 0, 0, None, capture_space_bits=65 + ) + right = bytes.fromhex("FAB6868A") + record = right if matches else bytes([right[0] ^ 1]) + right[1:] + monkeypatch.setattr(module.sys, "stdin", _TTYInput()) + monkeypatch.setattr(module, "_editable_input", lambda prompt: "n") + + assert not module._confirm_correction(candidate, [first], False, fingerprint, record) + shown = capsys.readouterr().err + verdict = "matches" if matches else "does not match" + assert f"Master fingerprint {verdict} your wallet record.\n\n" in shown + assert "FAB6868A" not in shown.upper() + matcher = module._fingerprint_matcher(fingerprint, record, [first]) + assert matcher(candidate) is matches + assert module._fingerprint_matcher(fingerprint, record, [])(candidate) is None + + +def test_wallet_checks_a_corrected_final_share_against_the_typed_record(monkeypatch) -> None: + cli = importlib.import_module("codex32.cli") + monkeypatch.setattr(cli, "_record", _RECORD) + monkeypatch.setattr(cli, "_recorded_fingerprint", _RECORDED_FINGERPRINT) + share_c = VECTOR_2["share_C"] + damaged = share_c[:20] + ("q" if share_c[20] != "q" else "p") + share_c[21:] + + result, core = _invoke_initialized_wallet(["wallet"], "fab6868a", VECTOR_2["share_A"], damaged, "y") + + assert result.exit_code == 0 + preview = result.stderr.split("Possible correction:", 1)[1].split("Bitcoin Core spending wallet", 1)[0] + assert "Master fingerprint matches your wallet record." in preview + assert "FAB6868A" not in preview.upper() + assert core.expected == bytes.fromhex("FAB6868A") + + def test_failed_fingerprint_never_offers_confirmation(monkeypatch, capsys): module = importlib.import_module("codex32._cli_input") from codex32.errors import CodexError @@ -2904,3 +2967,315 @@ def test_incomplete_candidate_has_no_search_warning_and_is_never_accepted_automa assert "Search incomplete" not in result.stderr assert "may not be unique" not in result.stderr assert "only a correction suggestion" in result.stderr + + +def _record_answers(monkeypatch: pytest.MonkeyPatch, *answers: str) -> list[str]: + prompts: list[str] = [] + remaining = iter(answers) + + def answer(prompt: str, **_options: object) -> str: + prompts.append(prompt) + return next(remaining) + + monkeypatch.setattr(importlib.import_module("codex32.cli"), "_text", answer) + return prompts + + +def test_restore_record_prompt_retries_until_the_library_accepts( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + core, secret = _FakeBitcoinCore(), parse_codex32(VECTOR_1["secret_s"]) + assert isinstance(secret, MasterSeed) + right = core.fingerprint(secret) + wrong = bytes([right[0] ^ 1]) + right[1:] + monkeypatch.setattr(importlib.import_module("codex32.cli"), "_record", _RECORD) + prompts = _record_answers(monkeypatch, "not hex", right.hex().upper()) + + assert _RECORDED_FINGERPRINT(core, secret, wrong) == right + assert prompts == ["Type the master fingerprint from your wallet record (Enter if none)"] * 2 + errors = capsys.readouterr().err + assert "8 characters" in errors and "does not match" in errors + assert right.hex() not in errors.lower() + + +def test_restore_without_a_record_shows_what_the_cards_say_and_asks( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + core, secret = _FakeBitcoinCore(), parse_codex32(VECTOR_1["secret_s"]) + assert isinstance(secret, MasterSeed) + fingerprint = core.fingerprint(secret) + + prompts = _record_answers(monkeypatch, "n") + interrupted = importlib.import_module("codex32.cli")._WalletSetupInterrupted + with pytest.raises(interrupted): + _RECORDED_FINGERPRINT(core, secret, None) + assert prompts == ["Restore without a wallet record? [y/N]"] + shown = capsys.readouterr().err + assert shown.count(f"Master fingerprint: {fingerprint.hex().upper()}") == 1 + assert "was not made from this seed" in shown and "nothing can prove" in shown + + prompts = _record_answers(monkeypatch, "y") + assert _RECORDED_FINGERPRINT(core, secret, None) is None + assert prompts == ["Restore without a wallet record? [y/N]"] + + derived = MasterSeed.from_seed(secret.seed_bytes, identifier=_fingerprint_identifier(fingerprint)) + _record_answers(monkeypatch, "yes") + assert _RECORDED_FINGERPRINT(core, derived, None) is None + assert "matches this seed (codex32 rule)" in capsys.readouterr().err + + +@pytest.mark.parametrize("typed", (False, True)) +def test_wallet_restore_asks_for_the_record_before_the_shares( + monkeypatch: pytest.MonkeyPatch, typed: bool +) -> None: + cli = importlib.import_module("codex32.cli") + core, secret = _FakeBitcoinCore(), parse_codex32(VECTOR_1["secret_s"]) + assert isinstance(secret, MasterSeed) + record = core.fingerprint(secret) if typed else None + seen: list[object] = [] + + def master_seed(fingerprint=None, recorded=None): + seen.append((fingerprint, recorded)) + return secret + + monkeypatch.setattr(cli.sys, "stdin", _TTYInput()) + monkeypatch.setattr(cli, "_connected_core", lambda: core) + monkeypatch.setattr(cli, "_record", lambda: seen.append("record") or record) + monkeypatch.setattr(cli, "_master_seed", master_seed) + monkeypatch.setattr( + cli, "_initialize_wallet", lambda *_args, **kwargs: seen.append(kwargs["expected_fingerprint"]) + ) + + cli._bitcoin_core(0, "now") + # Without a record the recovered fingerprint stays hidden until the recordless gate. + assert seen == ["record", (core.fingerprint if typed else None, record), record] + + +def test_create_only_requires_acknowledging_that_the_fingerprint_was_recorded( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + core, secret = _FakeBitcoinCore(), parse_codex32(VECTOR_1["secret_s"]) + assert isinstance(secret, MasterSeed) + right = core.fingerprint(secret) + prompts = _record_answers(monkeypatch, "") + + _SHOW_FINGERPRINT(core, secret, "Write it on the wallet record") + assert prompts[0] == "Write it on the wallet record, then press Enter" + shown = capsys.readouterr().err + assert f"Master fingerprint: {right.hex().upper()}" in shown + + +def test_create_initialization_does_not_authenticate_against_a_preexisting_wallet( + monkeypatch: pytest.MonkeyPatch, +) -> None: + core, secret = _FakeBitcoinCore(), parse_codex32(VECTOR_1["secret_s"]) + assert isinstance(secret, MasterSeed) + shown: list[str] = [] + monkeypatch.setattr( + "codex32.cli._show_fingerprint", + lambda _core, _secret, action: shown.append(action), + ) + + assert importlib.import_module("codex32.cli")._initialize_wallet(core, secret, confirmed=False) == 0 + assert shown == ["Write it on the wallet record"] + assert core.expected is None + + +@pytest.mark.parametrize("header", (None, "2")) +def test_create_existing_checks_the_record_before_import(monkeypatch: pytest.MonkeyPatch, header: str | None): + cli = importlib.import_module("codex32.cli") + secret = parse_codex32(VECTOR_1["secret_s"]) + core = _FakeBitcoinCore() + checked = [] + source_fingerprints = [] + emitted_fingerprints = [] + + def source(_profile, fingerprint=None, _record=None): + source_fingerprints.append(fingerprint) + return secret + + def record(_core, recovered, _expected): + assert core.imported is None + checked.append(recovered.seed_bytes) + return core.fingerprint(recovered) + + def confirm(artifact, accept=None): + if accept: + accept(artifact.text) + + def emit(_artifact, _plain, **kwargs): + emitted_fingerprints.append(kwargs.get("fingerprint")) + + monkeypatch.setattr(sys, "stdin", _TTYInput()) + monkeypatch.setattr(sys, "stdout", _TTYOutput()) + monkeypatch.setattr(cli, "_creation_source", source) + monkeypatch.setattr(cli, "_emit", emit) + monkeypatch.setattr(cli, "_confirm_card", confirm) + monkeypatch.setattr(cli, "_recorded_fingerprint", record) + monkeypatch.setattr(cli.BitcoinCore, "connect", lambda *args: core) + assert ms_main(["create", "--existing", *([header] if header else [])]) == 0 + assert source_fingerprints == [None] + assert emitted_fingerprints and all(fingerprint is None for fingerprint in emitted_fingerprints) + assert checked == [secret.seed_bytes] + assert core.expected == core.fingerprint(secret) + + +@pytest.mark.parametrize("encoding", ("hex", "codex32")) +@pytest.mark.parametrize("shared", (False, True)) +def test_create_existing_checks_record_before_card_output( + monkeypatch: pytest.MonkeyPatch, encoding: str, shared: bool +) -> None: + cli = importlib.import_module("codex32.cli") + secret = parse_codex32(VECTOR_1["secret_s"]) + assert isinstance(secret, MasterSeed) + core = _FakeBitcoinCore() + source = secret.seed_bytes.hex() if encoding == "hex" else secret.text + answers = iter((core.fingerprint(secret).hex().upper(), source)) + events: list[str] = [] + + def check_record( + selected: _FakeBitcoinCore, supplied: MasterSeed, expected: bytes | None + ) -> bytes | None: + assert events == [] + checked = _RECORDED_FINGERPRINT(selected, supplied, expected) + events.append("record") + return checked + + def confirm_card( + artifact: Share | Secret, + confirm: Callable[[str], ConfirmationResult] | None = None, + ) -> None: + if confirm is not None: + assert confirm(artifact.text).accepted + + monkeypatch.setattr(sys, "stdin", _TTYInput()) + monkeypatch.setattr(sys, "stdout", _TTYOutput()) + monkeypatch.setattr(cli, "_text", lambda _prompt, **_options: next(answers)) + monkeypatch.setattr(cli, "_record", _RECORD) + monkeypatch.setattr(cli, "_recorded_fingerprint", check_record) + monkeypatch.setattr(cli, "_emit", lambda *_args, **_kwargs: events.append("card")) + monkeypatch.setattr(cli, "_confirm_card", confirm_card) + monkeypatch.setattr(cli.BitcoinCore, "connect", lambda *args: core) + + args = ["create", "2", "--indices", "ac", "--existing"] if shared else ["create", "--existing"] + assert ms_main(args) == 0 + assert events == (["record", "card", "card"] if shared else ["record", "card"]) + assert core.expected == core.fingerprint(secret) + assert core.imported is not None and core.imported.seed_bytes == secret.seed_bytes + + +@pytest.mark.parametrize("encoding", ("hex", "codex32")) +def test_create_existing_rejects_wrong_record_before_sharing( + monkeypatch: pytest.MonkeyPatch, encoding: str +) -> None: + cli = importlib.import_module("codex32.cli") + secret = parse_codex32(VECTOR_1["secret_s"]) + assert isinstance(secret, MasterSeed) + core = _FakeBitcoinCore() + source = secret.seed_bytes.hex() if encoding == "hex" else secret.text + right = core.fingerprint(secret) + wrong = bytes([right[0] ^ 1]) + right[1:] + answers = iter((wrong.hex(), source, "", "n")) + + monkeypatch.setattr(sys, "stdin", _TTYInput()) + monkeypatch.setattr(sys, "stdout", _TTYOutput()) + monkeypatch.setattr(cli, "_text", lambda _prompt, **_options: next(answers)) + monkeypatch.setattr(cli, "_record", _RECORD) + monkeypatch.setattr(cli, "_recorded_fingerprint", _RECORDED_FINGERPRINT) + monkeypatch.setattr(cli.BitcoinCore, "connect", lambda *args: core) + with ( + patch("codex32.cli.CreationCeremony.from_secret") as split, + patch("codex32.cli._emit") as emit, + ): + assert ms_main(["create", "2", "--indices", "ac", "--existing"]) == 130 + split.assert_not_called() + emit.assert_not_called() + assert core.imported is None + + +def test_create_existing_record_gate_interruption_keeps_existing_backup_valid( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + cli = importlib.import_module("codex32.cli") + secret = parse_codex32(VECTOR_1["secret_s"]) + assert isinstance(secret, MasterSeed) + core = _FakeBitcoinCore() + + def interrupt_record(*_args: object) -> bytes | None: + raise KeyboardInterrupt + + monkeypatch.setattr(sys, "stdin", _TTYInput()) + monkeypatch.setattr(sys, "stdout", _TTYOutput()) + monkeypatch.setattr(cli, "_creation_source", lambda *_args: secret) + monkeypatch.setattr(cli, "_recorded_fingerprint", interrupt_record) + monkeypatch.setattr(cli.BitcoinCore, "connect", lambda *args: core) + with ( + patch("codex32.cli.CreationCeremony.from_secret") as split, + patch("codex32.cli._emit") as emit, + ): + assert ms_main(["create", "2", "--indices", "ac", "--existing"]) == 130 + + split.assert_not_called() + emit.assert_not_called() + message = capsys.readouterr().err + assert "recovery cards are valid" in message + assert "Mark every card" not in message + assert core.imported is None + + +def test_create_existing_interrupt_at_the_record_keeps_existing_backup_valid(monkeypatch, capsys) -> None: + cli = importlib.import_module("codex32.cli") + + def interrupt(*_args: object, **_kwargs: object) -> str: + raise KeyboardInterrupt + + monkeypatch.setattr(sys, "stdin", _TTYInput()) + monkeypatch.setattr(sys, "stdout", _TTYOutput()) + monkeypatch.setattr(cli, "_text", interrupt) + monkeypatch.setattr(cli, "_record", _RECORD) + assert ms_main(["create", "2", "--indices", "ac", "--existing"]) == 130 + message = capsys.readouterr().err + assert "recovery cards are valid" in message and "Mark every card" not in message + + +def test_create_existing_recordless_choice_precedes_sharing( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + cli = importlib.import_module("codex32.cli") + secret = parse_codex32(VECTOR_1["secret_s"]) + assert isinstance(secret, MasterSeed) + core = _FakeBitcoinCore() + answers = iter(("", secret.seed_bytes.hex(), "y")) + events: list[str] = [] + emitted: list[Share | Secret] = [] + + def confirm_card(artifact: Share | Secret, confirm=None) -> None: + if confirm is not None: + assert confirm(artifact.text).accepted + + def answer(_prompt: str, **_options: object) -> str: + assert events == [] + return next(answers) + + def emit(artifact: Share | Secret, *_args: object, **_kwargs: object) -> None: + events.append("card") + emitted.append(artifact) + + monkeypatch.setattr(sys, "stdin", _TTYInput()) + monkeypatch.setattr(sys, "stdout", _TTYOutput()) + monkeypatch.setattr(cli, "_text", answer) + monkeypatch.setattr(cli, "_record", _RECORD) + monkeypatch.setattr(cli, "_recorded_fingerprint", _RECORDED_FINGERPRINT) + monkeypatch.setattr(cli, "_emit", emit) + monkeypatch.setattr(cli, "_confirm_card", confirm_card) + monkeypatch.setattr(cli.BitcoinCore, "connect", lambda *args: core) + + assert ms_main(["create", "2", "--indices", "ac", "--existing"]) == 0 + assert events == ["card", "card"] + identifier = emitted[0].header.identifier + assert all(artifact.header.identifier == identifier for artifact in emitted) + assert core.expected is None + assert core.imported is not None and core.imported.seed_bytes == secret.seed_bytes + assert core.imported.header.identifier == identifier + assert capsys.readouterr().err.count(f"Backup identifier: {identifier.upper()}") >= 2 diff --git a/tests/test_correction_disclosure.py b/tests/test_correction_disclosure.py index 08592ee..cbbf80c 100644 --- a/tests/test_correction_disclosure.py +++ b/tests/test_correction_disclosure.py @@ -118,6 +118,8 @@ def test_declining_gate_aborts_every_flow_without_metadata(monkeypatch, answer, prompts = [] def respond(prompt, prefill=""): + if prompt.startswith("Type the master fingerprint"): + return "" # create --existing asks for the wallet record first; there is none here. prompts.append(prompt) if len(prompts) == 1: return source[:-1] + "?" diff --git a/tests/test_correction_indel.py b/tests/test_correction_indel.py index 646e8b5..2c860cf 100644 --- a/tests/test_correction_indel.py +++ b/tests/test_correction_indel.py @@ -158,7 +158,7 @@ def test_automatic_secondary_search_recovers_two_group_indels( damaged = _group_damage(source, inserted, omitted) contexts = tuple( CorrectionContext(Profile.MS, target, "ms1") - for target in _correction_plan(Profile.MS, None, len(damaged), None)[0] + for target in _correction_plan(Profile.MS, None, len(damaged), None) ) candidates, complete = _search_many( @@ -507,7 +507,6 @@ def search(state, _frontier, results, _deadline): # type: ignore[no-untyped-def contexts, damaged, primary=frozenset((48, 74, 127)), - reduced=frozenset((54, 61, 67)), ) return calls diff --git a/tools/bitcoin_core_main_smoke.py b/tools/bitcoin_core_main_smoke.py index 34775fd..6314720 100644 --- a/tools/bitcoin_core_main_smoke.py +++ b/tools/bitcoin_core_main_smoke.py @@ -115,12 +115,14 @@ def rpc(*rpc_arguments: str, wallet: str | None = None, stdin: str | None = None if not isinstance(secret, MasterSeed): raise TypeError("synthetic fixture was not a master seed") client = BitcoinCore.connect() + expected_fingerprint = client.fingerprint(secret) answers = iter(("yes",)) if ( client.initialize( secret, lambda _prompt: next(answers), lambda _message: None, + expected_fingerprint=expected_fingerprint, account=0, timestamp=0, ) diff --git a/tools/bitcoin_core_regtest.py b/tools/bitcoin_core_regtest.py index e7c8e3e..56c9742 100644 --- a/tools/bitcoin_core_regtest.py +++ b/tools/bitcoin_core_regtest.py @@ -132,12 +132,14 @@ def rpc(*rpc_arguments: str, wallet: str | None = None, stdin: str | None = None for seed, expected_fingerprint in CORE_FINGERPRINTS.items(): if client.fingerprint_seed(seed) != expected_fingerprint: raise RuntimeError("Bitcoin Core fingerprint fixture mismatch") + expected_fingerprint = CORE_FINGERPRINTS[secret.seed_bytes] answers = iter(("yes",)) if ( client.initialize( secret, lambda _prompt: next(answers), lambda _message: None, + expected_fingerprint=expected_fingerprint, account=0, timestamp=0, ) @@ -177,6 +179,7 @@ def rpc(*rpc_arguments: str, wallet: str | None = None, stdin: str | None = None secret, lambda _prompt: next(restore_answers), lambda _message: None, + expected_fingerprint=expected_fingerprint, account=0, timestamp=0, ) @@ -204,6 +207,7 @@ def rpc(*rpc_arguments: str, wallet: str | None = None, stdin: str | None = None lambda _prompt: "yes", lambda _message: None, account=0, + expected_fingerprint=expected_fingerprint, timestamp=recent_timestamp, ) != "restore_recent" diff --git a/tools/correction_benchmark.py b/tools/correction_benchmark.py index 922632d..c8d4842 100644 --- a/tools/correction_benchmark.py +++ b/tools/correction_benchmark.py @@ -108,7 +108,6 @@ def benchmark_cross_length(observed_length: int, *, unknown: bool = False) -> di contexts, damaged, primary=frozenset((48, 74, 127)), - reduced=frozenset() if unknown else frozenset((54, 61, 67)), ) seconds = perf_counter() - started if not complete or candidates: