32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
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
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
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
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
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403 | class DirectConnectionMixin:
"""Mixin class providing direct connection handling methods for MakerBot.
These methods handle incoming connections from takers via the hidden service,
including message parsing, handshake protocol, and message routing.
"""
# -- Attributes provided by MakerBot --
running: bool
config: MakerConfig
backend: BlockchainBackend
nick: str
current_offers: list[Offer]
directory_clients: dict[str, DirectoryClient]
direct_connections: dict[str, TCPConnection]
_direct_connection_rate_limiter: DirectConnectionRateLimiter
def _parse_direct_message(self, data: bytes) -> tuple[str, str, str] | None:
"""Parse a direct connection message supporting both formats.
The reference implementation uses OnionCustomMessage format:
{"type": 685, "line": "from_nick!to_nick!command data"}
Where type 685 = PRIVMSG, type 687 = PUBMSG.
Our internal format (for future use):
{"nick": "sender", "cmd": "command", "data": "..."}
Returns:
(sender_nick, command, message_data) tuple or None if parsing fails.
For PUBMSG (orderbook), returns (sender_nick, "PUBLIC:orderbook", "").
"""
try:
message = json.loads(data.decode("utf-8"))
except json.JSONDecodeError:
return None
# Check for reference implementation format: {"type": int, "line": str}
if "type" in message and "line" in message:
msg_type = message.get("type")
line = message.get("line", "")
# Handle PUBMSG (687) - typically orderbook requests
if msg_type == MessageType.PUBMSG.value:
# Parse line format: from_nick!PUBLIC!command
parts = line.split(COMMAND_PREFIX)
if len(parts) < 3:
logger.debug(f"Invalid PUBMSG line format: {line[:50]}...")
return None
sender_nick = parts[0]
to_nick = parts[1]
rest = COMMAND_PREFIX.join(parts[2:]).strip().lstrip("!")
if to_nick == "PUBLIC":
# Return special marker for public messages
logger.trace(
f"Received PUBMSG from {sender_nick} via direct connection: {rest}"
)
return (sender_nick, f"PUBLIC:{rest}", "")
else:
logger.debug(f"Ignoring PUBMSG with non-PUBLIC target: {to_nick}")
return None
# Handle PRIVMSG (685) for CoinJoin protocol
if msg_type != MessageType.PRIVMSG.value:
logger.debug(f"Ignoring message type {msg_type} on direct connection")
return None
# Parse line format: from_nick!to_nick!command data
parts = line.split(COMMAND_PREFIX)
if len(parts) < 3:
logger.warning(f"Invalid line format: {line[:50]}...")
return None
sender_nick = parts[0]
to_nick = parts[1]
rest = COMMAND_PREFIX.join(parts[2:])
# Check if message is for us
if to_nick != self.nick:
logger.debug(f"Ignoring message not for us: to={to_nick}, us={self.nick}")
return None
# Strip leading "!" and parse command
rest = rest.strip().lstrip("!")
# Extract command and data
cmd_parts = rest.split(" ", 1)
cmd = cmd_parts[0]
msg_data = cmd_parts[1] if len(cmd_parts) > 1 else ""
return (sender_nick, cmd, msg_data)
# Check for our internal format: {"nick": str, "cmd": str, "data": str}
elif "nick" in message or "cmd" in message:
sender_nick = message.get("nick", "unknown")
cmd = message.get("cmd", "")
msg_data = message.get("data", "")
return (sender_nick, cmd, msg_data)
return None
async def _try_handle_handshake(
self, connection: TCPConnection, data: bytes, peer_str: str
) -> bool:
"""Try to handle a handshake request on a direct connection.
In the reference implementation, when a non-directory peer (maker) receives
a HANDSHAKE (793) from a connecting peer (taker), it responds with its own
HANDSHAKE (793) using the client handshake format -- NOT a DN_HANDSHAKE (795).
Only directory nodes respond with DN_HANDSHAKE.
Both sides send HANDSHAKE (793) to each other (symmetric handshake).
The taker sends first (on connection), the maker responds with its own.
Both sides then mark the peer as handshaked.
If the maker were to send DN_HANDSHAKE (795), the reference taker would
reject it with "Unexpected dn-handshake from non-dn node" because it only
accepts DN_HANDSHAKE from peers marked as directories.
Args:
connection: The TCP connection
data: Raw message data
peer_str: Peer identifier string for logging
Returns:
True if this was a handshake message (handled), False otherwise.
"""
try:
message = json.loads(data.decode("utf-8"))
except (json.JSONDecodeError, UnicodeDecodeError):
return False
# Check for handshake message type (793 = HANDSHAKE)
if message.get("type") != MessageType.HANDSHAKE.value:
return False
# Parse the handshake request
try:
line = message.get("line", "")
handshake_data = json.loads(line) if isinstance(line, str) else line
except json.JSONDecodeError:
logger.warning(f"Invalid handshake JSON from {peer_str}")
return True # Was a handshake message, just malformed
peer_nick = handshake_data.get("nick", "unknown")
peer_network = handshake_data.get("network", "")
# Parse peer's advertised features (supports both dict and comma-string formats)
peer_features_raw = handshake_data.get("features", "")
peer_features = FeatureSet()
if isinstance(peer_features_raw, dict):
# Reference implementation format: {"peerlist_features": True, ...}
for feature_name, enabled in peer_features_raw.items():
if enabled:
peer_features.features.add(feature_name)
elif isinstance(peer_features_raw, str) and peer_features_raw:
# Comma-separated string format: "neutrino_compat,peerlist_features"
peer_features = FeatureSet.from_comma_string(peer_features_raw)
peer_version = handshake_data.get("version", handshake_data.get("proto-ver", "unknown"))
logger.info(f"Received handshake from {peer_nick} at {peer_str}")
logger.debug(
f"Peer {peer_nick} handshake details: version={peer_version}, "
f"network={peer_network or 'unspecified'}, "
f"features={peer_features.to_comma_string() or 'none'}"
)
# Validate network
if peer_network and peer_network != self.config.network.value:
logger.warning(
f"Network mismatch from {peer_nick}: "
f"{peer_network} != {self.config.network.value}. "
f"Not responding to handshake."
)
return True
# Build our feature set for the handshake
features = FeatureSet()
if self.backend.can_provide_neutrino_metadata():
features.features.add(FEATURE_NEUTRINO_COMPAT)
features.features.add(FEATURE_PEERLIST_FEATURES)
# Determine our location string (onion address or NOT-SERVING-ONION)
onion_host = self.config.onion_host
if onion_host:
our_location = f"{onion_host}:{self.config.onion_serving_port}"
else:
our_location = "NOT-SERVING-ONION"
# Respond with HANDSHAKE (793) using client handshake format.
# In the reference implementation, both peers send HANDSHAKE (793) to each
# other -- it is a symmetric exchange. Only directories use DN_HANDSHAKE (795).
response_data = create_handshake_request(
nick=self.nick,
location=our_location,
network=self.config.network.value,
directory=False,
features=features,
)
response_msg = {
"type": MessageType.HANDSHAKE.value,
"line": json.dumps(response_data),
}
try:
await connection.send(json.dumps(response_msg).encode("utf-8"))
logger.info(
f"Sent handshake to {peer_nick} (features: {features.to_comma_string() or 'none'})"
)
except Exception as e:
logger.warning(f"Failed to send handshake to {peer_str}: {e}")
return True
async def _on_direct_connection(
self: MakerBotProtocol, connection: TCPConnection, peer_str: str
) -> None:
"""Handle incoming direct connection from a taker via hidden service.
Direct connections support three message formats:
1. Handshake request (health check / feature discovery):
{"type": 793, "line": "<json handshake data>"}
Maker responds with handshake response including features.
2. Reference implementation format (OnionCustomMessage):
{"type": 685, "line": "from_nick!to_nick!command data"}
Where type 685 = PRIVMSG.
3. Our simplified format:
{"nick": "sender", "cmd": "command", "data": "..."}
This bypasses the directory server for lower latency once the taker
knows the maker's onion address (from the peerlist).
Rate Limiting Strategy:
- Direct connections are rate limited by connection address (peer_str), not by nick
- This prevents nick rotation attacks where attackers use different nicks per request
- Attackers connecting directly to the onion bypass directory-level protections
- Connection-based limiting is stricter: faster bans, longer intervals
"""
logger.info(f"Handling direct connection from {peer_str}")
# Check if this connection is already banned
if self._direct_connection_rate_limiter.is_banned(peer_str):
logger.debug(f"Rejecting direct connection from banned address {peer_str}")
await connection.close()
return
try:
# Keep connection open and process messages
while self.running and connection.is_connected():
try:
# Receive message with timeout
data = await asyncio.wait_for(connection.receive(), timeout=60.0)
if not data:
logger.info(f"Direct connection from {peer_str} closed")
break
# Apply connection-based message rate limiting FIRST
# This catches general floods before any processing
if not self._direct_connection_rate_limiter.check_message(peer_str):
logger.debug(f"Rate limiting message from {peer_str} (message flood)")
continue
# Check for handshake request first (health check / feature discovery)
handshake_handled = await self._try_handle_handshake(connection, data, peer_str)
if handshake_handled:
# Handshake was handled, connection may close after response
# Continue to allow follow-up messages or clean disconnect
continue
# Parse the message (supports both formats)
parsed = self._parse_direct_message(data)
if parsed is None:
# Log message content for debugging
# data is bytes, decode for display (replace errors to handle binary)
data_str = (
data.decode("utf-8", errors="replace")
if isinstance(data, bytes)
else str(data)
)
# Full message at DEBUG level for troubleshooting
logger.debug(f"Unparseable direct message from {peer_str}: {data_str!r}")
# Rate-limited WARNING with truncated preview
msg_preview = data_str[:100] + "..." if len(data_str) > 100 else data_str
self._log_rate_limited(
f"direct_parse_fail:{peer_str}",
f"Failed to parse direct message from {peer_str}: {msg_preview!r}",
interval_sec=10,
)
continue
sender_nick, cmd, msg_data = parsed
logger.debug(f"Direct message from {sender_nick}: cmd={cmd}")
# Track this connection by nick for sending responses
if sender_nick and sender_nick != "unknown":
self.direct_connections[sender_nick] = connection
# Handle PUBLIC messages (orderbook requests via direct connection)
if cmd.startswith("PUBLIC:"):
public_cmd = cmd[7:] # Strip "PUBLIC:" prefix
if public_cmd == "orderbook":
# Apply CONNECTION-BASED rate limiting (not nick-based!)
# This prevents nick rotation attacks
if not self._direct_connection_rate_limiter.check_orderbook(peer_str):
violations = (
self._direct_connection_rate_limiter.get_violation_count(
peer_str
)
)
is_banned = self._direct_connection_rate_limiter.is_banned(peer_str)
if is_banned:
logger.debug(
f"Ignoring orderbook request from banned connection "
f"{peer_str} (nick: {sender_nick})"
)
# Close connection to banned peer
await connection.close()
return
else:
logger.debug(
f"Rate limiting orderbook request from {peer_str} "
f"(nick: {sender_nick}, violations: {violations})"
)
continue
logger.info(
f"Received !orderbook request from {sender_nick} via direct "
f"connection, sending offers"
)
await self._send_offers_via_direct_connection(sender_nick, connection)
else:
logger.debug(
f"Unknown PUBLIC command from {sender_nick} via direct: "
f"{public_cmd}"
)
continue
# Process the command - reuse existing handlers
# Commands: fill, auth, tx (same as via directory)
full_msg = f"{cmd} {msg_data}" if msg_data else cmd
if cmd == "fill":
await self._handle_fill(sender_nick, full_msg, source="direct")
elif cmd == "auth":
await self._handle_auth(sender_nick, full_msg, source="direct")
elif cmd == "tx":
await self._handle_tx(sender_nick, full_msg, source="direct")
elif cmd == "push":
await self._handle_push(sender_nick, full_msg, source="direct")
else:
logger.debug(f"Unknown direct command from {sender_nick}: {cmd}")
except TimeoutError:
# No message received, continue waiting
continue
except Exception as e:
logger.error(f"Error processing direct message from {peer_str}: {e}")
break
except Exception as e:
logger.error(f"Error in direct connection handler for {peer_str}: {e}")
finally:
await connection.close()
# Clean up nick -> connection mapping
for nick, conn in list(self.direct_connections.items()):
if conn == connection:
del self.direct_connections[nick]
logger.info(f"Direct connection from {peer_str} closed")
|