netget/server/dc/mod.rs
1//! DC (Direct Connect) server implementation
2pub mod actions;
3
4use crate::llm::action_helper::call_llm;
5use crate::llm::actions::protocol_trait::ActionResult;
6use crate::llm::ollama_client::OllamaClient;
7use crate::logging::emit::Log;
8use crate::protocol::Event;
9use crate::server::connection::ConnectionId;
10use crate::server::DcProtocol;
11use crate::state::app_state::AppState;
12use actions::DC_COMMAND_RECEIVED_EVENT;
13use anyhow::Result;
14use std::net::SocketAddr;
15use std::sync::Arc;
16use std::time::Duration;
17use tokio::io::{AsyncReadExt, AsyncWriteExt};
18use tokio::sync::mpsc;
19
20/// The `Pk=` value inside the `$Lock` challenge, and the hub name used when only
21/// `hub_topic` was given — so a topic-only configuration still names the hub it belongs to
22/// rather than presenting the topic as the name.
23const DEFAULT_HUB_NAME: &str = "NetGetHub";
24
25/// Largest single NMDC command accepted from a client, in bytes.
26///
27/// NMDC frames by `|`, and the accumulator grew until one arrived — with no cap and no timeout,
28/// an **unauthenticated** peer that connects and never sends a `|` grows it as fast as its link
29/// allows until the process dies. That is the whole attack: `nc host 411 < /dev/zero`. Real
30/// commands are tens of bytes; `$MyINFO` with a long description is the largest and stays well
31/// under a kilobyte, so 64 KiB is generous and still finite.
32pub const MAX_COMMAND_LEN: usize = 64 * 1024;
33
34/// How long to wait for the first byte from a client that has only connected.
35///
36/// NMDC is hub-speaks-first: the hub sends `$Lock` and the client answers with `$Key` and
37/// `$ValidateNick` from inside its own connect path, with no human in the loop. A minute is far
38/// longer than that takes. The `$Lock` itself is written before the read loop begins, so the
39/// model's time over it — and a `manual` rule parking it for a human — is outside this bound by
40/// construction.
41const FIRST_BYTE_READ_TIMEOUT: Duration = Duration::from_secs(60);
42
43/// How long to wait for further bytes once the client has sent a complete command.
44///
45/// Five minutes. An NMDC client in a hub is never silent for long: a bare `|` is the protocol's
46/// keepalive and DC++ sends one about once a minute when it has nothing else to say, so five
47/// minutes is five missed keepalives — a link that is gone, not a user who is quiet. Making it
48/// longer would not help a real client and would only widen the window an unauthenticated peer
49/// can hold.
50///
51/// The same bound covers a peer that has begun a command and stalled before its `|`, which is
52/// the other way this loop can be held: [`MAX_COMMAND_LEN`] bounds how much such a peer can
53/// make the hub buffer, and this bounds how long it can sit there having sent less.
54///
55/// The LLM round-trip and a `manual` rule parking a command for a human
56/// (`src/state/intercepts.rs`, 300s by default) both happen after a complete command has been
57/// read, so neither can be timed out from under itself.
58const IDLE_BETWEEN_COMMANDS_TIMEOUT: Duration = Duration::from_secs(300);
59
60/// Concurrent connections this hub admits.
61///
62/// [`crate::server::accept_bounded::DEFAULT_MAX_CONNECTIONS`]. Each connection may buffer up to
63/// [`MAX_COMMAND_LEN`], so this is the multiplier that turns that per-connection bound into a
64/// total one.
65const MAX_CONNECTIONS: usize = crate::server::accept_bounded::DEFAULT_MAX_CONNECTIONS;
66
67/// What a peer over [`MAX_CONNECTIONS`] is told before the socket closes.
68///
69/// `$HubIsFull` is NMDC's own way for a hub to turn a client away, and it is what a client
70/// expects in place of the `$Lock` it was waiting for. A client that does not recognise it sees
71/// a well-formed, `|`-terminated command followed by a clean close rather than an unexplained
72/// disconnection. Fixed text, so there is no placeholder an internal error could reach.
73const CONNECTION_CAP_REFUSAL: &[u8] = b"$HubIsFull|";
74
75/// Strip the NMDC framing characters from a value that goes inside a command.
76///
77/// Every command this hub writes is `format!("$Something {}|", value)`, so a `|` anywhere in
78/// `value` ends the command early and everything after it is parsed by the client as a *second*
79/// command — `$Kick`, `$ForceMove`, anything. `\r` and `\n` go for the same reason. The values
80/// come from the model, and a hub that echoes a user's chat line back is putting peer-controlled
81/// text through that format string.
82///
83/// `hub_name_command` has always done this for its two startup parameters; the ten action
84/// executors did not. `send_dc_raw` is deliberately exempt — sending an arbitrary frame is its
85/// entire declared purpose.
86pub(crate) fn nmdc_field(value: &str) -> String {
87 value
88 .chars()
89 .filter(|c| *c != '|' && *c != '\r' && *c != '\n')
90 .collect()
91}
92
93/// Build the `$HubName` command from the `hub_name` / `hub_topic` startup parameters, or
94/// `None` when neither was supplied.
95///
96/// `$HubName <name> - <topic>` is how NMDC carries a topic: there is no separate topic
97/// command in the base protocol, and DC++ splits the name at the first ` - `.
98///
99/// Both values are sanitised. NMDC frames commands by `|`, so an unescaped pipe in either
100/// would end the command early and leave the remainder to be parsed as a second one; `\r`
101/// and `\n` are stripped for the same reason. Everything else is left alone — this is the
102/// operator's text, not ours to rewrite.
103fn hub_name_command(hub_name: Option<&str>, hub_topic: Option<&str>) -> Option<String> {
104 fn clean(value: &str) -> String {
105 value
106 .chars()
107 .filter(|c| *c != '|' && *c != '\r' && *c != '\n')
108 .collect::<String>()
109 .trim()
110 .to_string()
111 }
112
113 let name = hub_name.map(clean).filter(|s| !s.is_empty());
114 let topic = hub_topic.map(clean).filter(|s| !s.is_empty());
115
116 match (name, topic) {
117 (None, None) => None,
118 (Some(name), None) => Some(format!("$HubName {name}|")),
119 (name, Some(topic)) => {
120 let name = name.unwrap_or_else(|| DEFAULT_HUB_NAME.to_string());
121 Some(format!("$HubName {name} - {topic}|"))
122 }
123 }
124}
125
126/// DC server that forwards commands to LLM
127pub struct DcServer;
128
129impl DcServer {
130 /// Spawn DC server with integrated LLM actions
131 pub async fn spawn_with_llm_actions(
132 listen_addr: SocketAddr,
133 llm_client: OllamaClient,
134 app_state: Arc<AppState>,
135 status_tx: mpsc::UnboundedSender<String>,
136 server_id: crate::state::ServerId,
137 startup_params: Option<crate::protocol::StartupParams>,
138 ) -> Result<SocketAddr> {
139 // `hub_name` and `hub_topic` are declared as what this hub calls itself, and both
140 // were read by nothing: the only identity a client ever saw was the hardcoded
141 // `Pk=NetGetHub` inside the `$Lock` challenge, so setting either changed nothing an
142 // NMDC client could observe.
143 let hub_announcement = match &startup_params {
144 Some(params) => hub_name_command(
145 params.get_optional_string("hub_name")?.as_deref(),
146 params.get_optional_string("hub_topic")?.as_deref(),
147 ),
148 None => None,
149 };
150
151 let listener =
152 crate::server::socket_helpers::create_reusable_tcp_listener(listen_addr).await?;
153 let local_addr = listener.local_addr()?;
154 Log::new(Some(&status_tx)).info(format!(
155 "DC server (action-based) listening on {}",
156 local_addr
157 ));
158
159 let protocol = Arc::new(DcProtocol::new());
160
161 let task_registrar = app_state.clone();
162 let limiter = crate::server::accept_bounded::ConnectionLimiter::new(MAX_CONNECTIONS);
163 let accept_handle = tokio::spawn(async move {
164 loop {
165 match crate::server::accept_bounded::accept_bounded(
166 &listener,
167 &limiter,
168 CONNECTION_CAP_REFUSAL,
169 "DC",
170 Some(&status_tx),
171 )
172 .await
173 {
174 Ok((stream, remote_addr, permit)) => {
175 let connection_id =
176 ConnectionId::new(app_state.get_next_unified_id().await);
177 let local_addr_conn = stream.local_addr().unwrap_or(local_addr);
178 let llm_clone = llm_client.clone();
179 let state_clone = app_state.clone();
180 let status_clone = status_tx.clone();
181 let protocol_clone = protocol.clone();
182 let hub_announcement = hub_announcement.clone();
183
184 // Tracked, not detached: stop_server must abort this task too.
185 let task_owner = app_state.clone();
186 task_owner
187 .spawn_server_task(server_id, async move {
188 // Held for the life of the connection: releasing it here would
189 // cap the accept rate rather than the number of live
190 // connections.
191 let _permit = permit;
192 let (read_half, write_half) = tokio::io::split(stream);
193 let write_half_arc = Arc::new(tokio::sync::Mutex::new(write_half));
194
195 // Add connection to ServerInstance
196 use crate::state::server::{
197 ConnectionState as ServerConnectionState, ConnectionStatus,
198 ProtocolConnectionInfo,
199 };
200 let now = crate::utils::clock::Instant::now();
201 let conn_state = ServerConnectionState {
202 id: connection_id,
203 remote_addr,
204 local_addr: local_addr_conn,
205 bytes_sent: 0,
206 bytes_received: 0,
207 packets_sent: 0,
208 packets_received: 0,
209 last_activity: now,
210 status: ConnectionStatus::Active,
211 status_changed_at: now,
212 protocol_info: ProtocolConnectionInfo::empty(),
213 };
214 state_clone
215 .add_connection_to_server(server_id, conn_state)
216 .await;
217 let _ = status_clone.send("__UPDATE_UI__".to_string());
218
219 // Register the client in the protocol's own state map. This was
220 // never called from anywhere, so `DcProtocol::clients` stayed
221 // permanently empty and every accessor that goes through
222 // `get_mut` - set_nickname, set_client_info, set_operator - was a
223 // silent no-op, while get_nickname always returned None.
224 protocol_clone.add_connection(connection_id).await;
225
226 // Peer messaging: the dashboard's "message this peer" /
227 // "disconnect this peer" inject actions into THIS connection
228 // through the same executor the LLM path uses. Registered
229 // before the $Lock goes out so the operator can reach the
230 // connection from its first moment, including while a manual
231 // `*` rule parks the first command's answer.
232 let peer_rx = crate::server::peer_support::register_peer_channel(
233 &state_clone,
234 server_id,
235 connection_id.as_u32(),
236 )
237 .await;
238 crate::server::peer_support::spawn_peer_command_task(
239 peer_rx,
240 protocol_clone.clone(),
241 state_clone.clone(),
242 server_id,
243 connection_id.as_u32(),
244 write_half_arc.clone(),
245 status_clone.clone(),
246 );
247
248 Self::run_connection(
249 read_half,
250 &write_half_arc,
251 connection_id,
252 server_id,
253 &llm_clone,
254 &state_clone,
255 &status_clone,
256 &protocol_clone,
257 hub_announcement.as_deref(),
258 )
259 .await;
260
261 // Every exit path - EOF, read error, close_connection, a failed
262 // $Lock write - lands here.
263 state_clone
264 .remove_peer_handle(server_id, connection_id.as_u32())
265 .await;
266 protocol_clone.remove_connection(&connection_id).await;
267 state_clone
268 .remove_connection_from_server(server_id, connection_id)
269 .await;
270 let _ = status_clone.send("__UPDATE_UI__".to_string());
271 })
272 .await;
273 }
274 Err(e) => {
275 Log::new(Some(&status_tx))
276 .error(format!("Failed to accept DC connection: {}", e));
277 }
278 }
279 }
280 });
281
282 // Register the accept loop so stop_server can abort it and release the port.
283 task_registrar
284 .register_server_task(server_id, accept_handle)
285 .await;
286
287 Ok(local_addr)
288 }
289
290 /// Send the $Lock challenge, then read pipe-delimited commands until the peer goes
291 /// away or an action closes the connection. Returns on every exit so the caller runs
292 /// the one cleanup path.
293 #[allow(clippy::too_many_arguments)]
294 async fn run_connection<R, W>(
295 read_half: R,
296 write_half_arc: &Arc<tokio::sync::Mutex<W>>,
297 connection_id: ConnectionId,
298 server_id: crate::state::ServerId,
299 llm_clone: &OllamaClient,
300 state_clone: &Arc<AppState>,
301 status_clone: &mpsc::UnboundedSender<String>,
302 protocol_clone: &Arc<DcProtocol>,
303 hub_announcement: Option<&str>,
304 ) where
305 R: tokio::io::AsyncRead + Unpin,
306 W: tokio::io::AsyncWrite + Unpin,
307 {
308 // Send initial $Lock challenge, followed by the hub's own name/topic when the
309 // operator configured one. NMDC hubs announce `$HubName` right after `$Lock`, and
310 // it is the first thing a client displays.
311 let lock_command =
312 format!("$Lock EXTENDEDPROTOCOLABCABCABCABCABCABC Pk={DEFAULT_HUB_NAME}|");
313 {
314 let mut writer = write_half_arc.lock().await;
315 if let Err(e) = writer.write_all(lock_command.as_bytes()).await {
316 Log::new(Some(status_clone)).error(format!("Failed to send initial Lock: {}", e));
317 return;
318 }
319 Log::new(Some(status_clone))
320 .debug(format!("DC sent {} bytes (Lock)", lock_command.len()));
321
322 // Update stats
323 state_clone
324 .update_connection_stats(
325 server_id,
326 connection_id,
327 Some(0),
328 Some(lock_command.len() as u64),
329 Some(0),
330 Some(1),
331 )
332 .await;
333 }
334
335 // `$HubName` from the `hub_name` / `hub_topic` startup parameters. Sent only when
336 // one of them was set, so a hub configured with neither puts exactly the bytes on
337 // the wire it always did.
338 if let Some(announcement) = hub_announcement {
339 let mut writer = write_half_arc.lock().await;
340 if let Err(e) = writer.write_all(announcement.as_bytes()).await {
341 Log::new(Some(status_clone))
342 .error(format!("Failed to send initial HubName: {}", e));
343 return;
344 }
345 Log::new(Some(status_clone)).debug(format!("DC sent HubName: {}", announcement));
346 state_clone
347 .update_connection_stats(
348 server_id,
349 connection_id,
350 Some(0),
351 Some(announcement.len() as u64),
352 Some(0),
353 Some(1),
354 )
355 .await;
356 }
357
358 // Buffered: this loop reads a byte at a time looking for `|`, and unbuffered that was
359 // one syscall per byte of every command.
360 let mut read_half = tokio::io::BufReader::new(read_half);
361 let mut buffer = Vec::new();
362
363 // Labelled so `close_connection` can leave the *connection* loop. It used to `break`
364 // out of the inner `for protocol_result in ...` only, so the model asking to hang up
365 // logged "DC closing connection N", skipped the remaining actions in that one batch,
366 // and then went straight back to reading commands. The action's own description says
367 // "Disconnect this client without sending anything further"; it did neither.
368 let mut seen_a_command = false;
369 'connection: loop {
370 let mut byte = [0u8; 1];
371 // The deadline wraps this read and nothing else. Everything that can legitimately
372 // take minutes — the LLM round-trip, and a `manual` rule parking the command for a
373 // human to answer — happens below, after a whole `|`-terminated command has been
374 // read.
375 let read_timeout = if seen_a_command {
376 IDLE_BETWEEN_COMMANDS_TIMEOUT
377 } else {
378 FIRST_BYTE_READ_TIMEOUT
379 };
380 let read =
381 match tokio::time::timeout(read_timeout, read_half.read_exact(&mut byte)).await {
382 Ok(read) => read,
383 Err(_) => {
384 Log::new(Some(status_clone)).info(format!(
385 "DC connection {} sent nothing for {}s; closing idle connection",
386 connection_id,
387 read_timeout.as_secs()
388 ));
389 break;
390 }
391 };
392 match read {
393 Ok(_) => {
394 buffer.push(byte[0]);
395
396 // A client that never sends the `|` terminator must not be able to grow
397 // this buffer without limit; see MAX_COMMAND_LEN.
398 if buffer.len() > MAX_COMMAND_LEN {
399 Log::new(Some(status_clone)).warn(format!(
400 "DC connection {} sent {} bytes with no '|' terminator (limit {}); \
401 closing: decision=oversize_command",
402 connection_id,
403 buffer.len(),
404 MAX_COMMAND_LEN
405 ));
406 break;
407 }
408
409 // Check for pipe delimiter
410 if byte[0] == b'|' {
411 // We have a complete command
412 seen_a_command = true;
413 let command_bytes = buffer.clone();
414 buffer.clear();
415
416 // Convert to string
417 let command_str = match String::from_utf8(command_bytes.clone()) {
418 Ok(s) => s,
419 Err(e) => {
420 Log::new(Some(status_clone))
421 .warn(format!("DC received non-UTF8 data: {}", e));
422 continue;
423 }
424 };
425
426 // Remove trailing pipe
427 let command = command_str.trim_end_matches('|');
428
429 // Byte-count summary and full command are FileOnly.
430 let preview = crate::utils::truncate_for_log(command, 100);
431 let log = Log::new(Some(status_clone));
432 log.debug(format!(
433 "DC received {} bytes on connection {}: {}",
434 command_bytes.len(),
435 connection_id,
436 preview
437 ));
438 log.trace(format!("DC command: {:?}", command));
439
440 // Update receive stats
441 state_clone
442 .update_connection_stats(
443 server_id,
444 connection_id,
445 Some(command_bytes.len() as u64),
446 Some(0),
447 Some(1),
448 Some(0),
449 )
450 .await;
451
452 // Record the nickname the client just claimed.
453 //
454 // Nothing ever called `set_nickname`, so
455 // `client_nickname` below was null on every event
456 // this hub has ever raised - the field is
457 // documented as carrying the nickname "if set",
458 // and it never was. NMDC announces it in exactly
459 // two places: `$ValidateNick <nick>` at login, and
460 // the second field of `$MyINFO $ALL <nick> ...`.
461 if let Some(rest) = command.strip_prefix("$ValidateNick ") {
462 if let Some(nick) = rest.split_whitespace().next() {
463 protocol_clone
464 .set_nickname(connection_id, nick.to_string())
465 .await;
466 }
467 } else if let Some(rest) = command.strip_prefix("$MyINFO ") {
468 let mut fields = rest.split_whitespace();
469 if fields.next() == Some("$ALL") {
470 if let Some(nick) = fields.next() {
471 protocol_clone
472 .set_nickname(connection_id, nick.to_string())
473 .await;
474 }
475 }
476 }
477
478 // Parse command type
479 let command_type = if command.starts_with('$') {
480 command
481 .split_whitespace()
482 .next()
483 .unwrap_or("$Unknown")
484 .trim_start_matches('$')
485 } else if command.starts_with('<') {
486 "Chat"
487 } else {
488 "Unknown"
489 };
490
491 // Get client nickname if available
492 let client_nickname = protocol_clone.get_nickname(&connection_id).await;
493
494 let event = Event::new(
495 &DC_COMMAND_RECEIVED_EVENT,
496 serde_json::json!({
497 "command": command,
498 "command_type": command_type,
499 "client_nickname": client_nickname,
500 }),
501 );
502
503 Log::new(Some(status_clone))
504 .debug(format!("DC calling LLM for connection {}", connection_id));
505
506 let result = call_llm(
507 llm_clone,
508 state_clone,
509 server_id,
510 Some(connection_id),
511 &event,
512 protocol_clone.as_ref(),
513 )
514 .await;
515
516 match result {
517 Ok(execution_result) => {
518 let log = Log::new(Some(status_clone));
519 for message in &execution_result.messages {
520 log.info(format!("{}", message));
521 }
522
523 log.debug(format!(
524 "DC got {} protocol results",
525 execution_result.protocol_results.len()
526 ));
527
528 // Answering with nothing is a real answer in NMDC - plenty
529 // of client commands ($Version, $MyINFO) want no reply - so
530 // it is tagged, not treated as a failure. The tag is what
531 // separates it in the log from the backend erroring below.
532 if execution_result.protocol_results.is_empty() {
533 log.debug(format!(
534 "DC connection {} decision=model_no_actions",
535 connection_id
536 ));
537 }
538
539 for protocol_result in execution_result.protocol_results {
540 match protocol_result {
541 ActionResult::Output(data) => {
542 let mut write = write_half_arc.lock().await;
543 if let Err(e) = write.write_all(&data).await {
544 // The socket is gone; reading on is pointless
545 // and every further LLM call would be spent on
546 // answers nobody can receive.
547 log.error(format!(
548 "Failed to write DC response: {}",
549 e
550 ));
551 break 'connection;
552 }
553 let _ = write.flush().await;
554 drop(write);
555 state_clone
556 .update_connection_stats(
557 server_id,
558 connection_id,
559 None,
560 Some(data.len() as u64),
561 None,
562 Some(1),
563 )
564 .await;
565
566 log.debug(format!(
567 "DC sent {} bytes on connection {}",
568 data.len(),
569 connection_id
570 ));
571 log.trace(format!(
572 "DC sent data: {:?}",
573 String::from_utf8_lossy(&data)
574 ));
575 }
576 ActionResult::CloseConnection => {
577 log.info(format!(
578 "DC closing connection {}: decision=model_close",
579 connection_id
580 ));
581 break 'connection;
582 }
583 _ => {}
584 }
585 }
586 }
587 Err(e) => {
588 let log = Log::new(Some(status_clone));
589 let failure = crate::utils::WireFailure::classify(&e);
590 let class = if failure.is_overloaded() {
591 "overloaded"
592 } else {
593 "unavailable"
594 };
595 // The whole error - backend URL, model name, anyhow chain -
596 // goes to the log and the status stream, where an operator
597 // looks. Nothing derived from it reaches the peer.
598 log.warn(format!(
599 "DC command failed on connection {}: decision=fail_closed_llm_error class={} error={}",
600 connection_id, class, e
601 ));
602
603 // Silence would leave a half-finished NMDC handshake
604 // hanging until the client's own timeout. The hub answers
605 // with a category instead, in two distinct NMDC shapes so
606 // the client can tell a transient overload (chat notice,
607 // session continues, retry is sensible) from a hard
608 // failure ($Error, the protocol's own error command).
609 let frame = match failure {
610 crate::utils::WireFailure::Overloaded => {
611 format!("<Hub> {}|", failure.prefixed_text())
612 }
613 crate::utils::WireFailure::Unavailable => {
614 format!("$Error {}|", failure.prefixed_text())
615 }
616 };
617 let bytes = frame.into_bytes();
618 let mut write = write_half_arc.lock().await;
619 if let Err(werr) = write.write_all(&bytes).await {
620 log.error(format!(
621 "Failed to write DC failure notice on connection {}: {}",
622 connection_id, werr
623 ));
624 } else {
625 let _ = write.flush().await;
626 drop(write);
627 state_clone
628 .update_connection_stats(
629 server_id,
630 connection_id,
631 None,
632 Some(bytes.len() as u64),
633 None,
634 Some(1),
635 )
636 .await;
637 }
638 }
639 }
640 }
641 }
642 Err(e) => {
643 if e.kind() == std::io::ErrorKind::UnexpectedEof {
644 Log::new(Some(status_clone))
645 .debug(format!("DC connection {} closed by client", connection_id));
646 } else {
647 Log::new(Some(status_clone)).error(format!(
648 "DC read error on connection {}: {}",
649 connection_id, e
650 ));
651 }
652 break;
653 }
654 }
655 }
656 }
657}