Skip to content

jmcore.protocol

jmcore.protocol

JoinMarket protocol definitions, message types, and serialization.

Feature Flag System

This implementation uses feature flags for capability negotiation instead of protocol version bumping. This allows incremental feature adoption while maintaining full compatibility with the reference implementation from joinmarket-clientserver.

Features are advertised in the handshake features dict and negotiated per-CoinJoin session via extended !fill/!pubkey messages.

Available Features: - neutrino_compat: Extended UTXO metadata (scriptpubkey, blockheight) for light client verification. Required for Neutrino backend takers. - push_encrypted: Encrypted !push command with session binding. Prevents abuse of makers as unauthenticated broadcast bots.

Feature Dependencies: - neutrino_compat: No dependencies - push_encrypted: Requires active NaCl encryption session (implicit)

Nick Format:

JoinMarket nicks encode the protocol version: J{version}{hash} All nicks use version 5 for maximum compatibility with reference implementation. Feature detection happens via handshake and !fill/!pubkey exchange, not nick.

Cross-Implementation Compatibility:

Our Implementation ↔ Reference (JAM): - We use J5 nicks and proto-ver=5 in handshake - Features field is ignored by reference implementation - Legacy UTXO format used unless both peers advertise neutrino_compat - Graceful fallback to v5 behavior for all features

Feature Negotiation During CoinJoin: - Taker advertises features in !fill (optional JSON suffix) - Maker responds with features in !pubkey (optional JSON suffix) - Extended formats used only when both peers support the feature

Peerlist Feature Extension: Our directory server extends the peerlist format to include features: - Legacy format: nick;location (or nick;location;D for disconnected) - Extended format: nick;location;F:feature1+feature2 (features as plus-separated list) The extended format is backward compatible - legacy clients will ignore the F: suffix. Note: Plus separator is used because the peerlist itself uses commas to separate entries.

Attributes

ALL_FEATURES = {FEATURE_NEUTRINO_COMPAT, FEATURE_PUSH_ENCRYPTED, FEATURE_PEERLIST_FEATURES, FEATURE_PING} module-attribute

COMMAND_PREFIX = '!' module-attribute

FEATURE_DEPENDENCIES: dict[str, list[str]] = {FEATURE_NEUTRINO_COMPAT: [], FEATURE_PUSH_ENCRYPTED: [], FEATURE_PEERLIST_FEATURES: [], FEATURE_PING: []} module-attribute

FEATURE_NEUTRINO_COMPAT = 'neutrino_compat' module-attribute

FEATURE_PEERLIST_FEATURES = 'peerlist_features' module-attribute

FEATURE_PING = 'ping' module-attribute

FEATURE_PUSH_ENCRYPTED = 'push_encrypted' module-attribute

JM_VERSION = 5 module-attribute

JM_VERSION_MIN = JM_VERSION module-attribute

NICK_HASH_LENGTH = 10 module-attribute

NICK_MAX_ENCODED = 14 module-attribute

NICK_PEERLOCATOR_SEPARATOR = ';' module-attribute

NOT_SERVING_ONION_HOSTNAME = 'NOT-SERVING-ONION' module-attribute

ONION_VIRTUAL_PORT = 5222 module-attribute

Classes

FeatureSet

Represents a set of protocol features advertised by a peer.

Used for feature negotiation during handshake and CoinJoin sessions.

