Skip to content

jmcore.models

jmcore.models

Core data models using Pydantic for validation and serialization.

Attributes

DIRECTORY_NODES_MAINNET: list[str] = ['satoshi2vcg5e2ept7tjkzlkpomkobqmgtsjzegg6wipnoajadissead.onion:5222', 'coinjointovy3eq5fjygdwpkbcdx63d7vd4g32mw7y553uj3kjjzkiqd.onion:5222', 'nakamotourflxwjnjpnrk7yc2nhkf6r62ed4gdfxmmn5f4saw5q5qoyd.onion:5222', 'odpwaf67rs5226uabcamvypg3y4bngzmfk7255flcdodesqhsvkptaid.onion:5222', 'jmarketxf5wc4aldf3slm5u6726zsky52bqnfv6qyxe5hnafgly6yuyd.onion:5222', 'jmrust7bgdbdl6skkvuzhqost4jkikrluj6alemspeifm5hvgqz2qaad.onion:5222', 'nok55gjlqw6h76zi6gigukoztpx7xgo5r3w5csu362nys5yukzrxpgad.onion:5222'] module-attribute

DIRECTORY_NODES_SIGNET: list[str] = ['signetvaxgd3ivj4tml4g6ed3samaa2rscre2gyeyohncmwk4fbesiqd.onion:5222', 'u5oj5etqex3vh7jagljf3e2lo4awmmtcw3klbrlt2fonzyozpn5txrqd.onion:5222', 'frqp4m6yveiagow73dplmnyv2abrqhte2grbhgw7hkbkhrv6zagrqbqd.onion:5222'] module-attribute

DIRECTORY_NODES_TESTNET: list[str] = [] module-attribute

MAX_RELATIVE_FEE_EXPONENT = 64 module-attribute

MAX_RELATIVE_FEE_INPUT_LENGTH = 128 module-attribute

MAX_RELATIVE_FEE_PRECISION = 18 module-attribute

Classes

FidelityBond

Bases: BaseModel

Source code in jmcore/src/jmcore/models.py 
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
class FidelityBond(BaseModel):
    counterparty: str
    utxo_txid: str = Field(..., pattern=r"^[0-9a-fA-F]{64}$")
    utxo_vout: int = Field(..., ge=0)
    bond_value: int | None = Field(default=None, ge=0)
    verification_valid: bool | None = Field(
        default=None,
        description="Whether backend verification definitively accepted or rejected this bond",
    )
    verification_stale: bool = Field(
        default=False,
        description="Whether the displayed value is retained past its verification TTL",
    )
    locktime: int = Field(..., ge=0)
    amount: int = Field(default=0, ge=0)
    script: str
    utxo_confirmations: int = Field(..., ge=0)
    utxo_confirmation_timestamp: int = Field(default=0, ge=0)
    cert_expiry: int = Field(..., ge=0)
    directory_node: str | None = None
    directory_nodes: list[str] = Field(
        default_factory=list,
        description="All directory nodes that announced this bond (for statistics)",
    )
    fidelity_bond_data: dict[str, Any] | None = None
Attributes
amount: int = Field(default=0, ge=0) class-attribute instance-attribute
bond_value: int | None = Field(default=None, ge=0) class-attribute instance-attribute
cert_expiry: int = Field(..., ge=0) class-attribute instance-attribute
counterparty: str instance-attribute
directory_node: str | None = None class-attribute instance-attribute
directory_nodes: list[str] = Field(default_factory=list, description='All directory nodes that announced this bond (for statistics)') class-attribute instance-attribute
fidelity_bond_data: dict[str, Any] | None = None class-attribute instance-attribute
locktime: int = Field(..., ge=0) class-attribute instance-attribute
script: str instance-attribute
utxo_confirmation_timestamp: int = Field(default=0, ge=0) class-attribute instance-attribute
utxo_confirmations: int = Field(..., ge=0) class-attribute instance-attribute
utxo_txid: str = Field(..., pattern='^[0-9a-fA-F]{64}$') class-attribute instance-attribute
utxo_vout: int = Field(..., ge=0) class-attribute instance-attribute
verification_stale: bool = Field(default=False, description='Whether the displayed value is retained past its verification TTL') class-attribute instance-attribute
verification_valid: bool | None = Field(default=None, description='Whether backend verification definitively accepted or rejected this bond') class-attribute instance-attribute

