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,
resolve_tor_settings,
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, ...)
Classes
ResolvedBackendSettings
dataclass
Resolved backend settings ready for use.
Source code in jmcore/src/jmcore/cli_common.py
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75 | @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
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_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
90
91
92
93
94
95
96
97
98
99
100
101 | @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
|
Attributes
bip39_passphrase: str
instance-attribute
creation_height: int | None = None
class-attribute
instance-attribute
mnemonic: str
instance-attribute
source: str
instance-attribute
ResolvedTorSettings
dataclass
Resolved Tor settings ready for use.
Source code in jmcore/src/jmcore/cli_common.py
78
79
80
81
82
83
84
85
86
87 | @dataclass
class ResolvedTorSettings:
"""Resolved Tor settings ready for use."""
socks_host: str
socks_port: int
control_enabled: bool
control_host: str
control_port: int
cookie_path: Path | None
|
Attributes
control_enabled: bool
instance-attribute
control_host: str
instance-attribute
control_port: int
instance-attribute
cookie_path: Path | None
instance-attribute
socks_host: str
instance-attribute
socks_port: int
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
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167 | 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
131
132
133
134
135
136
137
138 | 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:
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
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
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044 | 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,
)
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
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069 | 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)
|
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
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
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 | 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: ResolvedTorSettings | 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
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113 | def log_resolved_settings(
backend: ResolvedBackendSettings,
tor: ResolvedTorSettings | 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(f"Neutrino URL: {backend.neutrino_url}")
else:
logger.info(f"RPC URL: {backend.rpc_url}")
if backend.rpc_user:
logger.info(f"RPC user: {backend.rpc_user}")
if tor:
logger.info(f"Tor SOCKS: {tor.socks_host}:{tor.socks_port}")
if tor.control_enabled:
logger.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(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
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
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 | 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,
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
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 | 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 (--mnemonic/MNEMONIC env) have no backing
file and are intentionally ignored here. Returns None when no
file-based wallet is configured or the default does not exist.
Source code in jmcore/src/jmcore/cli_common.py
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761 | 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 (``--mnemonic``/``MNEMONIC`` env) have no backing
file and are intentionally ignored here. Returns ``None`` when no
file-based wallet is configured or the default does not exist.
"""
if env_file := os.environ.get("MNEMONIC_FILE"):
return Path(env_file)
if settings.wallet.mnemonic_file:
return Path(settings.wallet.mnemonic_file)
default_wallet = settings.get_data_dir() / "wallets" / "default.mnemonic"
if default_wallet.exists():
return default_wallet
return 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
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 | 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) -> 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
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
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 | 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,
) -> 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
"""
resolved_mnemonic: str | None = None
source = ""
mnemonic_file_path: Path | None = None # Track file path for .meta loading
# 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()
# Priority 1: Direct mnemonic argument
if mnemonic:
resolved_mnemonic = mnemonic
source = "--mnemonic argument"
# Priority 2: Mnemonic file argument
elif mnemonic_file:
resolved_mnemonic = load_mnemonic_from_file(mnemonic_file, effective_password)
mnemonic_file_path = mnemonic_file
source = f"--mnemonic-file ({mnemonic_file})"
# Priority 3: MNEMONIC_FILE environment variable
elif env_file := os.environ.get("MNEMONIC_FILE"):
env_path = Path(env_file)
resolved_mnemonic = load_mnemonic_from_file(env_path, effective_password)
mnemonic_file_path = env_path
source = f"MNEMONIC_FILE env ({env_path})"
# Priority 4: MNEMONIC environment variable
elif env_mnemonic := os.environ.get("MNEMONIC"):
resolved_mnemonic = env_mnemonic
source = "MNEMONIC env"
# Priority 5: Config file wallet.mnemonic_file
elif settings.wallet.mnemonic_file:
config_path = Path(settings.wallet.mnemonic_file)
resolved_mnemonic = load_mnemonic_from_file(config_path, effective_password)
mnemonic_file_path = config_path
source = f"config file ({config_path})"
# Priority 6: Default wallet path
else:
default_wallet = settings.get_data_dir() / "wallets" / "default.mnemonic"
if default_wallet.exists():
resolved_mnemonic = load_mnemonic_from_file(default_wallet, effective_password)
mnemonic_file_path = default_wallet
source = f"default wallet ({default_wallet})"
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."
)
# Resolve BIP39 passphrase
# Priority: CLI arg > env var > config > prompt > empty
resolved_passphrase = ""
if bip39_passphrase:
resolved_passphrase = 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 prompt_bip39_passphrase:
# 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): ")
# Load wallet metadata (creation_height) from companion .meta file
creation_height: int | None = None
if mnemonic_file_path is not None:
try:
from jmwallet.cli.mnemonic import load_mnemonic_meta
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,
)
|
resolve_tor_settings(settings: JoinMarketSettings, *, socks_host: str | None = None, socks_port: int | None = None, control_host: str | None = None, control_port: int | None = None, cookie_path: Path | None = None, disable_control: bool = False) -> ResolvedTorSettings
Resolve Tor settings with priority: CLI > Settings > Defaults.
Args:
settings: JoinMarketSettings instance
socks_host: CLI override for SOCKS host
socks_port: CLI override for SOCKS port
control_host: CLI override for control host
control_port: CLI override for control port
cookie_path: CLI override for cookie path
disable_control: Whether to disable Tor control
Returns:
ResolvedTorSettings with all values resolved
Source code in jmcore/src/jmcore/cli_common.py
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 | def resolve_tor_settings(
settings: JoinMarketSettings,
*,
socks_host: str | None = None,
socks_port: int | None = None,
control_host: str | None = None,
control_port: int | None = None,
cookie_path: Path | None = None,
disable_control: bool = False,
) -> ResolvedTorSettings:
"""
Resolve Tor settings with priority: CLI > Settings > Defaults.
Args:
settings: JoinMarketSettings instance
socks_host: CLI override for SOCKS host
socks_port: CLI override for SOCKS port
control_host: CLI override for control host
control_port: CLI override for control port
cookie_path: CLI override for cookie path
disable_control: Whether to disable Tor control
Returns:
ResolvedTorSettings with all values resolved
"""
resolved_socks_host = socks_host if socks_host is not None else settings.tor.socks_host
resolved_socks_port = socks_port if socks_port is not None else settings.tor.socks_port
# Control port settings
control_enabled = not disable_control and settings.tor.control_enabled
resolved_control_host = control_host if control_host is not None else settings.tor.control_host
resolved_control_port = control_port if control_port is not None else settings.tor.control_port
resolved_cookie_path: Path | None = None
if cookie_path is not None:
resolved_cookie_path = cookie_path
elif settings.tor.cookie_path:
resolved_cookie_path = Path(settings.tor.cookie_path)
return ResolvedTorSettings(
socks_host=resolved_socks_host,
socks_port=resolved_socks_port,
control_enabled=control_enabled,
control_host=resolved_control_host,
control_port=resolved_control_port,
cookie_path=resolved_cookie_path,
)
|
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
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
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260 | 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.
early_level = log_level or os.environ.get("LOGGING__LEVEL") or "INFO"
setup_logging(early_level)
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)
return settings
|
setup_logging(level: str = 'INFO') -> None
Configure loguru logging with consistent format.
Args:
level: Log level (TRACE, DEBUG, INFO, WARNING, ERROR)
Source code in jmcore/src/jmcore/cli_common.py
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193 | def setup_logging(level: str = "INFO") -> None:
"""
Configure loguru logging with consistent format.
Args:
level: Log level (TRACE, DEBUG, INFO, WARNING, ERROR)
"""
logger.remove()
logger.add(
sys.stderr,
format=(
"<green>{time:YYYY-MM-DD HH:mm:ss}</green> | "
"<level>{level: <8}</level> | "
"<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - "
"<level>{message}</level>"
),
level=level.upper(),
colorize=True,
)
|