Source code in jmcore/src/jmcore/protocol.py 
 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
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
@dataclass
class FeatureSet:
    """
    Represents a set of protocol features advertised by a peer.

    Used for feature negotiation during handshake and CoinJoin sessions.
    """

    features: set[str] = Field(default_factory=set)

    @classmethod
    def from_handshake(cls, handshake_data: dict[str, Any]) -> FeatureSet:
        """Extract features from a handshake payload."""
        features_dict = handshake_data.get("features", {})
        # Only include features that are set to True
        features = {k for k, v in features_dict.items() if v is True}
        return cls(features=features)

    @classmethod
    def from_list(cls, feature_list: list[str]) -> FeatureSet:
        """Create from a list of feature names."""
        return cls(features=set(feature_list))

    @classmethod
    def from_comma_string(cls, s: str) -> FeatureSet:
        """Parse from plus-separated string (e.g., 'neutrino_compat+push_encrypted').

        Note: Despite the method name, uses '+' as separator because the peerlist
        itself uses ',' to separate entries. The name is kept for backward compatibility.
        Also accepts ',' for legacy/handshake use cases.
        """
        if not s or not s.strip():
            return cls(features=set())
        # Support both + (peerlist) and , (legacy/handshake) separators
        if "+" in s:
            return cls(features={f.strip() for f in s.split("+") if f.strip()})
        return cls(features={f.strip() for f in s.split(",") if f.strip()})

    def to_dict(self) -> dict[str, bool]:
        """Convert to dict for JSON serialization."""
        return dict.fromkeys(sorted(self.features), True)

    def to_comma_string(self) -> str:
        """Convert to plus-separated string for peerlist F: suffix.

        Note: Uses '+' as separator instead of ',' because the peerlist
        itself uses ',' to separate entries. Using ',' for features would
        cause parsing ambiguity.
        """
        return "+".join(sorted(self.features))

    def supports(self, feature: str) -> bool:
        """Check if this set includes a specific feature."""
        return feature in self.features

    def supports_neutrino_compat(self) -> bool:
        """Check if neutrino_compat is supported."""
        return FEATURE_NEUTRINO_COMPAT in self.features

    def supports_push_encrypted(self) -> bool:
        """Check if push_encrypted is supported."""
        return FEATURE_PUSH_ENCRYPTED in self.features

    def supports_peerlist_features(self) -> bool:
        """Check if peer supports extended peerlist with features (F: suffix)."""
        return FEATURE_PEERLIST_FEATURES in self.features

    def supports_ping(self) -> bool:
        """Check if peer supports application-level PING/PONG heartbeat."""
        return FEATURE_PING in self.features

    def validate_dependencies(self) -> tuple[bool, str]:
        """Check that all feature dependencies are satisfied."""
        for feature in self.features:
            deps = FEATURE_DEPENDENCIES.get(feature, [])
            for dep in deps:
                if dep not in self.features:
                    return False, f"Feature '{feature}' requires '{dep}'"
        return True, ""

    def intersection(self, other: FeatureSet) -> FeatureSet:
        """Return features supported by both sets."""
        return FeatureSet(features=self.features & other.features)

    def __bool__(self) -> bool:
        """True if any features are set."""
        return bool(self.features)

    def __contains__(self, feature: str) -> bool:
        return feature in self.features

    def __iter__(self):
        return iter(self.features)

    def __len__(self) -> int:
        return len(self.features)
Attributes
features: set[str] = Field(default_factory=set) class-attribute instance-attribute
Functions
__bool__() -> bool

True if any features are set.

Source code in jmcore/src/jmcore/protocol.py 
182
183
184
def __bool__(self) -> bool:
    """True if any features are set."""
    return bool(self.features)
__contains__(feature: str) -> bool
Source code in jmcore/src/jmcore/protocol.py 
186
187
def __contains__(self, feature: str) -> bool:
    return feature in self.features
__iter__()
Source code in jmcore/src/jmcore/protocol.py 
189
190
def __iter__(self):
    return iter(self.features)
__len__() -> int
Source code in jmcore/src/jmcore/protocol.py 
192
193
def __len__(self) -> int:
    return len(self.features)
from_comma_string(s: str) -> FeatureSet classmethod

Parse from plus-separated string (e.g., 'neutrino_compat+push_encrypted').

Note: Despite the method name, uses '+' as separator because the peerlist itself uses ',' to separate entries. The name is kept for backward compatibility. Also accepts ',' for legacy/handshake use cases.

Source code in jmcore/src/jmcore/protocol.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
@classmethod
def from_comma_string(cls, s: str) -> FeatureSet:
    """Parse from plus-separated string (e.g., 'neutrino_compat+push_encrypted').

    Note: Despite the method name, uses '+' as separator because the peerlist
    itself uses ',' to separate entries. The name is kept for backward compatibility.
    Also accepts ',' for legacy/handshake use cases.
    """
    if not s or not s.strip():
        return cls(features=set())
    # Support both + (peerlist) and , (legacy/handshake) separators
    if "+" in s:
        return cls(features={f.strip() for f in s.split("+") if f.strip()})
    return cls(features={f.strip() for f in s.split(",") if f.strip()})
from_handshake(handshake_data: dict[str, Any]) -> FeatureSet classmethod

Extract features from a handshake payload.

Source code in jmcore/src/jmcore/protocol.py 
108
109
110
111
112
113
114
@classmethod
def from_handshake(cls, handshake_data: dict[str, Any]) -> FeatureSet:
    """Extract features from a handshake payload."""
    features_dict = handshake_data.get("features", {})
    # Only include features that are set to True
    features = {k for k, v in features_dict.items() if v is True}
    return cls(features=features)
from_list(feature_list: list[str]) -> FeatureSet classmethod

Create from a list of feature names.

Source code in jmcore/src/jmcore/protocol.py 
116
117
118
119
@classmethod
def from_list(cls, feature_list: list[str]) -> FeatureSet:
    """Create from a list of feature names."""
    return cls(features=set(feature_list))
