Common CLI components for JoinMarket NG.
This module provides reusable CLI helper functions to reduce duplication
across jmwallet, maker, and taker CLIs.
Architecture:
- Resolver functions: Take CLI args + settings and return resolved values
- Setup functions: Common initialization (logging, settings, etc.)
- Mnemonic loading: Unified mnemonic resolution from multiple sources
The CLI parameter definitions remain in each CLI module for now, but the
resolution logic is centralized here. This approach:
- Avoids typer dependency in jmcore
- Allows each CLI to customize parameter names/help text if needed
- Centralizes the complex resolution logic that was duplicated
Usage:
from jmcore.cli_common import (
resolve_backend_settings,
resolve_mnemonic,
setup_cli,
)
@app.command()
def my_command(
network: Annotated[str | None, typer.Option("--network")] = None,
rpc_url: Annotated[str | None, typer.Option("--rpc-url")] = None,
...
):
settings = setup_cli(log_level, data_dir=data_dir)
backend = resolve_backend_settings(settings, network=network, rpc_url=rpc_url, ...)
Attributes
Classes
MnemonicSource
dataclass
Source provenance shared by public identity lookup and secret resolution.
Source code in jmcore/src/jmcore/cli_common.py
110
111
112
113
114
115
116 | @dataclass(frozen=True)
class MnemonicSource:
"""Source provenance shared by public identity lookup and secret resolution."""
mnemonic: str | None
path: Path | None
description: str
|
Attributes
description: str
instance-attribute
mnemonic: str | None
instance-attribute
path: Path | None
instance-attribute
ResolvedBackendSettings
dataclass
Resolved backend settings ready for use.
Source code in jmcore/src/jmcore/cli_common.py
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92 | @dataclass
class ResolvedBackendSettings:
"""Resolved backend settings ready for use."""
network: str
bitcoin_network: str
backend_type: str
rpc_url: str
rpc_user: str
rpc_password: str
neutrino_url: str
neutrino_add_peers: list[str]
data_dir: Path
scan_start_height: int | None = None
scan_lookback_blocks: int = 52_560
neutrino_tls_cert: str | None = None
neutrino_auth_token: str | None = None
fee_estimate_url: str | None = None
fee_estimate_proxy: str | None = None
|
Attributes
backend_type: str
instance-attribute
bitcoin_network: str
instance-attribute
data_dir: Path
instance-attribute
fee_estimate_proxy: str | None = None
class-attribute
instance-attribute
fee_estimate_url: str | None = None
class-attribute
instance-attribute
network: str
instance-attribute
neutrino_add_peers: list[str]
instance-attribute
neutrino_auth_token: str | None = None
class-attribute
instance-attribute
neutrino_tls_cert: str | None = None
class-attribute
instance-attribute
neutrino_url: str
instance-attribute
rpc_password: str
instance-attribute
rpc_url: str
instance-attribute
rpc_user: str
instance-attribute
scan_lookback_blocks: int = 52560
class-attribute
instance-attribute
scan_start_height: int | None = None
class-attribute
instance-attribute
ResolvedMnemonic
dataclass
Resolved mnemonic and BIP39 passphrase.
Note: bip39_passphrase is the optional BIP39 passphrase (13th/25th word),
NOT the password used to decrypt an encrypted mnemonic file.
Source code in jmcore/src/jmcore/cli_common.py
95
96
97
98
99
100
101
102
103
104
105
106
107 | @dataclass
class ResolvedMnemonic:
"""Resolved mnemonic and BIP39 passphrase.
Note: bip39_passphrase is the optional BIP39 passphrase (13th/25th word),
NOT the password used to decrypt an encrypted mnemonic file.
"""
mnemonic: str
bip39_passphrase: str
source: str # Where the mnemonic came from (for logging)
creation_height: int | None = None # Block height at wallet creation time
mnemonic_file: Path | None = None # File backing the mnemonic, when present
|
Attributes
bip39_passphrase: str
instance-attribute
creation_height: int | None = None
class-attribute
instance-attribute
mnemonic: str
instance-attribute
mnemonic_file: Path | None = None
class-attribute
instance-attribute
source: str
instance-attribute
Bases: HelpFormatter
Argparse help formatter that lists options and subcommands alphabetically.
- Optional arguments are sorted by their first long name (fallback: first
short name), case-insensitively, both in the options list and in the
usage synopsis.
- Positional arguments keep their declared order (it defines call syntax)
and stay ahead of the sorted options within mixed groups.
- Subcommand listings (from
add_subparsers) are sorted by name.
Typer-based CLIs get the same behavior from :mod:jmcore.cli_help.
Source code in jmcore/src/jmcore/cli_common.py
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231 | class SortedHelpFormatter(argparse.HelpFormatter):
"""Argparse help formatter that lists options and subcommands alphabetically.
- Optional arguments are sorted by their first long name (fallback: first
short name), case-insensitively, both in the options list and in the
usage synopsis.
- Positional arguments keep their declared order (it defines call syntax)
and stay ahead of the sorted options within mixed groups.
- Subcommand listings (from ``add_subparsers``) are sorted by name.
Typer-based CLIs get the same behavior from :mod:`jmcore.cli_help`.
"""
@staticmethod
def _sort_key(action: argparse.Action) -> tuple[int, str]:
if not action.option_strings:
# Stable sort keeps positionals in declaration order, first.
return (0, "")
long_opts = [opt for opt in action.option_strings if opt.startswith("--")]
name = long_opts[0] if long_opts else action.option_strings[0]
return (1, name.lstrip("-").casefold())
def add_usage(
self,
usage: str | None,
actions: Iterable[argparse.Action],
groups: Iterable[Any],
prefix: str | None = None,
) -> None:
super().add_usage(usage, sorted(actions, key=self._sort_key), groups, prefix)
def add_arguments(self, actions: Iterable[argparse.Action]) -> None:
super().add_arguments(sorted(actions, key=self._sort_key))
def _metavar_formatter(
self, action: argparse.Action, default_metavar: str
) -> Callable[[int], tuple[str, ...]]:
if (
isinstance(action, argparse._SubParsersAction) # noqa: SLF001
and action.metavar is None
and action.choices is not None
):
# Sort the "{cmd1,cmd2,...}" usage metavar. Regular choices (e.g.
# log levels) keep their declared, often meaningful, order.
choices = ",".join(sorted(str(choice) for choice in action.choices))
result = f"{{{choices}}}"
return lambda tuple_size: (result,) * tuple_size
return super()._metavar_formatter(action, default_metavar)
def _iter_indented_subactions(
self, action: argparse.Action
) -> Generator[argparse.Action, None, None]:
try:
get_subactions = action._get_subactions # type: ignore[attr-defined]
except AttributeError:
return
self._indent()
yield from sorted(get_subactions(), key=lambda subaction: subaction.dest)
self._dedent()
|
Source code in jmcore/src/jmcore/cli_common.py
| def add_arguments(self, actions: Iterable[argparse.Action]) -> None:
super().add_arguments(sorted(actions, key=self._sort_key))
|
Source code in jmcore/src/jmcore/cli_common.py
195
196
197
198
199
200
201
202 | def add_usage(
self,
usage: str | None,
actions: Iterable[argparse.Action],
groups: Iterable[Any],
prefix: str | None = None,
) -> None:
super().add_usage(usage, sorted(actions, key=self._sort_key), groups, prefix)
|
Functions:
bip39_prompt_required(settings: JoinMarketSettings, identity: WalletIdentity | None, *, explicitly_requested: bool = False) -> bool
Prompt policy shared by CLI resolution and the TUI credential dialog.
Source code in jmcore/src/jmcore/cli_common.py
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165 | def bip39_prompt_required(
settings: JoinMarketSettings,
identity: WalletIdentity | None,
*,
explicitly_requested: bool = False,
) -> bool:
"""Prompt policy shared by CLI resolution and the TUI credential dialog."""
return (
explicitly_requested
or (identity is not None and identity.bip39 == "required")
or (
(identity is None or identity.bip39 == "unknown")
and settings.wallet.bip39_passphrase_enabled
)
)
|
create_backend(backend_settings: ResolvedBackendSettings, *, wallet_name: str | None = None, creation_height: int | None = None) -> Any
Create a backend instance based on resolved settings.
Args:
backend_settings: Resolved backend settings
wallet_name: Wallet name for descriptor_wallet backend
creation_height: Block height at wallet creation time (used as scan start hint)
Returns:
Backend instance (DescriptorWalletBackend or NeutrinoBackend)
Raises:
ValueError: If backend type is invalid
ImportError: If backend module not available
Source code in jmcore/src/jmcore/cli_common.py
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150 | def create_backend(
backend_settings: ResolvedBackendSettings,
*,
wallet_name: str | None = None,
creation_height: int | None = None,
) -> Any:
"""
Create a backend instance based on resolved settings.
Args:
backend_settings: Resolved backend settings
wallet_name: Wallet name for descriptor_wallet backend
creation_height: Block height at wallet creation time (used as scan start hint)
Returns:
Backend instance (DescriptorWalletBackend or NeutrinoBackend)
Raises:
ValueError: If backend type is invalid
ImportError: If backend module not available
"""
# Import backends lazily to avoid circular imports
from jmwallet.backends.descriptor_wallet import DescriptorWalletBackend
from jmwallet.backends.neutrino import NeutrinoBackend
backend_type = backend_settings.backend_type
backend: DescriptorWalletBackend | NeutrinoBackend
if backend_type == "neutrino":
backend = NeutrinoBackend(
neutrino_url=backend_settings.neutrino_url,
network=backend_settings.bitcoin_network,
scan_start_height=backend_settings.scan_start_height,
add_peers=backend_settings.neutrino_add_peers,
tls_cert_path=backend_settings.neutrino_tls_cert,
auth_token=backend_settings.neutrino_auth_token,
fee_estimate_url=backend_settings.fee_estimate_url,
fee_estimate_proxy=backend_settings.fee_estimate_proxy,
)
elif backend_type == "descriptor_wallet":
if not wallet_name:
raise ValueError("wallet_name required for descriptor_wallet backend")
backend = DescriptorWalletBackend(
rpc_url=backend_settings.rpc_url,
rpc_user=backend_settings.rpc_user,
rpc_password=backend_settings.rpc_password,
wallet_name=wallet_name,
scan_start_height=backend_settings.scan_start_height,
scan_lookback_blocks=backend_settings.scan_lookback_blocks,
)
else:
raise ValueError(
f"Invalid backend type: {backend_type}. Valid options: descriptor_wallet, neutrino"
)
if creation_height is not None:
backend.set_wallet_creation_height(creation_height)
return backend
|
generate_descriptor_wallet_name(mnemonic: str, network: str, passphrase: str = '') -> str
Generate a deterministic wallet name from mnemonic fingerprint.
Args:
mnemonic: BIP39 mnemonic
network: Network name (mainnet, testnet, etc.)
passphrase: BIP39 passphrase
Returns:
Wallet name in format "jm-{fingerprint}-{network}"
Source code in jmcore/src/jmcore/cli_common.py
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175 | def generate_descriptor_wallet_name(
mnemonic: str,
network: str,
passphrase: str = "",
) -> str:
"""
Generate a deterministic wallet name from mnemonic fingerprint.
Args:
mnemonic: BIP39 mnemonic
network: Network name (mainnet, testnet, etc.)
passphrase: BIP39 passphrase
Returns:
Wallet name in format "jm-{fingerprint}-{network}"
"""
from jmwallet.backends.descriptor_wallet import (
generate_wallet_name,
get_mnemonic_fingerprint,
)
fingerprint = get_mnemonic_fingerprint(mnemonic, passphrase)
return generate_wallet_name(fingerprint, network)
|
has_bip39_credential(settings: JoinMarketSettings) -> bool
Match resolution policy for explicit credentials, including enabled empty env.
Source code in jmcore/src/jmcore/cli_common.py
142
143
144
145
146
147
148 | def has_bip39_credential(settings: JoinMarketSettings) -> bool:
"""Match resolution policy for explicit credentials, including enabled empty env."""
return (
bool(os.environ.get("BIP39_PASSPHRASE"))
or (settings.wallet.bip39_passphrase_enabled and "BIP39_PASSPHRASE" in os.environ)
or settings.wallet.bip39_passphrase is not None
)
|
load_mnemonic_from_file(path: Path, password: str | None = None, auto_prompt: bool = True, max_prompt_attempts: int = 3) -> str
Load mnemonic from a file (plain text or Fernet encrypted).
Args:
path: Path to mnemonic file
password: Password for decrypting the file (NOT BIP39 passphrase)
auto_prompt: If True, prompt for password when encrypted file is detected
max_prompt_attempts: When interactively prompting for a password, how
many times to retry on wrong-password errors before giving up.
A value of 1 disables retry. Only applies to the interactive
prompt path: explicit password arguments and passwords coming
from the MNEMONIC_PASSWORD env var still fail fast on mismatch.
Returns:
The mnemonic phrase
Raises:
FileNotFoundError: If file doesn't exist
ValueError: If file format is invalid or decryption fails
Source code in jmcore/src/jmcore/cli_common.py
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727 | def load_mnemonic_from_file(
path: Path,
password: str | None = None,
auto_prompt: bool = True,
max_prompt_attempts: int = 3,
) -> str:
"""
Load mnemonic from a file (plain text or Fernet encrypted).
Args:
path: Path to mnemonic file
password: Password for decrypting the file (NOT BIP39 passphrase)
auto_prompt: If True, prompt for password when encrypted file is detected
max_prompt_attempts: When interactively prompting for a password, how
many times to retry on wrong-password errors before giving up.
A value of 1 disables retry. Only applies to the interactive
prompt path: explicit ``password`` arguments and passwords coming
from the MNEMONIC_PASSWORD env var still fail fast on mismatch.
Returns:
The mnemonic phrase
Raises:
FileNotFoundError: If file doesn't exist
ValueError: If file format is invalid or decryption fails
"""
if not path.exists():
raise FileNotFoundError(f"Mnemonic file not found: {path}")
content = path.read_bytes()
# Try to decode as plain text first
try:
text = content.decode("utf-8")
# Check if it looks like a valid mnemonic (words separated by spaces)
words = text.strip().split()
if len(words) in (12, 15, 18, 21, 24) and all(w.isalpha() for w in words):
return text.strip()
except UnicodeDecodeError:
pass
# If not plain text, assume it's Fernet encrypted
prompt_used = False
if not password:
password = os.environ.get("MNEMONIC_PASSWORD")
if not password:
if auto_prompt:
password = _prompt_for_password(path)
prompt_used = True
else:
raise ValueError(
f"Mnemonic file appears to be encrypted. "
f"Set MNEMONIC_PASSWORD env, wallet.mnemonic_password in config, "
f"or use interactive prompt: {path}"
)
# Retry budget: only honoured for passwords the user is typing interactively
# so that scripted callers with an explicit password/env var still fail
# fast (issue #456 applies to the TUI's manual prompt path).
attempts_remaining = max_prompt_attempts if prompt_used else 1
mnemonic: str | None = None
last_error: ValueError | None = None
while attempts_remaining > 0:
try:
mnemonic = _decrypt_fernet_mnemonic(content, password, path)
break
except ValueError as e:
last_error = e
attempts_remaining -= 1
# Only retry wrong-password errors from the interactive prompt.
if not prompt_used or attempts_remaining <= 0:
raise
if "decryption failed" not in str(e).lower():
# Corruption or encoding error -- pointless to retry.
raise
try:
import typer
typer.echo(f"Decryption failed. {attempts_remaining} attempt(s) remaining.")
except ImportError:
print(f"Decryption failed. {attempts_remaining} attempt(s) remaining.")
password = _prompt_for_password(path)
if mnemonic is None:
# Should not happen because the loop either breaks or raises, but keep
# type checkers happy and surface the last error.
assert last_error is not None
raise last_error
# Basic validation
words = mnemonic.split()
if len(words) not in (12, 15, 18, 21, 24):
raise ValueError(
f"Invalid mnemonic: expected 12-24 words, got {len(words)}. "
f"File may be corrupted or in wrong format: {path}"
)
return mnemonic
|
log_resolved_settings(backend: ResolvedBackendSettings, tor: TorSettings | None = None, directory_servers: list[str] | None = None, mnemonic_source: str | None = None) -> None
Log resolved settings for debugging/transparency.
Args:
backend: Resolved backend settings
tor: Resolved Tor settings (optional)
directory_servers: Resolved directory servers (optional)
mnemonic_source: Source of mnemonic (optional)
Source code in jmcore/src/jmcore/cli_common.py
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224 | def log_resolved_settings(
backend: ResolvedBackendSettings,
tor: TorSettings | None = None,
directory_servers: list[str] | None = None,
mnemonic_source: str | None = None,
) -> None:
"""
Log resolved settings for debugging/transparency.
Args:
backend: Resolved backend settings
tor: Resolved Tor settings (optional)
directory_servers: Resolved directory servers (optional)
mnemonic_source: Source of mnemonic (optional)
"""
logger.info(f"Network: {backend.network}")
if backend.bitcoin_network != backend.network:
logger.info(f"Bitcoin network: {backend.bitcoin_network}")
logger.info(f"Backend: {backend.backend_type}")
if backend.backend_type == "neutrino":
logger.info("Neutrino backend endpoint configured")
logger.bind(sensitive=True).info(f"Neutrino URL: {backend.neutrino_url}")
else:
logger.info("Bitcoin Core RPC endpoint configured")
logger.bind(sensitive=True).info(f"RPC URL: {backend.rpc_url}")
if backend.rpc_user:
logger.bind(sensitive=True).info(f"RPC user: {backend.rpc_user}")
if tor:
logger.info("Tor SOCKS proxy configured")
logger.bind(sensitive=True).info(f"Tor SOCKS: {tor.socks_host}:{tor.socks_port}")
if tor.control_enabled:
logger.info("Tor control endpoint configured")
logger.bind(sensitive=True).info(f"Tor control: {tor.control_host}:{tor.control_port}")
if directory_servers:
logger.info(f"Directory servers: {len(directory_servers)} configured")
if mnemonic_source:
logger.info("Mnemonic loaded")
logger.bind(sensitive=True).info(f"Mnemonic loaded from: {mnemonic_source}")
|
resolve_backend_settings(settings: JoinMarketSettings, *, network: NetworkType | str | None = None, bitcoin_network: NetworkType | str | None = None, backend_type: str | None = None, rpc_url: str | None = None, rpc_user: str | None = None, rpc_password: str | None = None, neutrino_url: str | None = None, neutrino_tls_cert: str | None = None, neutrino_auth_token: str | None = None, data_dir: Path | None = None) -> ResolvedBackendSettings
Resolve backend settings with priority: CLI > Settings (env + config) > Defaults.
Args:
settings: JoinMarketSettings instance
network: CLI override for network
bitcoin_network: CLI override for bitcoin network
backend_type: CLI override for backend type
rpc_url: CLI override for RPC URL
rpc_user: CLI override for RPC user
rpc_password: CLI override for RPC password
neutrino_url: CLI override for Neutrino URL
neutrino_tls_cert: CLI override for Neutrino TLS certificate path
neutrino_auth_token: CLI override for Neutrino API auth token
data_dir: CLI override for data directory
Returns:
ResolvedBackendSettings with all values resolved
Source code in jmcore/src/jmcore/cli_common.py
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570 | def resolve_backend_settings(
settings: JoinMarketSettings,
*,
network: NetworkType | str | None = None,
bitcoin_network: NetworkType | str | None = None,
backend_type: str | None = None,
rpc_url: str | None = None,
rpc_user: str | None = None,
rpc_password: str | None = None,
neutrino_url: str | None = None,
neutrino_tls_cert: str | None = None,
neutrino_auth_token: str | None = None,
data_dir: Path | None = None,
) -> ResolvedBackendSettings:
"""
Resolve backend settings with priority: CLI > Settings (env + config) > Defaults.
Args:
settings: JoinMarketSettings instance
network: CLI override for network
bitcoin_network: CLI override for bitcoin network
backend_type: CLI override for backend type
rpc_url: CLI override for RPC URL
rpc_user: CLI override for RPC user
rpc_password: CLI override for RPC password
neutrino_url: CLI override for Neutrino URL
neutrino_tls_cert: CLI override for Neutrino TLS certificate path
neutrino_auth_token: CLI override for Neutrino API auth token
data_dir: CLI override for data directory
Returns:
ResolvedBackendSettings with all values resolved
"""
# Resolve network
if network is not None:
resolved_network = network.value if isinstance(network, NetworkType) else network
else:
resolved_network = settings.network_config.network.value
# Resolve bitcoin network (defaults to network if not specified)
if bitcoin_network is not None:
resolved_bitcoin_network = (
bitcoin_network.value if isinstance(bitcoin_network, NetworkType) else bitcoin_network
)
elif settings.network_config.bitcoin_network is not None:
resolved_bitcoin_network = settings.network_config.bitcoin_network.value
else:
resolved_bitcoin_network = resolved_network
# Resolve backend type
resolved_backend_type = (
backend_type if backend_type is not None else settings.bitcoin.backend_type
)
# Resolve RPC settings
resolved_rpc_url = rpc_url if rpc_url is not None else settings.bitcoin.rpc_url
resolved_rpc_user = rpc_user if rpc_user is not None else settings.bitcoin.rpc_user
# Handle SecretStr for password
if rpc_password is not None:
resolved_rpc_password = rpc_password
else:
pwd = settings.bitcoin.rpc_password
resolved_rpc_password = pwd.get_secret_value() if isinstance(pwd, SecretStr) else str(pwd)
# Resolve Neutrino URL
resolved_neutrino_url = (
neutrino_url if neutrino_url is not None else settings.bitcoin.neutrino_url
)
# Resolve Neutrino add peers
resolved_neutrino_add_peers = settings.get_neutrino_add_peers()
# Resolve data directory (used both as the resolved value AND as a base
# for resolving relative neutrino paths below).
resolved_data_dir = data_dir if data_dir is not None else settings.get_data_dir()
# Resolve Neutrino TLS cert path. Relative paths (e.g. "neutrino/tls.cert")
# are resolved against the data directory so a single config.toml works
# regardless of where the data dir lives.
resolved_neutrino_tls_cert = _resolve_data_dir_path(
neutrino_tls_cert if neutrino_tls_cert is not None else settings.bitcoin.neutrino_tls_cert,
resolved_data_dir,
)
# Resolve Neutrino auth token (string, not a path).
resolved_neutrino_auth_token = (
neutrino_auth_token
if neutrino_auth_token is not None
else settings.bitcoin.neutrino_auth_token
)
# If only an auth token *file* is configured, read it now so the backend
# never has to know where the data dir lives. Relative paths are resolved
# against the data directory. A missing file is not an error: the default
# path may simply not exist on insecure/legacy neutrino-api deployments.
if not resolved_neutrino_auth_token and settings.bitcoin.neutrino_auth_token_file:
token_path_str = _resolve_data_dir_path(
settings.bitcoin.neutrino_auth_token_file, resolved_data_dir
)
if token_path_str is not None:
token_path = Path(token_path_str)
if token_path.is_file():
try:
resolved_neutrino_auth_token = token_path.read_text().strip() or None
except OSError as exc:
logger.warning(
f"Could not read neutrino auth token from {token_path_str}: {exc}"
)
else:
logger.debug(f"Neutrino auth token file not found at {token_path_str}; skipping")
# neutrino-api serves HTTPS (with a self-signed certificate) whenever API
# authentication is enabled. When a token is present but the URL is still
# plain HTTP, upgrade it so requests don't fail with "client sent an HTTP
# request to an HTTPS server". The backend pins the certificate on first
# use (trust-on-first-use), so no manual cert wiring is required.
if resolved_neutrino_auth_token and resolved_neutrino_url.startswith("http://"):
upgraded_url = "https://" + resolved_neutrino_url.removeprefix("http://")
logger.info(
f"Neutrino auth token detected; upgrading URL from "
f"{resolved_neutrino_url} to {upgraded_url}"
)
resolved_neutrino_url = upgraded_url
# External fee source for backends without native estimation (neutrino).
from jmcore.fee_source import build_fee_source_proxy
resolved_fee_estimate_proxy = build_fee_source_proxy(
settings.tor.socks_host,
settings.tor.socks_port,
settings.tor.stream_isolation,
)
return ResolvedBackendSettings(
network=resolved_network,
bitcoin_network=resolved_bitcoin_network,
backend_type=resolved_backend_type,
rpc_url=resolved_rpc_url,
rpc_user=resolved_rpc_user,
rpc_password=resolved_rpc_password,
neutrino_url=resolved_neutrino_url,
neutrino_add_peers=resolved_neutrino_add_peers,
data_dir=resolved_data_dir,
scan_start_height=settings.wallet.scan_start_height,
scan_lookback_blocks=settings.wallet.scan_lookback_blocks,
neutrino_tls_cert=resolved_neutrino_tls_cert,
neutrino_auth_token=resolved_neutrino_auth_token,
fee_estimate_url=settings.bitcoin.fee_estimate_url,
fee_estimate_proxy=resolved_fee_estimate_proxy,
)
|
resolve_bip39_passphrase(bip39_passphrase: str | None = None, prompt: bool = False) -> str
Resolve BIP39 passphrase from argument or prompt.
Args:
bip39_passphrase: Direct passphrase value
prompt: Whether to prompt interactively
Returns:
Resolved passphrase (empty string if none)
Source code in jmcore/src/jmcore/cli_common.py
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084 | def resolve_bip39_passphrase(
bip39_passphrase: str | None = None,
prompt: bool = False,
) -> str:
"""
Resolve BIP39 passphrase from argument or prompt.
Args:
bip39_passphrase: Direct passphrase value
prompt: Whether to prompt interactively
Returns:
Resolved passphrase (empty string if none)
"""
if bip39_passphrase:
return bip39_passphrase
if prompt:
try:
import typer
return typer.prompt(
"Enter BIP39 passphrase (leave empty for none)",
default="",
hide_input=True,
)
except ImportError:
import getpass
return getpass.getpass("Enter BIP39 passphrase (leave empty for none): ")
return ""
|
Resolve the active wallet's mnemonic file path without decrypting it.
Mirrors the file-based portion of :func:resolve_mnemonic's priority
chain so per-wallet read commands can locate the configured wallet and
read its companion .meta fingerprint without prompting for a
password:
MNEMONIC_FILE environment variable
- Config file
wallet.mnemonic_file setting
- Default wallet path (
<data_dir>/wallets/default.mnemonic)
Raw-mnemonic sources have no backing file and prevent using an unrelated
configured file. Returns None when the selected source is not a file.
Source code in jmcore/src/jmcore/cli_common.py
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795 | def resolve_configured_mnemonic_file(settings: JoinMarketSettings) -> Path | None:
"""Resolve the active wallet's mnemonic file path without decrypting it.
Mirrors the file-based portion of :func:`resolve_mnemonic`'s priority
chain so per-wallet read commands can locate the configured wallet and
read its companion ``.meta`` fingerprint without prompting for a
password:
1. ``MNEMONIC_FILE`` environment variable
2. Config file ``wallet.mnemonic_file`` setting
3. Default wallet path (``<data_dir>/wallets/default.mnemonic``)
Raw-mnemonic sources have no backing file and prevent using an unrelated
configured file. Returns ``None`` when the selected source is not a file.
"""
source = select_mnemonic_source(settings)
return source.path if source is not None else None
|
resolve_directory_servers(settings: JoinMarketSettings, *, directory_servers: str | None = None, network: str | None = None) -> list[str]
Resolve directory servers with priority: CLI > Settings > Network defaults.
Args:
settings: JoinMarketSettings instance
directory_servers: CLI override (comma-separated)
network: Network to use for defaults (if not in settings)
Returns:
List of directory server addresses
Source code in jmcore/src/jmcore/cli_common.py
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600 | def resolve_directory_servers(
settings: JoinMarketSettings,
*,
directory_servers: str | None = None,
network: str | None = None,
) -> list[str]:
"""
Resolve directory servers with priority: CLI > Settings > Network defaults.
Args:
settings: JoinMarketSettings instance
directory_servers: CLI override (comma-separated)
network: Network to use for defaults (if not in settings)
Returns:
List of directory server addresses
"""
if directory_servers:
return [s.strip() for s in directory_servers.split(",") if s.strip()]
if settings.network_config.directory_servers:
return settings.network_config.directory_servers
# Use network-specific defaults
from jmcore.settings import DEFAULT_DIRECTORY_SERVERS
effective_network = network or settings.network_config.network.value
return DEFAULT_DIRECTORY_SERVERS.get(effective_network, [])
|
resolve_mnemonic(settings: JoinMarketSettings, *, mnemonic: str | None = None, mnemonic_file: Path | None = None, password: str | None = None, bip39_passphrase: str | None = None, prompt_bip39_passphrase: bool = False, required: bool = True, validate_identity: bool = True, expected_fingerprint: str | None = None) -> ResolvedMnemonic | None
Resolve mnemonic from various sources with priority.
Mnemonic priority:
1. --mnemonic argument
2. --mnemonic-file argument
3. MNEMONIC_FILE environment variable
4. MNEMONIC environment variable
5. Config file wallet.mnemonic_file setting
6. Default wallet path (~/.joinmarket-ng/wallets/default.mnemonic)
BIP39 passphrase priority:
1. --bip39-passphrase argument
2. BIP39_PASSPHRASE environment variable
3. Config file wallet.bip39_passphrase setting
4. Interactive prompt (if --prompt-bip39-passphrase is set)
5. Empty string (default - no passphrase)
For encrypted mnemonic files, the password is resolved as:
1. Config file wallet.mnemonic_password setting (or password param)
2. MNEMONIC_PASSWORD environment variable
3. Interactive prompt (if auto_prompt is enabled)
Args:
settings: JoinMarketSettings instance
mnemonic: CLI mnemonic string
mnemonic_file: CLI mnemonic file path
password: Password for encrypted mnemonic file (NOT BIP39 passphrase)
bip39_passphrase: BIP39 passphrase (13th/25th word, NOT file encryption password)
prompt_bip39_passphrase: Whether to prompt for BIP39 passphrase interactively
required: Whether mnemonic is required (raises error if not found)
Returns:
ResolvedMnemonic or None if not required and not found
Raises:
ValueError: If required but not found, or if loading fails
Source code in jmcore/src/jmcore/cli_common.py
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021 | def resolve_mnemonic(
settings: JoinMarketSettings,
*,
mnemonic: str | None = None,
mnemonic_file: Path | None = None,
password: str | None = None,
bip39_passphrase: str | None = None,
prompt_bip39_passphrase: bool = False,
required: bool = True,
validate_identity: bool = True,
expected_fingerprint: str | None = None,
) -> ResolvedMnemonic | None:
"""
Resolve mnemonic from various sources with priority.
Mnemonic priority:
1. --mnemonic argument
2. --mnemonic-file argument
3. MNEMONIC_FILE environment variable
4. MNEMONIC environment variable
5. Config file wallet.mnemonic_file setting
6. Default wallet path (~/.joinmarket-ng/wallets/default.mnemonic)
BIP39 passphrase priority:
1. --bip39-passphrase argument
2. BIP39_PASSPHRASE environment variable
3. Config file wallet.bip39_passphrase setting
4. Interactive prompt (if --prompt-bip39-passphrase is set)
5. Empty string (default - no passphrase)
For encrypted mnemonic files, the password is resolved as:
1. Config file wallet.mnemonic_password setting (or password param)
2. MNEMONIC_PASSWORD environment variable
3. Interactive prompt (if auto_prompt is enabled)
Args:
settings: JoinMarketSettings instance
mnemonic: CLI mnemonic string
mnemonic_file: CLI mnemonic file path
password: Password for encrypted mnemonic file (NOT BIP39 passphrase)
bip39_passphrase: BIP39 passphrase (13th/25th word, NOT file encryption password)
prompt_bip39_passphrase: Whether to prompt for BIP39 passphrase interactively
required: Whether mnemonic is required (raises error if not found)
Returns:
ResolvedMnemonic or None if not required and not found
Raises:
ValueError: If required but not found, or if loading fails
"""
selected_source = select_mnemonic_source(
settings, mnemonic=mnemonic, mnemonic_file=mnemonic_file
)
resolved_mnemonic: str | None = None
source = selected_source.description if selected_source else ""
mnemonic_file_path = selected_source.path if selected_source else None
# Resolve the encryption password once for all file-based paths so that
# ``wallet.mnemonic_password`` from config.toml works no matter whether
# the file path came from --mnemonic-file, MNEMONIC_FILE env, the config,
# or the default wallet location. Without this fallback an explicit
# --mnemonic-file (as used by the systemd unit template documented in
# docs/README-maker.md) silently bypasses the config password and falls
# through to an interactive prompt, which fails under systemd.
# See issue #498.
effective_password = password
if effective_password is None and settings.wallet.mnemonic_password:
effective_password = settings.wallet.mnemonic_password.get_secret_value()
if selected_source is not None:
resolved_mnemonic = (
load_mnemonic_from_file(mnemonic_file_path, effective_password)
if mnemonic_file_path is not None
else selected_source.mnemonic
)
if resolved_mnemonic is None:
if required:
raise ValueError(
"No mnemonic provided. Use --mnemonic-file, "
"MNEMONIC env, or set wallet.mnemonic_file in config."
)
return None
# Report an invalid BIP39 checksum at ERROR so the diagnostic remains
# visible even when the configured log level suppresses warnings. A typo
# (wrong, missing, or reordered word) still derives a wallet, just a *different*
# one than intended; depositing into it without noticing would strand
# funds. We deliberately do not hard-fail: users may intentionally keep a
# non-checksummed phrase (e.g. imported via `jm-wallet import` with
# "Continue anyway", possibly already holding funds), and locking them out
# would be worse.
if not validate_bip39_checksum(resolved_mnemonic):
logger.error(
f"SECURITY WARNING: the mnemonic from {source} has an INVALID BIP39 "
"checksum. This usually means a word is misspelled, missing, or in "
"the wrong order. The wallet will still load, but it derives a "
"DIFFERENT wallet than the words were intended to encode. Do NOT "
"deposit funds until you have verified the seed phrase (e.g. with "
"'jm-wallet validate'). Ignore this only if you intentionally use "
"a phrase without a valid checksum."
)
identity = (
selected_identity(mnemonic_file_path)
if mnemonic_file_path is not None and validate_identity
else None
)
legacy_file = mnemonic_file_path if identity is None and validate_identity else None
should_prompt = bip39_prompt_required(
settings, identity, explicitly_requested=prompt_bip39_passphrase
)
# Resolve BIP39 passphrase
# Priority: CLI arg > env var > config > prompt > empty
resolved_passphrase = ""
if bip39_passphrase is not None:
resolved_passphrase = bip39_passphrase
elif "BIP39_PASSPHRASE" in os.environ and settings.wallet.bip39_passphrase_enabled:
resolved_passphrase = os.environ["BIP39_PASSPHRASE"]
elif env_passphrase := os.environ.get("BIP39_PASSPHRASE"):
resolved_passphrase = env_passphrase
elif settings.wallet.bip39_passphrase is not None:
resolved_passphrase = settings.wallet.bip39_passphrase.get_secret_value()
elif should_prompt:
if settings.wallet.bip39_passphrase_enabled and not prompt_bip39_passphrase:
from jmcore.confirmation import is_interactive_mode
if not is_interactive_mode():
raise ValueError(
"wallet.wallet_with_passphrase is enabled "
"(compatibility alias for wallet.bip39_passphrase_enabled), "
"but no passphrase was provided and no interactive terminal is available. "
"Provide BIP39_PASSPHRASE or wallet.bip39_passphrase explicitly."
)
# Lazy import typer only when needed for prompting
try:
import typer
resolved_passphrase = typer.prompt(
"Enter BIP39 passphrase (leave empty for none)",
default="",
hide_input=True,
)
except ImportError:
# Fall back to getpass if typer not available
import getpass
resolved_passphrase = getpass.getpass("Enter BIP39 passphrase (leave empty for none): ")
_confirm_prompted_wallet(resolved_mnemonic, resolved_passphrase)
expected_fingerprints = (expected_fingerprint, os.environ.get("EXPECTED_FINGERPRINT"))
if (
identity is not None
or legacy_file is not None
or any(value is not None for value in expected_fingerprints)
):
from jmwallet.backends.descriptor_wallet import get_mnemonic_fingerprint
derived = get_mnemonic_fingerprint(resolved_mnemonic, resolved_passphrase)
if identity is not None and (
derived != identity.fingerprint
or (identity.bip39 == "required" and not resolved_passphrase)
):
raise ValueError(
"Derived wallet does not match the selected identity. No wallet activity started. "
"Check the BIP39 passphrase, or run `jm-wallet identity register --select` "
"with the intended passphrase to switch wallets."
)
for fingerprint in expected_fingerprints:
if fingerprint is None:
continue
expected = WalletIdentity.validate_fingerprint(fingerprint)
if derived != expected:
raise ValueError(
"Staged wallet fingerprint mismatch. No wallet activity started. "
"Restage the maker credentials for the intended wallet."
)
if legacy_file is not None:
_adopt_or_warn(legacy_file, derived, resolved_passphrase)
# Load wallet metadata (creation_height) from companion .meta file
creation_height: int | None = None
if mnemonic_file_path is not None:
try:
meta = load_mnemonic_meta(mnemonic_file_path)
raw_creation_height = meta.get("creation_height")
if isinstance(raw_creation_height, int) and not isinstance(raw_creation_height, bool):
if raw_creation_height >= 0:
creation_height = raw_creation_height
logger.debug(f"Loaded wallet creation height: {creation_height}")
else:
logger.warning(
f"Ignoring negative creation_height in metadata: {raw_creation_height}"
)
elif raw_creation_height is not None:
logger.warning(
f"Ignoring non-integer creation_height in metadata: {raw_creation_height!r}"
)
except Exception as exc:
logger.debug(f"Could not load mnemonic metadata: {exc}")
return ResolvedMnemonic(
mnemonic=resolved_mnemonic,
bip39_passphrase=resolved_passphrase,
source=source,
creation_height=creation_height,
mnemonic_file=mnemonic_file_path,
)
|
select_mnemonic_source(settings: JoinMarketSettings, *, mnemonic: str | None = None, mnemonic_file: Path | None = None) -> MnemonicSource | None
Source code in jmcore/src/jmcore/cli_common.py
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139 | def select_mnemonic_source(
settings: JoinMarketSettings,
*,
mnemonic: str | None = None,
mnemonic_file: Path | None = None,
) -> MnemonicSource | None:
if mnemonic:
return MnemonicSource(mnemonic, None, "--mnemonic argument")
if mnemonic_file is not None:
return MnemonicSource(None, mnemonic_file, f"--mnemonic-file ({mnemonic_file})")
if env_file := os.environ.get("MNEMONIC_FILE"):
return MnemonicSource(None, Path(env_file), f"MNEMONIC_FILE env ({env_file})")
if env_mnemonic := os.environ.get("MNEMONIC"):
return MnemonicSource(env_mnemonic, None, "MNEMONIC env")
if settings.wallet.mnemonic_file:
path = Path(settings.wallet.mnemonic_file)
return MnemonicSource(None, path, f"config file ({path})")
path = settings.get_data_dir() / "wallets" / "default.mnemonic"
if path.exists():
return MnemonicSource(None, path, f"default wallet ({path})")
return None
|
setup_cli(log_level: str | None = None, data_dir: Path | None = None, config_file: Path | None = None) -> JoinMarketSettings
Common CLI setup: reset settings cache, configure logging, return settings.
Log level priority: CLI argument > settings (env/config) > default "INFO"
When data_dir is provided (typically from a CLI --data-dir flag),
the JOINMARKET_DATA_DIR environment variable is set to that path
BEFORE settings are loaded. This ensures that <data_dir>/config.toml
is read, that wallet.mnemonic_file defaults resolve to that data
directory, and that any other env-driven defaults (paths, etc.) point at
the user-specified data directory.
When config_file is provided (typically from a CLI --config-file
flag), the JOINMARKET_CONFIG_FILE environment variable is set so that
the config is read from (and created at) that explicit path instead of
<data_dir>/config.toml. This decouples the configuration file location
from the data directory for FHS-style deployments (e.g. /etc/joinmarket
config with /var/lib/joinmarket data); see issue #537.
Args:
log_level: Log level override from CLI (None means use settings)
data_dir: Optional CLI override for data directory. When provided,
JOINMARKET_DATA_DIR is set in os.environ so that all
subsequent settings-loading code paths agree on the same data
directory.
config_file: Optional CLI override for the config file path. When
provided, JOINMARKET_CONFIG_FILE is set in os.environ so
that settings are read from and created at that explicit path.
Returns:
JoinMarketSettings instance with all sources loaded
Source code in jmcore/src/jmcore/cli_common.py
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350 | def setup_cli(
log_level: str | None = None,
data_dir: Path | None = None,
config_file: Path | None = None,
) -> JoinMarketSettings:
"""
Common CLI setup: reset settings cache, configure logging, return settings.
Log level priority: CLI argument > settings (env/config) > default "INFO"
When ``data_dir`` is provided (typically from a CLI ``--data-dir`` flag),
the ``JOINMARKET_DATA_DIR`` environment variable is set to that path
BEFORE settings are loaded. This ensures that ``<data_dir>/config.toml``
is read, that ``wallet.mnemonic_file`` defaults resolve to that data
directory, and that any other env-driven defaults (paths, etc.) point at
the user-specified data directory.
When ``config_file`` is provided (typically from a CLI ``--config-file``
flag), the ``JOINMARKET_CONFIG_FILE`` environment variable is set so that
the config is read from (and created at) that explicit path instead of
``<data_dir>/config.toml``. This decouples the configuration file location
from the data directory for FHS-style deployments (e.g. ``/etc/joinmarket``
config with ``/var/lib/joinmarket`` data); see issue #537.
Args:
log_level: Log level override from CLI (None means use settings)
data_dir: Optional CLI override for data directory. When provided,
``JOINMARKET_DATA_DIR`` is set in ``os.environ`` so that all
subsequent settings-loading code paths agree on the same data
directory.
config_file: Optional CLI override for the config file path. When
provided, ``JOINMARKET_CONFIG_FILE`` is set in ``os.environ`` so
that settings are read from and created at that explicit path.
Returns:
JoinMarketSettings instance with all sources loaded
"""
if data_dir is not None:
# Honour the user's CLI choice across every code path that reads
# JOINMARKET_DATA_DIR (settings, paths, subprocesses, etc.).
os.environ["JOINMARKET_DATA_DIR"] = str(data_dir)
if config_file is not None:
# Decouple the config file from the data dir. Expand ``~`` so a
# ``--config-file ~/foo.toml`` does not become a literal path.
os.environ["JOINMARKET_CONFIG_FILE"] = str(Path(config_file).expanduser())
# Apply an early logging configuration before loading settings so that
# informational messages emitted while parsing config.toml (e.g.
# "Loaded config from ...") honor the caller's intended log level.
# Priority for this early pass: CLI arg > LOGGING__LEVEL env > INFO default.
# Sensitive records stay hidden before parsing config, except when the
# dedicated environment variable is cleanly parseable.
early_level = log_level or os.environ.get("LOGGING__LEVEL") or "INFO"
early_sensitive = _parse_sensitive_logging_env(os.environ.get("LOGGING__SENSITIVE"))
setup_logging(early_level, sensitive=early_sensitive is True)
reset_settings()
try:
settings = get_settings()
except ValidationError as exc:
_exit_on_invalid_config(exc)
# Resolve log level: CLI > settings > default
effective_log_level = log_level if log_level is not None else settings.logging.level
setup_logging(effective_log_level, sensitive=settings.logging.sensitive)
return settings
|
setup_logging(level: str = 'INFO', sensitive: bool = False) -> None
Configure loguru logging with consistent format.
Args:
level: Log level (TRACE, DEBUG, INFO, WARNING, ERROR)
sensitive: Whether to emit records bound with sensitive=True
Source code in jmcore/src/jmcore/cli_common.py
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267 | def setup_logging(level: str = "INFO", sensitive: bool = False) -> None:
"""
Configure loguru logging with consistent format.
Args:
level: Log level (TRACE, DEBUG, INFO, WARNING, ERROR)
sensitive: Whether to emit records bound with ``sensitive=True``
"""
logger.remove()
logger.add(
sys.stderr,
format=_format_log_record,
level=level.upper(),
colorize=True,
filter=sensitive_log_filter(sensitive),
)
|