HandshakeRequest

Bases: BaseModel

Source code in jmcore/src/jmcore/models.py 
189
190
191
192
193
194
195
196
class HandshakeRequest(BaseModel):
    app_name: str = "JoinMarket"
    directory: bool = False
    location_string: str
    proto_ver: int
    features: dict[str, Any] = Field(default_factory=dict)
    nick: str = Field(..., min_length=1)
    network: NetworkType
Attributes
app_name: str = 'JoinMarket' class-attribute instance-attribute
directory: bool = False class-attribute instance-attribute
features: dict[str, Any] = Field(default_factory=dict) class-attribute instance-attribute
location_string: str instance-attribute
network: NetworkType instance-attribute
nick: str = Field(..., min_length=1) class-attribute instance-attribute
proto_ver: int instance-attribute

HandshakeResponse

Bases: BaseModel

Source code in jmcore/src/jmcore/models.py 
199
200
201
202
203
204
205
206
207
208
class HandshakeResponse(BaseModel):
    app_name: str = "JoinMarket"
    directory: bool = True
    proto_ver_min: int
    proto_ver_max: int
    features: dict[str, Any] = Field(default_factory=dict)
    accepted: bool
    nick: str = Field(..., min_length=1)
    network: NetworkType
    motd: str = "JoinMarket Directory Server"
Attributes
accepted: bool instance-attribute
app_name: str = 'JoinMarket' class-attribute instance-attribute
directory: bool = True class-attribute instance-attribute
features: dict[str, Any] = Field(default_factory=dict) class-attribute instance-attribute
motd: str = 'JoinMarket Directory Server' class-attribute instance-attribute
network: NetworkType instance-attribute
nick: str = Field(..., min_length=1) class-attribute instance-attribute
proto_ver_max: int instance-attribute
proto_ver_min: int instance-attribute

MessageEnvelope

Bases: BaseModel

Source code in jmcore/src/jmcore/models.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
class MessageEnvelope(BaseModel):
    message_type: int = Field(..., ge=0)
    payload: str
    timestamp: datetime = Field(default_factory=lambda: datetime.now(UTC))

    def to_bytes(self) -> bytes:
        import json

        result = json.dumps({"type": self.message_type, "line": self.payload}).encode("utf-8")
        return result

    @classmethod
    def from_bytes(
        cls, data: bytes, max_line_length: int = 65536, max_json_nesting_depth: int = 10
    ) -> MessageEnvelope:
        """
        Parse a message envelope from bytes with security limits.

        Args:
            data: Raw message bytes (without \\r\\n terminator)
            max_line_length: Maximum allowed line length in bytes (default 64KB)
            max_json_nesting_depth: Maximum JSON nesting depth (default 10)

        Returns:
            Parsed MessageEnvelope

        Raises:
            MessageParsingError: If message exceeds security limits
            json.JSONDecodeError: If JSON is malformed
        """
        import json

        # Check line length BEFORE parsing to prevent DoS
        if len(data) > max_line_length:
            raise MessageParsingError(
                f"Message line length {len(data)} exceeds maximum of {max_line_length} bytes"
            )

        try:
            # Parse and validate nesting depth before creating the model.
            obj = json.loads(data)
            validate_json_nesting_depth(obj, max_json_nesting_depth)
        except RecursionError as exc:
            raise MessageParsingError("JSON nesting depth exceeds parser limits") from exc

        return cls(message_type=obj["type"], payload=obj["line"])
