jmwalletd.wallet_ops

jmwalletd.wallet_ops

Wallet operations bridge.

Thin adapter layer between the HTTP API and our jmwallet.WalletService. These functions handle wallet creation, opening, and recovery, returning the initialised WalletService instances that the daemon state holds.

The actual wallet implementation lives in the jmwallet package; this module only wires things together in the way the HTTP daemon needs.

Attributes

P = ParamSpec('P') module-attribute

T = TypeVar('T') module-attribute

Classes

Functions:

create_wallet(*, wallet_path: Path, password: str, wallet_type: str, data_dir: Path, bip39_passphrase: str = '') -> tuple[Any, str] async

Create a new wallet and return (wallet_service, seedphrase).

Args: wallet_path: Full path for the new .jmdat wallet file. password: Encryption password. wallet_type: One of "sw", "sw-legacy", "sw-fb". data_dir: Application data directory.

Returns: Tuple of (WalletService, seed_phrase_string).

Raises: FileExistsError: If the wallet file already exists. ValueError: If the wallet type is invalid.

Source code in jmwalletd/src/jmwalletd/wallet_ops.py

async def create_wallet(
    *,
    wallet_path: Path,
    password: str,
    wallet_type: str,
    data_dir: Path,
    bip39_passphrase: str = "",
) -> tuple[Any, str]:
    """Create a new wallet and return ``(wallet_service, seedphrase)``.

    Args:
        wallet_path: Full path for the new .jmdat wallet file.
        password: Encryption password.
        wallet_type: One of ``"sw"``, ``"sw-legacy"``, ``"sw-fb"``.
        data_dir: Application data directory.

    Returns:
        Tuple of (WalletService, seed_phrase_string).

    Raises:
        FileExistsError: If the wallet file already exists.
        ValueError: If the wallet type is invalid.
    """
    if wallet_path.exists():
        raise FileExistsError(f"Wallet file already exists: {wallet_path}")

    valid_types = {"sw", "sw-legacy", "sw-fb"}
    if wallet_type not in valid_types:
        msg = f"Invalid wallet type: {wallet_type}. Must be one of {valid_types}"
        raise ValueError(msg)

    from jmwallet.mnemonic import generate_wallet_mnemonic
    from jmwallet.wallet.service import WalletService
    from jmwalletd._backend import get_backend

    with _reserve_wallet_path(wallet_path):
        seedphrase = generate_wallet_mnemonic(strength=128)

        backend = await get_backend(
            data_dir=data_dir,
            mnemonic=seedphrase,
            network=_get_network(),
            passphrase=bip39_passphrase,
        )

        # Record current block height as the wallet birthday. Since this is a
        # brand-new wallet, it cannot have received funds before this point.
        creation_height: int | None = None
        try:
            creation_height = await backend.get_block_height()
            logger.info(f"Recording wallet creation height: {creation_height}")
        except Exception as exc:
            logger.warning(f"Could not fetch block height for wallet birthday: {exc}")

        wallet_settings = _get_wallet_settings()
        ws = WalletService(
            mnemonic=seedphrase,
            passphrase=bip39_passphrase,
            backend=backend,
            data_dir=data_dir,
            network=_get_network(),
            mixdepth_count=wallet_settings.mixdepth_count,
            gap_limit=wallet_settings.gap_limit,
            scan_range=wallet_settings.scan_range,
            max_sats_freeze_reuse=wallet_settings.max_sats_freeze_reuse,
            reconstruct_history=wallet_settings.reconstruct_history,
        )

        # Ensure the watch-only descriptor wallet is loaded in Bitcoin Core
        # and import HD descriptors. No rescan is needed for a new wallet.
        if _is_descriptor_backend(backend):
            await ws.setup_descriptor_wallet(rescan=False)

        # Initial sync to populate caches.
        await ws.sync()

        await _run_wallet_file_operation(
            _save_wallet_file,
            wallet_path=wallet_path,
            mnemonic=seedphrase,
            password=password,
            wallet_type=wallet_type,
            creation_height=creation_height,
            bip39_passphrase=bip39_passphrase,
        )

    logger.info("Created wallet")
    logger.bind(sensitive=True).info("Created wallet: {}", wallet_path.name)
    return ws, seedphrase