intersection(other: FeatureSet) -> FeatureSet

Return features supported by both sets.

Source code in jmcore/src/jmcore/protocol.py 
178
179
180
def intersection(self, other: FeatureSet) -> FeatureSet:
    """Return features supported by both sets."""
    return FeatureSet(features=self.features & other.features)
supports(feature: str) -> bool

Check if this set includes a specific feature.

Source code in jmcore/src/jmcore/protocol.py 
149
150
151
def supports(self, feature: str) -> bool:
    """Check if this set includes a specific feature."""
    return feature in self.features
supports_neutrino_compat() -> bool

Check if neutrino_compat is supported.

Source code in jmcore/src/jmcore/protocol.py 
153
154
155
def supports_neutrino_compat(self) -> bool:
    """Check if neutrino_compat is supported."""
    return FEATURE_NEUTRINO_COMPAT in self.features
supports_peerlist_features() -> bool

Check if peer supports extended peerlist with features (F: suffix).

Source code in jmcore/src/jmcore/protocol.py 
161
162
163
def supports_peerlist_features(self) -> bool:
    """Check if peer supports extended peerlist with features (F: suffix)."""
    return FEATURE_PEERLIST_FEATURES in self.features
supports_ping() -> bool

Check if peer supports application-level PING/PONG heartbeat.

Source code in jmcore/src/jmcore/protocol.py 
165
166
167
def supports_ping(self) -> bool:
    """Check if peer supports application-level PING/PONG heartbeat."""
    return FEATURE_PING in self.features
supports_push_encrypted() -> bool

Check if push_encrypted is supported.

Source code in jmcore/src/jmcore/protocol.py 
157
158
159
def supports_push_encrypted(self) -> bool:
    """Check if push_encrypted is supported."""
    return FEATURE_PUSH_ENCRYPTED in self.features
to_comma_string() -> str

Convert to plus-separated string for peerlist F: suffix.

Note: Uses '+' as separator instead of ',' because the peerlist itself uses ',' to separate entries. Using ',' for features would cause parsing ambiguity.

Source code in jmcore/src/jmcore/protocol.py 
140
141
142
143
144
145
146
147
def to_comma_string(self) -> str:
    """Convert to plus-separated string for peerlist F: suffix.

    Note: Uses '+' as separator instead of ',' because the peerlist
    itself uses ',' to separate entries. Using ',' for features would
    cause parsing ambiguity.
    """
    return "+".join(sorted(self.features))
to_dict() -> dict[str, bool]

Convert to dict for JSON serialization.

Source code in jmcore/src/jmcore/protocol.py 
136
137
138
def to_dict(self) -> dict[str, bool]:
    """Convert to dict for JSON serialization."""
    return dict.fromkeys(sorted(self.features), True)
validate_dependencies() -> tuple[bool, str]

Check that all feature dependencies are satisfied.

Source code in jmcore/src/jmcore/protocol.py 
169
170
171
172
173
174
175
176
def validate_dependencies(self) -> tuple[bool, str]:
    """Check that all feature dependencies are satisfied."""
    for feature in self.features:
        deps = FEATURE_DEPENDENCIES.get(feature, [])
        for dep in deps:
            if dep not in self.features:
                return False, f"Feature '{feature}' requires '{dep}'"
    return True, ""

MessageType

Bases: IntEnum

Source code in jmcore/src/jmcore/protocol.py 
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
class MessageType(IntEnum):
    PRIVMSG = 685
    PUBMSG = 687
    PEERLIST = 789
    GETPEERLIST = 791
    HANDSHAKE = 793
    DN_HANDSHAKE = 795
    PING = 798
    PONG = 799
    DISCONNECT = 801

    # Local-only control messages, never sent over the wire.
    # The reference implementation uses 797 for CONNECT_IN (a local-only
    # control type).  PING was moved to 798 (JMP-0004) to avoid the collision.
    CONNECT = 785
    CONNECT_IN = 797
Attributes
CONNECT = 785 class-attribute instance-attribute
CONNECT_IN = 797 class-attribute instance-attribute
DISCONNECT = 801 class-attribute instance-attribute
DN_HANDSHAKE = 795 class-attribute instance-attribute
GETPEERLIST = 791 class-attribute instance-attribute
HANDSHAKE = 793 class-attribute instance-attribute
PEERLIST = 789 class-attribute instance-attribute
PING = 798 class-attribute instance-attribute
PONG = 799 class-attribute instance-attribute
PRIVMSG = 685 class-attribute instance-attribute
PUBMSG = 687 class-attribute instance-attribute