Attributes
message_type: int = Field(..., ge=0) class-attribute instance-attribute
payload: str instance-attribute
timestamp: datetime = Field(default_factory=lambda: datetime.now(UTC)) class-attribute instance-attribute
Methods:
from_bytes(data: bytes, max_line_length: int = 65536, max_json_nesting_depth: int = 10) -> MessageEnvelope classmethod

Parse a message envelope from bytes with security limits.

Args: data: Raw message bytes (without \r\n terminator) max_line_length: Maximum allowed line length in bytes (default 64KB) max_json_nesting_depth: Maximum JSON nesting depth (default 10)

Returns: Parsed MessageEnvelope

Raises: MessageParsingError: If message exceeds security limits json.JSONDecodeError: If JSON is malformed

Source code in jmcore/src/jmcore/models.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
@classmethod
def from_bytes(
    cls, data: bytes, max_line_length: int = 65536, max_json_nesting_depth: int = 10
) -> MessageEnvelope:
    """
    Parse a message envelope from bytes with security limits.

    Args:
        data: Raw message bytes (without \\r\\n terminator)
        max_line_length: Maximum allowed line length in bytes (default 64KB)
        max_json_nesting_depth: Maximum JSON nesting depth (default 10)

    Returns:
        Parsed MessageEnvelope

    Raises:
        MessageParsingError: If message exceeds security limits
        json.JSONDecodeError: If JSON is malformed
    """
    import json

    # Check line length BEFORE parsing to prevent DoS
    if len(data) > max_line_length:
        raise MessageParsingError(
            f"Message line length {len(data)} exceeds maximum of {max_line_length} bytes"
        )

    try:
        # Parse and validate nesting depth before creating the model.
        obj = json.loads(data)
        validate_json_nesting_depth(obj, max_json_nesting_depth)
    except RecursionError as exc:
        raise MessageParsingError("JSON nesting depth exceeds parser limits") from exc

    return cls(message_type=obj["type"], payload=obj["line"])
to_bytes() -> bytes
Source code in jmcore/src/jmcore/models.py 
146
147
148
149
150
def to_bytes(self) -> bytes:
    import json

    result = json.dumps({"type": self.message_type, "line": self.payload}).encode("utf-8")
    return result

MessageParsingError

Bases: Exception

Exception raised when message parsing fails due to security limits.

Source code in jmcore/src/jmcore/models.py 
19
20
21
22
class MessageParsingError(Exception):
    """Exception raised when message parsing fails due to security limits."""

    pass

NetworkType

Bases: StrEnum

Source code in jmcore/src/jmcore/models.py 
89
90
91
92
93
class NetworkType(StrEnum):
    MAINNET = "mainnet"
    TESTNET = "testnet"
    SIGNET = "signet"
    REGTEST = "regtest"
Attributes
MAINNET = 'mainnet' class-attribute instance-attribute
REGTEST = 'regtest' class-attribute instance-attribute
SIGNET = 'signet' class-attribute instance-attribute
TESTNET = 'testnet' class-attribute instance-attribute

Offer

Bases: BaseModel

