jmwalletd.routers.wallet_data

jmwalletd.routers.wallet_data

Wallet data endpoints: display, UTXOs, addresses, seed, freeze, config, rescan, sign.

Attributes

router = APIRouter() module-attribute

Classes

Functions:

config_get(walletname: str, body: ConfigGetRequest, _auth: dict[str, Any] = Depends(require_auth), _wallet: None = Depends(require_wallet_match), state: DaemonState = Depends(get_daemon_state)) -> ConfigGetResponse async

Read a config variable (in-memory override takes priority).

Source code in jmwalletd/src/jmwalletd/routers/wallet_data.py

@router.post("/wallet/{walletname}/configget", operation_id="configget")
async def config_get(
    walletname: str,
    body: ConfigGetRequest,
    _auth: dict[str, Any] = Depends(require_auth),
    _wallet: None = Depends(require_wallet_match),
    state: DaemonState = Depends(get_daemon_state),
) -> ConfigGetResponse:
    """Read a config variable (in-memory override takes priority)."""
    section = body.section.upper()
    field_name = body.field.lower()

    # Check in-memory overrides first.
    if section in state.config_overrides and field_name in state.config_overrides[section]:
        return ConfigGetResponse(configvalue=state.config_overrides[section][field_name])

    # Fall back to the settings system.
    from jmcore.settings import get_settings

    try:
        settings = get_settings()

        # Special handling for POLICY fields that JAM requests.
        if section == "POLICY":
            if field_name in _POLICY_DEFAULTS:
                return ConfigGetResponse(configvalue=_POLICY_DEFAULTS[field_name])

            if field_name in _POLICY_FIELD_MAP:
                attr, real_field = _POLICY_FIELD_MAP[field_name]
                subsettings = getattr(settings, attr)
                value = getattr(subsettings, real_field)
                return ConfigGetResponse(configvalue=str(value))

        value = _get_setting_value(settings, section, field_name)
        return ConfigGetResponse(configvalue=str(value))
    except (AttributeError, KeyError) as exc:
        raise ConfigNotPresent(f"Config not found: [{section}] {field_name}") from exc

config_set(walletname: str, body: ConfigSetRequest, _auth: dict[str, Any] = Depends(require_auth), _wallet: None = Depends(require_wallet_match), state: DaemonState = Depends(get_daemon_state)) -> dict[str, str] async

Set a config variable (in-memory only, not persisted).

Source code in jmwalletd/src/jmwalletd/routers/wallet_data.py

@router.post("/wallet/{walletname}/configset", operation_id="configsetting")
async def config_set(
    walletname: str,
    body: ConfigSetRequest,
    _auth: dict[str, Any] = Depends(require_auth),
    _wallet: None = Depends(require_wallet_match),
    state: DaemonState = Depends(get_daemon_state),
) -> dict[str, str]:
    """Set a config variable (in-memory only, not persisted)."""
    section = body.section.upper()
    field_name = body.field.lower()

    if section not in state.config_overrides:
        state.config_overrides[section] = {}

    state.config_overrides[section][field_name] = body.value
    logger.info("Config override set: [{}] {} = {}", section, field_name, body.value)
    return {}

freeze_utxo(walletname: str, body: FreezeRequest, _auth: dict[str, Any] = Depends(require_auth), _wallet: None = Depends(require_wallet_match), state: DaemonState = Depends(get_daemon_state)) -> dict[str, str] async

Freeze or unfreeze a UTXO.

Source code in jmwalletd/src/jmwalletd/routers/wallet_data.py