ProtocolMessage

Bases: BaseModel

Source code in jmcore/src/jmcore/protocol.py 
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
class ProtocolMessage(BaseModel):
    type: MessageType
    payload: dict[str, Any]

    def to_json(self) -> str:
        return json.dumps({"type": self.type.value, "data": self.payload})

    @classmethod
    def from_json(cls, data: str) -> ProtocolMessage:
        obj = json.loads(data)
        return cls(type=MessageType(obj["type"]), payload=obj["data"])

    def to_bytes(self) -> bytes:
        return self.to_json().encode("utf-8")

    @classmethod
    def from_bytes(cls, data: bytes) -> ProtocolMessage:
        return cls.from_json(data.decode("utf-8"))
Attributes
payload: dict[str, Any] instance-attribute
type: MessageType instance-attribute
Functions
from_bytes(data: bytes) -> ProtocolMessage classmethod
Source code in jmcore/src/jmcore/protocol.py 
413
414
415
@classmethod
def from_bytes(cls, data: bytes) -> ProtocolMessage:
    return cls.from_json(data.decode("utf-8"))
from_json(data: str) -> ProtocolMessage classmethod
Source code in jmcore/src/jmcore/protocol.py 
405
406
407
408
@classmethod
def from_json(cls, data: str) -> ProtocolMessage:
    obj = json.loads(data)
    return cls(type=MessageType(obj["type"]), payload=obj["data"])
to_bytes() -> bytes
Source code in jmcore/src/jmcore/protocol.py 
410
411
def to_bytes(self) -> bytes:
    return self.to_json().encode("utf-8")
to_json() -> str
Source code in jmcore/src/jmcore/protocol.py 
402
403
def to_json(self) -> str:
    return json.dumps({"type": self.type.value, "data": self.payload})

RequiredFeatures

Features that this peer requires from counterparties.

Used to filter incompatible peers during maker selection.

Source code in jmcore/src/jmcore/protocol.py 
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
@dataclass
class RequiredFeatures:
    """
    Features that this peer requires from counterparties.

    Used to filter incompatible peers during maker selection.
    """

    required: set[str] = Field(default_factory=set)

    @classmethod
    def for_neutrino_taker(cls) -> RequiredFeatures:
        """Create requirements for a taker using Neutrino backend."""
        return cls(required={FEATURE_NEUTRINO_COMPAT})

    @classmethod
    def none(cls) -> RequiredFeatures:
        """No required features."""
        return cls(required=set())

    def is_compatible(self, peer_features: FeatureSet) -> tuple[bool, str]:
        """Check if peer supports all required features."""
        missing = self.required - peer_features.features
        if missing:
            return False, f"Missing required features: {missing}"
        return True, ""

    def __bool__(self) -> bool:
        return bool(self.required)
Attributes
required: set[str] = Field(default_factory=set) class-attribute instance-attribute
Functions
__bool__() -> bool
Source code in jmcore/src/jmcore/protocol.py 
223
224
def __bool__(self) -> bool:
    return bool(self.required)
for_neutrino_taker() -> RequiredFeatures classmethod

Create requirements for a taker using Neutrino backend.

Source code in jmcore/src/jmcore/protocol.py 
206
207
208
209
@classmethod
def for_neutrino_taker(cls) -> RequiredFeatures:
    """Create requirements for a taker using Neutrino backend."""
    return cls(required={FEATURE_NEUTRINO_COMPAT})
is_compatible(peer_features: FeatureSet) -> tuple[bool, str]

Check if peer supports all required features.

Source code in jmcore/src/jmcore/protocol.py 
216
217
218
219
220
221
def is_compatible(self, peer_features: FeatureSet) -> tuple[bool, str]:
    """Check if peer supports all required features."""
    missing = self.required - peer_features.features
    if missing:
        return False, f"Missing required features: {missing}"
    return True, ""
none() -> RequiredFeatures classmethod

No required features.

Source code in jmcore/src/jmcore/protocol.py 
211
212
213
214
@classmethod
def none(cls) -> RequiredFeatures:
    """No required features."""
    return cls(required=set())

UTXOMetadata

Extended UTXO metadata for Neutrino-compatible verification.

This allows light clients to verify UTXOs without arbitrary blockchain queries by providing the scriptPubKey (for Neutrino watch list) and block height (for efficient rescan starting point).