Source code in jmcore/src/jmcore/models.py 
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
class Offer(BaseModel):
    counterparty: str = Field(..., min_length=1)
    oid: int = Field(..., ge=0)
    ordertype: OfferType
    minsize: int = Field(..., ge=0, le=MAX_MONEY)
    maxsize: int = Field(..., ge=0, le=MAX_MONEY)
    txfee: int = Field(..., ge=0, le=MAX_MONEY)
    cjfee: str | int
    fidelity_bond_value: int = Field(default=0, ge=0)
    fidelity_bond_verified: bool | None = Field(
        default=None,
        description="Whether the watcher definitively verified or rejected the advertised bond",
    )
    fidelity_bond_verification_stale: bool = Field(
        default=False,
        description="Whether watcher verification exceeded its revalidation TTL",
    )
    directory_node: str | None = None
    directory_nodes: list[str] = Field(
        default_factory=list,
        description="All directory nodes that announced this offer (for statistics)",
    )
    fidelity_bond_data: dict[str, Any] | None = None
    neutrino_compat: bool = Field(
        default=False,
        description="Maker requires extended UTXO format (neutrino-compatible backend)",
    )
    features: dict[str, bool] = Field(
        default_factory=dict,
        description="Features supported by this maker (from handshake)",
    )
    directly_reachable: bool | None = Field(
        default=None,
        description="Whether maker is directly reachable via their onion address (None = not checked)",
    )

    @field_validator("cjfee")
    @classmethod
    def validate_cjfee(cls, v: str | int, info) -> str | int:
        ordertype = info.data.get("ordertype")
        if ordertype in (OfferType.SW0_ABSOLUTE, OfferType.SWA_ABSOLUTE):
            # Absolute fees are integer satoshis; reject negatives and absurdly
            # large values that could overflow downstream amount arithmetic.
            iv = int(v)
            if iv < 0:
                raise ValueError("absolute cjfee must be non-negative")
            # 21M BTC in satoshis - any single-offer fee above this is nonsense
            # and could be used to trick takers into oversized fee calculations.
            if iv > MAX_MONEY:
                raise ValueError("absolute cjfee exceeds maximum money supply")
            return iv
        return normalize_relative_fee(v, "relative cjfee")

    @model_validator(mode="after")
    def validate_size_range(self) -> Offer:
        """Reject offers whose advertised minimum exceeds their maximum."""
        if self.minsize > self.maxsize:
            raise ValueError("minsize must be less than or equal to maxsize")
        return self

    def is_absolute_fee(self) -> bool:
        return is_absolute_offer_type(self.ordertype)

    def calculate_fee(self, amount: int) -> int:
        return calculate_cj_fee(self.ordertype, self.cjfee, amount)
Attributes
cjfee: str | int instance-attribute
counterparty: str = Field(..., min_length=1) class-attribute instance-attribute
directly_reachable: bool | None = Field(default=None, description='Whether maker is directly reachable via their onion address (None = not checked)') class-attribute instance-attribute
directory_node: str | None = None class-attribute instance-attribute
directory_nodes: list[str] = Field(default_factory=list, description='All directory nodes that announced this offer (for statistics)') class-attribute instance-attribute
features: dict[str, bool] = Field(default_factory=dict, description='Features supported by this maker (from handshake)') class-attribute instance-attribute
fidelity_bond_data: dict[str, Any] | None = None class-attribute instance-attribute
fidelity_bond_value: int = Field(default=0, ge=0) class-attribute instance-attribute
fidelity_bond_verification_stale: bool = Field(default=False, description='Whether watcher verification exceeded its revalidation TTL') class-attribute instance-attribute
fidelity_bond_verified: bool | None = Field(default=None, description='Whether the watcher definitively verified or rejected the advertised bond') class-attribute instance-attribute
maxsize: int = Field(..., ge=0, le=MAX_MONEY) class-attribute instance-attribute
minsize: int = Field(..., ge=0, le=MAX_MONEY) class-attribute instance-attribute
neutrino_compat: bool = Field(default=False, description='Maker requires extended UTXO format (neutrino-compatible backend)') class-attribute instance-attribute
oid: int = Field(..., ge=0) class-attribute instance-attribute
ordertype: OfferType instance-attribute
txfee: int = Field(..., ge=0, le=MAX_MONEY) class-attribute instance-attribute
Methods:
calculate_fee(amount: int) -> int
Source code in jmcore/src/jmcore/models.py 
349
350
def calculate_fee(self, amount: int) -> int:
    return calculate_cj_fee(self.ordertype, self.cjfee, amount)
is_absolute_fee() -> bool
Source code in jmcore/src/jmcore/models.py 
346
347
def is_absolute_fee(self) -> bool:
    return is_absolute_offer_type(self.ordertype)