@router.post("/wallet/{walletname}/freeze", operation_id="freeze")
async def freeze_utxo(
    walletname: str,
    body: FreezeRequest,
    _auth: dict[str, Any] = Depends(require_auth),
    _wallet: None = Depends(require_wallet_match),
    state: DaemonState = Depends(get_daemon_state),
) -> dict[str, str]:
    """Freeze or unfreeze a UTXO."""
    if state.taker_running:
        raise ActionNotAllowed("Cannot freeze/unfreeze UTXOs while a coinjoin is in progress.")

    # Validate utxo format: "txid:vout"
    utxo_str = body.utxo_string
    parts = utxo_str.split(":")
    if len(parts) != 2:
        raise InvalidRequestFormat(f"Invalid UTXO format: {utxo_str}. Expected txid:vout.")

    try:
        int(parts[1])
    except ValueError as exc:
        raise InvalidRequestFormat(f"Invalid vout in UTXO: {utxo_str}") from exc

    ws = state.wallet_service
    if body.freeze:
        ws.freeze_utxo(utxo_str)
    else:
        ws.unfreeze_utxo(utxo_str)

    return {}

get_new_address(walletname: str, mixdepth: str, _auth: dict[str, Any] = Depends(require_auth), _wallet: None = Depends(require_wallet_match), state: DaemonState = Depends(get_daemon_state)) -> GetAddressResponse async

Get a new receive address for the specified mixdepth.

Source code in jmwalletd/src/jmwalletd/routers/wallet_data.py

@router.get("/wallet/{walletname}/address/new/{mixdepth}", operation_id="getaddress")
async def get_new_address(
    walletname: str,
    mixdepth: str,
    _auth: dict[str, Any] = Depends(require_auth),
    _wallet: None = Depends(require_wallet_match),
    state: DaemonState = Depends(get_daemon_state),
) -> GetAddressResponse:
    """Get a new receive address for the specified mixdepth."""
    try:
        md = int(mixdepth)
    except ValueError as exc:
        raise InvalidRequestFormat(f"Invalid mixdepth: {mixdepth}") from exc

    ws = state.wallet_service

    if md < 0 or md >= ws.mixdepth_count:
        raise InvalidRequestFormat(f"Mixdepth {md} out of range (0-{ws.mixdepth_count - 1})")

    address = await ws.get_new_address_verified(md)
    return GetAddressResponse(address=address)

get_rescan_info(walletname: str, _auth: dict[str, Any] = Depends(require_auth), _wallet: None = Depends(require_wallet_match), state: DaemonState = Depends(get_daemon_state)) -> RescanInfoResponse async

Get rescan progress information (Bitcoin Core is the source of truth).

Source code in jmwalletd/src/jmwalletd/routers/wallet_data.py

@router.get("/wallet/{walletname}/getrescaninfo", operation_id="getrescaninfo")
async def get_rescan_info(
    walletname: str,
    _auth: dict[str, Any] = Depends(require_auth),
    _wallet: None = Depends(require_wallet_match),
    state: DaemonState = Depends(get_daemon_state),
) -> RescanInfoResponse:
    """Get rescan progress information (Bitcoin Core is the source of truth)."""
    rescanning, progress = await state.live_rescan_status()
    if rescanning:
        return RescanInfoResponse(rescanning=True, progress=progress)
    return RescanInfoResponse(rescanning=False, error=state.rescan_error)

get_seed(walletname: str, _auth: dict[str, Any] = Depends(require_auth), _wallet: None = Depends(require_wallet_match), state: DaemonState = Depends(get_daemon_state)) -> GetSeedResponse async

Return the wallet's BIP39 mnemonic seed phrase.

Source code in jmwalletd/src/jmwalletd/routers/wallet_data.py

@router.get("/wallet/{walletname}/getseed", operation_id="getseed")
async def get_seed(
    walletname: str,
    _auth: dict[str, Any] = Depends(require_auth),
    _wallet: None = Depends(require_wallet_match),
    state: DaemonState = Depends(get_daemon_state),
) -> GetSeedResponse:
    """Return the wallet's BIP39 mnemonic seed phrase."""
    if not state.wallet_mnemonic:
        raise ActionNotAllowed("Seed phrase is not available in daemon state.")
    return GetSeedResponse(seedphrase=state.wallet_mnemonic)

