jmcore.settings

jmcore.settings

Unified settings management for JoinMarket components.

This module provides a centralized configuration system using pydantic-settings that supports: 1. TOML configuration file (~/.joinmarket-ng/config.toml) 2. Environment variables 3. CLI arguments (via typer, handled by components)

Priority (highest to lowest): 1. CLI arguments 2. Environment variables 3. Config file 4. Default values

The config file is auto-generated on first run with all settings commented out, allowing users to selectively override only the settings they want to change. This approach facilitates software updates since unchanged defaults can be updated without user intervention.

Usage: from jmcore.settings import get_settings, JoinMarketSettings

# Get settings (loads from all sources with proper priority)
settings = get_settings()

# Access common settings
print(settings.tor.socks_host)
print(settings.bitcoin.rpc_url)

Environment Variable Naming: - Use uppercase with double underscore for nested settings - Examples: TOR__SOCKS_HOST, BITCOIN__RPC_URL, MAKER__MIN_SIZE - Maps to TOML sections: TOR__SOCKS_HOST -> [tor] socks_host

Attributes

DEFAULT_DIRECTORY_SERVERS: dict[str, list[str]] = {'mainnet': DIRECTORY_NODES_MAINNET, 'signet': DIRECTORY_NODES_SIGNET, 'testnet': DIRECTORY_NODES_TESTNET, 'regtest': []} module-attribute

__all__ = ['JoinMarketSettings', 'TorSettings', 'BitcoinSettings', 'NetworkSettings', 'WalletSettings', 'NotificationSettings', 'MakerSettings', 'TakerSettings', 'DirectoryServerSettings', 'OrderbookWatcherSettings', 'LoggingSettings', 'TuiSettings', 'get_settings', 'reset_settings', 'get_config_path', 'generate_config_template', 'generate_config_starter', 'ensure_config_file', 'config_diff', 'migrate_config', 'DEFAULT_DIRECTORY_SERVERS'] module-attribute

Classes

BitcoinSettings

Bases: BaseModel

Bitcoin backend configuration.

Source code in jmcore/src/jmcore/settings.py

class BitcoinSettings(BaseModel):
    """Bitcoin backend configuration."""

    backend_type: str = Field(
        default="descriptor_wallet",
        description="Backend type: descriptor_wallet or neutrino",
    )
    rpc_url: str = Field(
        default="http://127.0.0.1:8332",
        description="Bitcoin Core RPC URL",
    )
    rpc_user: str = Field(
        default="",
        description="Bitcoin Core RPC username",
    )
    rpc_password: SecretStr = Field(
        default=SecretStr(""),
        description="Bitcoin Core RPC password",
    )
    rpc_cookie_file: str | None = Field(
        default=None,
        description=(
            "Path to Bitcoin Core .cookie file for cookie-based RPC authentication. "
            "When set, the cookie file is read at startup and rpc_user/rpc_password "
            "are populated automatically. This is mutually exclusive with setting "
            "rpc_user/rpc_password manually. The cookie file is typically located "
            "at ~/.bitcoin/.cookie (mainnet) or ~/.bitcoin/regtest/.cookie (regtest)."
        ),
    )
    neutrino_url: str = Field(
        default="http://127.0.0.1:8334",
        description="Neutrino REST API URL (for neutrino backend)",
    )
    neutrino_add_peers: list[str] = Field(
        default_factory=list,
        description=(
            "Preferred peer addresses for neutrino (host:port) while still allowing "
            "DNS/discovery peers. Only takes effect when JoinMarket manages the "
            "neutrino process (e.g., flatpak deployment). When neutrino-api runs as "
            "a standalone service, configure peers directly via its ADD_PEERS env var "
            "or --addpeer flag."
        ),
    )
    neutrino_clearnet_initial_sync: bool = Field(
        default=True,
        description=(
            "Sync block headers over clearnet before switching to Tor. "
            "Headers are public deterministic data identical for all nodes, "
            "so downloading them over clearnet does not reveal watched addresses. "
            "Typically around 2x faster than doing the full initial header sync via Tor."
        ),
    )
    neutrino_prefetch_filters: bool = Field(
        default=True,
        description=(
            "Enable background prefetch of compact block filters. "
            "Enabled by default because jm-wallet info scans these filters anyway, "
            "so prefetching saves time on the initial scan. With the default "
            "lookback of ~2 years, this takes ~3 hours on clearnet and ~3GB disk "
            "on mainnet. Disable to fetch filters on-demand only. "
            "When false, neutrino_prefetch_lookback_blocks is ignored."
        ),
    )
    neutrino_prefetch_lookback_blocks: int = Field(
        default=105120,
        description=(
            "When neutrino_prefetch_filters is true, only prefetch filters "
            "for this many recent blocks (~2 years at 105120 blocks). "
            "Set to 0 to prefetch all filters from genesis. Ignored when "
            "neutrino_prefetch_filters is false."
        ),
    )
    neutrino_scan_lookback_blocks: int = Field(
        default=105120,
        description=(
            "Number of blocks to look back from tip for wallet rescans when "
            "no explicit scan_start_height is set. Default ~2 years (105120 blocks)."
        ),
    )
    neutrino_tls_cert: str | None = Field(
        default="neutrino/tls.cert",
        description=(
            "Path to neutrino-api TLS certificate for HTTPS verification. "
            "When the file exists, the neutrino backend pins it. When it is "
            "missing but the server is HTTPS, the backend fetches and pins the "
            "certificate on first use (trust-on-first-use) and writes it here. "
            "Relative paths are resolved against the data directory. "
            "Set to an empty string to disable certificate pinning."
        ),
    )
    neutrino_auth_token: str | None = Field(
        default=None,
        description=(
            "API bearer token for neutrino-api authentication. "
            "Sent as 'Authorization: Bearer <token>' on every request. "
            "Generated automatically by neutrino-api on first start "
            "and stored in its data directory as 'auth_token'."
        ),
    )
    neutrino_auth_token_file: str | None = Field(
        default="neutrino/auth_token",
        description=(
            "Path to a file containing the neutrino-api auth token. "
            "If set and neutrino_auth_token is not provided, the token "
            "is read from this file at startup. Relative paths are resolved "
            "against the data directory. Defaults to 'neutrino/auth_token'; "
            "missing files are ignored. Set to an empty string to disable."
        ),
    )
    neutrino_include_mempool: bool = Field(
        default=True,
        description=(
            "Include unconfirmed entries from the neutrino-api watched "
            "mempool tracker in UTXO listings, and overlay mempool "
            "spends on single-UTXO checks. Has no effect when the "
            "server does not expose the tracker (older neutrino-api or "
            "operator-disabled). Set to false for chain-only behaviour."
        ),
    )
    fee_estimate_url: str | None = Field(
        default=None,
        description=(
            "External HTTP fee estimate source used when the backend cannot "
            "estimate fees itself (neutrino). Supports mempool.space "
            "recommended-fees, Esplora /fee-estimates, and LND fee.url JSON "
            "formats. Multiple comma-separated URLs are tried in order. "
            "Leave unset for the onion-first default fallback chain for the "
            "current network, fetched over Tor; disabled on regtest or when "
            "no Tor proxy is available. Set to 'off' to disable external "
            "fee estimation entirely. Resolved estimates remain subject to "
            "wallet.max_fee_rate_sat_vb."
        ),
    )

    @model_validator(mode="after")
    def _load_rpc_cookie_from_file(self) -> Self:
        """Read rpc_user/rpc_password from Bitcoin Core .cookie file.

        Bitcoin Core writes a ``.cookie`` file containing
        ``__cookie__:<random_hex>`` in its data directory.  When
        ``rpc_cookie_file`` is set and ``rpc_user`` has not been
        explicitly provided, the cookie file is parsed and the
        credentials are populated automatically.
        """
        if self.rpc_cookie_file is not None:
            cookie_path = Path(self.rpc_cookie_file).expanduser()
            # Only override if user hasn't explicitly set credentials
            if self.rpc_user == "" and self.rpc_password.get_secret_value() == "":
                if cookie_path.is_file():
                    content = cookie_path.read_text().strip()
                    if ":" in content:
                        user, password = content.split(":", 1)
                        self.rpc_user = user
                        self.rpc_password = SecretStr(password)
                    else:
                        logger.warning(
                            f"Cookie file {cookie_path} has unexpected format "
                            "(expected 'user:password')"
                        )
                else:
                    logger.warning(f"Cookie file not found: {cookie_path}")
            else:
                logger.warning(
                    "Both rpc_cookie_file and rpc_user/rpc_password are set; "
                    "ignoring rpc_cookie_file in favor of explicit credentials"
                )
        return self

    @model_validator(mode="after")
    def _load_auth_token_from_file(self) -> Self:
        """Read neutrino_auth_token from file when neutrino_auth_token_file is set.

        Only loads the token at settings-init time when the configured path is
        absolute or starts with ``~``. Plain relative paths are deferred to the
        CLI layer (``resolve_backend_settings``) which knows the resolved data
        directory and can join the path correctly.
        """
        if self.neutrino_auth_token is None and self.neutrino_auth_token_file is not None:
            raw = self.neutrino_auth_token_file
            if raw.startswith("~") or Path(raw).is_absolute():
                token_path = Path(raw).expanduser()
                if token_path.is_file():
                    self.neutrino_auth_token = token_path.read_text().strip()
        return self

Attributes

backend_type: str = Field(default='descriptor_wallet', description='Backend type: descriptor_wallet or neutrino') class-attribute instance-attribute

fee_estimate_url: str | None = Field(default=None, description="External HTTP fee estimate source used when the backend cannot estimate fees itself (neutrino). Supports mempool.space recommended-fees, Esplora /fee-estimates, and LND fee.url JSON formats. Multiple comma-separated URLs are tried in order. Leave unset for the onion-first default fallback chain for the current network, fetched over Tor; disabled on regtest or when no Tor proxy is available. Set to 'off' to disable external fee estimation entirely. Resolved estimates remain subject to wallet.max_fee_rate_sat_vb.") class-attribute instance-attribute

neutrino_add_peers: list[str] = Field(default_factory=list, description='Preferred peer addresses for neutrino (host:port) while still allowing DNS/discovery peers. Only takes effect when JoinMarket manages the neutrino process (e.g., flatpak deployment). When neutrino-api runs as a standalone service, configure peers directly via its ADD_PEERS env var or --addpeer flag.') class-attribute instance-attribute

neutrino_auth_token: str | None = Field(default=None, description="API bearer token for neutrino-api authentication. Sent as 'Authorization: Bearer <token>' on every request. Generated automatically by neutrino-api on first start and stored in its data directory as 'auth_token'.") class-attribute instance-attribute

neutrino_auth_token_file: str | None = Field(default='neutrino/auth_token', description="Path to a file containing the neutrino-api auth token. If set and neutrino_auth_token is not provided, the token is read from this file at startup. Relative paths are resolved against the data directory. Defaults to 'neutrino/auth_token'; missing files are ignored. Set to an empty string to disable.") class-attribute instance-attribute

neutrino_clearnet_initial_sync: bool = Field(default=True, description='Sync block headers over clearnet before switching to Tor. Headers are public deterministic data identical for all nodes, so downloading them over clearnet does not reveal watched addresses. Typically around 2x faster than doing the full initial header sync via Tor.') class-attribute instance-attribute

neutrino_include_mempool: bool = Field(default=True, description='Include unconfirmed entries from the neutrino-api watched mempool tracker in UTXO listings, and overlay mempool spends on single-UTXO checks. Has no effect when the server does not expose the tracker (older neutrino-api or operator-disabled). Set to false for chain-only behaviour.') class-attribute instance-attribute

neutrino_prefetch_filters: bool = Field(default=True, description='Enable background prefetch of compact block filters. Enabled by default because jm-wallet info scans these filters anyway, so prefetching saves time on the initial scan. With the default lookback of ~2 years, this takes ~3 hours on clearnet and ~3GB disk on mainnet. Disable to fetch filters on-demand only. When false, neutrino_prefetch_lookback_blocks is ignored.') class-attribute instance-attribute

neutrino_prefetch_lookback_blocks: int = Field(default=105120, description='When neutrino_prefetch_filters is true, only prefetch filters for this many recent blocks (~2 years at 105120 blocks). Set to 0 to prefetch all filters from genesis. Ignored when neutrino_prefetch_filters is false.') class-attribute instance-attribute

neutrino_scan_lookback_blocks: int = Field(default=105120, description='Number of blocks to look back from tip for wallet rescans when no explicit scan_start_height is set. Default ~2 years (105120 blocks).') class-attribute instance-attribute

neutrino_tls_cert: str | None = Field(default='neutrino/tls.cert', description='Path to neutrino-api TLS certificate for HTTPS verification. When the file exists, the neutrino backend pins it. When it is missing but the server is HTTPS, the backend fetches and pins the certificate on first use (trust-on-first-use) and writes it here. Relative paths are resolved against the data directory. Set to an empty string to disable certificate pinning.') class-attribute instance-attribute

neutrino_url: str = Field(default='http://127.0.0.1:8334', description='Neutrino REST API URL (for neutrino backend)') class-attribute instance-attribute

rpc_cookie_file: str | None = Field(default=None, description='Path to Bitcoin Core .cookie file for cookie-based RPC authentication. When set, the cookie file is read at startup and rpc_user/rpc_password are populated automatically. This is mutually exclusive with setting rpc_user/rpc_password manually. The cookie file is typically located at ~/.bitcoin/.cookie (mainnet) or ~/.bitcoin/regtest/.cookie (regtest).') class-attribute instance-attribute

rpc_password: SecretStr = Field(default=SecretStr(''), description='Bitcoin Core RPC password') class-attribute instance-attribute

rpc_url: str = Field(default='http://127.0.0.1:8332', description='Bitcoin Core RPC URL') class-attribute instance-attribute

rpc_user: str = Field(default='', description='Bitcoin Core RPC username') class-attribute instance-attribute

DirectoryServerSettings

Bases: BaseModel

Directory server specific settings.

Source code in jmcore/src/jmcore/settings.py