Source code in jmcore/src/jmcore/protocol.py 
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
@dataclass
class UTXOMetadata:
    """
    Extended UTXO metadata for Neutrino-compatible verification.

    This allows light clients to verify UTXOs without arbitrary blockchain queries
    by providing the scriptPubKey (for Neutrino watch list) and block height
    (for efficient rescan starting point).
    """

    txid: str
    vout: int
    scriptpubkey: str | None = None  # Hex-encoded scriptPubKey
    blockheight: int | None = None  # Block height where UTXO was confirmed

    def __post_init__(self) -> None:
        """Strict validation of UTXO fields to match legacy protocol."""
        # TXID validation
        if len(self.txid) != 64:
            raise ValueError(f"Invalid TXID length: {len(self.txid)} (expected 64)")
        try:
            binascii.unhexlify(self.txid)
        except (binascii.Error, TypeError) as exc:
            raise ValueError(f"Invalid TXID hex: {self.txid}") from exc

        # Vout validation (Bitcoin uint32)
        if self.vout < 0:
            raise ValueError(f"Invalid vout (must be non-negative): {self.vout}")
        if self.vout > 0xFFFFFFFF:
            raise ValueError(f"Invalid vout (overflow, max 4294967295): {self.vout}")

        # Optional blockheight validation
        if self.blockheight is not None and self.blockheight < 0:
            raise ValueError(f"Invalid blockheight (must be non-negative): {self.blockheight}")

    def to_legacy_str(self) -> str:
        """Format as legacy string: txid:vout"""
        return f"{self.txid}:{self.vout}"

    def to_extended_str(self) -> str:
        """Format as extended string: txid:vout:scriptpubkey:blockheight"""
        if self.scriptpubkey is None or self.blockheight is None:
            return self.to_legacy_str()
        return f"{self.txid}:{self.vout}:{self.scriptpubkey}:{self.blockheight}"

    @classmethod
    def from_str(cls, s: str) -> UTXOMetadata:
        """
        Parse UTXO string in either legacy or extended format.

        Legacy format: txid:vout
        Extended format: txid:vout:scriptpubkey:blockheight
        """
        parts = s.split(":")
        if len(parts) not in [2, 4]:
            raise ValueError(f"Invalid UTXO format: {s}")

        txid = parts[0]
        try:
            vout = int(parts[1])
        except (ValueError, TypeError) as exc:
            raise ValueError(f"Invalid vout (not an integer): {parts[1]}") from exc

        if len(parts) == 2:
            return cls(txid=txid, vout=vout)

        # Extended format
        scriptpubkey = parts[2]
        try:
            blockheight = int(parts[3])
        except (ValueError, TypeError) as exc:
            raise ValueError(f"Invalid blockheight (not an integer): {parts[3]}") from exc

        return cls(
            txid=txid,
            vout=vout,
            scriptpubkey=scriptpubkey,
            blockheight=blockheight,
        )

    def has_neutrino_metadata(self) -> bool:
        """Check if this UTXO has the metadata needed for Neutrino verification."""
        return self.scriptpubkey is not None and self.blockheight is not None

    @staticmethod
    def is_valid_scriptpubkey(scriptpubkey: str) -> bool:
        """Validate scriptPubKey format (hex string)."""
        if not scriptpubkey:
            return False
        # Must be valid hex
        if not re.match(r"^[0-9a-fA-F]+$", scriptpubkey):
            return False
        # Common scriptPubKey lengths (in hex chars):
        # P2PKH: 50 (25 bytes), P2SH: 46 (23 bytes)
        # P2WPKH: 44 (22 bytes), P2WSH: 68 (34 bytes)
        # P2TR: 68 (34 bytes)
        return not (len(scriptpubkey) < 4 or len(scriptpubkey) > 200)
Attributes
blockheight: int | None = None class-attribute instance-attribute
scriptpubkey: str | None = None class-attribute instance-attribute
txid: str instance-attribute
vout: int instance-attribute
Functions
__post_init__() -> None

Strict validation of UTXO fields to match legacy protocol.

Source code in jmcore/src/jmcore/protocol.py 
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
def __post_init__(self) -> None:
    """Strict validation of UTXO fields to match legacy protocol."""
    # TXID validation
    if len(self.txid) != 64:
        raise ValueError(f"Invalid TXID length: {len(self.txid)} (expected 64)")
    try:
        binascii.unhexlify(self.txid)
    except (binascii.Error, TypeError) as exc:
        raise ValueError(f"Invalid TXID hex: {self.txid}") from exc

    # Vout validation (Bitcoin uint32)
    if self.vout < 0:
        raise ValueError(f"Invalid vout (must be non-negative): {self.vout}")
    if self.vout > 0xFFFFFFFF:
        raise ValueError(f"Invalid vout (overflow, max 4294967295): {self.vout}")

    # Optional blockheight validation
    if self.blockheight is not None and self.blockheight < 0:
        raise ValueError(f"Invalid blockheight (must be non-negative): {self.blockheight}")