get_timelock_address(walletname: str, lockdate: str, _auth: dict[str, Any] = Depends(require_auth), _wallet: None = Depends(require_wallet_match), state: DaemonState = Depends(get_daemon_state)) -> GetAddressResponse async

Get a new timelocked (fidelity bond) address.

The lockdate should be in YYYY-mm format (e.g., "2025-06").

Source code in jmwalletd/src/jmwalletd/routers/wallet_data.py

@router.get(
    "/wallet/{walletname}/address/timelock/new/{lockdate}", operation_id="gettimelockaddress"
)
async def get_timelock_address(
    walletname: str,
    lockdate: str,
    _auth: dict[str, Any] = Depends(require_auth),
    _wallet: None = Depends(require_wallet_match),
    state: DaemonState = Depends(get_daemon_state),
) -> GetAddressResponse:
    """Get a new timelocked (fidelity bond) address.

    The lockdate should be in YYYY-mm format (e.g., "2025-06").
    """
    try:
        parts = lockdate.split("-")
        if len(parts) != 2:
            msg = "Expected YYYY-mm"
            raise ValueError(msg)
        year, month = int(parts[0]), int(parts[1])
        if month < 1 or month > 12:
            msg = "Month must be 1-12"
            raise ValueError(msg)
    except (ValueError, IndexError) as exc:
        msg = f"Invalid lockdate format: {lockdate}. Expected YYYY-mm."
        raise InvalidRequestFormat(msg) from exc

    ws = state.wallet_service
    # Convert lockdate to locktime (first day of the month as UNIX timestamp).
    dt = datetime.datetime(year, month, 1, tzinfo=datetime.UTC)
    locktime = calendar.timegm(dt.timetuple())

    # Each locktime maps to exactly one timenumber (BIP32 child index).
    # The reference JoinMarket implementation does not support multiple
    # addresses per locktime.
    from jmcore.timenumber import timestamp_to_timenumber

    timenumber = timestamp_to_timenumber(locktime)

    address = ws.get_fidelity_bond_address(timenumber, locktime)

    # Persist the bond to the registry so the maker can pick it up at startup.
    # Disable the legacy fallback so foreign bonds are never copied into this
    # wallet's per-wallet file on save (#492).
    registry = load_registry(state.data_dir, ws.wallet_fingerprint, allow_legacy_fallback=False)
    if not registry.get_bond_by_address(address):
        pubkey_hex = (
            ws.get_fidelity_bond_key(timenumber, locktime)
            .get_public_key_bytes(compressed=True)
            .hex()
        )
        witness_script = ws.get_fidelity_bond_script(timenumber, locktime)
        path = f"{ws.root_path}/0'/{FIDELITY_BOND_BRANCH}/{timenumber}"
        bond_info = create_bond_info(
            address=address,
            locktime=locktime,
            index=timenumber,
            path=path,
            pubkey_hex=pubkey_hex,
            witness_script=witness_script,
            network=ws.network,
        )
        registry.add_bond(bond_info)
        save_registry(registry, state.data_dir, ws.wallet_fingerprint)
        logger.debug(
            "Registered fidelity bond address {} (timenumber={}, locktime={}) in registry",
            address,
            timenumber,
            locktime,
        )

    return GetAddressResponse(address=address)

list_utxos(walletname: str, _auth: dict[str, Any] = Depends(require_auth), _wallet: None = Depends(require_wallet_match), state: DaemonState = Depends(get_daemon_state)) -> ListUtxosResponse async

List all UTXOs in the wallet.

Source code in jmwalletd/src/jmwalletd/routers/wallet_data.py