class DirectoryServerSettings(BaseModel):
    """Directory server specific settings."""

    host: str = Field(
        default="127.0.0.1",
        description="Host address to bind to",
    )
    port: int = Field(
        default=5222,
        ge=0,
        le=65535,
        description="Port to listen on (0 = let OS assign)",
    )
    nick_auth_mode: NickAuthMode = Field(
        default=NickAuthMode.PREFER_VERIFIED,
        description="Server policy for verifying client nick ownership",
    )
    nick_auth_directory_id: str | None = Field(
        default=None,
        description="Stable JMP-0005 identifier for this directory endpoint",
    )
    nick_auth_timeout: float = Field(
        default=30.0,
        gt=0.0,
        le=30.0,
        description="Seconds to wait for a negotiated nick authentication proof",
    )

    @field_validator("nick_auth_directory_id")
    @classmethod
    def validate_nick_auth_directory_id(cls, value: str | None) -> str | None:
        return validate_directory_id(value) if value is not None else None

    @model_validator(mode="after")
    def require_nick_auth_directory_id(self) -> Self:
        if (
            self.nick_auth_mode is NickAuthMode.REQUIRE_VERIFIED
            and self.nick_auth_directory_id is None
        ):
            raise ValueError("require_verified nick authentication needs nick_auth_directory_id")
        return self

    max_peers: int = Field(
        default=10000,
        ge=1,
        description="Maximum number of connected peers",
    )
    max_message_size: int = Field(
        default=2097152,
        ge=1024,
        description="Maximum message size in bytes (2MB default)",
    )
    max_line_length: int = Field(
        default=65536,
        ge=1024,
        description="Maximum JSON-line message length (64KB default)",
    )
    max_json_nesting_depth: int = Field(
        default=10,
        ge=1,
        description="Maximum nesting depth for JSON parsing",
    )
    message_rate_limit: int = Field(
        default=500,
        ge=1,
        description="Messages per second (sustained)",
    )
    message_burst_limit: int = Field(
        default=1000,
        ge=1,
        description="Maximum burst size",
    )
    rate_limit_disconnect_threshold: int = Field(
        default=0,
        ge=0,
        description="Disconnect after N rate limit violations (0 = never disconnect)",
    )
    broadcast_batch_size: int = Field(
        default=50,
        ge=1,
        description="Batch size for concurrent broadcasts",
    )
    health_check_host: str = Field(
        default="127.0.0.1",
        description="Host for health check endpoint",
    )
    health_check_port: int = Field(
        default=8080,
        ge=0,
        le=65535,
        description="Port for health check endpoint (0 = let OS assign)",
    )
    motd: str = Field(
        default="JoinMarket NG Directory Server https://github.com/joinmarket-ng/joinmarket-ng/",
        description="Message of the day sent to clients",
    )
    heartbeat_sweep_interval: float = Field(
        default=60.0,
        gt=0.0,
        description="Seconds between heartbeat sweep cycles (default 60)",
    )
    heartbeat_idle_threshold: float = Field(
        default=600.0,
        gt=0.0,
        description="Seconds of inactivity before probing a peer (default 600 = 10 min)",
    )
    heartbeat_hard_evict: float = Field(
        default=1500.0,
        gt=0.0,
        description="Seconds of inactivity before unconditional eviction (default 1500 = 25 min)",
    )
    heartbeat_pong_wait: float = Field(
        default=30.0,
        gt=0.0,
        description="Seconds to wait for PONG after sending PING (default 30)",
    )

Attributes

broadcast_batch_size: int = Field(default=50, ge=1, description='Batch size for concurrent broadcasts') class-attribute instance-attribute

health_check_host: str = Field(default='127.0.0.1', description='Host for health check endpoint') class-attribute instance-attribute

health_check_port: int = Field(default=8080, ge=0, le=65535, description='Port for health check endpoint (0 = let OS assign)') class-attribute instance-attribute

heartbeat_hard_evict: float = Field(default=1500.0, gt=0.0, description='Seconds of inactivity before unconditional eviction (default 1500 = 25 min)') class-attribute instance-attribute

heartbeat_idle_threshold: float = Field(default=600.0, gt=0.0, description='Seconds of inactivity before probing a peer (default 600 = 10 min)') class-attribute instance-attribute

heartbeat_pong_wait: float = Field(default=30.0, gt=0.0, description='Seconds to wait for PONG after sending PING (default 30)') class-attribute instance-attribute

heartbeat_sweep_interval: float = Field(default=60.0, gt=0.0, description='Seconds between heartbeat sweep cycles (default 60)') class-attribute instance-attribute

host: str = Field(default='127.0.0.1', description='Host address to bind to') class-attribute instance-attribute

max_json_nesting_depth: int = Field(default=10, ge=1, description='Maximum nesting depth for JSON parsing') class-attribute instance-attribute

max_line_length: int = Field(default=65536, ge=1024, description='Maximum JSON-line message length (64KB default)') class-attribute instance-attribute

max_message_size: int = Field(default=2097152, ge=1024, description='Maximum message size in bytes (2MB default)') class-attribute instance-attribute

max_peers: int = Field(default=10000, ge=1, description='Maximum number of connected peers') class-attribute instance-attribute

message_burst_limit: int = Field(default=1000, ge=1, description='Maximum burst size') class-attribute instance-attribute

message_rate_limit: int = Field(default=500, ge=1, description='Messages per second (sustained)') class-attribute instance-attribute

motd: str = Field(default='JoinMarket NG Directory Server https://github.com/joinmarket-ng/joinmarket-ng/', description='Message of the day sent to clients') class-attribute instance-attribute

nick_auth_directory_id: str | None = Field(default=None, description='Stable JMP-0005 identifier for this directory endpoint') class-attribute instance-attribute

nick_auth_mode: NickAuthMode = Field(default=NickAuthMode.PREFER_VERIFIED, description='Server policy for verifying client nick ownership') class-attribute instance-attribute

nick_auth_timeout: float = Field(default=30.0, gt=0.0, le=30.0, description='Seconds to wait for a negotiated nick authentication proof') class-attribute instance-attribute

port: int = Field(default=5222, ge=0, le=65535, description='Port to listen on (0 = let OS assign)') class-attribute instance-attribute

rate_limit_disconnect_threshold: int = Field(default=0, ge=0, description='Disconnect after N rate limit violations (0 = never disconnect)') class-attribute instance-attribute

Methods:

require_nick_auth_directory_id() -> Self

Source code in jmcore/src/jmcore/settings.py

@model_validator(mode="after")
def require_nick_auth_directory_id(self) -> Self:
    if (
        self.nick_auth_mode is NickAuthMode.REQUIRE_VERIFIED
        and self.nick_auth_directory_id is None
    ):
        raise ValueError("require_verified nick authentication needs nick_auth_directory_id")
    return self

validate_nick_auth_directory_id(value: str | None) -> str | None classmethod

Source code in jmcore/src/jmcore/settings.py

@field_validator("nick_auth_directory_id")
@classmethod
def validate_nick_auth_directory_id(cls, value: str | None) -> str | None:
    return validate_directory_id(value) if value is not None else None

JoinMarketSettings

Bases: BaseSettings

Main JoinMarket settings class.

Loads configuration from multiple sources with the following priority: 1. CLI arguments (not handled here, passed to component init) 2. Environment variables 3. TOML config file (~/.joinmarket-ng/config.toml) 4. Default values

Source code in jmcore/src/jmcore/settings.py

class JoinMarketSettings(BaseSettings):
    """
    Main JoinMarket settings class.

    Loads configuration from multiple sources with the following priority:
    1. CLI arguments (not handled here, passed to component __init__)
    2. Environment variables
    3. TOML config file (~/.joinmarket-ng/config.toml)
    4. Default values
    """

    model_config = SettingsConfigDict(
        env_prefix="",  # No prefix by default, use env_nested_delimiter for nested
        env_nested_delimiter="__",
        case_sensitive=False,
        extra="ignore",  # Ignore unknown fields (for forward compatibility)
    )

    # Marker for config file path discovery
    _config_file_path: ClassVar[Path | None] = None

    # Core settings
    data_dir: Path | None = Field(
        default=None,
        description="Data directory (defaults to ~/.joinmarket-ng)",
    )

    @field_validator("data_dir", mode="before")
    @classmethod
    def _expand_data_dir(cls, v: Any) -> Any:
        """Expand ``~``/``~user`` in ``data_dir`` from any source.

        Users frequently write ``data_dir = "~/.joinmarket-ng"`` in
        ``config.toml`` (or pass it via env). Without expansion pydantic
        stores it as a literal ``Path("~/.joinmarket-ng")``, which later makes
        ``migrate_config()`` create a bogus ``./~`` directory and causes the
        real config to be ignored (issue #536). Expand the home directory here
        so every consumer of ``data_dir`` sees an absolute path.
        """
        if v is None:
            return v
        if isinstance(v, str):
            v = v.strip()
            if not v:
                return None
        if isinstance(v, (str, Path)):
            return Path(v).expanduser()
        return v

    # Nested settings groups
    tor: TorSettings = Field(default_factory=TorSettings)
    bitcoin: BitcoinSettings = Field(default_factory=BitcoinSettings)
    network_config: NetworkSettings = Field(default_factory=NetworkSettings)
    wallet: WalletSettings = Field(default_factory=WalletSettings)
    notifications: NotificationSettings = Field(default_factory=NotificationSettings)
    logging: LoggingSettings = Field(default_factory=LoggingSettings)
    tui: TuiSettings = Field(default_factory=TuiSettings)

    # Component-specific settings
    maker: MakerSettings = Field(default_factory=MakerSettings)
    taker: TakerSettings = Field(default_factory=TakerSettings)
    tumbler: TumblerSettings = Field(default_factory=TumblerSettings)
    directory_server: DirectoryServerSettings = Field(default_factory=DirectoryServerSettings)
    orderbook_watcher: OrderbookWatcherSettings = Field(default_factory=OrderbookWatcherSettings)

    @classmethod
    def settings_customise_sources(
        cls,
        settings_cls: type[BaseSettings],
        init_settings: PydanticBaseSettingsSource,
        env_settings: PydanticBaseSettingsSource,
        dotenv_settings: PydanticBaseSettingsSource,
        file_secret_settings: PydanticBaseSettingsSource,
    ) -> tuple[PydanticBaseSettingsSource, ...]:
        """
        Customize settings sources and their priority.

        Priority (highest to lowest):
        1. init_settings (CLI arguments passed to constructor)
        2. env_settings (environment variables with __ delimiter)
        3. toml_settings (config.toml file)
        4. defaults (in field definitions)
        """
        toml_source = TomlConfigSettingsSource(settings_cls)
        comma_env = _CommaListEnvSettingsSource(settings_cls)
        return (
            InitSettingsSource(settings_cls, _normalize_wallet_preference_source(init_settings())),
            comma_env,
            toml_source,
        )

    def get_data_dir(self) -> Path:
        """Get the data directory, using default if not set."""
        if self.data_dir is not None:
            return self.data_dir
        return get_default_data_dir()

    def get_directory_servers(self) -> list[str]:
        """Get directory servers, using network defaults if not set."""
        if self.network_config.directory_servers:
            return self.network_config.directory_servers
        network_name = self.network_config.network.value
        return DEFAULT_DIRECTORY_SERVERS.get(network_name, [])

    def get_neutrino_add_peers(self) -> list[str]:
        """Get the configured neutrino add peers."""
        return self.bitcoin.neutrino_add_peers

Attributes

bitcoin: BitcoinSettings = Field(default_factory=BitcoinSettings) class-attribute instance-attribute

data_dir: Path | None = Field(default=None, description='Data directory (defaults to ~/.joinmarket-ng)') class-attribute instance-attribute

directory_server: DirectoryServerSettings = Field(default_factory=DirectoryServerSettings) class-attribute instance-attribute

logging: LoggingSettings = Field(default_factory=LoggingSettings) class-attribute instance-attribute

maker: MakerSettings = Field(default_factory=MakerSettings) class-attribute instance-attribute

model_config = SettingsConfigDict(env_prefix='', env_nested_delimiter='__', case_sensitive=False, extra='ignore') class-attribute instance-attribute

network_config: NetworkSettings = Field(default_factory=NetworkSettings) class-attribute instance-attribute

notifications: NotificationSettings = Field(default_factory=NotificationSettings) class-attribute instance-attribute

orderbook_watcher: OrderbookWatcherSettings = Field(default_factory=OrderbookWatcherSettings) class-attribute instance-attribute

taker: TakerSettings = Field(default_factory=TakerSettings) class-attribute instance-attribute

tor: TorSettings = Field(default_factory=TorSettings) class-attribute instance-attribute

tui: TuiSettings = Field(default_factory=TuiSettings) class-attribute instance-attribute

tumbler: TumblerSettings = Field(default_factory=TumblerSettings) class-attribute instance-attribute

wallet: WalletSettings = Field(default_factory=WalletSettings) class-attribute instance-attribute

Methods:

get_data_dir() -> Path

Get the data directory, using default if not set.

Source code in jmcore/src/jmcore/settings.py

def get_data_dir(self) -> Path:
    """Get the data directory, using default if not set."""
    if self.data_dir is not None:
        return self.data_dir
    return get_default_data_dir()

get_directory_servers() -> list[str]

Get directory servers, using network defaults if not set.

Source code in jmcore/src/jmcore/settings.py

def get_directory_servers(self) -> list[str]:
    """Get directory servers, using network defaults if not set."""
    if self.network_config.directory_servers:
        return self.network_config.directory_servers
    network_name = self.network_config.network.value
    return DEFAULT_DIRECTORY_SERVERS.get(network_name, [])

get_neutrino_add_peers() -> list[str]

Get the configured neutrino add peers.

Source code in jmcore/src/jmcore/settings.py

def get_neutrino_add_peers(self) -> list[str]:
    """Get the configured neutrino add peers."""
    return self.bitcoin.neutrino_add_peers

settings_customise_sources(settings_cls: type[BaseSettings], init_settings: PydanticBaseSettingsSource, env_settings: PydanticBaseSettingsSource, dotenv_settings: PydanticBaseSettingsSource, file_secret_settings: PydanticBaseSettingsSource) -> tuple[PydanticBaseSettingsSource, ...] classmethod

Customize settings sources and their priority.

Priority (highest to lowest): 1. init_settings (CLI arguments passed to constructor) 2. env_settings (environment variables with __ delimiter) 3. toml_settings (config.toml file) 4. defaults (in field definitions)

Source code in jmcore/src/jmcore/settings.py

@classmethod
def settings_customise_sources(
    cls,
    settings_cls: type[BaseSettings],
    init_settings: PydanticBaseSettingsSource,
    env_settings: PydanticBaseSettingsSource,
    dotenv_settings: PydanticBaseSettingsSource,
    file_secret_settings: PydanticBaseSettingsSource,
) -> tuple[PydanticBaseSettingsSource, ...]:
    """
    Customize settings sources and their priority.

    Priority (highest to lowest):
    1. init_settings (CLI arguments passed to constructor)
    2. env_settings (environment variables with __ delimiter)
    3. toml_settings (config.toml file)
    4. defaults (in field definitions)
    """
    toml_source = TomlConfigSettingsSource(settings_cls)
    comma_env = _CommaListEnvSettingsSource(settings_cls)
    return (
        InitSettingsSource(settings_cls, _normalize_wallet_preference_source(init_settings())),
        comma_env,
        toml_source,
    )

LoggingSettings

Bases: BaseModel

Logging configuration.

Source code in jmcore/src/jmcore/settings.py

class LoggingSettings(BaseModel):
    """Logging configuration."""

    level: str = Field(
        default="INFO",
        description="Log level: TRACE, DEBUG, INFO, WARNING, ERROR",
    )
    sensitive: bool = Field(
        default=False,
        description=(
            "Enable privacy-sensitive diagnostics: wallet addresses, amounts, balances, "
            "transaction IDs, transaction data, descriptors, and detailed errors. "
            "Disabled by default; review logs before sharing"
        ),
    )

Attributes

level: str = Field(default='INFO', description='Log level: TRACE, DEBUG, INFO, WARNING, ERROR') class-attribute instance-attribute

sensitive: bool = Field(default=False, description='Enable privacy-sensitive diagnostics: wallet addresses, amounts, balances, transaction IDs, transaction data, descriptors, and detailed errors. Disabled by default; review logs before sharing') class-attribute instance-attribute

MakerSettings

Bases: BaseModel

Maker-specific settings.