from_str(s: str) -> UTXOMetadata classmethod

Parse UTXO string in either legacy or extended format.

Legacy format: txid:vout Extended format: txid:vout:scriptpubkey:blockheight

Source code in jmcore/src/jmcore/protocol.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
@classmethod
def from_str(cls, s: str) -> UTXOMetadata:
    """
    Parse UTXO string in either legacy or extended format.

    Legacy format: txid:vout
    Extended format: txid:vout:scriptpubkey:blockheight
    """
    parts = s.split(":")
    if len(parts) not in [2, 4]:
        raise ValueError(f"Invalid UTXO format: {s}")

    txid = parts[0]
    try:
        vout = int(parts[1])
    except (ValueError, TypeError) as exc:
        raise ValueError(f"Invalid vout (not an integer): {parts[1]}") from exc

    if len(parts) == 2:
        return cls(txid=txid, vout=vout)

    # Extended format
    scriptpubkey = parts[2]
    try:
        blockheight = int(parts[3])
    except (ValueError, TypeError) as exc:
        raise ValueError(f"Invalid blockheight (not an integer): {parts[3]}") from exc

    return cls(
        txid=txid,
        vout=vout,
        scriptpubkey=scriptpubkey,
        blockheight=blockheight,
    )
has_neutrino_metadata() -> bool

Check if this UTXO has the metadata needed for Neutrino verification.

Source code in jmcore/src/jmcore/protocol.py 
321
322
323
def has_neutrino_metadata(self) -> bool:
    """Check if this UTXO has the metadata needed for Neutrino verification."""
    return self.scriptpubkey is not None and self.blockheight is not None
is_valid_scriptpubkey(scriptpubkey: str) -> bool staticmethod

Validate scriptPubKey format (hex string).

Source code in jmcore/src/jmcore/protocol.py 
325
326
327
328
329
330
331
332
333
334
335
336
337
@staticmethod
def is_valid_scriptpubkey(scriptpubkey: str) -> bool:
    """Validate scriptPubKey format (hex string)."""
    if not scriptpubkey:
        return False
    # Must be valid hex
    if not re.match(r"^[0-9a-fA-F]+$", scriptpubkey):
        return False
    # Common scriptPubKey lengths (in hex chars):
    # P2PKH: 50 (25 bytes), P2SH: 46 (23 bytes)
    # P2WPKH: 44 (22 bytes), P2WSH: 68 (34 bytes)
    # P2TR: 68 (34 bytes)
    return not (len(scriptpubkey) < 4 or len(scriptpubkey) > 200)
to_extended_str() -> str

Format as extended string: txid:vout:scriptpubkey:blockheight

Source code in jmcore/src/jmcore/protocol.py 
280
281
282
283
284
def to_extended_str(self) -> str:
    """Format as extended string: txid:vout:scriptpubkey:blockheight"""
    if self.scriptpubkey is None or self.blockheight is None:
        return self.to_legacy_str()
    return f"{self.txid}:{self.vout}:{self.scriptpubkey}:{self.blockheight}"
to_legacy_str() -> str

Format as legacy string: txid:vout

Source code in jmcore/src/jmcore/protocol.py 
276
277
278
def to_legacy_str(self) -> str:
    """Format as legacy string: txid:vout"""
    return f"{self.txid}:{self.vout}"

Functions

create_handshake_request(nick: str, location: str, network: str, directory: bool = False, neutrino_compat: bool = False, features: FeatureSet | None = None) -> dict[str, Any]

Create a handshake request message.

Args: nick: Bot nickname location: Onion address or NOT-SERVING-ONION network: Bitcoin network (mainnet, testnet, signet, regtest) directory: True if this is a directory server neutrino_compat: True to advertise Neutrino-compatible UTXO metadata support features: FeatureSet to advertise (overrides neutrino_compat if provided)

Returns: Handshake request payload dict

Source code in jmcore/src/jmcore/protocol.py 
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
def create_handshake_request(
    nick: str,
    location: str,
    network: str,
    directory: bool = False,
    neutrino_compat: bool = False,
    features: FeatureSet | None = None,
) -> dict[str, Any]:
    """
    Create a handshake request message.

    Args:
        nick: Bot nickname
        location: Onion address or NOT-SERVING-ONION
        network: Bitcoin network (mainnet, testnet, signet, regtest)
        directory: True if this is a directory server
        neutrino_compat: True to advertise Neutrino-compatible UTXO metadata support
        features: FeatureSet to advertise (overrides neutrino_compat if provided)

    Returns:
        Handshake request payload dict
    """
    if features is not None:
        features_dict = features.to_dict()
    else:
        features_dict = {}
        if neutrino_compat:
            features_dict[FEATURE_NEUTRINO_COMPAT] = True

    return {
        "app-name": "joinmarket",
        "directory": directory,
        "location-string": location,
        "proto-ver": JM_VERSION,
        "features": features_dict,
        "nick": nick,
        "network": network,
    }