open_wallet(*, wallet_path: Path, password: str, data_dir: Path, sync_on_open: bool = True, bip39_passphrase: str = '') -> Any async

Open (unlock) an existing wallet file.

Returns: WalletService instance.

Raises: FileNotFoundError: If the wallet file doesn't exist. ValueError: If the password is wrong.

Source code in jmwalletd/src/jmwalletd/wallet_ops.py

async def open_wallet(
    *,
    wallet_path: Path,
    password: str,
    data_dir: Path,
    sync_on_open: bool = True,
    bip39_passphrase: str = "",
) -> Any:
    """Open (unlock) an existing wallet file.

    Returns:
        WalletService instance.

    Raises:
        FileNotFoundError: If the wallet file doesn't exist.
        ValueError: If the password is wrong.
    """
    ws, _ = await open_wallet_with_mnemonic(
        wallet_path=wallet_path,
        password=password,
        data_dir=data_dir,
        sync_on_open=sync_on_open,
        bip39_passphrase=bip39_passphrase,
    )
    return ws

open_wallet_with_mnemonic(*, wallet_path: Path, password: str, data_dir: Path, sync_on_open: bool = True, bip39_passphrase: str = '') -> tuple[Any, str] async

Open (unlock) an existing wallet file and return mnemonic.

Returns: Tuple of (WalletService, seedphrase).

Raises: FileNotFoundError: If the wallet file doesn't exist. ValueError: If the password is wrong.

Source code in jmwalletd/src/jmwalletd/wallet_ops.py

async def open_wallet_with_mnemonic(
    *,
    wallet_path: Path,
    password: str,
    data_dir: Path,
    sync_on_open: bool = True,
    bip39_passphrase: str = "",
) -> tuple[Any, str]:
    """Open (unlock) an existing wallet file and return mnemonic.

    Returns:
        Tuple of (WalletService, seedphrase).

    Raises:
        FileNotFoundError: If the wallet file doesn't exist.
        ValueError: If the password is wrong.
    """
    from jmwallet.wallet.service import WalletService
    from jmwalletd._backend import get_backend

    if not wallet_path.exists():
        raise FileNotFoundError(f"Wallet file not found: {wallet_path}")

    seedphrase, creation_height = await _run_wallet_file_operation(
        _load_wallet_file,
        wallet_path=wallet_path,
        password=password,
        **({"bip39_passphrase": bip39_passphrase} if bip39_passphrase else {}),
    )

    # Legacy or manually migrated wallet data may contain a phrase with an
    # invalid BIP39 checksum. Such a phrase still derives a wallet and may
    # already hold funds, so keep it accessible but report the problem at
    # ERROR so warning-filtering log configurations cannot hide it.
    if not validate_bip39_checksum(seedphrase):
        logger.error(
            "Wallet {} contains a mnemonic with an INVALID BIP39 checksum. "
            "This usually means a word is misspelled, missing, or in the wrong "
            "order, and the derived wallet is NOT the one the words were "
            "intended to encode. Do not deposit funds until the seed phrase "
            "has been verified.",
            wallet_path.name,
        )

    backend = await get_backend(
        data_dir=data_dir,
        mnemonic=seedphrase,
        network=_get_network(),
        passphrase=bip39_passphrase,
    )

    # Propagate wallet creation height hint to the backend.  Passing None
    # clears any stale hint from a previously opened wallet when backend
    # instances are reused.
    backend.set_wallet_creation_height(creation_height)

    wallet_settings = _get_wallet_settings()
    ws = WalletService(
        mnemonic=seedphrase,
        passphrase=bip39_passphrase,
        backend=backend,
        data_dir=data_dir,
        network=_get_network(),
        mixdepth_count=wallet_settings.mixdepth_count,
        gap_limit=wallet_settings.gap_limit,
        scan_range=wallet_settings.scan_range,
        max_sats_freeze_reuse=wallet_settings.max_sats_freeze_reuse,
        reconstruct_history=wallet_settings.reconstruct_history,
    )

    # Ensure the watch-only descriptor wallet is loaded in Bitcoin Core
    # and import HD descriptors.  Idempotent — skips if already set up.
    # Skipped for non-descriptor backends (e.g. neutrino).
    if _is_descriptor_backend(backend):
        await ws.setup_descriptor_wallet(
            smart_scan=wallet_settings.smart_scan,
            background_full_rescan=wallet_settings.background_full_rescan,
        )

    if sync_on_open:
        # Bond-aware sync so funded fidelity bonds from the registry are
        # imported and surfaced in /utxos and /display.
        await ws.sync_with_registered_bonds()

    logger.info("Opened wallet")
    logger.bind(sensitive=True).info("Opened wallet: {}", wallet_path.name)
    return ws, seedphrase