Source code in jmcore/src/jmcore/settings.py

class MakerSettings(BaseModel):
    """Maker-specific settings."""

    min_size: int = Field(
        default=100_000,
        ge=0,
        description=(
            "Minimum CoinJoin amount in satoshis. Default 100_000 matches the "
            "upstream JoinMarket reference (avoids fingerprinting jm-ng "
            "makers via different defaults -- see issue #468)."
        ),
    )
    min_fee_rate_sat_vb: float = Field(
        default=0.0,
        ge=0.0,
        allow_inf_nan=False,
        description="Static CoinJoin miner fee floor in sat/vB (0 disables the static floor)",
    )
    min_fee_block_target: int = Field(
        default=10,
        ge=-1,
        le=1008,
        description="Block target for the CoinJoin miner-fee floor (-1 disables estimation)",
    )

    @field_validator("min_fee_block_target")
    @classmethod
    def validate_fee_block_target(cls, value: int) -> int:
        return validate_min_fee_block_target(value)

    offer_type: str = Field(
        default="sw0reloffer",
        description="Offer type: sw0reloffer (relative) or sw0absoffer (absolute)",
    )
    cj_fee_relative: str = Field(
        default="0.0001",
        description=(
            "Relative CoinJoin fee. Default 0.0001 (0.01%) is a public fee-quantization quantum."
        ),
    )
    cj_fee_absolute: int = Field(
        default=500,
        ge=0,
        description="Absolute CoinJoin fee in satoshis",
    )
    tx_fee_contribution: int = Field(
        default=0,
        ge=0,
        description="Transaction fee contribution in satoshis",
    )
    cjfee_factor: float = Field(
        default=0.0,
        ge=0.0,
        description=(
            "Randomization factor applied to the CoinJoin fee on each offer "
            "announcement. The advertised fee is sampled from "
            "[cjfee*(1-f), cjfee*(1+f)]. Default 0 (no randomization) keeps a "
            "default maker exactly on its quantization quantum so it blends with "
            "other default makers. Enable randomization (e.g. 0.1, the upstream "
            "yg-privacyenhanced value) only when you use a non-quantized fee."
        ),
    )
    txfee_contribution_factor: float = Field(
        default=0.3,
        ge=0.0,
        description=(
            "Randomization factor applied to the tx fee contribution on each "
            "offer announcement. Default 0.3 matches the upstream JoinMarket "
            "reference."
        ),
    )
    size_factor: float = Field(
        default=0.1,
        ge=0.0,
        description=(
            "Downward randomization factor applied only to maxsize on each offer "
            "announcement. The minimum is not randomized. Set to 0 to disable. Default 0.1."
        ),
    )
    min_confirmations: int = Field(
        default=1,
        ge=1,
        description=(
            "Minimum confirmations for maker UTXOs offered into coinjoins. "
            "The base protocol requires at least one confirmation because "
            "takers reject unconfirmed maker inputs. The separate taker PoDLE "
            "commitment remains gated by taker_utxo_age (default 5)."
        ),
    )
    allow_mixdepth_zero_merge: bool = Field(
        default=False,
        description=(
            "Allow all mixdepth 0 UTXOs to merge instead of limiting automatic "
            "selection to maker-rotation lineage or one other UTXO "
            "(experienced makers only, reduces privacy)"
        ),
    )
    merge_algorithm: str = Field(
        default="default",
        description="UTXO selection: default, gradual, greedy, greediest, random",
    )
    mixdepth_selection_policy: str = Field(
        default="balanced",
        description=(
            "Source mixdepth policy: balanced (largest eligible balance) or "
            "concentrated (legacy cyclic-gap liquidity heuristic)"
        ),
    )
    session_timeout_sec: int = Field(
        default=300,
        ge=60,
        le=86_400,
        description="Maximum time for a CoinJoin session",
    )
    pre_sign_timeout_sec: int = Field(
        default=180,
        ge=60,
        le=3_600,
        description=(
            "Maximum seconds to wait for the taker's transaction after maker inputs "
            "have been reserved and disclosed"
        ),
    )
    identity_renewal_min_sec: int = Field(
        default=43_200,
        ge=60,
        description="Minimum randomized maker identity renewal interval in seconds",
    )
    identity_renewal_max_sec: int = Field(
        default=86_400,
        ge=60,
        description="Maximum randomized maker identity renewal interval in seconds",
    )
    identity_grace_sec: int = Field(
        default=300,
        ge=60,
        description="Fixed grace period for retired maker identity continuations",
    )
    identity_rotation_quiet_min_sec: int = Field(
        default=60,
        ge=0,
        description="Minimum quiet interval between old and replacement directory identities",
    )
    identity_rotation_quiet_max_sec: int = Field(
        default=600,
        ge=0,
        description="Maximum quiet interval between old and replacement directory identities",
    )
    pending_tx_timeout_min: int = Field(
        default=60,
        ge=10,
        le=1440,
        description=(
            "Minutes to monitor attempts without a recorded transaction ID; "
            "also sets the maker input reservation lifetime"
        ),
    )
    pending_tx_abandon_hours: int = Field(
        default=72,
        ge=1,
        le=8760,
        description=(
            "Hours to monitor a recorded transaction for confirmation before a local "
            "monitoring timeout. Explicit wallet refresh can still reconcile later "
            "confirmation. Default 72 h."
        ),
    )
    rescan_interval_sec: int = Field(
        default=600,
        ge=60,
        description="Interval for periodic wallet rescans",
    )
    # Hidden service settings
    onion_host: str | None = Field(
        default=None,
        description="Static hidden service address (e.g., 'mymaker...onion'). When not set, Tor control auto-generates one.",
    )
    onion_serving_host: str = Field(
        default="127.0.0.1",
        description="Bind address for incoming connections",
    )
    onion_serving_port: int = Field(
        default=5222,
        ge=0,
        le=65535,
        description="Port for incoming onion connections",
    )
    # Rate limiting
    message_rate_limit: int = Field(
        default=10,
        ge=1,
        description="Messages per second per peer (sustained)",
    )
    message_burst_limit: int = Field(
        default=100,
        ge=1,
        description="Maximum burst messages per peer",
    )
    offer_reannounce_delay_max: int = Field(
        default=600,
        ge=0,
        description=(
            "Maximum random delay in seconds before re-announcing offers "
            "after a balance change (0 = immediate, default: 600s = 10 minutes). "
            "Prevents observers from correlating block confirmations with "
            "offer updates."
        ),
    )
    # Directory reconnection
    directory_reconnect_interval: int = Field(
        default=300,
        ge=60,
        description="Interval in seconds between reconnection attempts for failed directories",
    )
    directory_reconnect_max_retries: int = Field(
        default=0,
        ge=0,
        description="Maximum reconnection attempts per directory (0 = unlimited)",
    )
    directory_startup_timeout: int = Field(
        default=120,
        ge=10,
        description=(
            "Seconds to keep retrying directory connections at startup before giving up "
            "and letting the background reconnect task take over"
        ),
    )
    # Aggregate response admissions, independent of per-peer limits.
    # Keep bursts within binary64's safe integer range for unit token decrements.
    directory_orderbook_response_burst: int = Field(
        default=200,
        ge=1,
        le=2**53 - 1,
        description="Maker-wide directory orderbook response burst capacity (requests)",
    )
    directory_orderbook_response_refill_per_second: float = Field(
        default=20.0,
        gt=0.0,
        allow_inf_nan=False,
        description="Maker-wide directory orderbook response refill rate (requests per second)",
    )
    direct_orderbook_response_burst: int = Field(
        default=20,
        ge=1,
        le=2**53 - 1,
        description="Maker-wide direct orderbook response burst capacity (requests)",
    )
    direct_orderbook_response_refill_per_second: float = Field(
        default=2.0,
        gt=0.0,
        allow_inf_nan=False,
        description="Maker-wide direct orderbook response refill rate (requests per second)",
    )

    # Orderbook rate limiting
    orderbook_rate_limit: int = Field(
        default=1,
        ge=1,
        description="Maximum orderbook responses per peer per interval",
    )
    orderbook_rate_interval: float = Field(
        default=10.0,
        ge=1.0,
        description="Interval in seconds for orderbook rate limiting",
    )
    orderbook_violation_ban_threshold: int = Field(
        default=100,
        ge=1,
        description="Ban peer after this many rate limit violations",
    )
    orderbook_violation_warning_threshold: int = Field(
        default=10,
        ge=1,
        description="Start exponential backoff after this many violations",
    )
    orderbook_violation_severe_threshold: int = Field(
        default=50,
        ge=1,
        description="Severe backoff threshold (higher penalty)",
    )
    orderbook_ban_duration: float = Field(
        default=3600.0,
        ge=60.0,
        description="Ban duration in seconds (default: 1 hour)",
    )
    # Dual-offer mode
    dual_offers: bool = Field(
        default=False,
        description=(
            "Advertise both a relative and an absolute fee offer simultaneously. "
            "Uses cj_fee_relative and cj_fee_absolute for the respective offer. "
            "The CLI flag --dual-offers overrides this setting."
        ),
    )

    @field_validator("cj_fee_relative", mode="before")
    @classmethod
    def normalize_cj_fee_relative(cls, v: str | float | int) -> str:
        """
        Normalize cj_fee_relative to avoid scientific notation.

        Pydantic may coerce float values (from env vars, TOML, or JSON) to strings,
        which can result in scientific notation for small values (e.g., 1e-05).
        The JoinMarket protocol expects decimal notation (e.g., 0.00001).
        """
        if isinstance(v, (int, float)):
            # Use Decimal to preserve precision and avoid scientific notation
            return format(Decimal(str(v)), "f")
        # Already a string - check if it contains scientific notation
        if "e" in v.lower():
            try:
                return format(Decimal(v), "f")
            except InvalidOperation:
                pass  # Let pydantic handle the validation error
        return v

    @model_validator(mode="after")
    def validate_identity_renewal(self) -> MakerSettings:
        """Ensure the configured randomized renewal interval is non-empty."""
        if self.identity_renewal_min_sec > self.identity_renewal_max_sec:
            raise ValueError("identity_renewal_min_sec must not exceed identity_renewal_max_sec")
        if self.identity_rotation_quiet_min_sec > self.identity_rotation_quiet_max_sec:
            raise ValueError(
                "identity_rotation_quiet_min_sec must not exceed identity_rotation_quiet_max_sec"
            )
        return self

Attributes

allow_mixdepth_zero_merge: bool = Field(default=False, description='Allow all mixdepth 0 UTXOs to merge instead of limiting automatic selection to maker-rotation lineage or one other UTXO (experienced makers only, reduces privacy)') class-attribute instance-attribute

cj_fee_absolute: int = Field(default=500, ge=0, description='Absolute CoinJoin fee in satoshis') class-attribute instance-attribute

cj_fee_relative: str = Field(default='0.0001', description='Relative CoinJoin fee. Default 0.0001 (0.01%) is a public fee-quantization quantum.') class-attribute instance-attribute

cjfee_factor: float = Field(default=0.0, ge=0.0, description='Randomization factor applied to the CoinJoin fee on each offer announcement. The advertised fee is sampled from [cjfee*(1-f), cjfee*(1+f)]. Default 0 (no randomization) keeps a default maker exactly on its quantization quantum so it blends with other default makers. Enable randomization (e.g. 0.1, the upstream yg-privacyenhanced value) only when you use a non-quantized fee.') class-attribute instance-attribute

direct_orderbook_response_burst: int = Field(default=20, ge=1, le=2 ** 53 - 1, description='Maker-wide direct orderbook response burst capacity (requests)') class-attribute instance-attribute

direct_orderbook_response_refill_per_second: float = Field(default=2.0, gt=0.0, allow_inf_nan=False, description='Maker-wide direct orderbook response refill rate (requests per second)') class-attribute instance-attribute

directory_orderbook_response_burst: int = Field(default=200, ge=1, le=2 ** 53 - 1, description='Maker-wide directory orderbook response burst capacity (requests)') class-attribute instance-attribute

directory_orderbook_response_refill_per_second: float = Field(default=20.0, gt=0.0, allow_inf_nan=False, description='Maker-wide directory orderbook response refill rate (requests per second)') class-attribute instance-attribute

directory_reconnect_interval: int = Field(default=300, ge=60, description='Interval in seconds between reconnection attempts for failed directories') class-attribute instance-attribute

directory_reconnect_max_retries: int = Field(default=0, ge=0, description='Maximum reconnection attempts per directory (0 = unlimited)') class-attribute instance-attribute

directory_startup_timeout: int = Field(default=120, ge=10, description='Seconds to keep retrying directory connections at startup before giving up and letting the background reconnect task take over') class-attribute instance-attribute

dual_offers: bool = Field(default=False, description='Advertise both a relative and an absolute fee offer simultaneously. Uses cj_fee_relative and cj_fee_absolute for the respective offer. The CLI flag --dual-offers overrides this setting.') class-attribute instance-attribute

identity_grace_sec: int = Field(default=300, ge=60, description='Fixed grace period for retired maker identity continuations') class-attribute instance-attribute

identity_renewal_max_sec: int = Field(default=86400, ge=60, description='Maximum randomized maker identity renewal interval in seconds') class-attribute instance-attribute

identity_renewal_min_sec: int = Field(default=43200, ge=60, description='Minimum randomized maker identity renewal interval in seconds') class-attribute instance-attribute

identity_rotation_quiet_max_sec: int = Field(default=600, ge=0, description='Maximum quiet interval between old and replacement directory identities') class-attribute instance-attribute

identity_rotation_quiet_min_sec: int = Field(default=60, ge=0, description='Minimum quiet interval between old and replacement directory identities') class-attribute instance-attribute

merge_algorithm: str = Field(default='default', description='UTXO selection: default, gradual, greedy, greediest, random') class-attribute instance-attribute

message_burst_limit: int = Field(default=100, ge=1, description='Maximum burst messages per peer') class-attribute instance-attribute

message_rate_limit: int = Field(default=10, ge=1, description='Messages per second per peer (sustained)') class-attribute instance-attribute

min_confirmations: int = Field(default=1, ge=1, description='Minimum confirmations for maker UTXOs offered into coinjoins. The base protocol requires at least one confirmation because takers reject unconfirmed maker inputs. The separate taker PoDLE commitment remains gated by taker_utxo_age (default 5).') class-attribute instance-attribute

min_fee_block_target: int = Field(default=10, ge=-1, le=1008, description='Block target for the CoinJoin miner-fee floor (-1 disables estimation)') class-attribute instance-attribute

min_fee_rate_sat_vb: float = Field(default=0.0, ge=0.0, allow_inf_nan=False, description='Static CoinJoin miner fee floor in sat/vB (0 disables the static floor)') class-attribute instance-attribute

min_size: int = Field(default=100000, ge=0, description='Minimum CoinJoin amount in satoshis. Default 100_000 matches the upstream JoinMarket reference (avoids fingerprinting jm-ng makers via different defaults -- see issue #468).') class-attribute instance-attribute

mixdepth_selection_policy: str = Field(default='balanced', description='Source mixdepth policy: balanced (largest eligible balance) or concentrated (legacy cyclic-gap liquidity heuristic)') class-attribute instance-attribute