create_handshake_response(nick: str, network: str, accepted: bool = True, motd: str = 'JoinMarket Directory Server', neutrino_compat: bool = False, features: FeatureSet | None = None) -> dict[str, Any]

Create a handshake response message.

Args: nick: Directory server nickname network: Bitcoin network accepted: Whether the connection is accepted motd: Message of the day neutrino_compat: True to advertise Neutrino-compatible UTXO metadata support features: FeatureSet to advertise (overrides neutrino_compat if provided)

Returns: Handshake response payload dict

Source code in jmcore/src/jmcore/protocol.py 
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
def create_handshake_response(
    nick: str,
    network: str,
    accepted: bool = True,
    motd: str = "JoinMarket Directory Server",
    neutrino_compat: bool = False,
    features: FeatureSet | None = None,
) -> dict[str, Any]:
    """
    Create a handshake response message.

    Args:
        nick: Directory server nickname
        network: Bitcoin network
        accepted: Whether the connection is accepted
        motd: Message of the day
        neutrino_compat: True to advertise Neutrino-compatible UTXO metadata support
        features: FeatureSet to advertise (overrides neutrino_compat if provided)

    Returns:
        Handshake response payload dict
    """
    if features is not None:
        features_dict = features.to_dict()
    else:
        features_dict = {}
        if neutrino_compat:
            features_dict[FEATURE_NEUTRINO_COMPAT] = True

    return {
        "app-name": "joinmarket",
        "directory": True,
        "proto-ver-min": JM_VERSION,
        "proto-ver-max": JM_VERSION,
        "features": features_dict,
        "accepted": accepted,
        "nick": nick,
        "network": network,
        "motd": motd,
    }

create_peerlist_entry(nick: str, location: str, disconnected: bool = False, features: FeatureSet | None = None) -> str

Create a peerlist entry string.

Format: - Legacy: nick;location or nick;location;D - Extended: nick;location;F:feature1,feature2 or nick;location;D;F:feature1,feature2

The F: prefix is used to identify the features field and maintain backward compatibility.

Source code in jmcore/src/jmcore/protocol.py 
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
def create_peerlist_entry(
    nick: str,
    location: str,
    disconnected: bool = False,
    features: FeatureSet | None = None,
) -> str:
    """
    Create a peerlist entry string.

    Format:
    - Legacy: nick;location or nick;location;D
    - Extended: nick;location;F:feature1,feature2 or nick;location;D;F:feature1,feature2

    The F: prefix is used to identify the features field and maintain backward compatibility.
    """
    entry = f"{nick}{NICK_PEERLOCATOR_SEPARATOR}{location}"
    if disconnected:
        entry += f"{NICK_PEERLOCATOR_SEPARATOR}D"
    if features and features.features:
        entry += f"{NICK_PEERLOCATOR_SEPARATOR}F:{features.to_comma_string()}"
    return entry

format_jm_message(from_nick: str, to_nick: str, cmd: str, message: str) -> str

Source code in jmcore/src/jmcore/protocol.py 
576
577
def format_jm_message(from_nick: str, to_nick: str, cmd: str, message: str) -> str:
    return f"{from_nick}{COMMAND_PREFIX}{to_nick}{COMMAND_PREFIX}{cmd} {message}"

format_utxo_list(utxos: list[UTXOMetadata], extended: bool = False) -> str

Format a list of UTXOs as comma-separated string.

Args: utxos: List of UTXOMetadata objects extended: If True, use extended format with scriptpubkey:blockheight

Returns: Comma-separated UTXO string

Source code in jmcore/src/jmcore/protocol.py 
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
def format_utxo_list(utxos: list[UTXOMetadata], extended: bool = False) -> str:
    """
    Format a list of UTXOs as comma-separated string.

    Args:
        utxos: List of UTXOMetadata objects
        extended: If True, use extended format with scriptpubkey:blockheight

    Returns:
        Comma-separated UTXO string
    """
    if extended:
        return ",".join(u.to_extended_str() for u in utxos)
    else:
        return ",".join(u.to_legacy_str() for u in utxos)

get_nick_version(nick: str) -> int

Extract protocol version from a JoinMarket nick.

Nick format: J{version}{hash} where version is a single digit. Example: J5abc123... (v5)

Returns JM_VERSION (5) if version cannot be determined.