validate_cjfee(v: str | int, info) -> str | int classmethod
Source code in jmcore/src/jmcore/models.py 
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
@field_validator("cjfee")
@classmethod
def validate_cjfee(cls, v: str | int, info) -> str | int:
    ordertype = info.data.get("ordertype")
    if ordertype in (OfferType.SW0_ABSOLUTE, OfferType.SWA_ABSOLUTE):
        # Absolute fees are integer satoshis; reject negatives and absurdly
        # large values that could overflow downstream amount arithmetic.
        iv = int(v)
        if iv < 0:
            raise ValueError("absolute cjfee must be non-negative")
        # 21M BTC in satoshis - any single-offer fee above this is nonsense
        # and could be used to trick takers into oversized fee calculations.
        if iv > MAX_MONEY:
            raise ValueError("absolute cjfee exceeds maximum money supply")
        return iv
    return normalize_relative_fee(v, "relative cjfee")
validate_size_range() -> Offer

Reject offers whose advertised minimum exceeds their maximum.

Source code in jmcore/src/jmcore/models.py 
339
340
341
342
343
344
@model_validator(mode="after")
def validate_size_range(self) -> Offer:
    """Reject offers whose advertised minimum exceeds their maximum."""
    if self.minsize > self.maxsize:
        raise ValueError("minsize must be less than or equal to maxsize")
    return self

OfferType

Bases: StrEnum

Source code in jmcore/src/jmcore/models.py 
211
212
213
214
215
class OfferType(StrEnum):
    SW0_ABSOLUTE = "sw0absoffer"
    SW0_RELATIVE = "sw0reloffer"
    SWA_ABSOLUTE = "swabsoffer"
    SWA_RELATIVE = "swreloffer"
Attributes
SW0_ABSOLUTE = 'sw0absoffer' class-attribute instance-attribute
SW0_RELATIVE = 'sw0reloffer' class-attribute instance-attribute
SWA_ABSOLUTE = 'swabsoffer' class-attribute instance-attribute
SWA_RELATIVE = 'swreloffer' class-attribute instance-attribute

OrderBook

Bases: BaseModel

Source code in jmcore/src/jmcore/models.py 
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
class OrderBook(BaseModel):
    offers: list[Offer] = Field(default_factory=list)
    fidelity_bonds: list[FidelityBond] = Field(default_factory=list)
    current_block_height: int | None = Field(default=None, ge=0)
    timestamp: datetime = Field(default_factory=lambda: datetime.now(UTC))
    directory_nodes: list[str] = Field(default_factory=list)

    def add_offers(self, offers: list[Offer], directory_node: str) -> None:
        for offer in offers:
            offer.directory_node = directory_node
        self.offers.extend(offers)
        if directory_node not in self.directory_nodes:
            self.directory_nodes.append(directory_node)

    def add_fidelity_bonds(self, bonds: list[FidelityBond], directory_node: str) -> None:
        for bond in bonds:
            bond.directory_node = directory_node
        self.fidelity_bonds.extend(bonds)

    def get_offers_by_directory(self) -> dict[str, list[Offer]]:
        """Get offers grouped by directory node.

        This uses the directory_nodes list (plural) which tracks all directories
        that announced each offer, so an offer will appear under multiple
        directories if it was announced by multiple directories.
        """
        result: dict[str, list[Offer]] = {}
        for offer in self.offers:
            # Use directory_nodes (plural) if populated, otherwise fallback to directory_node
            nodes = offer.directory_nodes if offer.directory_nodes else []
            if not nodes and offer.directory_node:
                nodes = [offer.directory_node]
            if not nodes:
                nodes = ["unknown"]
            for node in nodes:
                if node not in result:
                    result[node] = []
                result[node].append(offer)
        return result