offer_reannounce_delay_max: int = Field(default=600, ge=0, description='Maximum random delay in seconds before re-announcing offers after a balance change (0 = immediate, default: 600s = 10 minutes). Prevents observers from correlating block confirmations with offer updates.') class-attribute instance-attribute

offer_type: str = Field(default='sw0reloffer', description='Offer type: sw0reloffer (relative) or sw0absoffer (absolute)') class-attribute instance-attribute

onion_host: str | None = Field(default=None, description="Static hidden service address (e.g., 'mymaker...onion'). When not set, Tor control auto-generates one.") class-attribute instance-attribute

onion_serving_host: str = Field(default='127.0.0.1', description='Bind address for incoming connections') class-attribute instance-attribute

onion_serving_port: int = Field(default=5222, ge=0, le=65535, description='Port for incoming onion connections') class-attribute instance-attribute

orderbook_ban_duration: float = Field(default=3600.0, ge=60.0, description='Ban duration in seconds (default: 1 hour)') class-attribute instance-attribute

orderbook_rate_interval: float = Field(default=10.0, ge=1.0, description='Interval in seconds for orderbook rate limiting') class-attribute instance-attribute

orderbook_rate_limit: int = Field(default=1, ge=1, description='Maximum orderbook responses per peer per interval') class-attribute instance-attribute

orderbook_violation_ban_threshold: int = Field(default=100, ge=1, description='Ban peer after this many rate limit violations') class-attribute instance-attribute

orderbook_violation_severe_threshold: int = Field(default=50, ge=1, description='Severe backoff threshold (higher penalty)') class-attribute instance-attribute

orderbook_violation_warning_threshold: int = Field(default=10, ge=1, description='Start exponential backoff after this many violations') class-attribute instance-attribute

pending_tx_abandon_hours: int = Field(default=72, ge=1, le=8760, description='Hours to monitor a recorded transaction for confirmation before a local monitoring timeout. Explicit wallet refresh can still reconcile later confirmation. Default 72 h.') class-attribute instance-attribute

pending_tx_timeout_min: int = Field(default=60, ge=10, le=1440, description='Minutes to monitor attempts without a recorded transaction ID; also sets the maker input reservation lifetime') class-attribute instance-attribute

pre_sign_timeout_sec: int = Field(default=180, ge=60, le=3600, description="Maximum seconds to wait for the taker's transaction after maker inputs have been reserved and disclosed") class-attribute instance-attribute

rescan_interval_sec: int = Field(default=600, ge=60, description='Interval for periodic wallet rescans') class-attribute instance-attribute

session_timeout_sec: int = Field(default=300, ge=60, le=86400, description='Maximum time for a CoinJoin session') class-attribute instance-attribute

size_factor: float = Field(default=0.1, ge=0.0, description='Downward randomization factor applied only to maxsize on each offer announcement. The minimum is not randomized. Set to 0 to disable. Default 0.1.') class-attribute instance-attribute

tx_fee_contribution: int = Field(default=0, ge=0, description='Transaction fee contribution in satoshis') class-attribute instance-attribute

txfee_contribution_factor: float = Field(default=0.3, ge=0.0, description='Randomization factor applied to the tx fee contribution on each offer announcement. Default 0.3 matches the upstream JoinMarket reference.') class-attribute instance-attribute

Methods:

normalize_cj_fee_relative(v: str | float | int) -> str classmethod

Normalize cj_fee_relative to avoid scientific notation.

Pydantic may coerce float values (from env vars, TOML, or JSON) to strings, which can result in scientific notation for small values (e.g., 1e-05). The JoinMarket protocol expects decimal notation (e.g., 0.00001).

Source code in jmcore/src/jmcore/settings.py

@field_validator("cj_fee_relative", mode="before")
@classmethod
def normalize_cj_fee_relative(cls, v: str | float | int) -> str:
    """
    Normalize cj_fee_relative to avoid scientific notation.

    Pydantic may coerce float values (from env vars, TOML, or JSON) to strings,
    which can result in scientific notation for small values (e.g., 1e-05).
    The JoinMarket protocol expects decimal notation (e.g., 0.00001).
    """
    if isinstance(v, (int, float)):
        # Use Decimal to preserve precision and avoid scientific notation
        return format(Decimal(str(v)), "f")
    # Already a string - check if it contains scientific notation
    if "e" in v.lower():
        try:
            return format(Decimal(v), "f")
        except InvalidOperation:
            pass  # Let pydantic handle the validation error
    return v

validate_fee_block_target(value: int) -> int classmethod

Source code in jmcore/src/jmcore/settings.py

@field_validator("min_fee_block_target")
@classmethod
def validate_fee_block_target(cls, value: int) -> int:
    return validate_min_fee_block_target(value)

validate_identity_renewal() -> MakerSettings

Ensure the configured randomized renewal interval is non-empty.

Source code in jmcore/src/jmcore/settings.py

@model_validator(mode="after")
def validate_identity_renewal(self) -> MakerSettings:
    """Ensure the configured randomized renewal interval is non-empty."""
    if self.identity_renewal_min_sec > self.identity_renewal_max_sec:
        raise ValueError("identity_renewal_min_sec must not exceed identity_renewal_max_sec")
    if self.identity_rotation_quiet_min_sec > self.identity_rotation_quiet_max_sec:
        raise ValueError(
            "identity_rotation_quiet_min_sec must not exceed identity_rotation_quiet_max_sec"
        )
    return self

NetworkSettings

Bases: BaseModel

Network configuration.

Source code in jmcore/src/jmcore/settings.py

class NetworkSettings(BaseModel):
    """Network configuration."""

    network: NetworkType = Field(
        default=NetworkType.MAINNET,
        description="JoinMarket protocol network (mainnet, testnet, signet, regtest)",
    )
    bitcoin_network: NetworkType | None = Field(
        default=None,
        description="Bitcoin network for address generation (defaults to network)",
    )
    directory_servers: list[str] = Field(
        default_factory=list,
        description="Directory server addresses (host:port). Uses defaults if empty.",
    )
    allow_clearnet_connections: bool = Field(
        default=False,
        description=(
            "Allow direct TCP connections to non-onion JoinMarket directories and peers. "
            "Development and local testing only; regtest permits local connections without this."
        ),
    )
    nick_auth_mode: NickAuthMode = Field(
        default=NickAuthMode.PREFER_VERIFIED,
        description="Client policy for authenticating nick ownership to directory servers",
    )
    nick_auth_directory_ids: dict[str, str] = Field(
        default_factory=dict,
        description="Expected nick authentication identity by selected host:port endpoint",
    )

    @field_validator("directory_servers", mode="before")
    @classmethod
    def parse_directory_servers(cls, v: Any) -> Any:
        """Accept JSON arrays, comma-separated strings, or plain single values.

        pydantic-settings passes list[str] env vars through json.loads(), which
        requires JSON format. This validator also accepts plain comma-separated
        strings so that e.g. NETWORK_CONFIG__DIRECTORY_SERVERS=host:port works.
        """
        import json

        if isinstance(v, str):
            v = v.strip()
            # Try JSON first (handles '["a","b"]' or '"a"' forms)
            try:
                parsed = json.loads(v)
                if isinstance(parsed, list):
                    return [s.strip() for s in parsed if s.strip()]
                if isinstance(parsed, str):
                    return [parsed.strip()] if parsed.strip() else []
            except (json.JSONDecodeError, ValueError):
                pass
            # Fall back to comma-separated plain string
            return [s.strip() for s in v.split(",") if s.strip()]
        return v

    @field_validator("nick_auth_directory_ids")
    @classmethod
    def validate_nick_auth_directory_ids(cls, value: dict[str, str]) -> dict[str, str]:
        return {
            validate_directory_endpoint(endpoint): validate_directory_id(directory_id)
            for endpoint, directory_id in value.items()
        }

Attributes

allow_clearnet_connections: bool = Field(default=False, description='Allow direct TCP connections to non-onion JoinMarket directories and peers. Development and local testing only; regtest permits local connections without this.') class-attribute instance-attribute

bitcoin_network: NetworkType | None = Field(default=None, description='Bitcoin network for address generation (defaults to network)') class-attribute instance-attribute

directory_servers: list[str] = Field(default_factory=list, description='Directory server addresses (host:port). Uses defaults if empty.') class-attribute instance-attribute

network: NetworkType = Field(default=NetworkType.MAINNET, description='JoinMarket protocol network (mainnet, testnet, signet, regtest)') class-attribute instance-attribute

nick_auth_directory_ids: dict[str, str] = Field(default_factory=dict, description='Expected nick authentication identity by selected host:port endpoint') class-attribute instance-attribute

nick_auth_mode: NickAuthMode = Field(default=NickAuthMode.PREFER_VERIFIED, description='Client policy for authenticating nick ownership to directory servers') class-attribute instance-attribute

Methods:

parse_directory_servers(v: Any) -> Any classmethod

Accept JSON arrays, comma-separated strings, or plain single values.

pydantic-settings passes list[str] env vars through json.loads(), which requires JSON format. This validator also accepts plain comma-separated strings so that e.g. NETWORK_CONFIG__DIRECTORY_SERVERS=host:port works.

Source code in jmcore/src/jmcore/settings.py

@field_validator("directory_servers", mode="before")
@classmethod
def parse_directory_servers(cls, v: Any) -> Any:
    """Accept JSON arrays, comma-separated strings, or plain single values.

    pydantic-settings passes list[str] env vars through json.loads(), which
    requires JSON format. This validator also accepts plain comma-separated
    strings so that e.g. NETWORK_CONFIG__DIRECTORY_SERVERS=host:port works.
    """
    import json

    if isinstance(v, str):
        v = v.strip()
        # Try JSON first (handles '["a","b"]' or '"a"' forms)
        try:
            parsed = json.loads(v)
            if isinstance(parsed, list):
                return [s.strip() for s in parsed if s.strip()]
            if isinstance(parsed, str):
                return [parsed.strip()] if parsed.strip() else []
        except (json.JSONDecodeError, ValueError):
            pass
        # Fall back to comma-separated plain string
        return [s.strip() for s in v.split(",") if s.strip()]
    return v

validate_nick_auth_directory_ids(value: dict[str, str]) -> dict[str, str] classmethod

Source code in jmcore/src/jmcore/settings.py

@field_validator("nick_auth_directory_ids")
@classmethod
def validate_nick_auth_directory_ids(cls, value: dict[str, str]) -> dict[str, str]:
    return {
        validate_directory_endpoint(endpoint): validate_directory_id(directory_id)
        for endpoint, directory_id in value.items()
    }

NotificationSettings

Bases: BaseModel

Notification system configuration.

Source code in jmcore/src/jmcore/settings.py

class NotificationSettings(BaseModel):
    """Notification system configuration."""

    enabled: bool = Field(
        default=False,
        description="Enable notifications (requires urls to be set)",
    )
    urls: list[str] = Field(
        default_factory=list,
        description='Apprise notification URLs (e.g., ["tgram://bottoken/ChatID", "gotify://hostname/token"])',
    )
    title_prefix: str = Field(
        default="JoinMarket NG",
        description="Prefix for notification titles",
    )
    component_name: str = Field(
        default="",
        description="Component name in notification titles (e.g., 'Maker', 'Taker'). "
        "Usually set programmatically by each component.",
    )
    include_amounts: bool = Field(
        default=True,
        description="Include amounts in notifications",
    )
    include_txids: bool = Field(
        default=False,
        description="Include transaction IDs in notifications (privacy risk)",
    )
    include_coinjoin_id: bool = Field(
        default=True,
        description="Include commitment-derived CoinJoin IDs in notifications",
    )
    mempool_url: str = Field(
        default="",
        description=(
            "Base URL of a mempool/block explorer. When set (and include_txids is "
            "enabled), CoinJoin notifications render the full txid as a clickable "
            "link '<mempool_url>/tx/<txid>' instead of a bare id. Example: "
            "'https://mempool.space' (mainnet) or 'https://mempool.space/signet'."
        ),
    )
    include_nick: bool = Field(
        default=True,
        description="Include peer nicks in notifications",
    )
    use_tor: bool = Field(
        default=True,
        description=(
            "Route notification delivery through the Tor SOCKS proxy. Set to false for "
            "a LAN notification server. Does not affect update checks, which are "
            "always Tor-only."
        ),
    )
    verify_tls: bool = Field(
        default=True,
        description=(
            "Verify TLS certificates of notification servers. Set to false for "
            "servers using self-signed certificates or a private CA that is not "
            "in the certifi bundle (alternatively, point the REQUESTS_CA_BUNDLE "
            "environment variable at your CA bundle to keep verification on)."
        ),
    )
    # Event type toggles
    notify_fill: bool = Field(default=True, description="Notify on !fill requests")
    notify_rejection: bool = Field(default=True, description="Notify on rejections")
    notify_signing: bool = Field(default=True, description="Notify on transaction signing")
    notify_mempool: bool = Field(default=True, description="Notify on mempool detection")
    notify_confirmed: bool = Field(default=True, description="Notify on confirmation")
    notify_nick_change: bool = Field(default=False, description="Notify on nick change")
    notify_disconnect: bool = Field(
        default=False,
        description="Notify on individual directory server disconnect/reconnect (noisy)",
    )
    notify_all_disconnect: bool = Field(
        default=True,
        description="Notify when ALL directory servers are disconnected (critical)",
    )
    notify_coinjoin_start: bool = Field(default=True, description="Notify on CoinJoin start")
    notify_coinjoin_complete: bool = Field(default=True, description="Notify on CoinJoin complete")
    notify_coinjoin_failed: bool = Field(default=True, description="Notify on CoinJoin failure")
    notify_peer_events: bool = Field(default=False, description="Notify on peer connect/disconnect")
    notify_rate_limit: bool = Field(default=True, description="Notify on rate limit bans")
    notify_startup: bool = Field(default=True, description="Notify on component startup")
    notify_summary: bool = Field(
        default=True,
        description="Send periodic summary notifications with CoinJoin stats",
    )
    notify_summary_balance: bool = Field(
        default=False,
        description=(
            "Include total wallet balance and UTXO count in periodic summary notifications. "
            "Disabled by default for privacy."
        ),
    )
    summary_interval_hours: int = Field(
        default=24,
        ge=1,
        le=168,
        description=(
            "Interval in hours between summary notifications (1-168). "
            "Common values: 24 (daily), 168 (weekly)"
        ),
    )
    check_for_updates: bool = Field(
        default=False,
        description=(
            "Check GitHub for new releases and include version info in summary notifications. "
            "PRIVACY WARNING: This contacts github.com each summary interval. "
            "The request is always routed through Tor (independent of use_tor, which only "
            "affects notification delivery) and is skipped when the Tor proxy is unavailable. "
            "GitHub will still see the Tor exit node IP. Opt-in only."
        ),
    )
    # Retry settings
    retry_enabled: bool = Field(
        default=True,
        description=(
            "Retry failed notifications in the background with exponential backoff. "
            "Recommended when routing through Tor where transient failures are common."
        ),
    )
    retry_max_attempts: int = Field(
        default=3,
        ge=1,
        le=10,
        description="Maximum number of retry attempts for a failed notification (1-10)",
    )
    retry_base_delay: float = Field(
        default=5.0,
        ge=1.0,
        le=60.0,
        description=(
            "Base delay in seconds before the first retry (1-60). "
            "Subsequent retries double this delay (exponential backoff)."
        ),
    )

    @field_validator("urls", mode="before")
    @classmethod
    def parse_urls(cls, v: Any) -> Any:
        """Accept a single URL string in addition to a list of URLs.

        ``urls`` must be a TOML array (``urls = ["tgram://bottoken/ChatID"]``),
        but users frequently write a bare scalar instead
        (``urls = "tgram://bottoken/ChatID"``). Previously that raised a
        confusing ``ValidationError`` that crashed the component at startup
        (a systemd unit just reported ``status=1/FAILURE``). Coerce a single
        string (optionally comma-separated, or a JSON array) into a list so
        the friendlier form keeps working, mirroring ``parse_directory_servers``
        and the comma-separated env handling in ``_CommaListEnvSettingsSource``.
        """
        import json

        if isinstance(v, str):
            v = v.strip()
            if not v:
                return []
            # Try JSON first (handles '["a","b"]' or '"a"' forms).
            try:
                parsed = json.loads(v)
                if isinstance(parsed, list):
                    return [s.strip() for s in parsed if isinstance(s, str) and s.strip()]
                if isinstance(parsed, str):
                    return [parsed.strip()] if parsed.strip() else []
            except (json.JSONDecodeError, ValueError):
                pass
            # Fall back to a comma-separated plain string.
            return [s.strip() for s in v.split(",") if s.strip()]
        return v

