Skip to main content

netget/client/tor/
mod.rs

1//! Tor client implementation using arti
2pub mod actions;
3
4pub use actions::TorClientProtocol;
5
6use anyhow::{Context, Result};
7use arti_client::{TorClient as ArtiClient, TorClientConfig};
8use std::net::SocketAddr;
9use std::sync::Arc;
10use tokio::io::{AsyncReadExt, AsyncWriteExt};
11use tokio::sync::{mpsc, Mutex};
12use tracing::{error, info, trace, warn};
13
14#[cfg(unix)]
15use serde::Serialize;
16#[cfg(unix)]
17use tor_netdir::{NetDir, Relay};
18
19use crate::client::llm_budget::call_llm_for_client;
20use crate::client::tor::actions::{TOR_CLIENT_CONNECTED_EVENT, TOR_CLIENT_DATA_RECEIVED_EVENT};
21use crate::llm::ollama_client::OllamaClient;
22use crate::llm::ClientLlmResult;
23use crate::protocol::Event;
24use crate::state::app_state::AppState;
25use crate::state::{ClientId, ClientStatus};
26
27/// Connection state for LLM processing
28#[derive(Debug, Clone, PartialEq)]
29enum ConnectionState {
30    Idle,
31    Processing,
32    Accumulating,
33}
34
35/// Per-client data for LLM handling
36struct ClientData {
37    state: ConnectionState,
38    queued_data: Vec<u8>,
39    memory: String,
40}
41
42/// Relay filter criteria for directory queries
43#[cfg(unix)]
44#[derive(Debug, Clone, Default)]
45pub struct RelayFilter {
46    pub flags: Option<Vec<String>>,
47    pub min_bandwidth: Option<u64>,
48    pub nickname_pattern: Option<String>,
49    pub limit: Option<usize>,
50}
51
52#[cfg(unix)]
53impl RelayFilter {
54    /// Check if a relay matches this filter
55    fn matches(&self, _relay: &Relay<'_>) -> bool {
56        // NOTE: tor-netdir's Relay type doesn't expose flag-checking methods in the public API
57        // Even with experimental-api, the internal RouterStatus fields are not accessible
58        // For now, we match all relays - filtering would require deeper Arti API integration
59
60        // TODO: When Arti exposes flag/nickname access in experimental-api, implement:
61        // - Flag filtering (Guard, Exit, Fast, Stable, Running, Valid)
62        // - Nickname pattern matching
63        // - Bandwidth filtering
64
65        true
66    }
67}
68
69/// Simplified relay information for LLM
70#[cfg(unix)]
71#[derive(Debug, Clone, Serialize)]
72pub struct RelayInfo {
73    pub nickname: String,
74    pub fingerprint: String,
75    pub flags: Vec<String>,
76    pub is_guard: bool,
77    pub is_exit: bool,
78    pub is_fast: bool,
79    pub is_stable: bool,
80    pub is_running: bool,
81    pub is_valid: bool,
82}
83
84#[cfg(unix)]
85impl RelayInfo {
86    /// Create RelayInfo from a Tor relay
87    fn from_relay(relay: &Relay<'_>) -> Self {
88        // Get relay identity - rsa_id() is available via public trait
89        let fingerprint = format!("{:?}", relay.rsa_id());
90
91        // Use first 8 chars of fingerprint as a proxy for nickname
92        // NOTE: tor-netdir's Relay type doesn't expose nickname or flags in public API
93        let nickname = fingerprint.chars().take(8).collect::<String>();
94
95        // Return minimal info - flag details not accessible via current Arti API
96        // TODO: Update when Arti's experimental-api exposes RouterStatus fields
97        Self {
98            nickname,
99            fingerprint,
100            flags: vec![], // Not accessible via public API
101            is_guard: false,
102            is_exit: false,
103            is_fast: false,
104            is_stable: false,
105            is_running: false,
106            is_valid: false,
107        }
108    }
109}
110
111/// Startup parameter that opts a Tor client in to the public Tor network.
112///
113/// Named as a constant because it appears in four places that must agree: the parameter
114/// declaration, the refusal message, this module's check, and the test that pins all of it.
115pub const ALLOW_PUBLIC_TOR_NETWORK_PARAM: &str = "allow_public_tor_network";
116
117/// Where a Tor client is allowed to bootstrap from.
118#[derive(Debug, Clone, PartialEq, Eq)]
119pub enum BootstrapTarget {
120    /// A directory the caller named explicitly — normally a local `tor_relay`.
121    CustomDirectory(String),
122    /// The real Tor directory authorities, on the public internet. Opt-in only.
123    PublicNetwork,
124}
125
126/// Decide what `connect()` may bootstrap against, and refuse rather than reach the internet.
127///
128/// # Why this exists
129///
130/// `arti_client::TorClient::create_bootstrapped()` contacts the real Tor directory authorities
131/// **before it ever looks at the requested address** — measured at 14 seconds in an unguarded
132/// run. So merely *opening* a Tor client made outbound connections to third parties, whatever
133/// the user asked it to connect to. That is surprising for a tool that binds loopback
134/// everywhere else, and it made the client impossible to exercise offline: the whole-registry
135/// smoke test had to carve out a named exclusion for Tor because running it would have put a
136/// smoke test on the public internet.
137///
138/// # Why refuse rather than make it lazy
139///
140/// Deferring the bootstrap until a request "genuinely needs the network" buys nothing here:
141/// `connect()` is *given* the destination, so the need is immediate and the bootstrap would
142/// happen a few milliseconds later anyway — with the failure now surfacing somewhere with
143/// worse reporting than `connect()`'s `Err`. Deferral would hide the reach, not prevent it.
144/// The property worth having is that the reach is a **decision the caller made**, and the only
145/// way to express that is an explicit opt-in.
146///
147/// # The rule
148///
149/// * `directory_server` set → the caller named the directory to bootstrap from (normally a
150///   local `tor_relay`). That is already an explicit choice, so it is allowed as-is.
151/// * `allow_public_tor_network: true` → the caller explicitly asked for the real network.
152/// * Neither → `Err`, naming the parameter. **This is the default.**
153///
154/// Setting both is a contradiction and is refused rather than silently resolved: which one the
155/// caller meant is not inferable, and guessing is how a "local test" ends up on the public
156/// internet.
157pub fn bootstrap_target(
158    directory_server: Option<&str>,
159    allow_public_tor_network: bool,
160) -> Result<BootstrapTarget> {
161    match (directory_server, allow_public_tor_network) {
162        (Some(dir), false) => Ok(BootstrapTarget::CustomDirectory(dir.to_string())),
163        (None, true) => Ok(BootstrapTarget::PublicNetwork),
164        (Some(dir), true) => Err(anyhow::anyhow!(
165            "Tor client: `directory_server` ({dir}) and `{ALLOW_PUBLIC_TOR_NETWORK_PARAM}` are \
166             mutually exclusive — the first bootstraps from the directory you named, the second \
167             from the public Tor directory authorities. Pass exactly one."
168        )),
169        (None, false) => Err(anyhow::anyhow!(
170            "Tor client refused to start: bootstrapping contacts the real Tor directory \
171             authorities on the public internet, and does so before it looks at the address you \
172             asked for — so opening this client would make outbound connections to third \
173             parties regardless of the destination. NetGet does not do that by default. Either \
174             pass `directory_server` (e.g. \"127.0.0.1:9001\", a local `tor_relay`) to bootstrap \
175             locally, or pass `{ALLOW_PUBLIC_TOR_NETWORK_PARAM}: true` to opt in to the public \
176             Tor network."
177        )),
178    }
179}
180
181/// The write half of a Tor circuit, once one exists.
182type TorWriteHalf = Arc<Mutex<tokio::io::WriteHalf<arti_client::DataStream>>>;
183
184/// Run one directory verb. Needs no circuit — the consensus lives in `AppState`, put there
185/// when `create_bootstrapped` returned.
186#[cfg(unix)]
187async fn run_directory_action(
188    name: &str,
189    data: &serde_json::Value,
190    client_id: ClientId,
191    app_state: &Arc<AppState>,
192    status_tx: &mpsc::UnboundedSender<String>,
193) {
194    match name {
195        "get_consensus_info" => match TorClient::get_consensus_info(app_state, client_id).await {
196            Ok(info) => {
197                let _ = status_tx.send(format!(
198                    "[TOR] Consensus: {} relays, valid until {}",
199                    info["relay_count"], info["valid_until"]
200                ));
201                trace!("Consensus info: {}", info);
202            }
203            Err(e) => {
204                error!("Failed to get consensus info: {}", e);
205                let _ = status_tx.send(format!("[TOR] Error: {}", e));
206            }
207        },
208        "list_relays" | "search_relays" => {
209            let limit = data.get("limit").and_then(|v| v.as_u64()).unwrap_or(100) as usize;
210            let filter = RelayFilter {
211                flags: data.get("flags").and_then(|v| v.as_array()).map(|arr| {
212                    arr.iter()
213                        .filter_map(|v| v.as_str().map(String::from))
214                        .collect()
215                }),
216                nickname_pattern: data
217                    .get("nickname")
218                    .and_then(|v| v.as_str())
219                    .map(String::from),
220                limit: Some(limit),
221                ..Default::default()
222            };
223            match TorClient::query_relays(app_state, client_id, filter).await {
224                Ok(relays) => {
225                    let _ = status_tx.send(format!("[TOR] Found {} relays", relays.len()));
226                    for (i, relay) in relays.iter().take(10).enumerate() {
227                        trace!(
228                            "Relay {}: {} ({})",
229                            i + 1,
230                            relay.nickname,
231                            relay.flags.join(", ")
232                        );
233                    }
234                }
235                Err(e) => {
236                    error!("Failed to query relays: {}", e);
237                    let _ = status_tx.send(format!("[TOR] Error: {}", e));
238                }
239            }
240        }
241        other => {
242            warn!("Unknown custom action: {}", other);
243        }
244    }
245}
246
247/// Carry out what the model answered with. Returns `true` if it asked to disconnect.
248///
249/// `write_half` is `None` at `tor_bootstrap_complete`, which fires while `connect()` is still
250/// bootstrapping and no circuit exists yet. A `send_tor_data` there is **refused loudly**
251/// rather than dropped: the directory verbs are the ones that make sense at that moment, and
252/// the model needs to be told which of its answers could not be carried out. Every other
253/// caller has a circuit and passes `Some`.
254///
255/// `pub` so `tests/client/tor/` can pin the no-circuit contract directly: a live test would
256/// need a real Tor bootstrap and a circuit, which this repo has no cheap loopback for.
257pub async fn apply_actions(
258    actions: Vec<serde_json::Value>,
259    write_half: Option<&TorWriteHalf>,
260    client_id: ClientId,
261    app_state: &Arc<AppState>,
262    status_tx: &mpsc::UnboundedSender<String>,
263) -> bool {
264    use crate::llm::actions::client_trait::{Client, ClientActionResult};
265    let protocol = crate::client::tor::actions::TorClientProtocol::new();
266
267    for action in actions {
268        match protocol.execute_action(action) {
269            Ok(ClientActionResult::SendData(bytes)) => {
270                let Some(write_half) = write_half else {
271                    warn!(
272                        "Tor client {} cannot send {} bytes: no circuit yet",
273                        client_id,
274                        bytes.len()
275                    );
276                    let _ = status_tx.send(format!(
277                        "[TOR] ✖ send_tor_data ignored for client {}: the circuit is not open yet",
278                        client_id
279                    ));
280                    continue;
281                };
282                match write_half.lock().await.write_all(&bytes).await {
283                    Ok(()) => trace!("Tor client {} sent {} bytes", client_id, bytes.len()),
284                    Err(e) => error!("Tor client {} write failed: {}", client_id, e),
285                }
286            }
287            Ok(ClientActionResult::Disconnect) => {
288                info!("Tor client {} disconnecting", client_id);
289                return true;
290            }
291            #[cfg(unix)]
292            Ok(ClientActionResult::Custom { name, data }) => {
293                run_directory_action(&name, &data, client_id, app_state, status_tx).await;
294            }
295            Ok(_) => {}
296            Err(e) => {
297                warn!("Tor client {} rejected action: {}", client_id, e);
298                let _ = status_tx.send(format!(
299                    "[TOR] ✖ client {} rejected action: {}",
300                    client_id, e
301                ));
302            }
303        }
304    }
305    false
306}
307
308/// Tor client that connects through the Tor network
309pub struct TorClient;
310
311impl TorClient {
312    /// Connect to a destination through Tor with integrated LLM actions
313    pub async fn connect_with_llm_actions(
314        remote_addr: String,
315        llm_client: OllamaClient,
316        app_state: Arc<AppState>,
317        status_tx: mpsc::UnboundedSender<String>,
318        client_id: ClientId,
319        startup_params: Option<crate::protocol::StartupParams>,
320    ) -> Result<SocketAddr> {
321        info!("Tor client {} initializing...", client_id);
322        let _ = status_tx.send(format!("[CLIENT] Tor client {} initializing...", client_id));
323
324        // Choose the directory to bootstrap from — and refuse to reach the public Tor network
325        // unless the caller asked for it. See `bootstrap_target` for the reasoning.
326        let directory_server = match startup_params {
327            Some(ref params) => params.get_optional_string("directory_server")?,
328            None => None,
329        };
330        let allow_public = match startup_params {
331            Some(ref params) => params
332                .get_optional_bool(ALLOW_PUBLIC_TOR_NETWORK_PARAM)?
333                .unwrap_or(false),
334            None => false,
335        };
336
337        let config = match bootstrap_target(directory_server.as_deref(), allow_public)? {
338            BootstrapTarget::CustomDirectory(directory_server) => {
339                info!(
340                    "Tor client {} using custom directory server: {}",
341                    client_id, directory_server
342                );
343                let _ = status_tx.send(format!(
344                    "[CLIENT] Tor client {} using custom directory: {}",
345                    client_id, directory_server
346                ));
347                Self::create_custom_config(&directory_server)?
348            }
349            BootstrapTarget::PublicNetwork => {
350                // Loud, because this is the one path that leaves the machine.
351                warn!(
352                    "Tor client {} is bootstrapping against the PUBLIC Tor directory \
353                     authorities ({}=true): this makes outbound connections to third parties \
354                     before any traffic is sent to {}",
355                    client_id, ALLOW_PUBLIC_TOR_NETWORK_PARAM, remote_addr
356                );
357                let _ = status_tx.send(format!(
358                    "[CLIENT] ⚠ Tor client {} contacting the PUBLIC Tor directory authorities \
359                     (opted in via {})",
360                    client_id, ALLOW_PUBLIC_TOR_NETWORK_PARAM
361                ));
362                TorClientConfig::default()
363            }
364        };
365
366        let tor_client = ArtiClient::create_bootstrapped(config)
367            .await
368            .context("Failed to bootstrap Tor client")?;
369
370        info!("Tor client {} bootstrapped successfully", client_id);
371        let _ = status_tx.send(format!("[CLIENT] Tor client {} bootstrapped", client_id));
372
373        // Store Tor client for directory queries (requires experimental-api feature)
374        #[cfg(unix)]
375        app_state
376            .set_tor_client(client_id, Arc::new(tor_client.clone()))
377            .await;
378
379        // Emit bootstrap complete event with consensus info
380        #[cfg(unix)]
381        if let Some(instruction) = app_state.get_instruction_for_client(client_id).await {
382            let protocol = Arc::new(crate::client::tor::actions::TorClientProtocol::new());
383
384            // Try to get consensus info (may fail if bootstrap just completed)
385            match Self::get_consensus_info(&app_state, client_id).await {
386                Ok(info) => {
387                    use crate::client::tor::actions::TOR_BOOTSTRAP_COMPLETE_EVENT;
388                    let event = Event::new(
389                        &TOR_BOOTSTRAP_COMPLETE_EVENT,
390                        serde_json::json!({
391                            "relay_count": info["relay_count"],
392                            "valid_after": info["valid_after"],
393                        }),
394                    );
395
396                    // The answer is carried out, not just logged. At this point there is no
397                    // circuit — `connect()` has not run — so `apply_actions` gets `None` and
398                    // the directory verbs (`get_consensus_info`, `list_relays`,
399                    // `search_relays`) are the ones that can do anything. That is exactly
400                    // what this event is for: it reports `relay_count` and `valid_after`, and
401                    // its whole purpose is letting the model query the consensus it just
402                    // learned about.
403                    match call_llm_for_client(
404                        &llm_client,
405                        &app_state,
406                        client_id.to_string(),
407                        &instruction,
408                        "",
409                        Some(&event),
410                        protocol.as_ref(),
411                        &status_tx,
412                    )
413                    .await
414                    {
415                        Ok(ClientLlmResult {
416                            actions,
417                            memory_updates,
418                        }) => {
419                            if let Some(mem) = memory_updates {
420                                app_state.set_memory_for_client(client_id, mem).await;
421                            }
422                            apply_actions(actions, None, client_id, &app_state, &status_tx).await;
423                        }
424                        Err(e) => {
425                            warn!("Failed to call LLM for bootstrap event: {}", e);
426                        }
427                    }
428                }
429                Err(e) => {
430                    trace!(
431                        "Could not get consensus info at bootstrap (may not be available yet): {}",
432                        e
433                    );
434                }
435            }
436        }
437
438        // Parse target address (can be hostname:port or .onion:port)
439        let target = remote_addr.clone();
440
441        // Connect through Tor
442        let stream = tor_client
443            .connect(target.as_str())
444            .await
445            .context(format!("Failed to connect to {} through Tor", target))?;
446
447        // Get a dummy local address since Tor connections don't have real local addresses
448        let local_addr = SocketAddr::from(([127, 0, 0, 1], 0));
449
450        info!(
451            "Tor client {} connected to {} through Tor network",
452            client_id, remote_addr
453        );
454
455        // Update client state
456        app_state
457            .update_client_status(client_id, ClientStatus::Connected)
458            .await;
459        let _ = status_tx.send(format!(
460            "[CLIENT] Tor client {} connected to {}",
461            client_id, remote_addr
462        ));
463        let _ = status_tx.send("__UPDATE_UI__".to_string());
464
465        // The `tor_connected` LLM call happens inside the read-loop task below, not here.
466        // Two reasons, both of which were bugs at this spot: it needs the write half to carry
467        // out what the model answers with, and it must not precede the command channel's
468        // registration — a manual `*` routing rule parks this call until a human answers, and
469        // the dashboard's `[ send ]` has to reach the client for the whole park.
470
471        // Split stream into read/write halves
472        let (mut read_half, write_half) = tokio::io::split(stream);
473        let write_half_arc = Arc::new(Mutex::new(write_half));
474
475        // Initialize client data
476        let client_data = Arc::new(Mutex::new(ClientData {
477            state: ConnectionState::Idle,
478            queued_data: Vec::new(),
479            memory: String::new(),
480        }));
481
482        // Spawn read loop
483        // Registered with AppState so stop_client can abort this task —
484        // dropping a JoinHandle only detaches it in Tokio.
485        let task_registrar = app_state.clone();
486        // Command channel: lets the dashboard inject actions into this loop
487        // via AppState::send_to_client (see client/command_support.rs).
488        let mut command_rx =
489            crate::client::command_support::register_command_channel(&app_state, client_id).await;
490
491        let task_handle = tokio::spawn(async move {
492            // `tor_connected`: ask the model what to do with the circuit, then do it. The
493            // answer used to be `Ok(_) => trace!("LLM called successfully")` — the round-trip
494            // was paid for and every action the model chose was dropped on the floor, so a
495            // client told "send an HTTP GET and analyse the response" sent nothing and then
496            // waited forever for a reply to a request it never made.
497            if let Some(instruction) = app_state.get_instruction_for_client(client_id).await {
498                let protocol = crate::client::tor::actions::TorClientProtocol::new();
499                let event = Event::new(
500                    &TOR_CLIENT_CONNECTED_EVENT,
501                    serde_json::json!({ "target": remote_addr }),
502                );
503                match call_llm_for_client(
504                    &llm_client,
505                    &app_state,
506                    client_id.to_string(),
507                    &instruction,
508                    "",
509                    Some(&event),
510                    &protocol,
511                    &status_tx,
512                )
513                .await
514                {
515                    Ok(ClientLlmResult {
516                        actions,
517                        memory_updates,
518                    }) => {
519                        if let Some(mem) = memory_updates {
520                            client_data.lock().await.memory = mem;
521                        }
522                        if apply_actions(
523                            actions,
524                            Some(&write_half_arc),
525                            client_id,
526                            &app_state,
527                            &status_tx,
528                        )
529                        .await
530                        {
531                            app_state
532                                .update_client_status(client_id, ClientStatus::Disconnected)
533                                .await;
534                            let _ = status_tx.send("__UPDATE_UI__".to_string());
535                            app_state.remove_client_handle(client_id).await;
536                            return;
537                        }
538                    }
539                    Err(e) => {
540                        warn!(
541                            "Failed to call LLM for Tor client {} connection: {}",
542                            client_id, e
543                        );
544                    }
545                }
546            }
547
548            let mut buffer = vec![0u8; 8192];
549
550            loop {
551                let read_result = tokio::select! {
552                    read = read_half.read(&mut buffer) => read,
553                    Some(cmd) = command_rx.recv() => {
554                        let disconnect = crate::client::command_support::handle_stream_client_command(
555                            &crate::client::tor::actions::TorClientProtocol,
556                            &write_half_arc,
557                            cmd,
558                            client_id,
559                            &app_state,
560                            &status_tx,
561                        )
562                        .await;
563                        if disconnect {
564                            app_state
565                                .update_client_status(client_id, ClientStatus::Disconnected)
566                                .await;
567                            let _ = status_tx.send(format!(
568                                "[CLIENT] Tor client {} disconnected (injected action)",
569                                client_id
570                            ));
571                            let _ = status_tx.send("__UPDATE_UI__".to_string());
572                            break;
573                        }
574                        continue;
575                    }
576                };
577                match read_result {
578                    Ok(0) => {
579                        info!("Tor client {} disconnected", client_id);
580                        app_state
581                            .update_client_status(client_id, ClientStatus::Disconnected)
582                            .await;
583                        let _ = status_tx
584                            .send(format!("[CLIENT] Tor client {} disconnected", client_id));
585                        let _ = status_tx.send("__UPDATE_UI__".to_string());
586                        break;
587                    }
588                    Ok(n) => {
589                        let data = buffer[..n].to_vec();
590                        trace!("Tor client {} received {} bytes", client_id, n);
591
592                        // Handle data with LLM.
593                        //
594                        // A `disconnect` used to `break` the `for action in actions` loop and
595                        // nothing else, so the model could never actually close a Tor client:
596                        // it stopped executing the rest of that one answer and went straight
597                        // back to reading. The flag carries the decision out to the read loop.
598                        let mut disconnect_requested = false;
599                        let mut client_data_lock = client_data.lock().await;
600
601                        match client_data_lock.state {
602                            ConnectionState::Idle => {
603                                // Process immediately
604                                client_data_lock.state = ConnectionState::Processing;
605                                drop(client_data_lock);
606
607                                // Call LLM
608                                if let Some(instruction) =
609                                    app_state.get_instruction_for_client(client_id).await
610                                {
611                                    let protocol = Arc::new(
612                                        crate::client::tor::actions::TorClientProtocol::new(),
613                                    );
614                                    let event = Event::new(
615                                        &TOR_CLIENT_DATA_RECEIVED_EVENT,
616                                        crate::client::tor::actions::inbound_event_fields(&data),
617                                    );
618
619                                    match call_llm_for_client(
620                                        &llm_client,
621                                        &app_state,
622                                        client_id.to_string(),
623                                        &instruction,
624                                        &client_data.lock().await.memory,
625                                        Some(&event),
626                                        protocol.as_ref(),
627                                        &status_tx,
628                                    )
629                                    .await
630                                    {
631                                        Ok(ClientLlmResult {
632                                            actions,
633                                            memory_updates,
634                                        }) => {
635                                            if let Some(mem) = memory_updates {
636                                                client_data.lock().await.memory = mem;
637                                            }
638                                            // Shared with the connected-event and
639                                            // bootstrap-complete paths, so the Tor action
640                                            // vocabulary is executed in exactly one place.
641                                            disconnect_requested = apply_actions(
642                                                actions,
643                                                Some(&write_half_arc),
644                                                client_id,
645                                                &app_state,
646                                                &status_tx,
647                                            )
648                                            .await;
649                                        }
650                                        Err(e) => {
651                                            error!("LLM error for Tor client {}: {}", client_id, e);
652                                        }
653                                    }
654                                }
655
656                                // Process queued data if any
657                                let mut client_data_lock = client_data.lock().await;
658                                if !client_data_lock.queued_data.is_empty() {
659                                    client_data_lock.queued_data.clear();
660                                }
661                                client_data_lock.state = ConnectionState::Idle;
662                            }
663                            ConnectionState::Processing => {
664                                // Queue data
665                                client_data_lock.queued_data.extend_from_slice(&data);
666                                client_data_lock.state = ConnectionState::Accumulating;
667                            }
668                            ConnectionState::Accumulating => {
669                                // Continue queuing
670                                client_data_lock.queued_data.extend_from_slice(&data);
671                            }
672                        }
673
674                        if disconnect_requested {
675                            app_state
676                                .update_client_status(client_id, ClientStatus::Disconnected)
677                                .await;
678                            let _ = status_tx.send(format!(
679                                "[CLIENT] Tor client {} disconnected (model asked)",
680                                client_id
681                            ));
682                            let _ = status_tx.send("__UPDATE_UI__".to_string());
683                            break;
684                        }
685                    }
686                    Err(e) => {
687                        error!("Tor client {} read error: {}", client_id, e);
688                        app_state
689                            .update_client_status(client_id, ClientStatus::Error(e.to_string()))
690                            .await;
691                        let _ = status_tx.send("__UPDATE_UI__".to_string());
692                        break;
693                    }
694                }
695            }
696
697            // The loop owns the only receiver; dropping the registered handle
698            // makes later send_to_client calls fail fast instead of timing out.
699            app_state.remove_client_handle(client_id).await;
700        });
701        task_registrar
702            .register_client_task(client_id, task_handle)
703            .await;
704
705        Ok(local_addr)
706    }
707
708    /// Create custom TorClientConfig pointing to a localhost Tor relay
709    ///
710    /// This allows testing with a local tor_relay server (using BEGIN_DIR for directory queries)
711    /// instead of real Tor network. The directory_server parameter should be in format "127.0.0.1:9001"
712    #[cfg(unix)]
713    fn create_custom_config(directory_server: &str) -> Result<TorClientConfig> {
714        use arti_client::config::dir::FallbackDir;
715
716        // Parse the directory server address
717        let addr: SocketAddr = directory_server.parse().context(
718            "Invalid directory_server address format. Expected 'IP:port' like '127.0.0.1:9030'",
719        )?;
720
721        // WARNING: Arti FallbackDir requires an OR (relay) port, but we only have HTTP directory
722        // This configuration will likely fail during bootstrap because Arti expects a working
723        // Tor relay at the OR port, not just an HTTP directory server.
724        //
725        // For now, we configure the OR port to point to the same address as a workaround,
726        // but this is a fundamental incompatibility: Arti needs a full Tor relay to bootstrap,
727        // not just a directory server.
728
729        let mut fallback = FallbackDir::builder();
730        fallback
731            .rsa_identity([0x42; 20].into()) // Dummy RSA identity
732            .ed_identity([0x99; 32].into()) // Dummy Ed25519 identity
733            .orports() // Arti only supports orports(), not dirports()
734            .push(addr);
735
736        // Build config with custom fallback
737        let mut bld = TorClientConfig::builder();
738        bld.tor_network().set_fallback_caches(vec![fallback]);
739
740        // Note: We don't set custom authorities here because fallback_caches is sufficient
741        // for bootstrapping. The client will use the fallback to fetch the consensus.
742
743        let config = bld.build().context("Failed to build TorClientConfig")?;
744
745        Ok(config)
746    }
747
748    /// Get network directory from Arti (requires experimental-api feature)
749    #[cfg(unix)]
750    pub async fn get_netdir(
751        app_state: &Arc<crate::state::app_state::AppState>,
752        client_id: crate::state::ClientId,
753    ) -> Result<Arc<NetDir>> {
754        let tor_client = app_state
755            .get_tor_client(client_id)
756            .await
757            .context("Tor client not found in app state")?;
758
759        // Access directory manager (requires experimental-api feature)
760        let dirmgr = tor_client.dirmgr();
761
762        // Get current network directory
763        let netdir = dirmgr
764            .netdir(tor_netdir::Timeliness::Timely)
765            .context("No network directory available - client may still be bootstrapping")?;
766
767        Ok(netdir)
768    }
769
770    /// Query relays from network directory with optional filter
771    #[cfg(unix)]
772    pub async fn query_relays(
773        app_state: &Arc<crate::state::app_state::AppState>,
774        client_id: crate::state::ClientId,
775        filter: RelayFilter,
776    ) -> Result<Vec<RelayInfo>> {
777        let netdir = Self::get_netdir(app_state, client_id).await?;
778
779        let limit = filter.limit.unwrap_or(100);
780        let mut relays = Vec::new();
781
782        for relay in netdir.relays() {
783            if filter.matches(&relay) {
784                relays.push(RelayInfo::from_relay(&relay));
785                if relays.len() >= limit {
786                    break;
787                }
788            }
789        }
790
791        Ok(relays)
792    }
793
794    /// Get consensus metadata (relay count, validity times)
795    #[cfg(unix)]
796    pub async fn get_consensus_info(
797        app_state: &Arc<crate::state::app_state::AppState>,
798        client_id: crate::state::ClientId,
799    ) -> Result<serde_json::Value> {
800        let netdir = Self::get_netdir(app_state, client_id).await?;
801
802        let relay_count = netdir.relays().count();
803        let lifetime = netdir.lifetime();
804
805        Ok(serde_json::json!({
806            "relay_count": relay_count,
807            "valid_after": format!("{:?}", lifetime.valid_after()),
808            "fresh_until": format!("{:?}", lifetime.fresh_until()),
809            "valid_until": format!("{:?}", lifetime.valid_until()),
810        }))
811    }
812}