Attributes
current_block_height: int | None = Field(default=None, ge=0) class-attribute instance-attribute
directory_nodes: list[str] = Field(default_factory=list) class-attribute instance-attribute
fidelity_bonds: list[FidelityBond] = Field(default_factory=list) class-attribute instance-attribute
offers: list[Offer] = Field(default_factory=list) class-attribute instance-attribute
timestamp: datetime = Field(default_factory=lambda: datetime.now(UTC)) class-attribute instance-attribute
Methods:
add_fidelity_bonds(bonds: list[FidelityBond], directory_node: str) -> None
Source code in jmcore/src/jmcore/models.py 
394
395
396
397
def add_fidelity_bonds(self, bonds: list[FidelityBond], directory_node: str) -> None:
    for bond in bonds:
        bond.directory_node = directory_node
    self.fidelity_bonds.extend(bonds)
add_offers(offers: list[Offer], directory_node: str) -> None
Source code in jmcore/src/jmcore/models.py 
387
388
389
390
391
392
def add_offers(self, offers: list[Offer], directory_node: str) -> None:
    for offer in offers:
        offer.directory_node = directory_node
    self.offers.extend(offers)
    if directory_node not in self.directory_nodes:
        self.directory_nodes.append(directory_node)
get_offers_by_directory() -> dict[str, list[Offer]]

Get offers grouped by directory node.

This uses the directory_nodes list (plural) which tracks all directories that announced each offer, so an offer will appear under multiple directories if it was announced by multiple directories.

Source code in jmcore/src/jmcore/models.py 
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
def get_offers_by_directory(self) -> dict[str, list[Offer]]:
    """Get offers grouped by directory node.

    This uses the directory_nodes list (plural) which tracks all directories
    that announced each offer, so an offer will appear under multiple
    directories if it was announced by multiple directories.
    """
    result: dict[str, list[Offer]] = {}
    for offer in self.offers:
        # Use directory_nodes (plural) if populated, otherwise fallback to directory_node
        nodes = offer.directory_nodes if offer.directory_nodes else []
        if not nodes and offer.directory_node:
            nodes = [offer.directory_node]
        if not nodes:
            nodes = ["unknown"]
        for node in nodes:
            if node not in result:
                result[node] = []
            result[node].append(offer)
    return result

PeerInfo

Bases: BaseModel

Source code in jmcore/src/jmcore/models.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
class PeerInfo(BaseModel):
    nick: str = Field(..., min_length=1, max_length=64)
    onion_address: str = Field(..., pattern=r"^[a-z2-7]{56}\.onion$|^NOT-SERVING-ONION$")
    port: int = Field(..., ge=-1, le=65535)
    status: PeerStatus = PeerStatus.UNCONNECTED
    is_directory: bool = False
    network: NetworkType = NetworkType.MAINNET
    last_seen: datetime | None = None
    features: dict[str, Any] = Field(default_factory=dict)
    protocol_version: int = Field(default=5, ge=5, le=10)  # Negotiated protocol version
    neutrino_compat: bool = False  # True if peer supports extended UTXO metadata

    @field_validator("onion_address")
    @classmethod
    def validate_onion(cls, v: str) -> str:
        if v == "NOT-SERVING-ONION":
            return v
        if not v.endswith(".onion"):
            raise ValueError("Invalid onion address")
        return v

    @field_validator("port")
    @classmethod
    def validate_port(cls, v: int, info) -> int:
        if v == -1 and info.data.get("onion_address") == "NOT-SERVING-ONION":
            return v
        if v < 1 or v > 65535:
            raise ValueError("Port must be between 1 and 65535")
        return v

    @cached_property
    def location_string(self) -> str:
        if self.onion_address == "NOT-SERVING-ONION":
            return "NOT-SERVING-ONION"
        return f"{self.onion_address}:{self.port}"

    def supports_extended_utxo(self) -> bool:
        """Check if this peer supports extended UTXO format (neutrino_compat)."""
        # With feature-based detection, we check the neutrino_compat flag
        # which is set from the features dict during handshake
        return self.neutrino_compat

    model_config = {"frozen": False}