@router.get("/wallet/{walletname}/utxos", operation_id="listutxos")
async def list_utxos(
    walletname: str,
    _auth: dict[str, Any] = Depends(require_auth),
    _wallet: None = Depends(require_wallet_match),
    state: DaemonState = Depends(get_daemon_state),
) -> ListUtxosResponse:
    """List all UTXOs in the wallet."""
    ws = state.wallet_service
    rescanning, _progress = await state.live_rescan_status()
    if not rescanning:
        await ws.sync_with_registered_bonds()
    utxo_entries: list[UTXOEntry] = []

    for mixdepth in range(ws.mixdepth_count):
        utxos = ws.utxo_cache.get(mixdepth, [])
        for u in utxos:
            utxo_entries.append(
                UTXOEntry(
                    utxo=u.outpoint,
                    address=u.address,
                    path=u.path,
                    label=u.label or "",
                    value=u.value,
                    tries=0,
                    tries_remaining=3,
                    external=False,
                    mixdepth=u.mixdepth,
                    confirmations=u.confirmations,
                    frozen=u.frozen,
                    locktime=_format_bond_locktime(u.locktime),
                )
            )

    return ListUtxosResponse(utxos=utxo_entries)

rescan_blockchain(walletname: str, blockheight: int, _auth: dict[str, Any] = Depends(require_auth), _wallet: None = Depends(require_wallet_match), state: DaemonState = Depends(get_daemon_state)) -> RescanBlockchainResponse async

Trigger a blockchain rescan from the given block height.

Source code in jmwalletd/src/jmwalletd/routers/wallet_data.py

@router.get("/wallet/{walletname}/rescanblockchain/{blockheight}", operation_id="rescanblockchain")
async def rescan_blockchain(
    walletname: str,
    blockheight: int,
    _auth: dict[str, Any] = Depends(require_auth),
    _wallet: None = Depends(require_wallet_match),
    state: DaemonState = Depends(get_daemon_state),
) -> RescanBlockchainResponse:
    """Trigger a blockchain rescan from the given block height."""
    backend = state.wallet_service.backend

    # Rescan support is only available on DescriptorWalletBackend.
    if not (hasattr(backend, "start_background_rescan") or hasattr(backend, "rescan_blockchain")):
        raise ActionNotAllowed("Rescan not supported by the current backend.")

    if state._rescan_task is not None and not state._rescan_task.done():
        logger.info("Rescan tracking task already active; returning existing operation")
        return RescanBlockchainResponse(walletname=walletname)

    state.rescanning = True
    state.rescan_progress = 0.0
    state.rescan_error = None

    # Keep a reference so the task is not garbage-collected mid-flight and
    # can be cancelled on wallet lock.
    state._rescan_task = asyncio.create_task(_run_rescan(state, blockheight))
    return RescanBlockchainResponse(walletname=walletname)

sign_message(walletname: str, body: SignMessageRequest, _auth: dict[str, Any] = Depends(require_auth), _wallet: None = Depends(require_wallet_match), state: DaemonState = Depends(get_daemon_state)) -> SignMessageResponse async

Sign a message with a wallet key at the given HD path.

Source code in jmwalletd/src/jmwalletd/routers/wallet_data.py

@router.post("/wallet/{walletname}/signmessage", operation_id="signmessage")
async def sign_message(
    walletname: str,
    body: SignMessageRequest,
    _auth: dict[str, Any] = Depends(require_auth),
    _wallet: None = Depends(require_wallet_match),
    state: DaemonState = Depends(get_daemon_state),
) -> SignMessageResponse:
    """Sign a message with a wallet key at the given HD path."""
    ws = state.wallet_service

    try:
        # Parse the HD path to derive the address, then get the key.
        # The hd_path format is like "m/84'/0'/0'/0/5".
        # We look up the address from the path components.
        parts = body.hd_path.replace("'", "").split("/")
        if len(parts) < 5:
            raise InvalidRequestFormat(f"Invalid HD path: {body.hd_path}")

        # Extract mixdepth, change, index from the path.
        mixdepth = int(parts[3])
        change = int(parts[4]) if len(parts) > 4 else 0
        index = int(parts[5]) if len(parts) > 5 else 0

        address = ws.get_address(mixdepth, change, index)
        key = ws.get_key_for_address(address)
        if key is None:
            raise InvalidRequestFormat(f"Cannot derive key for path: {body.hd_path}")

        # Sign the message using Bitcoin message format (via coincurve).
        from jmcore.crypto import bitcoin_message_hash

        msg_hash = bitcoin_message_hash(body.message)

        from coincurve import PrivateKey

        privkey = PrivateKey(key.private_key)
        sig = privkey.sign_recoverable(msg_hash, hasher=None)
        import base64

        signature = base64.b64encode(sig).decode()

        return SignMessageResponse(
            signature=signature,
            message=body.message,
            address=address,
        )
    except (InvalidRequestFormat, ValueError) as exc:
        raise InvalidRequestFormat(f"Signing failed: {exc}") from exc
    except Exception as exc:
        raise InvalidRequestFormat(f"Signing failed: {exc}") from exc