Attributes

check_for_updates: bool = Field(default=False, description='Check GitHub for new releases and include version info in summary notifications. PRIVACY WARNING: This contacts github.com each summary interval. The request is always routed through Tor (independent of use_tor, which only affects notification delivery) and is skipped when the Tor proxy is unavailable. GitHub will still see the Tor exit node IP. Opt-in only.') class-attribute instance-attribute

component_name: str = Field(default='', description="Component name in notification titles (e.g., 'Maker', 'Taker'). Usually set programmatically by each component.") class-attribute instance-attribute

enabled: bool = Field(default=False, description='Enable notifications (requires urls to be set)') class-attribute instance-attribute

include_amounts: bool = Field(default=True, description='Include amounts in notifications') class-attribute instance-attribute

include_coinjoin_id: bool = Field(default=True, description='Include commitment-derived CoinJoin IDs in notifications') class-attribute instance-attribute

include_nick: bool = Field(default=True, description='Include peer nicks in notifications') class-attribute instance-attribute

include_txids: bool = Field(default=False, description='Include transaction IDs in notifications (privacy risk)') class-attribute instance-attribute

mempool_url: str = Field(default='', description="Base URL of a mempool/block explorer. When set (and include_txids is enabled), CoinJoin notifications render the full txid as a clickable link '<mempool_url>/tx/<txid>' instead of a bare id. Example: 'https://mempool.space' (mainnet) or 'https://mempool.space/signet'.") class-attribute instance-attribute

notify_all_disconnect: bool = Field(default=True, description='Notify when ALL directory servers are disconnected (critical)') class-attribute instance-attribute

notify_coinjoin_complete: bool = Field(default=True, description='Notify on CoinJoin complete') class-attribute instance-attribute

notify_coinjoin_failed: bool = Field(default=True, description='Notify on CoinJoin failure') class-attribute instance-attribute

notify_coinjoin_start: bool = Field(default=True, description='Notify on CoinJoin start') class-attribute instance-attribute

notify_confirmed: bool = Field(default=True, description='Notify on confirmation') class-attribute instance-attribute

notify_disconnect: bool = Field(default=False, description='Notify on individual directory server disconnect/reconnect (noisy)') class-attribute instance-attribute

notify_fill: bool = Field(default=True, description='Notify on !fill requests') class-attribute instance-attribute

notify_mempool: bool = Field(default=True, description='Notify on mempool detection') class-attribute instance-attribute

notify_nick_change: bool = Field(default=False, description='Notify on nick change') class-attribute instance-attribute

notify_peer_events: bool = Field(default=False, description='Notify on peer connect/disconnect') class-attribute instance-attribute

notify_rate_limit: bool = Field(default=True, description='Notify on rate limit bans') class-attribute instance-attribute

notify_rejection: bool = Field(default=True, description='Notify on rejections') class-attribute instance-attribute

notify_signing: bool = Field(default=True, description='Notify on transaction signing') class-attribute instance-attribute

notify_startup: bool = Field(default=True, description='Notify on component startup') class-attribute instance-attribute

notify_summary: bool = Field(default=True, description='Send periodic summary notifications with CoinJoin stats') class-attribute instance-attribute

notify_summary_balance: bool = Field(default=False, description='Include total wallet balance and UTXO count in periodic summary notifications. Disabled by default for privacy.') class-attribute instance-attribute

retry_base_delay: float = Field(default=5.0, ge=1.0, le=60.0, description='Base delay in seconds before the first retry (1-60). Subsequent retries double this delay (exponential backoff).') class-attribute instance-attribute

retry_enabled: bool = Field(default=True, description='Retry failed notifications in the background with exponential backoff. Recommended when routing through Tor where transient failures are common.') class-attribute instance-attribute

retry_max_attempts: int = Field(default=3, ge=1, le=10, description='Maximum number of retry attempts for a failed notification (1-10)') class-attribute instance-attribute

summary_interval_hours: int = Field(default=24, ge=1, le=168, description='Interval in hours between summary notifications (1-168). Common values: 24 (daily), 168 (weekly)') class-attribute instance-attribute

title_prefix: str = Field(default='JoinMarket NG', description='Prefix for notification titles') class-attribute instance-attribute

urls: list[str] = Field(default_factory=list, description='Apprise notification URLs (e.g., ["tgram://bottoken/ChatID", "gotify://hostname/token"])') class-attribute instance-attribute

use_tor: bool = Field(default=True, description='Route notification delivery through the Tor SOCKS proxy. Set to false for a LAN notification server. Does not affect update checks, which are always Tor-only.') class-attribute instance-attribute

verify_tls: bool = Field(default=True, description='Verify TLS certificates of notification servers. Set to false for servers using self-signed certificates or a private CA that is not in the certifi bundle (alternatively, point the REQUESTS_CA_BUNDLE environment variable at your CA bundle to keep verification on).') class-attribute instance-attribute

Methods:

parse_urls(v: Any) -> Any classmethod

Accept a single URL string in addition to a list of URLs.

urls must be a TOML array (urls = ["tgram://bottoken/ChatID"]), but users frequently write a bare scalar instead (urls = "tgram://bottoken/ChatID"). Previously that raised a confusing ValidationError that crashed the component at startup (a systemd unit just reported status=1/FAILURE). Coerce a single string (optionally comma-separated, or a JSON array) into a list so the friendlier form keeps working, mirroring parse_directory_servers and the comma-separated env handling in _CommaListEnvSettingsSource.

Source code in jmcore/src/jmcore/settings.py

@field_validator("urls", mode="before")
@classmethod
def parse_urls(cls, v: Any) -> Any:
    """Accept a single URL string in addition to a list of URLs.

    ``urls`` must be a TOML array (``urls = ["tgram://bottoken/ChatID"]``),
    but users frequently write a bare scalar instead
    (``urls = "tgram://bottoken/ChatID"``). Previously that raised a
    confusing ``ValidationError`` that crashed the component at startup
    (a systemd unit just reported ``status=1/FAILURE``). Coerce a single
    string (optionally comma-separated, or a JSON array) into a list so
    the friendlier form keeps working, mirroring ``parse_directory_servers``
    and the comma-separated env handling in ``_CommaListEnvSettingsSource``.
    """
    import json

    if isinstance(v, str):
        v = v.strip()
        if not v:
            return []
        # Try JSON first (handles '["a","b"]' or '"a"' forms).
        try:
            parsed = json.loads(v)
            if isinstance(parsed, list):
                return [s.strip() for s in parsed if isinstance(s, str) and s.strip()]
            if isinstance(parsed, str):
                return [parsed.strip()] if parsed.strip() else []
        except (json.JSONDecodeError, ValueError):
            pass
        # Fall back to a comma-separated plain string.
        return [s.strip() for s in v.split(",") if s.strip()]
    return v

OrderbookWatcherSettings

Bases: BaseModel

Orderbook watcher specific settings.

Source code in jmcore/src/jmcore/settings.py

class OrderbookWatcherSettings(BaseModel):
    """Orderbook watcher specific settings."""

    http_host: str = Field(
        default="127.0.0.1",
        description="HTTP server bind address",
    )
    http_port: int = Field(
        default=8000,
        ge=1,
        le=65535,
        description="HTTP server port",
    )
    update_interval: int = Field(
        default=60,
        ge=10,
        description="Update interval in seconds",
    )
    mempool_api_url: str = Field(
        default="",
        description="Mempool API URL for transaction lookups",
    )
    mempool_api_use_tor: bool = Field(
        default=True,
        description="Route mempool API requests through Tor SOCKS",
    )
    mempool_web_url: str | None = Field(
        default=None,
        description="Mempool web URL for human-readable links",
    )
    uptime_grace_period: int = Field(
        default=60,
        ge=0,
        description="Grace period before tracking uptime",
    )
    max_message_size: int = Field(
        default=2097152,
        ge=1024,
        description="Maximum message size in bytes (2MB default)",
    )
    connection_timeout: float = Field(
        default=120.0,
        gt=0.0,
        description=(
            "Timeout in seconds for Tor SOCKS5 connections. Covers TCP handshake, "
            "SOCKS5 negotiation, Tor circuit building, and PoW solving."
        ),
    )

Attributes

connection_timeout: float = Field(default=120.0, gt=0.0, description='Timeout in seconds for Tor SOCKS5 connections. Covers TCP handshake, SOCKS5 negotiation, Tor circuit building, and PoW solving.') class-attribute instance-attribute

http_host: str = Field(default='127.0.0.1', description='HTTP server bind address') class-attribute instance-attribute

http_port: int = Field(default=8000, ge=1, le=65535, description='HTTP server port') class-attribute instance-attribute

max_message_size: int = Field(default=2097152, ge=1024, description='Maximum message size in bytes (2MB default)') class-attribute instance-attribute

mempool_api_url: str = Field(default='', description='Mempool API URL for transaction lookups') class-attribute instance-attribute

mempool_api_use_tor: bool = Field(default=True, description='Route mempool API requests through Tor SOCKS') class-attribute instance-attribute

mempool_web_url: str | None = Field(default=None, description='Mempool web URL for human-readable links') class-attribute instance-attribute

update_interval: int = Field(default=60, ge=10, description='Update interval in seconds') class-attribute instance-attribute

uptime_grace_period: int = Field(default=60, ge=0, description='Grace period before tracking uptime') class-attribute instance-attribute

TakerSettings

Bases: BaseModel

Taker-specific settings.

Source code in jmcore/src/jmcore/settings.py

class TakerSettings(BaseModel):
    """Taker-specific settings."""

    counterparty_count: int | None = Field(
        default=None,
        ge=1,
        le=20,
        description=(
            "Number of makers to select for CoinJoin. When unset (the default), "
            "the taker picks a random value in [8, 10] for each CoinJoin. This "
            "matches the upstream JoinMarket sendpayment behaviour and prevents "
            "fingerprinting jm-ng takers via a fixed maker count."
        ),
    )
    min_fee_rate_sat_vb: float = Field(
        default=1.0,
        gt=0.0,
        allow_inf_nan=False,
        description="Minimum CoinJoin miner fee rate in sat/vB",
    )
    min_fee_block_target: int = Field(
        default=10,
        ge=1,
        le=1008,
        description="Block target for the conservative CoinJoin miner-fee floor",
    )
    max_cj_fee_abs: int = Field(
        default=500,
        ge=0,
        description="Maximum absolute CoinJoin fee in satoshis",
    )
    max_maker_utxos: int = Field(
        default=15,
        ge=0,
        description=(
            "Maximum number of inputs a single maker may contribute to the "
            "CoinJoin. The taker pays the mining fee for every input, so an "
            "unbounded count lets a counterparty consolidate its UTXOs at the "
            "taker's expense. Makers exceeding the cap are dropped (and replaced "
            "when possible). 0 disables the cap (not recommended)."
        ),
    )
    taker_utxo_age: int = Field(
        default=5,
        ge=1,
        description=(
            "Minimum confirmations required on a UTXO before the taker may use "
            "it as a PoDLE commitment input. Reference default: 5."
        ),
    )
    taker_utxo_retries: int = Field(
        default=3,
        ge=1,
        le=10,
        description="Maximum PoDLE index retries per UTXO (reference default: 3)",
    )
    taker_utxo_amtpercent: int = Field(
        default=20,
        ge=1,
        le=100,
        description=(
            "Minimum UTXO value as a percentage of the CoinJoin amount for "
            "PoDLE commitments (reference default: 20)."
        ),
    )
    max_cj_fee_rel: str = Field(
        default="0.001",
        description="Maximum relative CoinJoin fee (0.001 = 0.1%)",
    )
    require_quantized_cj_fees: bool = Field(
        default=True,
        description="Only consider offers whose advertised CoinJoin fee is on the public grid",
    )
    round_up_cj_fees: bool = Field(
        default=False,
        description="Round each selected maker fee up to the next public fee quantum",
    )
    equalize_cj_fees: bool = Field(
        default=False,
        description="Pay every selected maker the highest realized fee in the selected set",
    )
    max_sweep_fee_change: float = Field(
        default=0.8,
        ge=0.0,
        allow_inf_nan=False,
        description=("Maximum relative amount a sweep fee estimate may exceed its selected budget"),
    )

    @field_validator("max_cj_fee_rel", mode="before")
    @classmethod
    def normalize_max_cj_fee_rel(cls, v: str | float | int) -> str:
        """Normalize to avoid scientific notation (same as MakerSettings.cj_fee_relative)."""
        if isinstance(v, (int, float)):
            return format(Decimal(str(v)), "f")
        if isinstance(v, str) and "e" in v.lower():
            try:
                return format(Decimal(v), "f")
            except InvalidOperation:
                pass
        return v

    tx_fee_factor: float = Field(
        default=0.2,
        ge=0.0,
        description=(
            "Fee randomization factor. 0 disables randomization; "
            "0.2 randomizes up to 20% above the base rate."
        ),
    )
    fee_rate: float | None = Field(
        default=None,
        gt=0.0,
        description="Manual fee rate in sat/vB (mutually exclusive with fee_block_target)",
    )
    fee_block_target: int | None = Field(
        default=None,
        ge=1,
        le=1008,
        description="Target blocks for fee estimation (mutually exclusive with fee_rate)",
    )
    bondless_makers_allowance: float = Field(
        default=0.05,
        ge=0.0,
        le=1.0,
        description="Per-slot probability of selecting uniformly from zero-fee offers",
    )
    bond_value_exponent: float = Field(
        default=1.3,
        gt=0.0,
        description="Exponent for fidelity bond value calculation",
    )
    bondless_require_zero_fee: bool = Field(
        default=True,
        description=(
            "Restrict allowance spots to zero-fee offers and reject fee-charging bondless offers"
        ),
    )
    maker_timeout_sec: int = Field(
        default=60,
        ge=10,
        le=3600,
        description="Timeout for maker responses",
    )
    initial_confirmation_timeout_sec: int = Field(
        default=300,
        ge=0,
        le=3600,
        description=(
            "Maximum seconds to approve the initial maker and fee preview before the "
            "prepared CoinJoin expires. 0 disables expiry."
        ),
    )
    order_wait_time: float = Field(
        default=120.0,
        ge=1.0,
        le=3600.0,
        description=(
            "Maximum seconds to wait for orderbook responses (hard ceiling). "
            "Empirical testing shows 95th percentile response time over Tor is ~101s. "
            "Default 120s provides a 20% buffer."
        ),
    )
    orderbook_min_wait: float = Field(
        default=30.0,
        ge=0.0,
        description=(
            "Minimum seconds to listen before allowing early exit. "
            "Prevents cutting off slow Tor responses during the initial burst."
        ),
    )
    orderbook_quiet_period: float = Field(
        default=15.0,
        ge=1.0,
        description=(
            "Seconds without new offers before exiting early. "
            "After orderbook_min_wait, if no new offers arrive for this long, "
            "all responsive makers are assumed to have replied."
        ),
    )
    tx_broadcast: str = Field(
        default="random-peer",
        description="Broadcast policy: self, random-peer, multiple-peers, not-self",
    )
    broadcast_peer_count: int = Field(
        default=3,
        ge=1,
        description="Number of peers for multiple-peers broadcast",
    )
    minimum_makers: int = Field(
        default=4,
        ge=1,
        description=(
            "Final floor for makers after attempts to restore counterparty_count. "
            "Default 4 matches the upstream JoinMarket reference (POLICY.n)."
        ),
    )
    max_maker_replacement_attempts: int = Field(
        default=3,
        ge=0,
        le=10,
        description=(
            "Maximum fill/auth replacement attempts to restore counterparty_count "
            "before proceeding at minimum_makers (0 = disabled)."
        ),
    )
    rescan_interval_sec: int = Field(
        default=600,
        ge=60,
        description="Interval for periodic wallet rescans",
    )
    pending_tx_abandon_hours: int = Field(
        default=24,
        ge=1,
        le=336,
        description=(
            "Hours to monitor a recorded CoinJoin for confirmation before a local "
            "monitoring timeout. Also sets the pending allowance in the input reservation "
            "lifetime. Explicit wallet refresh can reconcile later confirmation. Default 24 h."
        ),
    )