Attributes
features: dict[str, Any] = Field(default_factory=dict) class-attribute instance-attribute
is_directory: bool = False class-attribute instance-attribute
last_seen: datetime | None = None class-attribute instance-attribute
location_string: str cached property
model_config = {'frozen': False} class-attribute instance-attribute
network: NetworkType = NetworkType.MAINNET class-attribute instance-attribute
neutrino_compat: bool = False class-attribute instance-attribute
nick: str = Field(..., min_length=1, max_length=64) class-attribute instance-attribute
onion_address: str = Field(..., pattern='^[a-z2-7]{56}\\.onion$|^NOT-SERVING-ONION$') class-attribute instance-attribute
port: int = Field(..., ge=-1, le=65535) class-attribute instance-attribute
protocol_version: int = Field(default=5, ge=5, le=10) class-attribute instance-attribute
status: PeerStatus = PeerStatus.UNCONNECTED class-attribute instance-attribute
Methods:
supports_extended_utxo() -> bool

Check if this peer supports extended UTXO format (neutrino_compat).

Source code in jmcore/src/jmcore/models.py 
132
133
134
135
136
def supports_extended_utxo(self) -> bool:
    """Check if this peer supports extended UTXO format (neutrino_compat)."""
    # With feature-based detection, we check the neutrino_compat flag
    # which is set from the features dict during handshake
    return self.neutrino_compat
validate_onion(v: str) -> str classmethod
Source code in jmcore/src/jmcore/models.py 
108
109
110
111
112
113
114
115
@field_validator("onion_address")
@classmethod
def validate_onion(cls, v: str) -> str:
    if v == "NOT-SERVING-ONION":
        return v
    if not v.endswith(".onion"):
        raise ValueError("Invalid onion address")
    return v
validate_port(v: int, info) -> int classmethod
Source code in jmcore/src/jmcore/models.py 
117
118
119
120
121
122
123
124
@field_validator("port")
@classmethod
def validate_port(cls, v: int, info) -> int:
    if v == -1 and info.data.get("onion_address") == "NOT-SERVING-ONION":
        return v
    if v < 1 or v > 65535:
        raise ValueError("Port must be between 1 and 65535")
    return v

PeerStatus

Bases: StrEnum

Source code in jmcore/src/jmcore/models.py 
82
83
84
85
86
class PeerStatus(StrEnum):
    UNCONNECTED = "unconnected"
    CONNECTED = "connected"
    HANDSHAKED = "handshaked"
    DISCONNECTED = "disconnected"
Attributes
CONNECTED = 'connected' class-attribute instance-attribute
DISCONNECTED = 'disconnected' class-attribute instance-attribute
HANDSHAKED = 'handshaked' class-attribute instance-attribute
UNCONNECTED = 'unconnected' class-attribute instance-attribute

Functions:

calculate_cj_fee(offer_type: OfferType, cjfee: str | int, amount: int) -> int

Calculate actual CoinJoin fee based on offer type.

This is the canonical fee calculation used by both makers and takers.

Args: offer_type: Absolute or relative offer type cjfee: Fee value (int for absolute, string decimal for relative) amount: CoinJoin amount in satoshis

Returns: Actual fee in satoshis

Source code in jmcore/src/jmcore/models.py 
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
def calculate_cj_fee(offer_type: OfferType, cjfee: str | int, amount: int) -> int:
    """
    Calculate actual CoinJoin fee based on offer type.

    This is the canonical fee calculation used by both makers and takers.

    Args:
        offer_type: Absolute or relative offer type
        cjfee: Fee value (int for absolute, string decimal for relative)
        amount: CoinJoin amount in satoshis

    Returns:
        Actual fee in satoshis
    """
    if is_absolute_offer_type(offer_type):
        return int(cjfee)
    else:
        return calculate_relative_fee(amount, str(cjfee))

get_default_directory_nodes(network: NetworkType) -> list[str]

Get default directory nodes for a given network.