wallet_display(walletname: str, _auth: dict[str, Any] = Depends(require_auth), _wallet: None = Depends(require_wallet_match), state: DaemonState = Depends(get_daemon_state)) -> WalletDisplayResponse async

Return full wallet display with accounts, branches, and entries.

Source code in jmwalletd/src/jmwalletd/routers/wallet_data.py

@router.get("/wallet/{walletname}/display", operation_id="displaywallet")
async def wallet_display(
    walletname: str,
    _auth: dict[str, Any] = Depends(require_auth),
    _wallet: None = Depends(require_wallet_match),
    state: DaemonState = Depends(get_daemon_state),
) -> WalletDisplayResponse:
    """Return full wallet display with accounts, branches, and entries."""
    ws = state.wallet_service
    rescanning, _progress = await state.live_rescan_status()
    if not rescanning:
        await ws.sync_with_registered_bonds()

    # Load history data so address statuses (cj-out, change, etc.) are
    # classified correctly.  Without this, all funded internal addresses
    # fall through to the "non-cj-change" default.
    used_addresses = get_used_addresses(state.data_dir, wallet_fingerprint=ws.wallet_fingerprint)
    history_addresses = get_address_history_types(
        state.data_dir, wallet_fingerprint=ws.wallet_fingerprint
    )

    accounts: list[WalletDisplayAccount] = []
    total_balance = 0
    total_available = 0

    for mixdepth in range(ws.mixdepth_count):
        balance = await ws.get_balance(mixdepth)
        total_balance += balance

        # Build external and internal branches.
        branches: list[WalletDisplayBranch] = []
        branch_defs = [(0, "external addresses\tm/84'"), (1, "internal addresses\tm/84'")]
        for change, branch_label in branch_defs:
            address_infos = ws.get_address_info_for_mixdepth(
                mixdepth,
                change,
                used_addresses=used_addresses,
                history_addresses=history_addresses,
            )
            branch_balance = sum(ai.balance for ai in address_infos)

            entries: list[WalletDisplayEntry] = []
            for ai in address_infos:
                entries.append(
                    WalletDisplayEntry(
                        hd_path=ai.path,
                        address=ai.address,
                        amount=f"{ai.balance / 1e8:.8f}",
                        available_balance=f"{ai.balance / 1e8:.8f}",
                        status=ai.status,
                        label="",
                        extradata="",
                    )
                )

            branches.append(
                WalletDisplayBranch(
                    branch=f"{branch_label}/{mixdepth}'/{change}",
                    balance=f"{branch_balance / 1e8:.8f}",
                    available_balance=f"{branch_balance / 1e8:.8f}",
                    entries=entries,
                )
            )

        account_balance = balance
        total_available += account_balance

        accounts.append(
            WalletDisplayAccount(
                account=str(mixdepth),
                account_balance=f"{account_balance / 1e8:.8f}",
                available_balance=f"{account_balance / 1e8:.8f}",
                branches=branches,
            )
        )

    return WalletDisplayResponse(
        walletname=walletname,
        walletinfo=WalletInfo(
            wallet_name="JM wallet",
            total_balance=f"{total_balance / 1e8:.8f}",
            available_balance=f"{total_available / 1e8:.8f}",
            accounts=accounts,
        ),
    )