Attributes

bond_value_exponent: float = Field(default=1.3, gt=0.0, description='Exponent for fidelity bond value calculation') class-attribute instance-attribute

bondless_makers_allowance: float = Field(default=0.05, ge=0.0, le=1.0, description='Per-slot probability of selecting uniformly from zero-fee offers') class-attribute instance-attribute

bondless_require_zero_fee: bool = Field(default=True, description='Restrict allowance spots to zero-fee offers and reject fee-charging bondless offers') class-attribute instance-attribute

broadcast_peer_count: int = Field(default=3, ge=1, description='Number of peers for multiple-peers broadcast') class-attribute instance-attribute

counterparty_count: int | None = Field(default=None, ge=1, le=20, description='Number of makers to select for CoinJoin. When unset (the default), the taker picks a random value in [8, 10] for each CoinJoin. This matches the upstream JoinMarket sendpayment behaviour and prevents fingerprinting jm-ng takers via a fixed maker count.') class-attribute instance-attribute

equalize_cj_fees: bool = Field(default=False, description='Pay every selected maker the highest realized fee in the selected set') class-attribute instance-attribute

fee_block_target: int | None = Field(default=None, ge=1, le=1008, description='Target blocks for fee estimation (mutually exclusive with fee_rate)') class-attribute instance-attribute

fee_rate: float | None = Field(default=None, gt=0.0, description='Manual fee rate in sat/vB (mutually exclusive with fee_block_target)') class-attribute instance-attribute

initial_confirmation_timeout_sec: int = Field(default=300, ge=0, le=3600, description='Maximum seconds to approve the initial maker and fee preview before the prepared CoinJoin expires. 0 disables expiry.') class-attribute instance-attribute

maker_timeout_sec: int = Field(default=60, ge=10, le=3600, description='Timeout for maker responses') class-attribute instance-attribute

max_cj_fee_abs: int = Field(default=500, ge=0, description='Maximum absolute CoinJoin fee in satoshis') class-attribute instance-attribute

max_cj_fee_rel: str = Field(default='0.001', description='Maximum relative CoinJoin fee (0.001 = 0.1%)') class-attribute instance-attribute

max_maker_replacement_attempts: int = Field(default=3, ge=0, le=10, description='Maximum fill/auth replacement attempts to restore counterparty_count before proceeding at minimum_makers (0 = disabled).') class-attribute instance-attribute

max_maker_utxos: int = Field(default=15, ge=0, description="Maximum number of inputs a single maker may contribute to the CoinJoin. The taker pays the mining fee for every input, so an unbounded count lets a counterparty consolidate its UTXOs at the taker's expense. Makers exceeding the cap are dropped (and replaced when possible). 0 disables the cap (not recommended).") class-attribute instance-attribute

max_sweep_fee_change: float = Field(default=0.8, ge=0.0, allow_inf_nan=False, description='Maximum relative amount a sweep fee estimate may exceed its selected budget') class-attribute instance-attribute

min_fee_block_target: int = Field(default=10, ge=1, le=1008, description='Block target for the conservative CoinJoin miner-fee floor') class-attribute instance-attribute

min_fee_rate_sat_vb: float = Field(default=1.0, gt=0.0, allow_inf_nan=False, description='Minimum CoinJoin miner fee rate in sat/vB') class-attribute instance-attribute

minimum_makers: int = Field(default=4, ge=1, description='Final floor for makers after attempts to restore counterparty_count. Default 4 matches the upstream JoinMarket reference (POLICY.n).') class-attribute instance-attribute

order_wait_time: float = Field(default=120.0, ge=1.0, le=3600.0, description='Maximum seconds to wait for orderbook responses (hard ceiling). Empirical testing shows 95th percentile response time over Tor is ~101s. Default 120s provides a 20% buffer.') class-attribute instance-attribute

orderbook_min_wait: float = Field(default=30.0, ge=0.0, description='Minimum seconds to listen before allowing early exit. Prevents cutting off slow Tor responses during the initial burst.') class-attribute instance-attribute

orderbook_quiet_period: float = Field(default=15.0, ge=1.0, description='Seconds without new offers before exiting early. After orderbook_min_wait, if no new offers arrive for this long, all responsive makers are assumed to have replied.') class-attribute instance-attribute

pending_tx_abandon_hours: int = Field(default=24, ge=1, le=336, description='Hours to monitor a recorded CoinJoin for confirmation before a local monitoring timeout. Also sets the pending allowance in the input reservation lifetime. Explicit wallet refresh can reconcile later confirmation. Default 24 h.') class-attribute instance-attribute

require_quantized_cj_fees: bool = Field(default=True, description='Only consider offers whose advertised CoinJoin fee is on the public grid') class-attribute instance-attribute

rescan_interval_sec: int = Field(default=600, ge=60, description='Interval for periodic wallet rescans') class-attribute instance-attribute

round_up_cj_fees: bool = Field(default=False, description='Round each selected maker fee up to the next public fee quantum') class-attribute instance-attribute

taker_utxo_age: int = Field(default=5, ge=1, description='Minimum confirmations required on a UTXO before the taker may use it as a PoDLE commitment input. Reference default: 5.') class-attribute instance-attribute

taker_utxo_amtpercent: int = Field(default=20, ge=1, le=100, description='Minimum UTXO value as a percentage of the CoinJoin amount for PoDLE commitments (reference default: 20).') class-attribute instance-attribute

taker_utxo_retries: int = Field(default=3, ge=1, le=10, description='Maximum PoDLE index retries per UTXO (reference default: 3)') class-attribute instance-attribute

tx_broadcast: str = Field(default='random-peer', description='Broadcast policy: self, random-peer, multiple-peers, not-self') class-attribute instance-attribute

tx_fee_factor: float = Field(default=0.2, ge=0.0, description='Fee randomization factor. 0 disables randomization; 0.2 randomizes up to 20% above the base rate.') class-attribute instance-attribute

Methods:

normalize_max_cj_fee_rel(v: str | float | int) -> str classmethod

Normalize to avoid scientific notation (same as MakerSettings.cj_fee_relative).

Source code in jmcore/src/jmcore/settings.py

@field_validator("max_cj_fee_rel", mode="before")
@classmethod
def normalize_max_cj_fee_rel(cls, v: str | float | int) -> str:
    """Normalize to avoid scientific notation (same as MakerSettings.cj_fee_relative)."""
    if isinstance(v, (int, float)):
        return format(Decimal(str(v)), "f")
    if isinstance(v, str) and "e" in v.lower():
        try:
            return format(Decimal(v), "f")
        except InvalidOperation:
            pass
    return v

TomlConfigSettingsSource

Bases: PydanticBaseSettingsSource

Custom settings source that reads from a TOML config file.

The config file is expected at ~/.joinmarket-ng/config.toml or $JOINMARKET_DATA_DIR/config.toml if the environment variable is set.

Source code in jmcore/src/jmcore/settings.py

class TomlConfigSettingsSource(PydanticBaseSettingsSource):
    """
    Custom settings source that reads from a TOML config file.

    The config file is expected at ~/.joinmarket-ng/config.toml or
    $JOINMARKET_DATA_DIR/config.toml if the environment variable is set.
    """

    def __init__(self, settings_cls: type[BaseSettings]) -> None:
        super().__init__(settings_cls)
        self._config: dict[str, Any] = {}
        self._load_config()

    def _get_config_path(self) -> Path:
        """Determine the config file path."""
        # Check for explicit config path in environment
        env_path = os.environ.get("JOINMARKET_CONFIG_FILE")
        if env_path:
            return Path(env_path).expanduser()

        # Use data directory
        data_dir_env = os.environ.get("JOINMARKET_DATA_DIR")
        data_dir = (
            Path(data_dir_env).expanduser() if data_dir_env else Path.home() / ".joinmarket-ng"
        )

        return data_dir / "config.toml"

    def _load_config(self) -> None:
        """Load configuration from TOML file."""
        config_path = self._get_config_path()

        if not config_path.exists():
            logger.debug(f"Config file not found at {config_path}, using defaults")
            return

        try:
            import tomllib

            self._config = tomllib.load(BytesIO(read_sensitive_file(config_path)))
            _warn_unknown_settings(self._config, self.settings_cls)

            logger.info(f"Loaded config from {config_path}")
        except tomllib.TOMLDecodeError as e:
            logger.error(f"Invalid TOML syntax in config file {config_path}")
            logger.error(f"Error: {e}")
            logger.error("Please fix the syntax errors in your config file and try again.")
            logger.error(
                "Tip: Make sure section headers like [bitcoin], [tor], etc. are uncommented"
            )
            logger.error(
                'Tip: String values must be quoted, e.g. urls = ["tgram://bottoken/ChatID"]'
            )
            import sys

            sys.exit(1)
        except Exception as e:
            logger.error(f"Failed to load config from {config_path}: {e}")
            logger.error("Please check your config file and try again.")
            import sys

            sys.exit(1)

    def get_field_value(self, field: Any, field_name: str) -> tuple[Any, str, bool]:
        """Get field value from TOML config."""
        # Handle nested fields by looking up in nested dicts
        value = self._config.get(field_name)
        return value, field_name, value is not None

    def __call__(self) -> dict[str, Any]:
        """Return all config values as a flat dict for pydantic-settings."""
        return _normalize_wallet_preference_source(self._config)

Methods:

__call__() -> dict[str, Any]

Return all config values as a flat dict for pydantic-settings.

Source code in jmcore/src/jmcore/settings.py

def __call__(self) -> dict[str, Any]:
    """Return all config values as a flat dict for pydantic-settings."""
    return _normalize_wallet_preference_source(self._config)

__init__(settings_cls: type[BaseSettings]) -> None

Source code in jmcore/src/jmcore/settings.py

def __init__(self, settings_cls: type[BaseSettings]) -> None:
    super().__init__(settings_cls)
    self._config: dict[str, Any] = {}
    self._load_config()

get_field_value(field: Any, field_name: str) -> tuple[Any, str, bool]

Get field value from TOML config.

Source code in jmcore/src/jmcore/settings.py

def get_field_value(self, field: Any, field_name: str) -> tuple[Any, str, bool]:
    """Get field value from TOML config."""
    # Handle nested fields by looking up in nested dicts
    value = self._config.get(field_name)
    return value, field_name, value is not None

TorSettings

Bases: BaseModel

Tor proxy and control port configuration.

Source code in jmcore/src/jmcore/settings.py

class TorSettings(BaseModel):
    """Tor proxy and control port configuration."""

    # SOCKS proxy settings
    socks_host: str = Field(
        default="127.0.0.1",
        description="Tor SOCKS5 proxy host",
    )
    socks_port: int = Field(
        default=9050,
        ge=1,
        le=65535,
        description="Tor SOCKS5 proxy port",
    )

    # Control port settings
    control_enabled: bool = Field(
        default=True,
        description="Enable Tor control port integration for ephemeral hidden services",
    )
    control_host: str = Field(
        default_factory=lambda data: data["socks_host"],
        description="Tor control port host (defaults to socks_host)",
    )
    control_port: int = Field(
        default=9051,
        ge=1,
        le=65535,
        description="Tor control port",
    )
    cookie_path: str | None = Field(
        default=None,
        description="Path to Tor cookie auth file",
    )
    password: SecretStr | None = Field(
        default=None,
        description="Tor control port password (use cookie auth instead if possible)",
    )

    # Hidden service target (for makers)
    target_host: str = Field(
        default="127.0.0.1",
        description="Target host for Tor hidden service (usually container name in Docker)",
    )

    # Stream isolation
    stream_isolation: bool = Field(
        default=True,
        description=(
            "Use SOCKS5 authentication to isolate different connection types "
            "onto separate Tor circuits.  This prevents traffic correlation "
            "between e.g. directory connections, peer connections, and "
            "notification traffic.  Requires IsolateSOCKSAuth on the Tor "
            "SocksPort (enabled by default)."
        ),
    )

    # Connection timeout
    connection_timeout: float = Field(
        default=120.0,
        gt=0.0,
        description=(
            "Timeout in seconds for Tor SOCKS5 connections. Covers TCP handshake, "
            "SOCKS5 negotiation, Tor circuit building, and PoW solving. "
            "Default 120s matches Tor's internal circuit timeout."
        ),
    )

Attributes

connection_timeout: float = Field(default=120.0, gt=0.0, description="Timeout in seconds for Tor SOCKS5 connections. Covers TCP handshake, SOCKS5 negotiation, Tor circuit building, and PoW solving. Default 120s matches Tor's internal circuit timeout.") class-attribute instance-attribute

control_enabled: bool = Field(default=True, description='Enable Tor control port integration for ephemeral hidden services') class-attribute instance-attribute

control_host: str = Field(default_factory=lambda data: data['socks_host'], description='Tor control port host (defaults to socks_host)') class-attribute instance-attribute

control_port: int = Field(default=9051, ge=1, le=65535, description='Tor control port') class-attribute instance-attribute

cookie_path: str | None = Field(default=None, description='Path to Tor cookie auth file') class-attribute instance-attribute

password: SecretStr | None = Field(default=None, description='Tor control port password (use cookie auth instead if possible)') class-attribute instance-attribute