Source code in jmcore/src/jmcore/protocol.py 
227
228
229
230
231
232
233
234
235
236
237
238
def get_nick_version(nick: str) -> int:
    """
    Extract protocol version from a JoinMarket nick.

    Nick format: J{version}{hash} where version is a single digit.
    Example: J5abc123... (v5)

    Returns JM_VERSION (5) if version cannot be determined.
    """
    if nick and len(nick) >= 2 and nick[0] == "J" and nick[1].isdigit():
        return int(nick[1])
    return JM_VERSION

parse_jm_message(msg: str) -> tuple[str, str, str] | None

Source code in jmcore/src/jmcore/protocol.py 
580
581
582
583
584
585
586
587
588
589
590
def parse_jm_message(msg: str) -> tuple[str, str, str] | None:
    try:
        parts = msg.split(COMMAND_PREFIX)
        if len(parts) < 3:
            return None
        from_nick = parts[0]
        to_nick = parts[1]
        rest = COMMAND_PREFIX.join(parts[2:])
        return (from_nick, to_nick, rest)
    except Exception:
        return None

parse_peer_location(location: str) -> tuple[str, int]

Source code in jmcore/src/jmcore/protocol.py 
514
515
516
517
518
519
520
521
522
523
524
def parse_peer_location(location: str) -> tuple[str, int]:
    if location == NOT_SERVING_ONION_HOSTNAME:
        return (location, -1)
    try:
        host, port_str = location.split(":")
        port = int(port_str)
        if port <= 0 or port > 65535:
            raise ValueError(f"Invalid port: {port}")
        return (host, port)
    except (ValueError, AttributeError) as e:
        raise ValueError(f"Invalid location string: {location}") from e

parse_peerlist_entry(entry: str) -> tuple[str, str, bool, FeatureSet]

Parse a peerlist entry string.

Returns: Tuple of (nick, location, disconnected, features)

Source code in jmcore/src/jmcore/protocol.py 
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
def parse_peerlist_entry(entry: str) -> tuple[str, str, bool, FeatureSet]:
    """
    Parse a peerlist entry string.

    Returns:
        Tuple of (nick, location, disconnected, features)
    """
    parts = entry.split(NICK_PEERLOCATOR_SEPARATOR)
    if len(parts) < 2:
        raise ValueError(f"Invalid peerlist entry: {entry}")

    nick = parts[0]
    location = parts[1]
    disconnected = False
    features = FeatureSet()

    # Parse remaining parts
    for part in parts[2:]:
        if part == "D":
            disconnected = True
        elif part.startswith("F:"):
            features = FeatureSet.from_comma_string(part[2:])

    return (nick, location, disconnected, features)

parse_utxo_list(utxo_list_str: str, require_metadata: bool = False) -> list[UTXOMetadata]

Parse a comma-separated list of UTXOs.

Args: utxo_list_str: Comma-separated UTXOs (legacy or extended format) require_metadata: If True, raise error if any UTXO lacks Neutrino metadata

Returns: List of UTXOMetadata objects

Source code in jmcore/src/jmcore/protocol.py 
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
def parse_utxo_list(utxo_list_str: str, require_metadata: bool = False) -> list[UTXOMetadata]:
    """
    Parse a comma-separated list of UTXOs.

    Args:
        utxo_list_str: Comma-separated UTXOs (legacy or extended format)
        require_metadata: If True, raise error if any UTXO lacks Neutrino metadata

    Returns:
        List of UTXOMetadata objects
    """
    if not utxo_list_str:
        return []

    utxos = []
    for utxo_str in utxo_list_str.split(","):
        utxo = UTXOMetadata.from_str(utxo_str.strip())
        if require_metadata and not utxo.has_neutrino_metadata():
            raise ValueError(f"UTXO {utxo.to_legacy_str()} missing Neutrino metadata")
        utxos.append(utxo)
    return utxos

peer_supports_neutrino_compat(handshake_data: dict[str, Any]) -> bool

Check if a peer supports Neutrino-compatible UTXO metadata.

Args: handshake_data: Handshake payload from peer

Returns: True if peer advertises neutrino_compat feature

Source code in jmcore/src/jmcore/protocol.py 
500
501
502
503
504
505
506
507
508
509
510
511
def peer_supports_neutrino_compat(handshake_data: dict[str, Any]) -> bool:
    """
    Check if a peer supports Neutrino-compatible UTXO metadata.

    Args:
        handshake_data: Handshake payload from peer

    Returns:
        True if peer advertises neutrino_compat feature
    """
    features = handshake_data.get("features", {})
    return features.get(FEATURE_NEUTRINO_COMPAT, False)