wallet_history(walletname: str, limit: int | None = None, _auth: dict[str, Any] = Depends(require_auth), _wallet: None = Depends(require_wallet_match), state: DaemonState = Depends(get_daemon_state)) -> WalletHistoryResponse async

Return the active wallet's CoinJoin/spend history (from history.csv).

Entries are scoped to the active wallet's master fingerprint and returned most-recent first. Pass ?limit=N to cap the number of entries. This is the same data the jm-wallet history CLI reads; it is read-only.

Source code in jmwalletd/src/jmwalletd/routers/wallet_data.py

@router.get("/wallet/{walletname}/history", operation_id="wallethistory")
async def wallet_history(
    walletname: str,
    limit: int | None = None,
    _auth: dict[str, Any] = Depends(require_auth),
    _wallet: None = Depends(require_wallet_match),
    state: DaemonState = Depends(get_daemon_state),
) -> WalletHistoryResponse:
    """Return the active wallet's CoinJoin/spend history (from history.csv).

    Entries are scoped to the active wallet's master fingerprint and returned
    most-recent first. Pass ``?limit=N`` to cap the number of entries. This is
    the same data the ``jm-wallet history`` CLI reads; it is read-only.
    """
    from jmwallet.history import read_history

    ws = state.wallet_service

    entries = read_history(
        state.data_dir,
        limit=limit,
        wallet_fingerprint=ws.wallet_fingerprint,
    )

    history = [
        HistoryEntry(
            timestamp=e.timestamp,
            completed_at=e.completed_at,
            confirmed_at=e.confirmed_at,
            role=e.role,
            success=e.success,
            failure_reason=e.failure_reason,
            confirmations=e.confirmations,
            txid=e.txid,
            cj_amount=e.cj_amount,
            peer_count=e.peer_count,
            counterparty_nicks=e.counterparty_nicks,
            fee_received=e.fee_received,
            txfee_contribution=e.txfee_contribution,
            total_maker_fees_paid=e.total_maker_fees_paid,
            mining_fee_paid=e.mining_fee_paid,
            net_fee=e.net_fee,
            source_mixdepth=e.source_mixdepth,
            destination_address=e.destination_address,
            change_address=e.change_address,
            utxos_used=e.utxos_used,
            broadcast_method=e.broadcast_method,
            network=e.network,
            source=e.source,
        )
        for e in entries
    ]
    return WalletHistoryResponse(history=history)

yieldgen_report(_auth: dict[str, Any] = Depends(require_auth), state: DaemonState = Depends(get_daemon_state)) -> YieldGenReportResponse async

Return the yield generator (maker earnings) report.

The joinmarket-ng maker logs its activity to history.csv (its single source of truth) rather than the legacy yigen-statement.csv file, so the report is synthesized from the successful maker rows of history.csv in the reference comma-separated format the Earn report UI expects (a header row followed by one row per CoinJoin).

Source code in jmwalletd/src/jmwalletd/routers/wallet_data.py

@router.get("/wallet/yieldgen/report", operation_id="yieldgenreport")
async def yieldgen_report(
    _auth: dict[str, Any] = Depends(require_auth),
    state: DaemonState = Depends(get_daemon_state),
) -> YieldGenReportResponse:
    """Return the yield generator (maker earnings) report.

    The joinmarket-ng maker logs its activity to ``history.csv`` (its single
    source of truth) rather than the legacy ``yigen-statement.csv`` file, so the
    report is synthesized from the successful maker rows of ``history.csv`` in
    the reference comma-separated format the Earn report UI expects (a header
    row followed by one row per CoinJoin).
    """
    from jmwallet.history import format_yield_generator_report

    if state.wallet_service is None:
        raise NoWalletFound("No wallet loaded.")

    try:
        rows = format_yield_generator_report(
            state.data_dir,
            wallet_fingerprint=state.wallet_service.wallet_fingerprint,
        )
    except Exception as exc:
        raise YieldGeneratorDataUnreadable(str(exc)) from exc

    return YieldGenReportResponse(yigen_data=rows)