socks_host: str = Field(default='127.0.0.1', description='Tor SOCKS5 proxy host') class-attribute instance-attribute

socks_port: int = Field(default=9050, ge=1, le=65535, description='Tor SOCKS5 proxy port') class-attribute instance-attribute

stream_isolation: bool = Field(default=True, description='Use SOCKS5 authentication to isolate different connection types onto separate Tor circuits. This prevents traffic correlation between e.g. directory connections, peer connections, and notification traffic. Requires IsolateSOCKSAuth on the Tor SocksPort (enabled by default).') class-attribute instance-attribute

target_host: str = Field(default='127.0.0.1', description='Target host for Tor hidden service (usually container name in Docker)') class-attribute instance-attribute

TuiSettings

Bases: BaseModel

Menu settings, consumed by the shell TUI rather than Python components.

Source code in jmcore/src/jmcore/settings.py

class TuiSettings(BaseModel):
    """Menu settings, consumed by the shell TUI rather than Python components."""

    log_level: str | None = Field(
        default=None,
        description="Optional log level override for TUI-launched commands; otherwise follow logging.level",
    )

Attributes

log_level: str | None = Field(default=None, description='Optional log level override for TUI-launched commands; otherwise follow logging.level') class-attribute instance-attribute

TumblerSettings

Bases: BaseModel

Tumbler runner timing knobs.

These mirror :class:tumbler.runner.RunnerContext fields that govern inter-phase pacing. Production defaults are intentionally long so the runner waits patiently for blockchain confirmations and UTXO age between CoinJoin phases. Test environments should shorten these via env vars (TUMBLER__RETRY_DELAY_SECONDS, TUMBLER__CONFIRMATION_POLL_INTERVAL, TUMBLER__MIN_CONFIRMATIONS_BETWEEN_PHASES) so e2e suites observe retries within a reasonable wall-clock budget.

Source code in jmcore/src/jmcore/settings.py

class TumblerSettings(BaseModel):
    """Tumbler runner timing knobs.

    These mirror :class:`tumbler.runner.RunnerContext` fields that govern
    inter-phase pacing. Production defaults are intentionally long so the
    runner waits patiently for blockchain confirmations and UTXO age between
    CoinJoin phases. Test environments should shorten these via env vars
    (``TUMBLER__RETRY_DELAY_SECONDS``, ``TUMBLER__CONFIRMATION_POLL_INTERVAL``,
    ``TUMBLER__MIN_CONFIRMATIONS_BETWEEN_PHASES``) so e2e suites observe
    retries within a reasonable wall-clock budget.
    """

    min_confirmations_between_phases: int = Field(
        default=6,
        ge=0,
        description=(
            "Minimum confirmations on a CoinJoin output before the next "
            "tumbler phase may spend it. Set to 0 to disable the gate."
        ),
    )
    confirmation_poll_interval: float = Field(
        default=30.0,
        gt=0.0,
        description="Polling interval (seconds) for the confirmation gate.",
    )
    retry_delay_seconds: float = Field(
        default=1800.0,
        ge=0.0,
        description=(
            "Base delay (seconds) before retrying a failed taker phase. "
            "Applied with linear backoff: ``retry_delay_seconds * "
            "attempt_count``. Set to 0 to retry immediately."
        ),
    )

Attributes

confirmation_poll_interval: float = Field(default=30.0, gt=0.0, description='Polling interval (seconds) for the confirmation gate.') class-attribute instance-attribute

min_confirmations_between_phases: int = Field(default=6, ge=0, description='Minimum confirmations on a CoinJoin output before the next tumbler phase may spend it. Set to 0 to disable the gate.') class-attribute instance-attribute

retry_delay_seconds: float = Field(default=1800.0, ge=0.0, description='Base delay (seconds) before retrying a failed taker phase. Applied with linear backoff: ``retry_delay_seconds * attempt_count``. Set to 0 to retry immediately.') class-attribute instance-attribute

WalletSettings

Bases: BaseModel

Wallet configuration.

Source code in jmcore/src/jmcore/settings.py

class WalletSettings(BaseModel):
    """Wallet configuration."""

    @model_validator(mode="before")
    @classmethod
    def reject_legacy_scan_name(cls, value: Any) -> Any:
        if isinstance(value, dict) and "background_full_scan" in value:
            raise ValueError(
                "wallet.background_full_scan is no longer supported; rename it to "
                "wallet.background_full_rescan (environment: WALLET__BACKGROUND_FULL_RESCAN)"
            )
        return value

    mixdepth_count: int = Field(
        default=5,
        ge=1,
        le=10,
        description="Number of mixdepths (privacy compartments)",
    )
    gap_limit: int = Field(
        default=20,
        ge=6,
        description=(
            "BIP44 gap limit: the number of consecutive empty trailing "
            "addresses kept beyond the highest used one (Electrum convention "
            "is 20). Also used as the buffer when auto-expanding the Bitcoin "
            "Core descriptor range as the wallet grows. Distinct from "
            "``scan_range`` (the initial lookahead window). See "
            "docs/technical/wallet-scanning.md."
        ),
    )
    scan_range: int = Field(
        default=1000,
        ge=100,
        # Bitcoin Core's importdescriptors rejects ranges spanning more than
        # 1,000,000 indices with "Range is too large", so this is the hard cap.
        le=1_000_000,
        description=(
            "Initial descriptor scan range: the max address index per branch "
            "imported into Bitcoin Core's descriptor wallet (default 1000, "
            "matching Core's keypool lookahead). It auto-expands as addresses "
            "are used. Capped at 1,000,000, which is Bitcoin Core's "
            "per-descriptor range limit (larger values are rejected with "
            "'Range is too large'). To widen the range for an already-imported "
            "wallet (e.g. one migrated from legacy joinmarket-clientserver with "
            "deep addresses), run ``jm-wallet rescan --scan-depth N`` once. See "
            "docs/technical/wallet-scanning.md."
        ),
    )
    dust_threshold: int = Field(
        default=DUST_THRESHOLD,
        ge=DUST_THRESHOLD,
        le=DUST_THRESHOLD,
        description="Fixed JoinMarket maker change threshold in satoshis",
    )
    max_sats_freeze_reuse: int = Field(
        default=-1,
        ge=-1,
        description=(
            "Threshold (in satoshis) below which an incoming UTXO landing on a "
            "previously-used wallet address that is now empty is AUTOMATICALLY "
            "frozen, to defend against forced address-reuse (dust) attacks that "
            "try to coerce linkage via the common-input-ownership heuristic "
            "(https://en.bitcoin.it/wiki/Privacy#Forced_address_reuse). Only "
            "re-funding of an already-spent (empty) used address is frozen; "
            "coins arriving on an address that still holds funds are left "
            "spendable so they can be fully spent together. "
            "The default -1 freezes ALL such reuse UTXOs regardless of value; "
            "a positive N freezes only reuse UTXOs with value <= N sats; "
            "0 disables auto-freezing. Frozen UTXOs can be released "
            "with ``jm-wallet unfreeze``."
        ),
    )
    reconstruct_history: bool = Field(
        default=True,
        description=(
            "Automatically reconstruct CoinJoin/send/deposit history from "
            "on-chain data for wallets imported from seed (wallets with no "
            "recorded history). Reconstructed rows are tagged as on-chain "
            "guesses and never override protocol-time history. Disable to "
            "keep history.csv strictly protocol-recorded; "
            "`jm-wallet reconstruct-history` remains available for manual runs."
        ),
    )
    smart_scan: bool = Field(
        default=True,
        description="Use smart scan for fast startup",
    )
    background_full_rescan: bool = Field(
        default=True,
        description="Run full blockchain rescan in background",
    )
    scan_lookback_blocks: int = Field(
        default=52560,
        ge=0,
        description="Blocks to look back for smart scan (~1 year default)",
    )
    scan_start_height: int | None = Field(
        default=None,
        ge=0,
        description="Explicit start height for initial scan (overrides scan_lookback_blocks if set)",
    )
    default_fee_block_target: int = Field(
        default=3,
        ge=1,
        le=1008,
        description="Default block target for fee estimation in wallet transactions",
    )
    max_fee_rate_sat_vb: float = Field(
        default=1_000.0,
        gt=0.0,
        description=(
            "Safety cap on the fee rate (sat/vB) used for any wallet-driven "
            "transaction (direct send and CoinJoin). Manual rates above this "
            "value, and backend fee estimates above this value, are rejected "
            "to prevent runaway-fee bugs and malicious or misconfigured fee "
            "oracles from draining the wallet."
        ),
    )
    info_show_fidelity_bonds: bool = Field(
        default=True,
        description=(
            "Show the fidelity bond branch in the extended wallet info views "
            "(``jm-wallet info --extended``). Bonds are fundamentally different "
            "from the day-to-day addresses and are managed via ``jm-wallet "
            "bonds``; set to false to keep the info views free of bond "
            "addresses (a one-line notice still points out funded hidden bonds)."
        ),
    )
    mnemonic_file: str | None = Field(
        default=None,
        description="Default path to mnemonic file",
    )
    mnemonic_password: SecretStr | None = Field(
        default=None,
        description="Password for encrypted mnemonic file",
    )
    bip39_passphrase: SecretStr | None = Field(
        default=None,
        description="BIP39 passphrase (13th/25th word). For security, prefer BIP39_PASSPHRASE env var.",
    )
    bip39_passphrase_enabled: bool = Field(
        default=False,
        validation_alias=AliasChoices("bip39_passphrase_enabled", "wallet_with_passphrase"),
        description=(
            "Enable BIP39 onboarding and prompts for unregistered wallets. Registered wallet "
            "requirements always apply; identity-only reads never need the passphrase."
        ),
    )

    @model_validator(mode="before")
    @classmethod
    def normalize_passphrase_preference(cls, data: Any) -> Any:
        if isinstance(data, dict) and "wallet_with_passphrase" in data:
            canonical = data.get("bip39_passphrase_enabled")
            legacy = data["wallet_with_passphrase"]
            if canonical is not None and (
                TypeAdapter(bool).validate_python(canonical)
                != TypeAdapter(bool).validate_python(legacy)
            ):
                raise ValueError("Conflicting BIP39 onboarding preferences; keep one spelling")
        return data

    @property
    def wallet_with_passphrase(self) -> bool:
        """Compatibility spelling, never an independent wallet requirement."""
        return self.bip39_passphrase_enabled

    @wallet_with_passphrase.setter
    def wallet_with_passphrase(self, value: bool) -> None:
        self.bip39_passphrase_enabled = value

Attributes

background_full_rescan: bool = Field(default=True, description='Run full blockchain rescan in background') class-attribute instance-attribute

bip39_passphrase: SecretStr | None = Field(default=None, description='BIP39 passphrase (13th/25th word). For security, prefer BIP39_PASSPHRASE env var.') class-attribute instance-attribute

bip39_passphrase_enabled: bool = Field(default=False, validation_alias=AliasChoices('bip39_passphrase_enabled', 'wallet_with_passphrase'), description='Enable BIP39 onboarding and prompts for unregistered wallets. Registered wallet requirements always apply; identity-only reads never need the passphrase.') class-attribute instance-attribute

default_fee_block_target: int = Field(default=3, ge=1, le=1008, description='Default block target for fee estimation in wallet transactions') class-attribute instance-attribute

dust_threshold: int = Field(default=DUST_THRESHOLD, ge=DUST_THRESHOLD, le=DUST_THRESHOLD, description='Fixed JoinMarket maker change threshold in satoshis') class-attribute instance-attribute

gap_limit: int = Field(default=20, ge=6, description='BIP44 gap limit: the number of consecutive empty trailing addresses kept beyond the highest used one (Electrum convention is 20). Also used as the buffer when auto-expanding the Bitcoin Core descriptor range as the wallet grows. Distinct from ``scan_range`` (the initial lookahead window). See docs/technical/wallet-scanning.md.') class-attribute instance-attribute

info_show_fidelity_bonds: bool = Field(default=True, description='Show the fidelity bond branch in the extended wallet info views (``jm-wallet info --extended``). Bonds are fundamentally different from the day-to-day addresses and are managed via ``jm-wallet bonds``; set to false to keep the info views free of bond addresses (a one-line notice still points out funded hidden bonds).') class-attribute instance-attribute

max_fee_rate_sat_vb: float = Field(default=1000.0, gt=0.0, description='Safety cap on the fee rate (sat/vB) used for any wallet-driven transaction (direct send and CoinJoin). Manual rates above this value, and backend fee estimates above this value, are rejected to prevent runaway-fee bugs and malicious or misconfigured fee oracles from draining the wallet.') class-attribute instance-attribute

max_sats_freeze_reuse: int = Field(default=-1, ge=-1, description='Threshold (in satoshis) below which an incoming UTXO landing on a previously-used wallet address that is now empty is AUTOMATICALLY frozen, to defend against forced address-reuse (dust) attacks that try to coerce linkage via the common-input-ownership heuristic (https://en.bitcoin.it/wiki/Privacy#Forced_address_reuse). Only re-funding of an already-spent (empty) used address is frozen; coins arriving on an address that still holds funds are left spendable so they can be fully spent together. The default -1 freezes ALL such reuse UTXOs regardless of value; a positive N freezes only reuse UTXOs with value <= N sats; 0 disables auto-freezing. Frozen UTXOs can be released with ``jm-wallet unfreeze``.') class-attribute instance-attribute

mixdepth_count: int = Field(default=5, ge=1, le=10, description='Number of mixdepths (privacy compartments)') class-attribute instance-attribute

mnemonic_file: str | None = Field(default=None, description='Default path to mnemonic file') class-attribute instance-attribute

mnemonic_password: SecretStr | None = Field(default=None, description='Password for encrypted mnemonic file') class-attribute instance-attribute

reconstruct_history: bool = Field(default=True, description='Automatically reconstruct CoinJoin/send/deposit history from on-chain data for wallets imported from seed (wallets with no recorded history). Reconstructed rows are tagged as on-chain guesses and never override protocol-time history. Disable to keep history.csv strictly protocol-recorded; `jm-wallet reconstruct-history` remains available for manual runs.') class-attribute instance-attribute

scan_lookback_blocks: int = Field(default=52560, ge=0, description='Blocks to look back for smart scan (~1 year default)') class-attribute instance-attribute

scan_range: int = Field(default=1000, ge=100, le=1000000, description="Initial descriptor scan range: the max address index per branch imported into Bitcoin Core's descriptor wallet (default 1000, matching Core's keypool lookahead). It auto-expands as addresses are used. Capped at 1,000,000, which is Bitcoin Core's per-descriptor range limit (larger values are rejected with 'Range is too large'). To widen the range for an already-imported wallet (e.g. one migrated from legacy joinmarket-clientserver with deep addresses), run ``jm-wallet rescan --scan-depth N`` once. See docs/technical/wallet-scanning.md.") class-attribute instance-attribute

scan_start_height: int | None = Field(default=None, ge=0, description='Explicit start height for initial scan (overrides scan_lookback_blocks if set)') class-attribute instance-attribute

smart_scan: bool = Field(default=True, description='Use smart scan for fast startup') class-attribute instance-attribute

wallet_with_passphrase: bool property writable

Compatibility spelling, never an independent wallet requirement.

Methods:

normalize_passphrase_preference(data: Any) -> Any classmethod

Source code in jmcore/src/jmcore/settings.py

@model_validator(mode="before")
@classmethod
def normalize_passphrase_preference(cls, data: Any) -> Any:
    if isinstance(data, dict) and "wallet_with_passphrase" in data:
        canonical = data.get("bip39_passphrase_enabled")
        legacy = data["wallet_with_passphrase"]
        if canonical is not None and (
            TypeAdapter(bool).validate_python(canonical)
            != TypeAdapter(bool).validate_python(legacy)
        ):
            raise ValueError("Conflicting BIP39 onboarding preferences; keep one spelling")
    return data

reject_legacy_scan_name(value: Any) -> Any classmethod

Source code in jmcore/src/jmcore/settings.py

@model_validator(mode="before")
@classmethod
def reject_legacy_scan_name(cls, value: Any) -> Any:
    if isinstance(value, dict) and "background_full_scan" in value:
        raise ValueError(
            "wallet.background_full_scan is no longer supported; rename it to "
            "wallet.background_full_rescan (environment: WALLET__BACKGROUND_FULL_RESCAN)"
        )
    return value

Functions:

config_diff(config_path: Path, template_text: str | None = None) -> list[str]

Compare the user's config against the template and report differences.

This is a read-only operation -- the user's config file is never modified. Returns a list of new sections and keys that exist in the template but are missing from the user's config, so the caller can display them.

Args: config_path: Path to the user's config.toml. template_text: Template text to compare against. When None, the bundled config.toml.template shipped with the package is used.

Returns: List of human-readable descriptions of missing items (empty if the config is up to date). Each entry is either "section:<name>" for a missing section or "key:<section>.<key>" for a missing key within an existing section.

Source code in jmcore/src/jmcore/settings.py

def config_diff(
    config_path: Path,
    template_text: str | None = None,
) -> list[str]:
    """Compare the user's config against the template and report differences.

    This is a **read-only** operation -- the user's config file is never
    modified.  Returns a list of new sections and keys that exist in the
    template but are missing from the user's config, so the caller can
    display them.

    Args:
        config_path: Path to the user's ``config.toml``.
        template_text: Template text to compare against.  When *None*,
            the bundled ``config.toml.template`` shipped with the package
            is used.

    Returns:
        List of human-readable descriptions of missing items (empty if the
        config is up to date).  Each entry is either ``"section:<name>"``
        for a missing section or ``"key:<section>.<key>"`` for a missing
        key within an existing section.
    """
    if template_text is None:
        template_text = _get_bundled_template()
    if template_text is None:
        template_text = generate_config_template()
    if not template_text:
        return []

    if not config_path.exists():
        return []

    user_text = config_path.read_text()
    user_sections = _get_user_sections(user_text)

    # Extract template section names.
    template_section_re = re.compile(r"^\[(\w+)]", re.MULTILINE)
    template_section_names = [m.group(1) for m in template_section_re.finditer(template_text)]

    diffs: list[str] = []

    # Report missing sections.
    for name in template_section_names:
        if name not in user_sections:
            diffs.append(f"section:{name}")

    # Report missing keys within existing sections.
    template_keys = _get_template_section_keys(template_text)
    user_keys = _get_user_section_keys(user_text)
    for section_name in template_section_names:
        if section_name not in user_keys:
            continue
        missing = template_keys.get(section_name, set()) - user_keys[section_name]
        for key in sorted(missing):
            diffs.append(f"key:{section_name}.{key}")

    return diffs

ensure_config_file(data_dir: Path | None = None, config_file: Path | None = None) -> Path

Ensure the config file exists.

On first run the config file is created from the bundled starter. Existing config files are never modified.

The config file location is resolved with this priority: 1. config_file argument (explicit override). 2. JOINMARKET_CONFIG_FILE environment variable (set e.g. by the --config-file CLI flag) so config can live outside the data dir for FHS-style deployments (issue #537). 3. <data_dir>/config.toml (the default, backwards-compatible behaviour).

Args: data_dir: Optional data directory path. Uses default if not provided. config_file: Optional explicit config file path that decouples the config location from data_dir.

Returns: Path to the config file.

Source code in jmcore/src/jmcore/settings.py

def ensure_config_file(
    data_dir: Path | None = None,
    config_file: Path | None = None,
) -> Path:
    """Ensure the config file exists.

    On first run the config file is created from the bundled starter.
    Existing config files are never modified.

    The config file location is resolved with this priority:
    1. ``config_file`` argument (explicit override).
    2. ``JOINMARKET_CONFIG_FILE`` environment variable (set e.g. by the
       ``--config-file`` CLI flag) so config can live outside the data dir
       for FHS-style deployments (issue #537).
    3. ``<data_dir>/config.toml`` (the default, backwards-compatible behaviour).

    Args:
        data_dir: Optional data directory path. Uses default if not provided.
        config_file: Optional explicit config file path that decouples the
            config location from ``data_dir``.

    Returns:
        Path to the config file.
    """
    if config_file is not None:
        config_path = Path(config_file).expanduser()
    elif os.environ.get("JOINMARKET_CONFIG_FILE"):
        # An explicit config file was requested via env (e.g. --config-file);
        # create/read it there rather than under the data dir.
        config_path = get_config_path()
    else:
        if data_dir is None:
            data_dir = get_default_data_dir()
        config_path = data_dir / "config.toml"

    migrate_config(config_path)
    return config_path

generate_config_starter() -> str

Return the small, override-only starter shipped with this release.

Source code in jmcore/src/jmcore/settings.py

def generate_config_starter() -> str:
    """Return the small, override-only starter shipped with this release."""
    ref = importlib.resources.files("jmcore") / "data" / "config-starter.toml.template"
    return ref.read_text(encoding="utf-8")

generate_config_template() -> str

Generate a config file template with all settings commented out.

This allows users to see all available settings with their defaults and descriptions, while only uncommenting what they want to change.

Source code in jmcore/src/jmcore/settings.py

def generate_config_template() -> str:
    """
    Generate a config file template with all settings commented out.

    This allows users to see all available settings with their defaults
    and descriptions, while only uncommenting what they want to change.
    """
    lines: list[str] = []

    lines.append("# JoinMarket NG Configuration")
    lines.append("#")
    lines.append("# This file contains all available settings with their default values.")
    lines.append("# Settings are commented out by default - uncomment to override.")
    lines.append("#")
    lines.append("# Priority (highest to lowest):")
    lines.append("#   1. CLI arguments")
    lines.append("#   2. Environment variables")
    lines.append("#   3. This config file")
    lines.append("#   4. Built-in defaults")
    lines.append("#")
    lines.append("# Environment variables use uppercase with double underscore for nesting:")
    lines.append("#   TOR__SOCKS_HOST=127.0.0.1")
    lines.append("#   BITCOIN__RPC_URL=http://localhost:8332")
    lines.append("#")
    lines.append("")

    # Generate sections for each nested model
    def add_section(title: str, model_cls: type[BaseModel], prefix: str = "") -> None:
        lines.append(f"# {'=' * 60}")
        lines.append(f"# {title}")
        lines.append(f"# {'=' * 60}")
        lines.append(f"[{prefix}]" if prefix else "")
        lines.append("")

        defaults: dict[str, Any] = {}
        for field_name, field_info in model_cls.model_fields.items():
            # Get description
            desc = field_info.description or ""
            if desc:
                lines.append(f"# {desc}")

            # Get default value
            default = field_info.get_default(call_default_factory=True, validated_data=defaults)
            defaults[field_name] = default

            # Format the value for TOML
            if isinstance(default, bool):
                value_str = str(default).lower()
            elif isinstance(default, str):
                value_str = f'"{default}"'
            elif isinstance(default, list):
                # For directory_servers, show example from defaults
                if field_name == "directory_servers" and prefix == "network_config":
                    lines.append("# Mainnet defaults (leave empty to use automatically):")
                    lines.append("# directory_servers = [")
                    for server in DEFAULT_DIRECTORY_SERVERS["mainnet"]:
                        lines.append(f'#   "{server}",')
                    lines.append("# ]")
                    lines.append("")
                    lines.append("# Signet defaults:")
                    lines.append("# directory_servers = [")
                    for server in DEFAULT_DIRECTORY_SERVERS["signet"]:
                        lines.append(f'#   "{server}",')
                    lines.append("# ]")
                    lines.append("")
                    continue
                if field_name == "neutrino_add_peers" and prefix == "bitcoin":
                    lines.append(
                        "# Preferred peers for neutrino (host:port), while discovery stays enabled."
                    )
                    lines.append("# neutrino_add_peers = [")
                    lines.append('#   "your-filter-peer:38333",')
                    lines.append("# ]")
                    lines.append("")
                    continue
                value_str = "[]" if not default else str(default).replace("'", '"')
            elif isinstance(default, SecretStr):
                value_str = '""'
            elif default is None:
                # Skip None values with a comment
                lines.append(f"# {field_name} = ")
                lines.append("")
                continue
            elif hasattr(default, "value"):  # Enum - use string value
                value_str = f'"{default.value}"'
            else:
                value_str = str(default)

            lines.append(f"# {field_name} = {value_str}")
            lines.append("")

    # Data directory (top-level)
    lines.append("# Data directory for JoinMarket files")
    lines.append("# Defaults to ~/.joinmarket-ng or $JOINMARKET_DATA_DIR")
    lines.append("# data_dir = ")
    lines.append("")

    # Add all sections
    add_section("Tor Settings", TorSettings, "tor")
    add_section("Bitcoin Backend Settings", BitcoinSettings, "bitcoin")
    add_section("Network Settings", NetworkSettings, "network_config")
    add_section("Wallet Settings", WalletSettings, "wallet")
    add_section("Notification Settings", NotificationSettings, "notifications")
    add_section("Logging Settings", LoggingSettings, "logging")
    add_section("TUI Settings", TuiSettings, "tui")
    add_section("Maker Settings", MakerSettings, "maker")
    add_section("Taker Settings", TakerSettings, "taker")
    add_section("Directory Server Settings", DirectoryServerSettings, "directory_server")
    add_section("Orderbook Watcher Settings", OrderbookWatcherSettings, "orderbook_watcher")

    return "\n".join(lines)

get_config_path() -> Path

Get the path to the config file.

Resolution order (highest priority first): 1. JOINMARKET_CONFIG_FILE (explicit config file, decoupled from data dir) 2. $JOINMARKET_DATA_DIR/config.toml 3. ~/.joinmarket-ng/config.toml

~ is expanded so a configured tilde path never becomes a literal ./~ segment (see issue #536).

Source code in jmcore/src/jmcore/settings.py

def get_config_path() -> Path:
    """Get the path to the config file.

    Resolution order (highest priority first):
    1. ``JOINMARKET_CONFIG_FILE`` (explicit config file, decoupled from data dir)
    2. ``$JOINMARKET_DATA_DIR/config.toml``
    3. ``~/.joinmarket-ng/config.toml``

    ``~`` is expanded so a configured tilde path never becomes a literal
    ``./~`` segment (see issue #536).
    """
    config_file_env = os.environ.get("JOINMARKET_CONFIG_FILE")
    if config_file_env:
        return Path(config_file_env).expanduser()

    data_dir_env = os.environ.get("JOINMARKET_DATA_DIR")
    data_dir = Path(data_dir_env).expanduser() if data_dir_env else Path.home() / ".joinmarket-ng"
    return data_dir / "config.toml"

get_settings(**overrides: Any) -> JoinMarketSettings

Get the JoinMarket settings instance.

On first call, loads settings from all sources. Subsequent calls return the cached instance unless reset_settings() is called.

Args: **overrides: Optional settings overrides (highest priority)

Returns: JoinMarketSettings instance

Source code in jmcore/src/jmcore/settings.py

def get_settings(**overrides: Any) -> JoinMarketSettings:
    """
    Get the JoinMarket settings instance.

    On first call, loads settings from all sources. Subsequent calls
    return the cached instance unless reset_settings() is called.

    Args:
        **overrides: Optional settings overrides (highest priority)

    Returns:
        JoinMarketSettings instance
    """
    global _settings
    if _settings is None or overrides:
        _settings = JoinMarketSettings(**overrides)
    return _settings

migrate_config(config_path: Path, template_text: str | None = None) -> list[str]

Create a small starter config if the config file does not exist.

When the config file already exists, no modifications are made. Use :func:config_diff to discover new settings that the user may want to add manually.

A reference copy of the template is also maintained as config.toml.template next to the config file, so users can compare their config against the current version's template. The copy is refreshed whenever its content is out of date; the user's config.toml itself is never modified.

Args: config_path: Path to the user's config.toml. template_text: Explicit template for both fresh creation and the reference copy (retained for compatibility). When None, create from the bundled starter and refresh the separate full reference.

Returns: Empty list (kept for backward compatibility).

Source code in jmcore/src/jmcore/settings.py

def migrate_config(
    config_path: Path,
    template_text: str | None = None,
) -> list[str]:
    """Create a small starter config if the config file does not exist.

    When the config file already exists, no modifications are made.
    Use :func:`config_diff` to discover new settings that the user
    may want to add manually.

    A reference copy of the template is also maintained as
    ``config.toml.template`` next to the config file, so users can
    compare their config against the current version's template. The
    copy is refreshed whenever its content is out of date; the user's
    ``config.toml`` itself is never modified.

    Args:
        config_path: Path to the user's ``config.toml``.
        template_text: Explicit template for both fresh creation and the
            reference copy (retained for compatibility). When *None*, create
            from the bundled starter and refresh the separate full reference.

    Returns:
        Empty list (kept for backward compatibility).
    """
    # Defensive expansion: a caller may still pass an un-expanded path (e.g.
    # ``~/.joinmarket-ng/config.toml``). Expanding here ensures we never create
    # a literal ``./~`` directory regardless of how data_dir was resolved (#536).
    config_path = config_path.expanduser()
    config_exists = config_path.exists()
    if config_exists:
        ensure_sensitive_file(config_path)

    custom_template = template_text is not None
    if template_text is None:
        template_text = _get_bundled_template()
    if template_text is None:
        template_text = generate_config_template()
    if not template_text:
        logger.warning("No config template available; skipping config creation")
        return []

    if not config_exists:
        logger.info(f"Config file missing; creating starter at {config_path}")
        starter_text = template_text if custom_template else generate_config_starter()
        atomic_write_sensitive_file(config_path, starter_text.encode("utf-8"))

    # Keep a reference copy of the template alongside the config so users
    # can diff their settings against the current version's template.
    # Failures here must never break startup or config creation.
    template_copy = config_path.with_name("config.toml.template")
    try:
        if template_copy.resolve() == config_path.resolve():
            raise OSError("Reference template aliases the user config; leaving it unchanged")
        if not template_copy.exists() or read_sensitive_file(template_copy) != template_text.encode(
            "utf-8"
        ):
            atomic_write_sensitive_file(template_copy, template_text.encode("utf-8"))
    except OSError as exc:
        logger.warning(f"Could not update template copy at {template_copy}: {exc}")

    return []

reset_settings() -> None

Reset the global settings instance (useful for testing).

Source code in jmcore/src/jmcore/settings.py

def reset_settings() -> None:
    """Reset the global settings instance (useful for testing)."""
    global _settings
    _settings = None