Source code in jmcore/src/jmcore/models.py 
70
71
72
73
74
75
76
77
78
79
def get_default_directory_nodes(network: NetworkType) -> list[str]:
    """Get default directory nodes for a given network."""
    if network == NetworkType.MAINNET:
        return DIRECTORY_NODES_MAINNET.copy()
    elif network == NetworkType.SIGNET:
        return DIRECTORY_NODES_SIGNET.copy()
    elif network == NetworkType.TESTNET:
        return DIRECTORY_NODES_TESTNET.copy()
    # Regtest has no default directory nodes - must be configured
    return []

is_absolute_offer_type(offer_type: OfferType) -> bool

Check if an offer type uses absolute fees.

Source code in jmcore/src/jmcore/models.py 
261
262
263
def is_absolute_offer_type(offer_type: OfferType) -> bool:
    """Check if an offer type uses absolute fees."""
    return offer_type in (OfferType.SW0_ABSOLUTE, OfferType.SWA_ABSOLUTE)

normalize_relative_fee(value: str | int | float | Decimal, field_name: str) -> str

Validate and normalize a relative fee without unbounded fixed-point formatting.

Source code in jmcore/src/jmcore/models.py 
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
def normalize_relative_fee(value: str | int | float | Decimal, field_name: str) -> str:
    """Validate and normalize a relative fee without unbounded fixed-point formatting."""
    text = str(value)
    if len(text) > MAX_RELATIVE_FEE_INPUT_LENGTH:
        raise ValueError(
            f"{field_name} input exceeds maximum length of {MAX_RELATIVE_FEE_INPUT_LENGTH}"
        )

    try:
        fee = Decimal(text)
    except InvalidOperation as exc:
        raise ValueError(f"{field_name} is not a valid decimal: {text!r}") from exc

    if not fee.is_finite():
        raise ValueError(f"{field_name} must be finite")

    decimal_tuple = fee.as_tuple()
    if len(decimal_tuple.digits) > MAX_RELATIVE_FEE_PRECISION:
        raise ValueError(
            f"{field_name} has too many significant digits (maximum {MAX_RELATIVE_FEE_PRECISION})"
        )

    exponent = decimal_tuple.exponent
    if not isinstance(exponent, int):  # Defensive, non-finite values were rejected above.
        raise ValueError(f"{field_name} must be finite")
    if not -MAX_RELATIVE_FEE_EXPONENT <= exponent <= MAX_RELATIVE_FEE_EXPONENT:
        raise ValueError(
            f"{field_name} exponent must be between "
            f"{-MAX_RELATIVE_FEE_EXPONENT} and {MAX_RELATIVE_FEE_EXPONENT}"
        )

    if fee < 0:
        raise ValueError(f"{field_name} must be non-negative")
    if fee >= 1:
        raise ValueError(f"{field_name} must be less than 1")
    return format(fee, "f")

validate_json_nesting_depth(obj: Any, max_depth: int = 10, current_depth: int = 0) -> None

Validate that a JSON object does not exceed maximum nesting depth.

Args: obj: The object to validate (dict, list, or primitive) max_depth: Maximum allowed nesting depth current_depth: Current depth in recursion

Raises: MessageParsingError: If nesting depth exceeds max_depth

Source code in jmcore/src/jmcore/models.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def validate_json_nesting_depth(obj: Any, max_depth: int = 10, current_depth: int = 0) -> None:
    """
    Validate that a JSON object does not exceed maximum nesting depth.

    Args:
        obj: The object to validate (dict, list, or primitive)
        max_depth: Maximum allowed nesting depth
        current_depth: Current depth in recursion

    Raises:
        MessageParsingError: If nesting depth exceeds max_depth
    """
    if current_depth > max_depth:
        raise MessageParsingError(f"JSON nesting depth exceeds maximum of {max_depth}")

    if isinstance(obj, dict):
        for value in obj.values():
            validate_json_nesting_depth(value, max_depth, current_depth + 1)
    elif isinstance(obj, list):
        for item in obj:
            validate_json_nesting_depth(item, max_depth, current_depth + 1)