recover_wallet(*, wallet_path: Path, password: str, wallet_type: str, seedphrase: str, data_dir: Path, scan_range: int | None = None, bip39_passphrase: str = '') -> Any async

Recover a wallet from a BIP39 seed phrase.

Returns: WalletService instance.

Raises: FileExistsError: If the wallet file already exists. ValueError: If the seed phrase or wallet type is invalid.

Source code in jmwalletd/src/jmwalletd/wallet_ops.py

async def recover_wallet(
    *,
    wallet_path: Path,
    password: str,
    wallet_type: str,
    seedphrase: str,
    data_dir: Path,
    scan_range: int | None = None,
    bip39_passphrase: str = "",
) -> Any:
    """Recover a wallet from a BIP39 seed phrase.

    Returns:
        WalletService instance.

    Raises:
        FileExistsError: If the wallet file already exists.
        ValueError: If the seed phrase or wallet type is invalid.
    """
    if wallet_path.exists():
        raise FileExistsError(f"Wallet file already exists: {wallet_path}")

    if not validate_bip39_checksum(seedphrase):
        msg = "Invalid BIP39 mnemonic seed phrase."
        raise ValueError(msg)

    valid_types = {"sw", "sw-legacy", "sw-fb"}
    if wallet_type not in valid_types:
        msg = f"Invalid wallet type: {wallet_type}. Must be one of {valid_types}"
        raise ValueError(msg)

    with _reserve_wallet_path(wallet_path):
        return await _recover_reserved_wallet(
            wallet_path=wallet_path,
            password=password,
            wallet_type=wallet_type,
            seedphrase=seedphrase,
            data_dir=data_dir,
            scan_range=scan_range,
            bip39_passphrase=bip39_passphrase,
        )

verify_wallet_password(*, wallet_path: Path, password: str, bip39_passphrase: str = '') -> None async

Verify a wallet password without initializing a backend service.

This lets a cross-wallet unlock authenticate its target before it tears down the currently active backend and its associated daemon state.

Source code in jmwalletd/src/jmwalletd/wallet_ops.py

async def verify_wallet_password(
    *, wallet_path: Path, password: str, bip39_passphrase: str = ""
) -> None:
    """Verify a wallet password without initializing a backend service.

    This lets a cross-wallet unlock authenticate its target before it tears
    down the currently active backend and its associated daemon state.
    """
    if not wallet_path.exists():
        raise FileNotFoundError(f"Wallet file not found: {wallet_path}")
    await _run_wallet_file_operation(
        _load_wallet_file,
        wallet_path=wallet_path,
        password=password,
        **({"bip39_passphrase": bip39_passphrase} if bip39_passphrase else {}),
    )