rns_net/
rpc.rs

1//! RPC server and client for cross-process daemon communication.
2//!
3//! Implements Python `multiprocessing.connection` wire protocol:
4//! - 4-byte big-endian signed i32 length prefix + payload
5//! - HMAC-SHA256 challenge-response authentication
6//! - msgpack serialization for request/response dictionaries
7//!
8//! Server translates RPC dicts into [`QueryRequest`] events, sends
9//! them through the driver event channel, and returns RPC responses.
10
11use std::io::{self, Read, Write};
12use std::net::{IpAddr, TcpListener, TcpStream};
13#[cfg(target_os = "linux")]
14use std::os::linux::net::SocketAddrExt;
15#[cfg(target_os = "linux")]
16use std::os::unix::net::{SocketAddr, UnixListener};
17use std::sync::mpsc;
18use std::sync::{
19    atomic::{AtomicBool, Ordering},
20    Arc,
21};
22use std::thread;
23use std::time::Duration;
24
25use rns_core::msgpack::{self, Value as MsgpackValue};
26use rns_core::transport::types::InterfaceId;
27use rns_crypto::hmac::hmac_sha256;
28use rns_crypto::sha256::sha256;
29
30#[cfg(test)]
31use crate::event::TrafficDetail;
32use crate::event::{
33    BackboneInterfaceEntry, BackbonePeerStateEntry, BlackholeInfo, BurstClassStats, DrainStatus,
34    DynamicBurstStats, Event, EventSender, HookInfo, InterfaceStatsResponse, KnownDestinationEntry,
35    LifecycleState, PathTableEntry, ProviderBridgeStats, QueryRequest, QueryResponse,
36    RateTableEntry, RuntimeConfigApplyMode, RuntimeConfigEntry, RuntimeConfigError,
37    RuntimeConfigErrorCode, RuntimeConfigSource, RuntimeConfigValue, SingleInterfaceStat,
38};
39use crate::md5::hmac_md5;
40use crate::pickle::{self, PickleValue};
41
42const CHALLENGE_PREFIX: &[u8] = b"#CHALLENGE#";
43const WELCOME: &[u8] = b"#WELCOME#";
44const FAILURE: &[u8] = b"#FAILURE#";
45const CHALLENGE_LEN: usize = 40;
46const MAX_CONCURRENT_RPC_CONNECTIONS: usize = 32;
47
48#[derive(Debug, Clone, Copy, PartialEq, Eq)]
49enum RpcCodec {
50    Msgpack,
51    Pickle,
52}
53
54/// RPC address types.
55#[derive(Debug, Clone)]
56pub enum RpcAddr {
57    Tcp(String, u16),
58}
59
60/// RPC server that listens for incoming connections and handles queries.
61pub struct RpcServer {
62    shutdown: Arc<AtomicBool>,
63    thread: Option<thread::JoinHandle<()>>,
64}
65
66impl RpcServer {
67    /// Start the RPC server on the given address.
68    pub fn start(addr: &RpcAddr, auth_key: [u8; 32], event_tx: EventSender) -> io::Result<Self> {
69        let shutdown = Arc::new(AtomicBool::new(false));
70        let shutdown2 = shutdown.clone();
71
72        let listener = match addr {
73            RpcAddr::Tcp(host, port) => {
74                let l = TcpListener::bind((host.as_str(), *port))?;
75                // Non-blocking so we can check shutdown flag
76                l.set_nonblocking(true)?;
77                l
78            }
79        };
80
81        let thread = thread::Builder::new()
82            .name("rpc-server".into())
83            .spawn(move || {
84                rpc_server_loop(listener, auth_key, event_tx, shutdown2);
85            })
86            .map_err(io::Error::other)?;
87
88        Ok(RpcServer {
89            shutdown,
90            thread: Some(thread),
91        })
92    }
93
94    /// Listen on the abstract socket used by Python Reticulum on Linux.
95    #[cfg(target_os = "linux")]
96    pub fn start_abstract_unix(
97        instance_name: &str,
98        auth_key: [u8; 32],
99        event_tx: EventSender,
100    ) -> io::Result<Self> {
101        let addr = SocketAddr::from_abstract_name(format!("rns/{instance_name}/rpc"))?;
102        let listener = UnixListener::bind_addr(&addr)?;
103        listener.set_nonblocking(true)?;
104        let shutdown = Arc::new(AtomicBool::new(false));
105        let shutdown2 = shutdown.clone();
106        let thread = thread::Builder::new()
107            .name("rpc-unix-server".into())
108            .spawn(move || rpc_server_loop_unix(listener, auth_key, event_tx, shutdown2))
109            .map_err(io::Error::other)?;
110        Ok(Self {
111            shutdown,
112            thread: Some(thread),
113        })
114    }
115
116    /// Stop the RPC server.
117    pub fn stop(&mut self) {
118        self.shutdown.store(true, Ordering::Relaxed);
119        if let Some(handle) = self.thread.take() {
120            let _ = handle.join();
121        }
122    }
123}
124
125#[cfg(target_os = "linux")]
126fn rpc_server_loop_unix(
127    listener: UnixListener,
128    auth_key: [u8; 32],
129    event_tx: EventSender,
130    shutdown: Arc<AtomicBool>,
131) {
132    let mut workers = Vec::new();
133    loop {
134        reap_finished_rpc_workers(&mut workers);
135        if shutdown.load(Ordering::Relaxed) {
136            break;
137        }
138        match listener.accept() {
139            Ok((stream, _)) => {
140                if workers.len() >= MAX_CONCURRENT_RPC_CONNECTIONS {
141                    log::warn!("rejecting Unix RPC connection: {MAX_CONCURRENT_RPC_CONNECTIONS} handlers already active");
142                    continue;
143                }
144                let _ = stream.set_read_timeout(Some(Duration::from_secs(10)));
145                let _ = stream.set_write_timeout(Some(Duration::from_secs(10)));
146                let worker_event_tx = event_tx.clone();
147                match thread::Builder::new()
148                    .name("rpc-unix-client".into())
149                    .spawn(move || {
150                        if let Err(e) = handle_connection(stream, &auth_key, &worker_event_tx) {
151                            log::debug!("Unix RPC connection error: {e}");
152                        }
153                    }) {
154                    Ok(worker) => workers.push(worker),
155                    Err(e) => log::error!("failed to start Unix RPC connection handler: {e}"),
156                }
157            }
158            Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
159                thread::sleep(Duration::from_millis(10))
160            }
161            Err(e) => {
162                log::error!("Unix RPC accept error: {e}");
163                thread::sleep(Duration::from_millis(100));
164            }
165        }
166    }
167    for worker in workers {
168        let _ = worker.join();
169    }
170}
171
172impl Drop for RpcServer {
173    fn drop(&mut self) {
174        self.stop();
175    }
176}
177
178fn rpc_server_loop(
179    listener: TcpListener,
180    auth_key: [u8; 32],
181    event_tx: EventSender,
182    shutdown: Arc<AtomicBool>,
183) {
184    let mut workers = Vec::new();
185
186    loop {
187        reap_finished_rpc_workers(&mut workers);
188
189        if shutdown.load(Ordering::Relaxed) {
190            break;
191        }
192
193        match listener.accept() {
194            Ok((stream, addr)) => {
195                if workers.len() >= MAX_CONCURRENT_RPC_CONNECTIONS {
196                    log::warn!(
197                        "rejecting RPC connection from {addr}: {MAX_CONCURRENT_RPC_CONNECTIONS} handlers already active"
198                    );
199                    continue;
200                }
201
202                // Set blocking for this connection
203                let _ = stream.set_nonblocking(false);
204                let _ = stream.set_read_timeout(Some(std::time::Duration::from_secs(10)));
205                let _ = stream.set_write_timeout(Some(std::time::Duration::from_secs(10)));
206
207                let worker_event_tx = event_tx.clone();
208                match thread::Builder::new()
209                    .name("rpc-client".into())
210                    .spawn(move || {
211                        if let Err(e) = handle_connection(stream, &auth_key, &worker_event_tx) {
212                            log::debug!("RPC connection error: {e}");
213                        }
214                    }) {
215                    Ok(worker) => workers.push(worker),
216                    Err(e) => log::error!("failed to start RPC connection handler: {e}"),
217                }
218            }
219            Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
220                // No pending connection, sleep briefly and retry
221                thread::sleep(std::time::Duration::from_millis(10));
222            }
223            Err(e) => {
224                log::error!("RPC accept error: {}", e);
225                thread::sleep(std::time::Duration::from_millis(100));
226            }
227        }
228    }
229
230    for worker in workers {
231        let _ = worker.join();
232    }
233}
234
235fn reap_finished_rpc_workers(workers: &mut Vec<thread::JoinHandle<()>>) {
236    let mut index = 0;
237    while index < workers.len() {
238        if workers[index].is_finished() {
239            let worker = workers.swap_remove(index);
240            let _ = worker.join();
241        } else {
242            index += 1;
243        }
244    }
245}
246
247fn handle_connection<S: Read + Write>(
248    mut stream: S,
249    auth_key: &[u8; 32],
250    event_tx: &EventSender,
251) -> io::Result<()> {
252    // Authentication: send challenge, verify response
253    server_auth(&mut stream, auth_key)?;
254    // multiprocessing.connection authenticates both peers before requests.
255    answer_challenge(&mut stream, auth_key)?;
256
257    // Read request.
258    let request_bytes = recv_bytes(&mut stream)?;
259    let (request, codec) = decode_rpc_payload(&request_bytes)?;
260
261    // Translate request dict to query, send to driver, get response.
262    let response = handle_rpc_request(&request, event_tx)?;
263
264    // Encode response and send
265    let response_bytes = encode_rpc_payload(&response, codec);
266    send_bytes(&mut stream, &response_bytes)?;
267
268    Ok(())
269}
270
271fn decode_rpc_payload(data: &[u8]) -> io::Result<(PickleValue, RpcCodec)> {
272    if let Ok(value) = msgpack::unpack_exact(data) {
273        return Ok((msgpack_to_pickle(&value), RpcCodec::Msgpack));
274    }
275    pickle::decode(data)
276        .map(|value| (value, RpcCodec::Pickle))
277        .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e.to_string()))
278}
279
280fn encode_rpc_payload(value: &PickleValue, codec: RpcCodec) -> Vec<u8> {
281    match codec {
282        RpcCodec::Msgpack => msgpack::pack(&pickle_to_msgpack(value)),
283        RpcCodec::Pickle => pickle::encode(value),
284    }
285}
286
287fn pickle_to_msgpack(value: &PickleValue) -> MsgpackValue {
288    match value {
289        PickleValue::None => MsgpackValue::Nil,
290        PickleValue::Bool(v) => MsgpackValue::Bool(*v),
291        PickleValue::Int(v) if *v >= 0 => MsgpackValue::UInt(*v as u64),
292        PickleValue::Int(v) => MsgpackValue::Int(*v),
293        PickleValue::Float(v) => MsgpackValue::Float(*v),
294        PickleValue::String(v) => MsgpackValue::Str(v.clone()),
295        PickleValue::Bytes(v) => MsgpackValue::Bin(v.clone()),
296        PickleValue::List(values) => {
297            MsgpackValue::Array(values.iter().map(pickle_to_msgpack).collect())
298        }
299        PickleValue::Dict(values) => MsgpackValue::Map(
300            values
301                .iter()
302                .map(|(key, value)| (pickle_to_msgpack(key), pickle_to_msgpack(value)))
303                .collect(),
304        ),
305    }
306}
307
308fn msgpack_to_pickle(value: &MsgpackValue) -> PickleValue {
309    match value {
310        MsgpackValue::Nil => PickleValue::None,
311        MsgpackValue::Bool(v) => PickleValue::Bool(*v),
312        MsgpackValue::UInt(v) if *v <= i64::MAX as u64 => PickleValue::Int(*v as i64),
313        MsgpackValue::UInt(v) => PickleValue::Float(*v as f64),
314        MsgpackValue::Int(v) => PickleValue::Int(*v),
315        MsgpackValue::Float(v) => PickleValue::Float(*v),
316        MsgpackValue::Bin(v) => PickleValue::Bytes(v.clone()),
317        MsgpackValue::Str(v) => PickleValue::String(v.clone()),
318        MsgpackValue::Array(values) => {
319            PickleValue::List(values.iter().map(msgpack_to_pickle).collect())
320        }
321        MsgpackValue::Map(values) => PickleValue::Dict(
322            values
323                .iter()
324                .map(|(key, value)| (msgpack_to_pickle(key), msgpack_to_pickle(value)))
325                .collect(),
326        ),
327    }
328}
329
330/// Server-side authentication: challenge-response.
331fn server_auth<S: Read + Write>(stream: &mut S, auth_key: &[u8; 32]) -> io::Result<()> {
332    // Generate challenge: #CHALLENGE#{sha256}<40 random bytes>
333    let mut random_bytes = [0u8; CHALLENGE_LEN];
334    // Use /dev/urandom for randomness
335    {
336        let mut f = std::fs::File::open("/dev/urandom")?;
337        f.read_exact(&mut random_bytes)?;
338    }
339
340    let mut challenge_message = Vec::with_capacity(CHALLENGE_PREFIX.len() + 8 + CHALLENGE_LEN);
341    challenge_message.extend_from_slice(CHALLENGE_PREFIX);
342    challenge_message.extend_from_slice(b"{sha256}");
343    challenge_message.extend_from_slice(&random_bytes);
344
345    send_bytes(stream, &challenge_message)?;
346
347    // Read response (max 256 bytes)
348    let response = recv_bytes(stream)?;
349
350    // Verify response
351    // The message to HMAC is everything after #CHALLENGE# (i.e. {sha256}<random>)
352    let message = &challenge_message[CHALLENGE_PREFIX.len()..];
353
354    if verify_response(auth_key, message, &response) {
355        send_bytes(stream, WELCOME)?;
356        Ok(())
357    } else {
358        send_bytes(stream, FAILURE)?;
359        Err(io::Error::new(
360            io::ErrorKind::PermissionDenied,
361            "auth failed",
362        ))
363    }
364}
365
366/// Verify a client's HMAC response.
367fn verify_response(auth_key: &[u8; 32], message: &[u8], response: &[u8]) -> bool {
368    // Modern protocol: response = {sha256}<hmac-sha256 digest>
369    if response.starts_with(b"{sha256}") {
370        let digest = &response[8..];
371        let expected = hmac_sha256(auth_key, message);
372        constant_time_eq(digest, &expected)
373    }
374    // Legacy protocol: response = raw 16-byte HMAC-MD5 digest
375    else if response.len() == 16 {
376        let expected = hmac_md5(auth_key, message);
377        constant_time_eq(response, &expected)
378    }
379    // Legacy with {md5} prefix
380    else if response.starts_with(b"{md5}") {
381        let digest = &response[5..];
382        let expected = hmac_md5(auth_key, message);
383        constant_time_eq(digest, &expected)
384    } else {
385        false
386    }
387}
388
389/// Constant-time byte comparison.
390fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
391    if a.len() != b.len() {
392        return false;
393    }
394    let mut diff = 0u8;
395    for (x, y) in a.iter().zip(b.iter()) {
396        diff |= x ^ y;
397    }
398    diff == 0
399}
400
401/// Send bytes with 4-byte big-endian length prefix.
402fn send_bytes<S: Write>(stream: &mut S, data: &[u8]) -> io::Result<()> {
403    let len = data.len() as i32;
404    stream.write_all(&len.to_be_bytes())?;
405    stream.write_all(data)?;
406    stream.flush()
407}
408
409/// Receive bytes with 4-byte big-endian length prefix.
410fn recv_bytes<S: Read>(stream: &mut S) -> io::Result<Vec<u8>> {
411    let mut len_buf = [0u8; 4];
412    stream.read_exact(&mut len_buf)?;
413    let len = i32::from_be_bytes(len_buf);
414
415    if len < 0 {
416        // Extended format: 8-byte length
417        let mut len8_buf = [0u8; 8];
418        stream.read_exact(&mut len8_buf)?;
419        let len = u64::from_be_bytes(len8_buf) as usize;
420        if len > 64 * 1024 * 1024 {
421            return Err(io::Error::new(
422                io::ErrorKind::InvalidData,
423                "message too large",
424            ));
425        }
426        let mut buf = vec![0u8; len];
427        stream.read_exact(&mut buf)?;
428        Ok(buf)
429    } else {
430        let len = len as usize;
431        if len > 64 * 1024 * 1024 {
432            return Err(io::Error::new(
433                io::ErrorKind::InvalidData,
434                "message too large",
435            ));
436        }
437        let mut buf = vec![0u8; len];
438        stream.read_exact(&mut buf)?;
439        Ok(buf)
440    }
441}
442
443/// Translate an RPC request dict to a query event and get response.
444fn handle_rpc_request(request: &PickleValue, event_tx: &EventSender) -> io::Result<PickleValue> {
445    if let Some(operation) = request.get("manage").and_then(PickleValue::as_str) {
446        let operation = match operation {
447            "attach_interface" => crate::event::InterfaceManagementOperation::Attach,
448            "detach_interface" => crate::event::InterfaceManagementOperation::Detach,
449            "reload_interface" => crate::event::InterfaceManagementOperation::Reload,
450            _ => return Ok(PickleValue::None),
451        };
452        let name = request
453            .get("name")
454            .and_then(PickleValue::as_str)
455            .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "missing interface name"))?;
456        let result = send_query(
457            event_tx,
458            QueryRequest::ManageInterface {
459                operation,
460                name: name.to_string(),
461            },
462        )?;
463        return Ok(match result {
464            QueryResponse::InterfaceManagement(Some(value)) => PickleValue::Bool(value),
465            _ => PickleValue::None,
466        });
467    }
468    // Handle "get" requests
469    if let Some(get_val) = request.get("get") {
470        if let Some(path) = get_val.as_str() {
471            return match path {
472                "interface_stats" => {
473                    let resp = send_query(event_tx, QueryRequest::InterfaceStats)?;
474                    if let QueryResponse::InterfaceStats(stats) = resp {
475                        let mut response = interface_stats_to_pickle(&stats);
476                        add_inbound_runtime_stats(&mut response, &stats, event_tx);
477                        Ok(response)
478                    } else {
479                        Ok(PickleValue::None)
480                    }
481                }
482                "profiling_results" => {
483                    Ok(crate::profiling::results_pickle().unwrap_or(PickleValue::None))
484                }
485                "path_table" => {
486                    let max_hops = request
487                        .get("max_hops")
488                        .and_then(|v| v.as_int().map(|n| n as u8));
489                    let resp = send_query(event_tx, QueryRequest::PathTable { max_hops })?;
490                    if let QueryResponse::PathTable(entries) = resp {
491                        Ok(path_table_to_pickle(&entries))
492                    } else {
493                        Ok(PickleValue::None)
494                    }
495                }
496                "rate_table" => {
497                    let resp = send_query(event_tx, QueryRequest::RateTable)?;
498                    if let QueryResponse::RateTable(entries) = resp {
499                        Ok(rate_table_to_pickle(&entries))
500                    } else {
501                        Ok(PickleValue::None)
502                    }
503                }
504                "lowest_interface_bitrate" => {
505                    let resp = send_query(event_tx, QueryRequest::LowestInterfaceBitrate)?;
506                    if let QueryResponse::LowestInterfaceBitrate(Some(bitrate)) = resp {
507                        Ok(PickleValue::Int(bitrate as i64))
508                    } else {
509                        Ok(PickleValue::None)
510                    }
511                }
512                "medium_path_timeout" => {
513                    let resp = send_query(event_tx, QueryRequest::MediumPathTimeout)?;
514                    if let QueryResponse::MediumPathTimeout(timeout) = resp {
515                        Ok(PickleValue::Float(timeout))
516                    } else {
517                        Ok(PickleValue::Int(0))
518                    }
519                }
520                "first_hop_timeout" => {
521                    let hash = extract_dest_hash(request, "destination_hash")?;
522                    let resp =
523                        send_query(event_tx, QueryRequest::FirstHopTimeout { dest_hash: hash })?;
524                    if let QueryResponse::FirstHopTimeout(timeout) = resp {
525                        Ok(PickleValue::Float(timeout))
526                    } else {
527                        Ok(PickleValue::Float(
528                            rns_core::constants::LINK_ESTABLISHMENT_TIMEOUT_PER_HOP,
529                        ))
530                    }
531                }
532                "next_hop" => {
533                    let hash = extract_dest_hash(request, "destination_hash")?;
534                    let resp = send_query(event_tx, QueryRequest::NextHop { dest_hash: hash })?;
535                    if let QueryResponse::NextHop(Some(nh)) = resp {
536                        Ok(PickleValue::Bytes(nh.next_hop.to_vec()))
537                    } else {
538                        Ok(PickleValue::None)
539                    }
540                }
541                "next_hop_if_name" => {
542                    let hash = extract_dest_hash(request, "destination_hash")?;
543                    let resp =
544                        send_query(event_tx, QueryRequest::NextHopIfName { dest_hash: hash })?;
545                    if let QueryResponse::NextHopIfName(Some(name)) = resp {
546                        Ok(PickleValue::String(name))
547                    } else {
548                        Ok(PickleValue::None)
549                    }
550                }
551                "link_count" => {
552                    let resp = send_query(event_tx, QueryRequest::LinkCount)?;
553                    if let QueryResponse::LinkCount(n) = resp {
554                        Ok(PickleValue::Int(n as i64))
555                    } else {
556                        Ok(PickleValue::None)
557                    }
558                }
559                "active_link_count" => {
560                    let resp = send_query(event_tx, QueryRequest::ActiveLinkCount)?;
561                    if let QueryResponse::ActiveLinkCount(n) = resp {
562                        Ok(PickleValue::Int(n as i64))
563                    } else {
564                        Ok(PickleValue::None)
565                    }
566                }
567                "transport_identity" => {
568                    let resp = send_query(event_tx, QueryRequest::TransportIdentity)?;
569                    if let QueryResponse::TransportIdentity(Some(hash)) = resp {
570                        Ok(PickleValue::Bytes(hash.to_vec()))
571                    } else {
572                        Ok(PickleValue::None)
573                    }
574                }
575                "blackholed" => {
576                    let resp = send_query(event_tx, QueryRequest::GetBlackholed)?;
577                    if let QueryResponse::Blackholed(entries) = resp {
578                        Ok(blackholed_to_pickle(&entries))
579                    } else {
580                        Ok(PickleValue::None)
581                    }
582                }
583                "discovered_interfaces" => {
584                    let only_available = request
585                        .get("only_available")
586                        .and_then(|v| v.as_bool())
587                        .unwrap_or(false);
588                    let only_transport = request
589                        .get("only_transport")
590                        .and_then(|v| v.as_bool())
591                        .unwrap_or(false);
592                    let resp = send_query(
593                        event_tx,
594                        QueryRequest::DiscoveredInterfaces {
595                            only_available,
596                            only_transport,
597                        },
598                    )?;
599                    if let QueryResponse::DiscoveredInterfaces(interfaces) = resp {
600                        Ok(discovered_interfaces_to_pickle(&interfaces))
601                    } else {
602                        Ok(PickleValue::None)
603                    }
604                }
605                "hooks" => {
606                    let (response_tx, response_rx) = mpsc::channel();
607                    try_send_event(event_tx, Event::ListHooks { response_tx })?;
608                    let hooks = response_rx
609                        .recv_timeout(std::time::Duration::from_secs(5))
610                        .map_err(|_| {
611                            io::Error::new(io::ErrorKind::TimedOut, "list hooks timed out")
612                        })?;
613                    Ok(hooks_to_pickle(&hooks))
614                }
615                "runtime_config" => {
616                    let resp = send_query(event_tx, QueryRequest::ListRuntimeConfig)?;
617                    if let QueryResponse::RuntimeConfigList(entries) = resp {
618                        Ok(runtime_config_list_to_pickle(&entries))
619                    } else {
620                        Ok(PickleValue::None)
621                    }
622                }
623                "known_destinations" => {
624                    let resp = send_query(event_tx, QueryRequest::KnownDestinations)?;
625                    if let QueryResponse::KnownDestinations(entries) = resp {
626                        Ok(known_destinations_to_pickle(&entries))
627                    } else {
628                        Ok(PickleValue::None)
629                    }
630                }
631                "runtime_config_entry" => {
632                    let key = request
633                        .get("key")
634                        .and_then(|v| v.as_str())
635                        .unwrap_or_default()
636                        .to_string();
637                    let resp = send_query(event_tx, QueryRequest::GetRuntimeConfig { key })?;
638                    if let QueryResponse::RuntimeConfigEntry(entry) = resp {
639                        Ok(entry
640                            .as_ref()
641                            .map(runtime_config_entry_to_pickle)
642                            .unwrap_or(PickleValue::None))
643                    } else {
644                        Ok(PickleValue::None)
645                    }
646                }
647                "backbone_peer_state" => {
648                    let interface_name = request
649                        .get("interface")
650                        .and_then(|v| v.as_str())
651                        .map(|s| s.to_string());
652                    let resp =
653                        send_query(event_tx, QueryRequest::BackbonePeerState { interface_name })?;
654                    if let QueryResponse::BackbonePeerState(entries) = resp {
655                        Ok(backbone_peer_state_to_pickle(&entries))
656                    } else {
657                        Ok(PickleValue::None)
658                    }
659                }
660                "backbone_interfaces" => {
661                    let resp = send_query(event_tx, QueryRequest::BackboneInterfaces)?;
662                    if let QueryResponse::BackboneInterfaces(entries) = resp {
663                        Ok(backbone_interfaces_to_pickle(&entries))
664                    } else {
665                        Ok(PickleValue::None)
666                    }
667                }
668                "provider_bridge_stats" => {
669                    let resp = send_query(event_tx, QueryRequest::ProviderBridgeStats)?;
670                    if let QueryResponse::ProviderBridgeStats(stats) = resp {
671                        Ok(stats
672                            .as_ref()
673                            .map(provider_bridge_stats_to_pickle)
674                            .unwrap_or(PickleValue::None))
675                    } else {
676                        Ok(PickleValue::None)
677                    }
678                }
679                "drain_status" => {
680                    let resp = send_query(event_tx, QueryRequest::DrainStatus)?;
681                    if let QueryResponse::DrainStatus(status) = resp {
682                        Ok(drain_status_to_pickle(&status))
683                    } else {
684                        Ok(PickleValue::None)
685                    }
686                }
687                _ => Ok(PickleValue::None),
688            };
689        }
690    }
691
692    if let Some(begin_val) = request.get("begin_drain") {
693        let timeout_secs = begin_val
694            .as_float()
695            .or_else(|| begin_val.as_int().map(|value| value as f64))
696            .unwrap_or(0.0)
697            .max(0.0);
698        let timeout = Duration::from_secs_f64(timeout_secs);
699        try_send_event(event_tx, Event::BeginDrain { timeout })?;
700        return Ok(PickleValue::Bool(true));
701    }
702
703    if let Some(set_val) = request.get("set").and_then(|v| v.as_str()) {
704        if set_val == "known_destination_retained" {
705            let dest_hash = extract_dest_hash(request, "dest_hash")?;
706            let resp = send_query(event_tx, QueryRequest::RetainKnownDestination { dest_hash })?;
707            return if let QueryResponse::RetainKnownDestination(ok) = resp {
708                Ok(PickleValue::Bool(ok))
709            } else {
710                Ok(PickleValue::None)
711            };
712        }
713        if set_val == "identity_retained" {
714            let identity_hash = extract_dest_hash(request, "identity_hash")?;
715            let resp = send_query(event_tx, QueryRequest::RetainIdentity { identity_hash })?;
716            return if let QueryResponse::RetainIdentity(ok) = resp {
717                Ok(PickleValue::Bool(ok))
718            } else {
719                Ok(PickleValue::None)
720            };
721        }
722        if set_val == "known_destination_used" {
723            let dest_hash = extract_dest_hash(request, "dest_hash")?;
724            let resp = send_query(
725                event_tx,
726                QueryRequest::MarkKnownDestinationUsed { dest_hash },
727            )?;
728            return if let QueryResponse::MarkKnownDestinationUsed(ok) = resp {
729                Ok(PickleValue::Bool(ok))
730            } else {
731                Ok(PickleValue::None)
732            };
733        }
734        if set_val == "runtime_config" {
735            let key = request
736                .get("key")
737                .and_then(|v| v.as_str())
738                .unwrap_or_default()
739                .to_string();
740            let Some(value) = request
741                .get("value")
742                .and_then(runtime_config_value_from_pickle)
743            else {
744                return Ok(runtime_config_error_to_pickle(&RuntimeConfigError {
745                    code: RuntimeConfigErrorCode::InvalidType,
746                    message: "runtime-config set requires a scalar value".into(),
747                }));
748            };
749            let resp = send_query(event_tx, QueryRequest::SetRuntimeConfig { key, value })?;
750            return if let QueryResponse::RuntimeConfigSet(result) = resp {
751                Ok(runtime_config_result_to_pickle(result))
752            } else {
753                Ok(PickleValue::None)
754            };
755        }
756    }
757
758    if let Some(reset_val) = request.get("reset").and_then(|v| v.as_str()) {
759        if reset_val == "runtime_config" {
760            let key = request
761                .get("key")
762                .and_then(|v| v.as_str())
763                .unwrap_or_default()
764                .to_string();
765            let resp = send_query(event_tx, QueryRequest::ResetRuntimeConfig { key })?;
766            return if let QueryResponse::RuntimeConfigReset(result) = resp {
767                Ok(runtime_config_result_to_pickle(result))
768            } else {
769                Ok(PickleValue::None)
770            };
771        }
772    }
773
774    if let Some(clear_val) = request.get("clear").and_then(|v| v.as_str()) {
775        if clear_val == "known_destination_retained" {
776            let dest_hash = extract_dest_hash(request, "dest_hash")?;
777            let resp = send_query(
778                event_tx,
779                QueryRequest::UnretainKnownDestination { dest_hash },
780            )?;
781            return if let QueryResponse::UnretainKnownDestination(ok) = resp {
782                Ok(PickleValue::Bool(ok))
783            } else {
784                Ok(PickleValue::None)
785            };
786        }
787        if clear_val == "backbone_peer_state" {
788            let interface_name = required_string(request, "interface")?;
789            let peer_ip = required_string(request, "ip")?;
790            let peer_ip = peer_ip
791                .parse()
792                .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "invalid peer IP"))?;
793            let resp = send_query(
794                event_tx,
795                QueryRequest::ClearBackbonePeerState {
796                    interface_name,
797                    peer_ip,
798                },
799            )?;
800            return if let QueryResponse::ClearBackbonePeerState(ok) = resp {
801                Ok(PickleValue::Bool(ok))
802            } else {
803                Ok(PickleValue::None)
804            };
805        }
806    }
807
808    if let Some(set_val) = request.get("set").and_then(|v| v.as_str()) {
809        if set_val == "backbone_peer_blacklist" {
810            let interface_name = required_string(request, "interface")?;
811            let peer_ip = required_string(request, "ip")?;
812            let peer_ip: IpAddr = peer_ip
813                .parse()
814                .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "invalid peer IP"))?;
815            let duration_secs = request
816                .get("duration_secs")
817                .and_then(|v| v.as_int())
818                .ok_or_else(|| {
819                    io::Error::new(io::ErrorKind::InvalidInput, "missing duration_secs")
820                })?;
821            let reason = request
822                .get("reason")
823                .and_then(|v| v.as_str())
824                .unwrap_or("sentinel blacklist")
825                .to_string();
826            let penalty_level = request
827                .get("penalty_level")
828                .and_then(|v| v.as_int())
829                .unwrap_or(0)
830                .clamp(0, u8::MAX as i64) as u8;
831            let resp = send_query(
832                event_tx,
833                QueryRequest::BlacklistBackbonePeer {
834                    interface_name,
835                    peer_ip,
836                    duration: Duration::from_secs(duration_secs as u64),
837                    reason,
838                    penalty_level,
839                },
840            )?;
841            return if let QueryResponse::BlacklistBackbonePeer(ok) = resp {
842                Ok(PickleValue::Bool(ok))
843            } else {
844                Ok(PickleValue::None)
845            };
846        }
847    }
848
849    // Handle "request_path" -- trigger a path request to the network
850    if let Some(hash_val) = request.get("request_path") {
851        if let Some(hash_bytes) = hash_val.as_bytes() {
852            if hash_bytes.len() >= 16 {
853                let mut dest_hash = [0u8; 16];
854                dest_hash.copy_from_slice(&hash_bytes[..16]);
855                try_send_event(event_tx, crate::event::Event::RequestPath { dest_hash })?;
856                return Ok(PickleValue::Bool(true));
857            }
858        }
859    }
860
861    // Handle "send_probe" requests
862    if let Some(hash_val) = request.get("send_probe") {
863        if let Some(hash_bytes) = hash_val.as_bytes() {
864            if hash_bytes.len() >= 16 {
865                let mut dest_hash = [0u8; 16];
866                dest_hash.copy_from_slice(&hash_bytes[..16]);
867                let payload_size = request
868                    .get("size")
869                    .and_then(|v| v.as_int())
870                    .and_then(|n| {
871                        if n > 0 && n <= 400 {
872                            Some(n as usize)
873                        } else {
874                            None
875                        }
876                    })
877                    .unwrap_or(16);
878                let resp = send_query(
879                    event_tx,
880                    QueryRequest::SendProbe {
881                        dest_hash,
882                        payload_size,
883                    },
884                )?;
885                if let QueryResponse::SendProbe(Some((packet_hash, hops))) = resp {
886                    return Ok(PickleValue::Dict(vec![
887                        (
888                            PickleValue::String("packet_hash".into()),
889                            PickleValue::Bytes(packet_hash.to_vec()),
890                        ),
891                        (
892                            PickleValue::String("hops".into()),
893                            PickleValue::Int(hops as i64),
894                        ),
895                    ]));
896                } else {
897                    return Ok(PickleValue::None);
898                }
899            }
900        }
901    }
902
903    // Handle "check_proof" requests
904    if let Some(hash_val) = request.get("check_proof") {
905        if let Some(hash_bytes) = hash_val.as_bytes() {
906            if hash_bytes.len() >= 32 {
907                let mut packet_hash = [0u8; 32];
908                packet_hash.copy_from_slice(&hash_bytes[..32]);
909                let resp = send_query(event_tx, QueryRequest::CheckProof { packet_hash })?;
910                if let QueryResponse::CheckProof(Some(rtt)) = resp {
911                    return Ok(PickleValue::Float(rtt));
912                } else {
913                    return Ok(PickleValue::None);
914                }
915            }
916        }
917    }
918
919    // Handle "blackhole" requests
920    if let Some(hash_val) = request.get("blackhole") {
921        if let Some(hash_bytes) = hash_val.as_bytes() {
922            if hash_bytes.len() >= 16 {
923                let mut identity_hash = [0u8; 16];
924                identity_hash.copy_from_slice(&hash_bytes[..16]);
925                let duration_hours = request.get("duration").and_then(|v| v.as_float());
926                let reason = request
927                    .get("reason")
928                    .and_then(|v| v.as_str())
929                    .map(|s| s.to_string());
930                let resp = send_query(
931                    event_tx,
932                    QueryRequest::BlackholeIdentity {
933                        identity_hash,
934                        duration_hours,
935                        reason,
936                    },
937                )?;
938                return Ok(PickleValue::Bool(matches!(
939                    resp,
940                    QueryResponse::BlackholeResult(true)
941                )));
942            }
943        }
944    }
945
946    // Handle "unblackhole" requests
947    if let Some(hash_val) = request.get("unblackhole") {
948        if let Some(hash_bytes) = hash_val.as_bytes() {
949            if hash_bytes.len() >= 16 {
950                let mut identity_hash = [0u8; 16];
951                identity_hash.copy_from_slice(&hash_bytes[..16]);
952                let resp = send_query(
953                    event_tx,
954                    QueryRequest::UnblackholeIdentity { identity_hash },
955                )?;
956                return Ok(PickleValue::Bool(matches!(
957                    resp,
958                    QueryResponse::UnblackholeResult(true)
959                )));
960            }
961        }
962    }
963
964    // Handle "drop" requests
965    if let Some(drop_val) = request.get("drop") {
966        if let Some(path) = drop_val.as_str() {
967            return match path {
968                "path" => {
969                    let hash = extract_dest_hash(request, "destination_hash")?;
970                    let resp = send_query(event_tx, QueryRequest::DropPath { dest_hash: hash })?;
971                    if let QueryResponse::DropPath(ok) = resp {
972                        Ok(PickleValue::Bool(ok))
973                    } else {
974                        Ok(PickleValue::None)
975                    }
976                }
977                "all_via" => {
978                    let hash = extract_dest_hash(request, "destination_hash")?;
979                    let resp = send_query(
980                        event_tx,
981                        QueryRequest::DropAllVia {
982                            transport_hash: hash,
983                        },
984                    )?;
985                    if let QueryResponse::DropAllVia(n) = resp {
986                        Ok(PickleValue::Int(n as i64))
987                    } else {
988                        Ok(PickleValue::None)
989                    }
990                }
991                "announce_queues" => {
992                    let resp = send_query(event_tx, QueryRequest::DropAnnounceQueues)?;
993                    if let QueryResponse::DropAnnounceQueues = resp {
994                        Ok(PickleValue::Bool(true))
995                    } else {
996                        Ok(PickleValue::None)
997                    }
998                }
999                _ => Ok(PickleValue::None),
1000            };
1001        }
1002    }
1003
1004    if let Some(hook_val) = request.get("hook").and_then(|v| v.as_str()) {
1005        return handle_hook_rpc_request(hook_val, request, event_tx);
1006    }
1007
1008    Ok(PickleValue::None)
1009}
1010
1011fn handle_hook_rpc_request(
1012    op: &str,
1013    request: &PickleValue,
1014    event_tx: &EventSender,
1015) -> io::Result<PickleValue> {
1016    match op {
1017        "load" => {
1018            let name = required_string(request, "name")?;
1019            let attach_point = required_string(request, "attach_point")?;
1020            let priority = request
1021                .get("priority")
1022                .and_then(|v| v.as_int())
1023                .unwrap_or(0) as i32;
1024            let wasm = request
1025                .get("wasm")
1026                .and_then(|v| v.as_bytes())
1027                .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "missing wasm"))?
1028                .to_vec();
1029            let (response_tx, response_rx) = mpsc::channel();
1030            try_send_event(
1031                event_tx,
1032                Event::LoadHook {
1033                    name,
1034                    wasm_bytes: wasm,
1035                    attach_point,
1036                    priority,
1037                    response_tx,
1038                },
1039            )?;
1040            let response = response_rx
1041                .recv_timeout(std::time::Duration::from_secs(5))
1042                .map_err(|_| io::Error::new(io::ErrorKind::TimedOut, "hook load timed out"))?;
1043            Ok(hook_result_to_pickle(response))
1044        }
1045        "load_builtin" => {
1046            let name = required_string(request, "name")?;
1047            let attach_point = required_string(request, "attach_point")?;
1048            let builtin_id = required_string(request, "builtin_id")?;
1049            let priority = request
1050                .get("priority")
1051                .and_then(|v| v.as_int())
1052                .unwrap_or(0) as i32;
1053            let (response_tx, response_rx) = mpsc::channel();
1054            try_send_event(
1055                event_tx,
1056                Event::LoadBuiltinHook {
1057                    name,
1058                    builtin_id,
1059                    attach_point,
1060                    priority,
1061                    response_tx,
1062                },
1063            )?;
1064            let response = response_rx
1065                .recv_timeout(std::time::Duration::from_secs(5))
1066                .map_err(|_| {
1067                    io::Error::new(io::ErrorKind::TimedOut, "built-in hook load timed out")
1068                })?;
1069            Ok(hook_result_to_pickle(response))
1070        }
1071        "unload" => {
1072            let name = required_string(request, "name")?;
1073            let attach_point = required_string(request, "attach_point")?;
1074            let (response_tx, response_rx) = mpsc::channel();
1075            try_send_event(
1076                event_tx,
1077                Event::UnloadHook {
1078                    name,
1079                    attach_point,
1080                    response_tx,
1081                },
1082            )?;
1083            let response = response_rx
1084                .recv_timeout(std::time::Duration::from_secs(5))
1085                .map_err(|_| io::Error::new(io::ErrorKind::TimedOut, "hook unload timed out"))?;
1086            Ok(hook_result_to_pickle(response))
1087        }
1088        "enable" | "disable" => {
1089            let name = required_string(request, "name")?;
1090            let attach_point = required_string(request, "attach_point")?;
1091            let enabled = op == "enable";
1092            let (response_tx, response_rx) = mpsc::channel();
1093            try_send_event(
1094                event_tx,
1095                Event::SetHookEnabled {
1096                    name,
1097                    attach_point,
1098                    enabled,
1099                    response_tx,
1100                },
1101            )?;
1102            let response = response_rx
1103                .recv_timeout(std::time::Duration::from_secs(5))
1104                .map_err(|_| {
1105                    io::Error::new(io::ErrorKind::TimedOut, "hook enable/disable timed out")
1106                })?;
1107            Ok(hook_result_to_pickle(response))
1108        }
1109        "set_priority" => {
1110            let name = required_string(request, "name")?;
1111            let attach_point = required_string(request, "attach_point")?;
1112            let priority = request
1113                .get("priority")
1114                .and_then(|v| v.as_int())
1115                .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "missing priority"))?
1116                as i32;
1117            let (response_tx, response_rx) = mpsc::channel();
1118            try_send_event(
1119                event_tx,
1120                Event::SetHookPriority {
1121                    name,
1122                    attach_point,
1123                    priority,
1124                    response_tx,
1125                },
1126            )?;
1127            let response = response_rx
1128                .recv_timeout(std::time::Duration::from_secs(5))
1129                .map_err(|_| io::Error::new(io::ErrorKind::TimedOut, "hook priority timed out"))?;
1130            Ok(hook_result_to_pickle(response))
1131        }
1132        _ => Ok(PickleValue::None),
1133    }
1134}
1135
1136/// Send a query to the driver and wait for the response.
1137fn send_query(event_tx: &EventSender, request: QueryRequest) -> io::Result<QueryResponse> {
1138    let (resp_tx, resp_rx) = mpsc::channel();
1139    try_send_event(event_tx, Event::query(request, resp_tx))?;
1140    resp_rx
1141        .recv_timeout(std::time::Duration::from_secs(5))
1142        .map_err(|_| io::Error::new(io::ErrorKind::TimedOut, "query timed out"))
1143}
1144
1145fn try_send_event(event_tx: &EventSender, event: Event) -> io::Result<()> {
1146    event_tx.try_send(event).map_err(|error| match error {
1147        mpsc::TrySendError::Full(_) => {
1148            io::Error::new(io::ErrorKind::WouldBlock, "driver event queue is full")
1149        }
1150        mpsc::TrySendError::Disconnected(_) => {
1151            io::Error::new(io::ErrorKind::BrokenPipe, "driver shut down")
1152        }
1153    })
1154}
1155
1156/// Extract a 16-byte destination hash from an RPC dict field.
1157fn extract_dest_hash(request: &PickleValue, key: &str) -> io::Result<[u8; 16]> {
1158    let bytes = request
1159        .get(key)
1160        .and_then(|v| v.as_bytes())
1161        .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "missing destination_hash"))?;
1162    if bytes.len() < 16 {
1163        return Err(io::Error::new(io::ErrorKind::InvalidData, "hash too short"));
1164    }
1165    let mut hash = [0u8; 16];
1166    hash.copy_from_slice(&bytes[..16]);
1167    Ok(hash)
1168}
1169
1170fn required_string(request: &PickleValue, key: &str) -> io::Result<String> {
1171    request
1172        .get(key)
1173        .and_then(|v| v.as_str())
1174        .map(|s| s.to_string())
1175        .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, format!("missing {}", key)))
1176}
1177
1178fn hook_result_to_pickle(result: Result<(), String>) -> PickleValue {
1179    match result {
1180        Ok(()) => PickleValue::Dict(vec![(
1181            PickleValue::String("ok".into()),
1182            PickleValue::Bool(true),
1183        )]),
1184        Err(error) => PickleValue::Dict(vec![
1185            (PickleValue::String("ok".into()), PickleValue::Bool(false)),
1186            (
1187                PickleValue::String("error".into()),
1188                PickleValue::String(error),
1189            ),
1190        ]),
1191    }
1192}
1193
1194// --- Pickle response builders ---
1195
1196fn interface_stats_to_pickle(stats: &InterfaceStatsResponse) -> PickleValue {
1197    let mut ifaces = Vec::new();
1198    for iface in &stats.interfaces {
1199        ifaces.push(single_iface_to_pickle(iface));
1200    }
1201
1202    let mut dict = vec![
1203        (
1204            PickleValue::String("interfaces".into()),
1205            PickleValue::List(ifaces),
1206        ),
1207        (
1208            PickleValue::String("transport_enabled".into()),
1209            PickleValue::Bool(stats.transport_enabled),
1210        ),
1211        (
1212            PickleValue::String("transport_uptime".into()),
1213            PickleValue::Float(stats.transport_uptime),
1214        ),
1215        (
1216            PickleValue::String("rxb".into()),
1217            PickleValue::Int(stats.total_rxb as i64),
1218        ),
1219        (
1220            PickleValue::String("txb".into()),
1221            PickleValue::Int(stats.total_txb as i64),
1222        ),
1223        (
1224            PickleValue::String("rxpps".into()),
1225            PickleValue::Float(stats.rx_pps),
1226        ),
1227        (
1228            PickleValue::String("txpps".into()),
1229            PickleValue::Float(stats.tx_pps),
1230        ),
1231        (
1232            PickleValue::String("rxs".into()),
1233            PickleValue::Float(stats.traffic.rxs),
1234        ),
1235        (
1236            PickleValue::String("txs".into()),
1237            PickleValue::Float(stats.traffic.txs),
1238        ),
1239        (
1240            PickleValue::String("arxb".into()),
1241            PickleValue::Int(stats.traffic.arxb as i64),
1242        ),
1243        (
1244            PickleValue::String("atxb".into()),
1245            PickleValue::Int(stats.traffic.atxb as i64),
1246        ),
1247        (
1248            PickleValue::String("arxc".into()),
1249            PickleValue::Int(stats.traffic.arxc as i64),
1250        ),
1251        (
1252            PickleValue::String("atxc".into()),
1253            PickleValue::Int(stats.traffic.atxc as i64),
1254        ),
1255        (
1256            PickleValue::String("arxs".into()),
1257            PickleValue::Float(stats.traffic.arxs),
1258        ),
1259        (
1260            PickleValue::String("atxs".into()),
1261            PickleValue::Float(stats.traffic.atxs),
1262        ),
1263        (
1264            PickleValue::String("prxb".into()),
1265            PickleValue::Int(stats.traffic.prxb as i64),
1266        ),
1267        (
1268            PickleValue::String("ptxb".into()),
1269            PickleValue::Int(stats.traffic.ptxb as i64),
1270        ),
1271        (
1272            PickleValue::String("prxc".into()),
1273            PickleValue::Int(stats.traffic.prxc as i64),
1274        ),
1275        (
1276            PickleValue::String("ptxc".into()),
1277            PickleValue::Int(stats.traffic.ptxc as i64),
1278        ),
1279        (
1280            PickleValue::String("prxs".into()),
1281            PickleValue::Float(stats.traffic.prxs),
1282        ),
1283        (
1284            PickleValue::String("ptxs".into()),
1285            PickleValue::Float(stats.traffic.ptxs),
1286        ),
1287        (
1288            PickleValue::String("arxf".into()),
1289            PickleValue::Float(stats.traffic.arxf),
1290        ),
1291        (
1292            PickleValue::String("atxf".into()),
1293            PickleValue::Float(stats.traffic.atxf),
1294        ),
1295        (
1296            PickleValue::String("prxf".into()),
1297            PickleValue::Float(stats.traffic.prxf),
1298        ),
1299        (
1300            PickleValue::String("ptxf".into()),
1301            PickleValue::Float(stats.traffic.ptxf),
1302        ),
1303    ];
1304
1305    if let Some(tid) = stats.transport_id {
1306        dict.push((
1307            PickleValue::String("transport_id".into()),
1308            PickleValue::Bytes(tid.to_vec()),
1309        ));
1310    } else {
1311        dict.push((
1312            PickleValue::String("transport_id".into()),
1313            PickleValue::None,
1314        ));
1315    }
1316
1317    if let Some(pr) = stats.probe_responder {
1318        dict.push((
1319            PickleValue::String("probe_responder".into()),
1320            PickleValue::Bytes(pr.to_vec()),
1321        ));
1322    } else {
1323        dict.push((
1324            PickleValue::String("probe_responder".into()),
1325            PickleValue::None,
1326        ));
1327    }
1328
1329    if let Some(pool) = &stats.backbone_peer_pool {
1330        let members = pool
1331            .members
1332            .iter()
1333            .map(|member| {
1334                let mut member_dict = vec![
1335                    (
1336                        PickleValue::String("name".into()),
1337                        PickleValue::String(member.name.clone()),
1338                    ),
1339                    (
1340                        PickleValue::String("remote".into()),
1341                        PickleValue::String(member.remote.clone()),
1342                    ),
1343                    (
1344                        PickleValue::String("source".into()),
1345                        PickleValue::String(member.source.clone()),
1346                    ),
1347                    (
1348                        PickleValue::String("priority".into()),
1349                        PickleValue::Int(member.priority as i64),
1350                    ),
1351                    (
1352                        PickleValue::String("state".into()),
1353                        PickleValue::String(member.state.clone()),
1354                    ),
1355                    (
1356                        PickleValue::String("failure_count".into()),
1357                        PickleValue::Int(member.failure_count as i64),
1358                    ),
1359                ];
1360                member_dict.push((
1361                    PickleValue::String("interface_id".into()),
1362                    member
1363                        .interface_id
1364                        .map(|id| PickleValue::Int(id as i64))
1365                        .unwrap_or(PickleValue::None),
1366                ));
1367                member_dict.push((
1368                    PickleValue::String("last_error".into()),
1369                    member
1370                        .last_error
1371                        .as_ref()
1372                        .map(|err| PickleValue::String(err.clone()))
1373                        .unwrap_or(PickleValue::None),
1374                ));
1375                member_dict.push((
1376                    PickleValue::String("cooldown_remaining_seconds".into()),
1377                    member
1378                        .cooldown_remaining_seconds
1379                        .map(PickleValue::Float)
1380                        .unwrap_or(PickleValue::None),
1381                ));
1382                PickleValue::Dict(member_dict)
1383            })
1384            .collect();
1385        dict.push((
1386            PickleValue::String("backbone_peer_pool".into()),
1387            PickleValue::Dict(vec![
1388                (
1389                    PickleValue::String("max_connected".into()),
1390                    PickleValue::Int(pool.max_connected as i64),
1391                ),
1392                (
1393                    PickleValue::String("active_count".into()),
1394                    PickleValue::Int(pool.active_count as i64),
1395                ),
1396                (
1397                    PickleValue::String("standby_count".into()),
1398                    PickleValue::Int(pool.standby_count as i64),
1399                ),
1400                (
1401                    PickleValue::String("cooldown_count".into()),
1402                    PickleValue::Int(pool.cooldown_count as i64),
1403                ),
1404                (
1405                    PickleValue::String("members".into()),
1406                    PickleValue::List(members),
1407                ),
1408            ]),
1409        ));
1410    } else {
1411        dict.push((
1412            PickleValue::String("backbone_peer_pool".into()),
1413            PickleValue::None,
1414        ));
1415    }
1416
1417    PickleValue::Dict(dict)
1418}
1419
1420fn add_inbound_runtime_stats(
1421    response: &mut PickleValue,
1422    stats: &InterfaceStatsResponse,
1423    event_tx: &EventSender,
1424) {
1425    let PickleValue::Dict(dict) = response else {
1426        return;
1427    };
1428    if let Some((_, PickleValue::List(ifaces))) = dict
1429        .iter_mut()
1430        .find(|(key, _)| key.as_str() == Some("interfaces"))
1431    {
1432        for (iface, stat) in ifaces.iter_mut().zip(&stats.interfaces) {
1433            let PickleValue::Dict(iface_dict) = iface else {
1434                continue;
1435            };
1436            let burst_stats = if stat.interface_type == "BackboneInterface" {
1437                event_tx
1438                    .dynamic_burst_stats(InterfaceId(stat.id))
1439                    .or(Some(DynamicBurstStats {
1440                        announce: BurstClassStats {
1441                            count: 0,
1442                            activated: None,
1443                        },
1444                        path_request: BurstClassStats {
1445                            count: 0,
1446                            activated: None,
1447                        },
1448                    }))
1449            } else {
1450                None
1451            };
1452            iface_dict.push((
1453                PickleValue::String("burst_count".into()),
1454                burst_stats
1455                    .map(|stats| PickleValue::Int(stats.announce.count as i64))
1456                    .unwrap_or(PickleValue::None),
1457            ));
1458            iface_dict.push((
1459                PickleValue::String("pr_burst_count".into()),
1460                burst_stats
1461                    .map(|stats| PickleValue::Int(stats.path_request.count as i64))
1462                    .unwrap_or(PickleValue::None),
1463            ));
1464            if let Some(stats) = burst_stats {
1465                set_pickle_dict_entry(
1466                    iface_dict,
1467                    "burst_active",
1468                    PickleValue::Bool(stats.announce.count > 0),
1469                );
1470                set_pickle_dict_entry(
1471                    iface_dict,
1472                    "burst_activated",
1473                    stats
1474                        .announce
1475                        .activated
1476                        .map(PickleValue::Float)
1477                        .unwrap_or(PickleValue::None),
1478                );
1479                set_pickle_dict_entry(
1480                    iface_dict,
1481                    "pr_burst_active",
1482                    PickleValue::Bool(stats.path_request.count > 0),
1483                );
1484                set_pickle_dict_entry(
1485                    iface_dict,
1486                    "pr_burst_activated",
1487                    stats
1488                        .path_request
1489                        .activated
1490                        .map(PickleValue::Float)
1491                        .unwrap_or(PickleValue::None),
1492                );
1493            }
1494        }
1495    }
1496
1497    let snapshot = event_tx.inbound_queue_snapshot();
1498    let total_height = snapshot.total_height();
1499    let total_capacity = snapshot.total_capacity();
1500    let total_dropped = snapshot.total_dropped();
1501    let pressure = |height: usize, capacity: usize| {
1502        if height == 0 {
1503            0.0
1504        } else {
1505            height as f64 / capacity as f64
1506        }
1507    };
1508    dict.extend([
1509        (
1510            PickleValue::String("rxqt".into()),
1511            PickleValue::Int(total_height as i64),
1512        ),
1513        (
1514            PickleValue::String("rxqd".into()),
1515            PickleValue::Int(snapshot.heights[0] as i64),
1516        ),
1517        (
1518            PickleValue::String("rxqa".into()),
1519            PickleValue::Int(snapshot.heights[1] as i64),
1520        ),
1521        (
1522            PickleValue::String("rxqp".into()),
1523            PickleValue::Int(snapshot.heights[2] as i64),
1524        ),
1525        (
1526            PickleValue::String("rxqil".into()),
1527            PickleValue::Int(snapshot.heights[3] as i64),
1528        ),
1529        (
1530            PickleValue::String("rxqtd".into()),
1531            PickleValue::Int(total_dropped as i64),
1532        ),
1533        (
1534            PickleValue::String("rxqdd".into()),
1535            PickleValue::Int(snapshot.dropped[0] as i64),
1536        ),
1537        (
1538            PickleValue::String("rxqad".into()),
1539            PickleValue::Int(snapshot.dropped[1] as i64),
1540        ),
1541        (
1542            PickleValue::String("rxqpd".into()),
1543            PickleValue::Int(snapshot.dropped[2] as i64),
1544        ),
1545        (
1546            PickleValue::String("rxqild".into()),
1547            PickleValue::Int(snapshot.dropped[3] as i64),
1548        ),
1549        (
1550            PickleValue::String("tqpressure".into()),
1551            PickleValue::Float(pressure(total_height, total_capacity)),
1552        ),
1553        (
1554            PickleValue::String("dqpressure".into()),
1555            PickleValue::Float(pressure(snapshot.heights[0], snapshot.capacities[0])),
1556        ),
1557        (
1558            PickleValue::String("aqpressure".into()),
1559            PickleValue::Float(pressure(snapshot.heights[1], snapshot.capacities[1])),
1560        ),
1561        (
1562            PickleValue::String("pqpressure".into()),
1563            PickleValue::Float(pressure(snapshot.heights[2], snapshot.capacities[2])),
1564        ),
1565        (
1566            PickleValue::String("ilqpressure".into()),
1567            PickleValue::Float(pressure(snapshot.heights[3], snapshot.capacities[3])),
1568        ),
1569        (PickleValue::String("txq".into()), PickleValue::None),
1570    ]);
1571}
1572
1573fn set_pickle_dict_entry(
1574    dict: &mut Vec<(PickleValue, PickleValue)>,
1575    key: &str,
1576    value: PickleValue,
1577) {
1578    if let Some((_, existing)) = dict
1579        .iter_mut()
1580        .find(|(candidate, _)| candidate.as_str() == Some(key))
1581    {
1582        *existing = value;
1583    } else {
1584        dict.push((PickleValue::String(key.into()), value));
1585    }
1586}
1587
1588fn single_iface_to_pickle(s: &SingleInterfaceStat) -> PickleValue {
1589    let mut dict = vec![
1590        (
1591            PickleValue::String("id".into()),
1592            PickleValue::Int(s.id as i64),
1593        ),
1594        (
1595            PickleValue::String("name".into()),
1596            PickleValue::String(s.name.clone()),
1597        ),
1598        (
1599            PickleValue::String("status".into()),
1600            PickleValue::Bool(s.status),
1601        ),
1602        (
1603            PickleValue::String("mode".into()),
1604            PickleValue::Int(s.mode as i64),
1605        ),
1606        (
1607            PickleValue::String("gravity".into()),
1608            PickleValue::Int(s.gravity),
1609        ),
1610        (
1611            PickleValue::String("announces_to_internal".into()),
1612            s.announces_to_internal
1613                .map(PickleValue::Bool)
1614                .unwrap_or(PickleValue::None),
1615        ),
1616        (
1617            PickleValue::String("rxb".into()),
1618            PickleValue::Int(s.rxb as i64),
1619        ),
1620        (
1621            PickleValue::String("txb".into()),
1622            PickleValue::Int(s.txb as i64),
1623        ),
1624        (
1625            PickleValue::String("txdrp".into()),
1626            PickleValue::Int(s.tx_drops as i64),
1627        ),
1628        (
1629            PickleValue::String("txdrb".into()),
1630            PickleValue::Int(s.tx_dropped_bytes as i64),
1631        ),
1632        (
1633            PickleValue::String("txstalled".into()),
1634            PickleValue::Bool(s.tx_stalled),
1635        ),
1636        (
1637            PickleValue::String("txbuffered".into()),
1638            PickleValue::Int(s.tx_buffered as i64),
1639        ),
1640        (
1641            PickleValue::String("rxs".into()),
1642            PickleValue::Float(s.traffic.rxs),
1643        ),
1644        (
1645            PickleValue::String("txs".into()),
1646            PickleValue::Float(s.traffic.txs),
1647        ),
1648        (
1649            PickleValue::String("arxb".into()),
1650            PickleValue::Int(s.traffic.arxb as i64),
1651        ),
1652        (
1653            PickleValue::String("atxb".into()),
1654            PickleValue::Int(s.traffic.atxb as i64),
1655        ),
1656        (
1657            PickleValue::String("arxc".into()),
1658            PickleValue::Int(s.traffic.arxc as i64),
1659        ),
1660        (
1661            PickleValue::String("atxc".into()),
1662            PickleValue::Int(s.traffic.atxc as i64),
1663        ),
1664        (
1665            PickleValue::String("arxs".into()),
1666            PickleValue::Float(s.traffic.arxs),
1667        ),
1668        (
1669            PickleValue::String("atxs".into()),
1670            PickleValue::Float(s.traffic.atxs),
1671        ),
1672        (
1673            PickleValue::String("prxb".into()),
1674            PickleValue::Int(s.traffic.prxb as i64),
1675        ),
1676        (
1677            PickleValue::String("ptxb".into()),
1678            PickleValue::Int(s.traffic.ptxb as i64),
1679        ),
1680        (
1681            PickleValue::String("prxc".into()),
1682            PickleValue::Int(s.traffic.prxc as i64),
1683        ),
1684        (
1685            PickleValue::String("ptxc".into()),
1686            PickleValue::Int(s.traffic.ptxc as i64),
1687        ),
1688        (
1689            PickleValue::String("prxs".into()),
1690            PickleValue::Float(s.traffic.prxs),
1691        ),
1692        (
1693            PickleValue::String("ptxs".into()),
1694            PickleValue::Float(s.traffic.ptxs),
1695        ),
1696        (
1697            PickleValue::String("protocol_violations".into()),
1698            PickleValue::Int(s.protocol_violations as i64),
1699        ),
1700        (
1701            PickleValue::String("ifac_violations".into()),
1702            PickleValue::Int(s.ifac_violations as i64),
1703        ),
1704        (
1705            PickleValue::String("packet_filter_hits".into()),
1706            PickleValue::Int(s.packet_filter_hits as i64),
1707        ),
1708        (
1709            PickleValue::String("rx_packets".into()),
1710            PickleValue::Int(s.rx_packets as i64),
1711        ),
1712        (
1713            PickleValue::String("tx_packets".into()),
1714            PickleValue::Int(s.tx_packets as i64),
1715        ),
1716        (
1717            PickleValue::String("cpu_load".into()),
1718            s.cpu_load
1719                .map(PickleValue::Float)
1720                .unwrap_or(PickleValue::None),
1721        ),
1722        (
1723            PickleValue::String("mem_load".into()),
1724            s.mem_load
1725                .map(PickleValue::Float)
1726                .unwrap_or(PickleValue::None),
1727        ),
1728        (
1729            PickleValue::String("switch_id".into()),
1730            s.switch_id
1731                .map(|v| PickleValue::Bytes(v.to_vec()))
1732                .unwrap_or(PickleValue::None),
1733        ),
1734        (
1735            PickleValue::String("endpoint_id".into()),
1736            s.endpoint_id
1737                .map(|v| PickleValue::Bytes(v.to_vec()))
1738                .unwrap_or(PickleValue::None),
1739        ),
1740        (
1741            PickleValue::String("via_switch_id".into()),
1742            s.via_switch_id
1743                .map(|v| PickleValue::Bytes(v.to_vec()))
1744                .unwrap_or(PickleValue::None),
1745        ),
1746        (
1747            PickleValue::String("peers".into()),
1748            s.peers
1749                .map(|v| PickleValue::Int(v as i64))
1750                .unwrap_or(PickleValue::None),
1751        ),
1752        (
1753            PickleValue::String("started".into()),
1754            PickleValue::Float(s.started),
1755        ),
1756        (
1757            PickleValue::String("ia_freq".into()),
1758            PickleValue::Float(s.ia_freq),
1759        ),
1760        (
1761            PickleValue::String("oa_freq".into()),
1762            PickleValue::Float(s.oa_freq),
1763        ),
1764        (
1765            PickleValue::String("ip_freq".into()),
1766            PickleValue::Float(s.ip_freq),
1767        ),
1768        (
1769            PickleValue::String("op_freq".into()),
1770            PickleValue::Float(s.op_freq),
1771        ),
1772        (
1773            PickleValue::String("burst_active".into()),
1774            PickleValue::Bool(s.burst_active),
1775        ),
1776        (
1777            PickleValue::String("burst_activated".into()),
1778            PickleValue::Float(s.burst_activated),
1779        ),
1780        (
1781            PickleValue::String("pr_burst_active".into()),
1782            PickleValue::Bool(s.pr_burst_active),
1783        ),
1784        (
1785            PickleValue::String("pr_burst_activated".into()),
1786            PickleValue::Float(s.pr_burst_activated),
1787        ),
1788        (
1789            PickleValue::String("clients".into()),
1790            s.clients
1791                .map(|clients| PickleValue::Int(clients as i64))
1792                .unwrap_or(PickleValue::None),
1793        ),
1794        (
1795            PickleValue::String("blocked_ips".into()),
1796            s.blocked_ips
1797                .map(|blocked| PickleValue::Int(blocked as i64))
1798                .unwrap_or(PickleValue::None),
1799        ),
1800        (
1801            PickleValue::String("blocked_ip_list".into()),
1802            s.blocked_ip_list
1803                .as_ref()
1804                .map(|blocked| {
1805                    PickleValue::List(blocked.iter().cloned().map(PickleValue::String).collect())
1806                })
1807                .unwrap_or(PickleValue::None),
1808        ),
1809        (
1810            PickleValue::String("announce_rate_grace".into()),
1811            PickleValue::Int(s.announce_rate_grace as i64),
1812        ),
1813        (
1814            PickleValue::String("announce_rate_penalty".into()),
1815            PickleValue::Float(s.announce_rate_penalty),
1816        ),
1817    ];
1818
1819    match s.announce_rate_target {
1820        Some(target) => dict.push((
1821            PickleValue::String("announce_rate_target".into()),
1822            PickleValue::Float(target),
1823        )),
1824        None => dict.push((
1825            PickleValue::String("announce_rate_target".into()),
1826            PickleValue::None,
1827        )),
1828    }
1829
1830    match s.bitrate {
1831        Some(br) => dict.push((
1832            PickleValue::String("bitrate".into()),
1833            PickleValue::Int(br as i64),
1834        )),
1835        None => dict.push((PickleValue::String("bitrate".into()), PickleValue::None)),
1836    }
1837
1838    match s.mtu {
1839        Some(mtu) => dict.push((
1840            PickleValue::String("mtu".into()),
1841            PickleValue::Int(mtu as i64),
1842        )),
1843        None => dict.push((PickleValue::String("mtu".into()), PickleValue::None)),
1844    }
1845
1846    match s.ifac_size {
1847        Some(sz) => dict.push((
1848            PickleValue::String("ifac_size".into()),
1849            PickleValue::Int(sz as i64),
1850        )),
1851        None => dict.push((PickleValue::String("ifac_size".into()), PickleValue::None)),
1852    }
1853
1854    PickleValue::Dict(dict)
1855}
1856
1857fn path_table_to_pickle(entries: &[PathTableEntry]) -> PickleValue {
1858    let list: Vec<PickleValue> = entries
1859        .iter()
1860        .map(|e| {
1861            PickleValue::Dict(vec![
1862                (
1863                    PickleValue::String("hash".into()),
1864                    PickleValue::Bytes(e.hash.to_vec()),
1865                ),
1866                (
1867                    PickleValue::String("timestamp".into()),
1868                    PickleValue::Float(e.timestamp),
1869                ),
1870                (
1871                    PickleValue::String("via".into()),
1872                    PickleValue::Bytes(e.via.to_vec()),
1873                ),
1874                (
1875                    PickleValue::String("hops".into()),
1876                    PickleValue::Int(e.hops as i64),
1877                ),
1878                (
1879                    PickleValue::String("expires".into()),
1880                    PickleValue::Float(e.expires),
1881                ),
1882                (
1883                    PickleValue::String("interface".into()),
1884                    PickleValue::String(e.interface_name.clone()),
1885                ),
1886            ])
1887        })
1888        .collect();
1889    PickleValue::List(list)
1890}
1891
1892fn rate_table_to_pickle(entries: &[RateTableEntry]) -> PickleValue {
1893    let list: Vec<PickleValue> = entries
1894        .iter()
1895        .map(|e| {
1896            PickleValue::Dict(vec![
1897                (
1898                    PickleValue::String("hash".into()),
1899                    PickleValue::Bytes(e.hash.to_vec()),
1900                ),
1901                (
1902                    PickleValue::String("last".into()),
1903                    PickleValue::Float(e.last),
1904                ),
1905                (
1906                    PickleValue::String("rate_violations".into()),
1907                    PickleValue::Int(e.rate_violations as i64),
1908                ),
1909                (
1910                    PickleValue::String("blocked_until".into()),
1911                    PickleValue::Float(e.blocked_until),
1912                ),
1913                (
1914                    PickleValue::String("timestamps".into()),
1915                    PickleValue::List(
1916                        e.timestamps
1917                            .iter()
1918                            .map(|&t| PickleValue::Float(t))
1919                            .collect(),
1920                    ),
1921                ),
1922            ])
1923        })
1924        .collect();
1925    PickleValue::List(list)
1926}
1927
1928fn blackholed_to_pickle(entries: &[BlackholeInfo]) -> PickleValue {
1929    let list: Vec<PickleValue> = entries
1930        .iter()
1931        .map(|e| {
1932            let mut dict = vec![
1933                (
1934                    PickleValue::String("identity_hash".into()),
1935                    PickleValue::Bytes(e.identity_hash.to_vec()),
1936                ),
1937                (
1938                    PickleValue::String("created".into()),
1939                    PickleValue::Float(e.created),
1940                ),
1941                (
1942                    PickleValue::String("expires".into()),
1943                    PickleValue::Float(e.expires),
1944                ),
1945            ];
1946            if let Some(ref reason) = e.reason {
1947                dict.push((
1948                    PickleValue::String("reason".into()),
1949                    PickleValue::String(reason.clone()),
1950                ));
1951            } else {
1952                dict.push((PickleValue::String("reason".into()), PickleValue::None));
1953            }
1954            PickleValue::Dict(dict)
1955        })
1956        .collect();
1957    PickleValue::List(list)
1958}
1959
1960fn discovered_interfaces_to_pickle(
1961    interfaces: &[crate::discovery::DiscoveredInterface],
1962) -> PickleValue {
1963    let list: Vec<PickleValue> = interfaces
1964        .iter()
1965        .map(|iface| {
1966            let mut dict = vec![
1967                (
1968                    PickleValue::String("type".into()),
1969                    PickleValue::String(iface.interface_type.clone()),
1970                ),
1971                (
1972                    PickleValue::String("impl_name".into()),
1973                    iface
1974                        .impl_name
1975                        .clone()
1976                        .map(PickleValue::String)
1977                        .unwrap_or(PickleValue::None),
1978                ),
1979                (
1980                    PickleValue::String("version".into()),
1981                    iface
1982                        .impl_version
1983                        .clone()
1984                        .map(PickleValue::String)
1985                        .unwrap_or(PickleValue::None),
1986                ),
1987                (
1988                    PickleValue::String("transport".into()),
1989                    PickleValue::Bool(iface.transport),
1990                ),
1991                (
1992                    PickleValue::String("name".into()),
1993                    PickleValue::String(iface.name.clone()),
1994                ),
1995                (
1996                    PickleValue::String("discovered".into()),
1997                    PickleValue::Float(iface.discovered),
1998                ),
1999                (
2000                    PickleValue::String("last_heard".into()),
2001                    PickleValue::Float(iface.last_heard),
2002                ),
2003                (
2004                    PickleValue::String("heard_count".into()),
2005                    PickleValue::Int(iface.heard_count as i64),
2006                ),
2007                (
2008                    PickleValue::String("status".into()),
2009                    PickleValue::String(iface.status.as_str().into()),
2010                ),
2011                (
2012                    PickleValue::String("stamp".into()),
2013                    PickleValue::Bytes(iface.stamp.clone()),
2014                ),
2015                (
2016                    PickleValue::String("value".into()),
2017                    PickleValue::Int(iface.stamp_value as i64),
2018                ),
2019                (
2020                    PickleValue::String("transport_id".into()),
2021                    PickleValue::Bytes(iface.transport_id.to_vec()),
2022                ),
2023                (
2024                    PickleValue::String("network_id".into()),
2025                    PickleValue::Bytes(iface.network_id.to_vec()),
2026                ),
2027                (
2028                    PickleValue::String("hops".into()),
2029                    PickleValue::Int(iface.hops as i64),
2030                ),
2031            ];
2032
2033            // Optional location fields
2034            if let Some(v) = iface.latitude {
2035                dict.push((
2036                    PickleValue::String("latitude".into()),
2037                    PickleValue::Float(v),
2038                ));
2039            } else {
2040                dict.push((PickleValue::String("latitude".into()), PickleValue::None));
2041            }
2042            if let Some(v) = iface.longitude {
2043                dict.push((
2044                    PickleValue::String("longitude".into()),
2045                    PickleValue::Float(v),
2046                ));
2047            } else {
2048                dict.push((PickleValue::String("longitude".into()), PickleValue::None));
2049            }
2050            if let Some(v) = iface.height {
2051                dict.push((PickleValue::String("height".into()), PickleValue::Float(v)));
2052            } else {
2053                dict.push((PickleValue::String("height".into()), PickleValue::None));
2054            }
2055            if let Some(v) = iface.operator_lxmf_address {
2056                dict.push((
2057                    PickleValue::String("operator_lxmf_address".into()),
2058                    PickleValue::String(crate::discovery::hex_encode(&v)),
2059                ));
2060            }
2061
2062            // Connection info
2063            if let Some(ref v) = iface.reachable_on {
2064                dict.push((
2065                    PickleValue::String("reachable_on".into()),
2066                    PickleValue::String(v.clone()),
2067                ));
2068            } else {
2069                dict.push((
2070                    PickleValue::String("reachable_on".into()),
2071                    PickleValue::None,
2072                ));
2073            }
2074            if let Some(v) = iface.port {
2075                dict.push((
2076                    PickleValue::String("port".into()),
2077                    PickleValue::Int(v as i64),
2078                ));
2079            } else {
2080                dict.push((PickleValue::String("port".into()), PickleValue::None));
2081            }
2082
2083            // RNode/RF specific
2084            if let Some(v) = iface.frequency {
2085                dict.push((
2086                    PickleValue::String("frequency".into()),
2087                    PickleValue::Int(v as i64),
2088                ));
2089            } else {
2090                dict.push((PickleValue::String("frequency".into()), PickleValue::None));
2091            }
2092            if let Some(v) = iface.bandwidth {
2093                dict.push((
2094                    PickleValue::String("bandwidth".into()),
2095                    PickleValue::Int(v as i64),
2096                ));
2097            } else {
2098                dict.push((PickleValue::String("bandwidth".into()), PickleValue::None));
2099            }
2100            if let Some(v) = iface.spreading_factor {
2101                dict.push((PickleValue::String("sf".into()), PickleValue::Int(v as i64)));
2102            } else {
2103                dict.push((PickleValue::String("sf".into()), PickleValue::None));
2104            }
2105            if let Some(v) = iface.coding_rate {
2106                dict.push((PickleValue::String("cr".into()), PickleValue::Int(v as i64)));
2107            } else {
2108                dict.push((PickleValue::String("cr".into()), PickleValue::None));
2109            }
2110            if let Some(ref v) = iface.modulation {
2111                dict.push((
2112                    PickleValue::String("modulation".into()),
2113                    PickleValue::String(v.clone()),
2114                ));
2115            } else {
2116                dict.push((PickleValue::String("modulation".into()), PickleValue::None));
2117            }
2118            if let Some(v) = iface.channel {
2119                dict.push((
2120                    PickleValue::String("channel".into()),
2121                    PickleValue::Int(v as i64),
2122                ));
2123            } else {
2124                dict.push((PickleValue::String("channel".into()), PickleValue::None));
2125            }
2126
2127            // IFAC info
2128            if let Some(ref v) = iface.ifac_netname {
2129                dict.push((
2130                    PickleValue::String("ifac_netname".into()),
2131                    PickleValue::String(v.clone()),
2132                ));
2133            } else {
2134                dict.push((
2135                    PickleValue::String("ifac_netname".into()),
2136                    PickleValue::None,
2137                ));
2138            }
2139            if let Some(ref v) = iface.ifac_netkey {
2140                dict.push((
2141                    PickleValue::String("ifac_netkey".into()),
2142                    PickleValue::String(v.clone()),
2143                ));
2144            } else {
2145                dict.push((PickleValue::String("ifac_netkey".into()), PickleValue::None));
2146            }
2147
2148            // Config entry
2149            if let Some(ref v) = iface.config_entry {
2150                dict.push((
2151                    PickleValue::String("config_entry".into()),
2152                    PickleValue::String(v.clone()),
2153                ));
2154            } else {
2155                dict.push((
2156                    PickleValue::String("config_entry".into()),
2157                    PickleValue::None,
2158                ));
2159            }
2160
2161            dict.push((
2162                PickleValue::String("discovery_hash".into()),
2163                PickleValue::Bytes(iface.discovery_hash.to_vec()),
2164            ));
2165
2166            PickleValue::Dict(dict)
2167        })
2168        .collect();
2169    PickleValue::List(list)
2170}
2171
2172fn hooks_to_pickle(hooks: &[HookInfo]) -> PickleValue {
2173    PickleValue::List(
2174        hooks
2175            .iter()
2176            .map(|hook| {
2177                PickleValue::Dict(vec![
2178                    (
2179                        PickleValue::String("name".into()),
2180                        PickleValue::String(hook.name.clone()),
2181                    ),
2182                    (
2183                        PickleValue::String("type".into()),
2184                        PickleValue::String(hook.hook_type.clone()),
2185                    ),
2186                    (
2187                        PickleValue::String("attach_point".into()),
2188                        PickleValue::String(hook.attach_point.clone()),
2189                    ),
2190                    (
2191                        PickleValue::String("priority".into()),
2192                        PickleValue::Int(hook.priority as i64),
2193                    ),
2194                    (
2195                        PickleValue::String("enabled".into()),
2196                        PickleValue::Bool(hook.enabled),
2197                    ),
2198                    (
2199                        PickleValue::String("consecutive_traps".into()),
2200                        PickleValue::Int(hook.consecutive_traps as i64),
2201                    ),
2202                ])
2203            })
2204            .collect(),
2205    )
2206}
2207
2208fn backbone_peer_state_to_pickle(entries: &[BackbonePeerStateEntry]) -> PickleValue {
2209    PickleValue::List(
2210        entries
2211            .iter()
2212            .map(|entry| {
2213                PickleValue::Dict(vec![
2214                    (
2215                        PickleValue::String("interface".into()),
2216                        PickleValue::String(entry.interface_name.clone()),
2217                    ),
2218                    (
2219                        PickleValue::String("ip".into()),
2220                        PickleValue::String(entry.peer_ip.to_string()),
2221                    ),
2222                    (
2223                        PickleValue::String("connected_count".into()),
2224                        PickleValue::Int(entry.connected_count as i64),
2225                    ),
2226                    (
2227                        PickleValue::String("blacklisted_remaining_secs".into()),
2228                        entry
2229                            .blacklisted_remaining_secs
2230                            .map(PickleValue::Float)
2231                            .unwrap_or(PickleValue::None),
2232                    ),
2233                    (
2234                        PickleValue::String("blacklist_reason".into()),
2235                        entry
2236                            .blacklist_reason
2237                            .as_ref()
2238                            .map(|v: &String| PickleValue::String(v.clone()))
2239                            .unwrap_or(PickleValue::None),
2240                    ),
2241                    (
2242                        PickleValue::String("reject_count".into()),
2243                        PickleValue::Int(entry.reject_count as i64),
2244                    ),
2245                ])
2246            })
2247            .collect(),
2248    )
2249}
2250
2251fn backbone_interfaces_to_pickle(entries: &[BackboneInterfaceEntry]) -> PickleValue {
2252    PickleValue::List(
2253        entries
2254            .iter()
2255            .map(|entry| {
2256                PickleValue::Dict(vec![
2257                    (
2258                        PickleValue::String("id".into()),
2259                        PickleValue::Int(entry.interface_id.0 as i64),
2260                    ),
2261                    (
2262                        PickleValue::String("name".into()),
2263                        PickleValue::String(entry.interface_name.clone()),
2264                    ),
2265                ])
2266            })
2267            .collect(),
2268    )
2269}
2270
2271fn provider_bridge_stats_to_pickle(stats: &ProviderBridgeStats) -> PickleValue {
2272    PickleValue::Dict(vec![
2273        (
2274            PickleValue::String("connected".into()),
2275            PickleValue::Bool(stats.connected),
2276        ),
2277        (
2278            PickleValue::String("consumer_count".into()),
2279            PickleValue::Int(stats.consumer_count as i64),
2280        ),
2281        (
2282            PickleValue::String("queue_max_events".into()),
2283            PickleValue::Int(stats.queue_max_events as i64),
2284        ),
2285        (
2286            PickleValue::String("queue_max_bytes".into()),
2287            PickleValue::Int(stats.queue_max_bytes as i64),
2288        ),
2289        (
2290            PickleValue::String("backlog_len".into()),
2291            PickleValue::Int(stats.backlog_len as i64),
2292        ),
2293        (
2294            PickleValue::String("backlog_bytes".into()),
2295            PickleValue::Int(stats.backlog_bytes as i64),
2296        ),
2297        (
2298            PickleValue::String("backlog_dropped_pending".into()),
2299            PickleValue::Int(stats.backlog_dropped_pending as i64),
2300        ),
2301        (
2302            PickleValue::String("backlog_dropped_total".into()),
2303            PickleValue::Int(stats.backlog_dropped_total as i64),
2304        ),
2305        (
2306            PickleValue::String("total_disconnect_count".into()),
2307            PickleValue::Int(stats.total_disconnect_count as i64),
2308        ),
2309        (
2310            PickleValue::String("consumers".into()),
2311            PickleValue::List(
2312                stats
2313                    .consumers
2314                    .iter()
2315                    .map(|consumer| {
2316                        PickleValue::Dict(vec![
2317                            (
2318                                PickleValue::String("id".into()),
2319                                PickleValue::Int(consumer.id as i64),
2320                            ),
2321                            (
2322                                PickleValue::String("connected".into()),
2323                                PickleValue::Bool(consumer.connected),
2324                            ),
2325                            (
2326                                PickleValue::String("queue_len".into()),
2327                                PickleValue::Int(consumer.queue_len as i64),
2328                            ),
2329                            (
2330                                PickleValue::String("queued_bytes".into()),
2331                                PickleValue::Int(consumer.queued_bytes as i64),
2332                            ),
2333                            (
2334                                PickleValue::String("dropped_pending".into()),
2335                                PickleValue::Int(consumer.dropped_pending as i64),
2336                            ),
2337                            (
2338                                PickleValue::String("dropped_total".into()),
2339                                PickleValue::Int(consumer.dropped_total as i64),
2340                            ),
2341                            (
2342                                PickleValue::String("queue_max_events".into()),
2343                                PickleValue::Int(consumer.queue_max_events as i64),
2344                            ),
2345                            (
2346                                PickleValue::String("queue_max_bytes".into()),
2347                                PickleValue::Int(consumer.queue_max_bytes as i64),
2348                            ),
2349                        ])
2350                    })
2351                    .collect(),
2352            ),
2353        ),
2354    ])
2355}
2356
2357fn lifecycle_state_name(state: LifecycleState) -> &'static str {
2358    match state {
2359        LifecycleState::Active => "active",
2360        LifecycleState::Draining => "draining",
2361        LifecycleState::Stopping => "stopping",
2362        LifecycleState::Stopped => "stopped",
2363    }
2364}
2365
2366fn drain_status_to_pickle(status: &DrainStatus) -> PickleValue {
2367    PickleValue::Dict(vec![
2368        (
2369            PickleValue::String("state".into()),
2370            PickleValue::String(lifecycle_state_name(status.state).into()),
2371        ),
2372        (
2373            PickleValue::String("drain_age_seconds".into()),
2374            status
2375                .drain_age_seconds
2376                .map(PickleValue::Float)
2377                .unwrap_or(PickleValue::None),
2378        ),
2379        (
2380            PickleValue::String("deadline_remaining_seconds".into()),
2381            status
2382                .deadline_remaining_seconds
2383                .map(PickleValue::Float)
2384                .unwrap_or(PickleValue::None),
2385        ),
2386        (
2387            PickleValue::String("drain_complete".into()),
2388            PickleValue::Bool(status.drain_complete),
2389        ),
2390        (
2391            PickleValue::String("interface_writer_queued_frames".into()),
2392            PickleValue::Int(status.interface_writer_queued_frames as i64),
2393        ),
2394        (
2395            PickleValue::String("provider_backlog_events".into()),
2396            PickleValue::Int(status.provider_backlog_events as i64),
2397        ),
2398        (
2399            PickleValue::String("provider_consumer_queued_events".into()),
2400            PickleValue::Int(status.provider_consumer_queued_events as i64),
2401        ),
2402        (
2403            PickleValue::String("detail".into()),
2404            status
2405                .detail
2406                .as_ref()
2407                .map(|detail| PickleValue::String(detail.clone()))
2408                .unwrap_or(PickleValue::None),
2409        ),
2410    ])
2411}
2412
2413fn runtime_config_value_to_pickle(value: &RuntimeConfigValue) -> PickleValue {
2414    match value {
2415        RuntimeConfigValue::Int(v) => PickleValue::Int(*v),
2416        RuntimeConfigValue::Float(v) => PickleValue::Float(*v),
2417        RuntimeConfigValue::Bool(v) => PickleValue::Bool(*v),
2418        RuntimeConfigValue::String(v) => PickleValue::String(v.clone()),
2419        RuntimeConfigValue::Null => PickleValue::None,
2420    }
2421}
2422
2423fn runtime_config_value_from_pickle(value: &PickleValue) -> Option<RuntimeConfigValue> {
2424    match value {
2425        PickleValue::Int(v) => Some(RuntimeConfigValue::Int(*v)),
2426        PickleValue::Float(v) => Some(RuntimeConfigValue::Float(*v)),
2427        PickleValue::Bool(v) => Some(RuntimeConfigValue::Bool(*v)),
2428        PickleValue::String(v) => Some(RuntimeConfigValue::String(v.clone())),
2429        PickleValue::None => Some(RuntimeConfigValue::Null),
2430        _ => None,
2431    }
2432}
2433
2434fn runtime_config_entry_to_pickle(entry: &RuntimeConfigEntry) -> PickleValue {
2435    PickleValue::Dict(vec![
2436        (
2437            PickleValue::String("key".into()),
2438            PickleValue::String(entry.key.clone()),
2439        ),
2440        (
2441            PickleValue::String("value".into()),
2442            runtime_config_value_to_pickle(&entry.value),
2443        ),
2444        (
2445            PickleValue::String("default".into()),
2446            runtime_config_value_to_pickle(&entry.default),
2447        ),
2448        (
2449            PickleValue::String("source".into()),
2450            PickleValue::String(match entry.source {
2451                RuntimeConfigSource::Startup => "startup".into(),
2452                RuntimeConfigSource::RuntimeOverride => "runtime_override".into(),
2453            }),
2454        ),
2455        (
2456            PickleValue::String("apply_mode".into()),
2457            PickleValue::String(match entry.apply_mode {
2458                RuntimeConfigApplyMode::Immediate => "immediate".into(),
2459                RuntimeConfigApplyMode::NewConnectionsOnly => "new_connections_only".into(),
2460                RuntimeConfigApplyMode::NextReconnect => "next_reconnect".into(),
2461                RuntimeConfigApplyMode::RestartRequired => "restart_required".into(),
2462            }),
2463        ),
2464        (
2465            PickleValue::String("description".into()),
2466            entry
2467                .description
2468                .as_ref()
2469                .map(|v| PickleValue::String(v.clone()))
2470                .unwrap_or(PickleValue::None),
2471        ),
2472    ])
2473}
2474
2475fn runtime_config_list_to_pickle(entries: &[RuntimeConfigEntry]) -> PickleValue {
2476    PickleValue::List(entries.iter().map(runtime_config_entry_to_pickle).collect())
2477}
2478
2479fn runtime_config_error_to_pickle(error: &RuntimeConfigError) -> PickleValue {
2480    PickleValue::Dict(vec![
2481        (
2482            PickleValue::String("error".into()),
2483            PickleValue::String(match error.code {
2484                RuntimeConfigErrorCode::UnknownKey => "unknown_key".into(),
2485                RuntimeConfigErrorCode::InvalidType => "invalid_type".into(),
2486                RuntimeConfigErrorCode::InvalidValue => "invalid_value".into(),
2487                RuntimeConfigErrorCode::Unsupported => "unsupported".into(),
2488                RuntimeConfigErrorCode::NotFound => "not_found".into(),
2489                RuntimeConfigErrorCode::ApplyFailed => "apply_failed".into(),
2490            }),
2491        ),
2492        (
2493            PickleValue::String("message".into()),
2494            PickleValue::String(error.message.clone()),
2495        ),
2496    ])
2497}
2498
2499fn runtime_config_result_to_pickle(
2500    result: Result<RuntimeConfigEntry, RuntimeConfigError>,
2501) -> PickleValue {
2502    match result {
2503        Ok(entry) => runtime_config_entry_to_pickle(&entry),
2504        Err(error) => runtime_config_error_to_pickle(&error),
2505    }
2506}
2507
2508fn known_destination_entry_to_pickle(entry: &KnownDestinationEntry) -> PickleValue {
2509    PickleValue::Dict(vec![
2510        (
2511            PickleValue::String("dest_hash".into()),
2512            PickleValue::Bytes(entry.dest_hash.to_vec()),
2513        ),
2514        (
2515            PickleValue::String("identity_hash".into()),
2516            PickleValue::Bytes(entry.identity_hash.to_vec()),
2517        ),
2518        (
2519            PickleValue::String("public_key".into()),
2520            PickleValue::Bytes(entry.public_key.to_vec()),
2521        ),
2522        (
2523            PickleValue::String("app_data".into()),
2524            entry
2525                .app_data
2526                .as_ref()
2527                .map(|data: &Vec<u8>| PickleValue::Bytes(data.clone()))
2528                .unwrap_or(PickleValue::None),
2529        ),
2530        (
2531            PickleValue::String("hops".into()),
2532            PickleValue::Int(entry.hops as i64),
2533        ),
2534        (
2535            PickleValue::String("received_at".into()),
2536            PickleValue::Float(entry.received_at),
2537        ),
2538        (
2539            PickleValue::String("receiving_interface".into()),
2540            PickleValue::Int(entry.receiving_interface.0 as i64),
2541        ),
2542        (
2543            PickleValue::String("was_used".into()),
2544            PickleValue::Bool(entry.was_used),
2545        ),
2546        (
2547            PickleValue::String("last_used_at".into()),
2548            entry
2549                .last_used_at
2550                .map(PickleValue::Float)
2551                .unwrap_or(PickleValue::None),
2552        ),
2553        (
2554            PickleValue::String("retained".into()),
2555            PickleValue::Bool(entry.retained),
2556        ),
2557    ])
2558}
2559
2560fn known_destinations_to_pickle(entries: &[KnownDestinationEntry]) -> PickleValue {
2561    PickleValue::List(
2562        entries
2563            .iter()
2564            .map(known_destination_entry_to_pickle)
2565            .collect(),
2566    )
2567}
2568
2569fn parse_known_destination_entry(value: &PickleValue) -> io::Result<KnownDestinationEntry> {
2570    let get_bytes = |key: &str, len: usize| -> io::Result<Vec<u8>> {
2571        let value = value.get(key).ok_or_else(|| {
2572            io::Error::new(io::ErrorKind::InvalidData, format!("missing {}", key))
2573        })?;
2574        let bytes = value.as_bytes().ok_or_else(|| {
2575            io::Error::new(io::ErrorKind::InvalidData, format!("invalid {}", key))
2576        })?;
2577        if bytes.len() != len {
2578            return Err(io::Error::new(
2579                io::ErrorKind::InvalidData,
2580                format!("invalid {} length", key),
2581            ));
2582        }
2583        Ok(bytes.to_vec())
2584    };
2585    let get_int = |key: &str| -> io::Result<i64> {
2586        value
2587            .get(key)
2588            .and_then(|v| v.as_int())
2589            .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, format!("invalid {}", key)))
2590    };
2591    let get_float = |key: &str| -> io::Result<f64> {
2592        value
2593            .get(key)
2594            .and_then(|v| v.as_float())
2595            .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, format!("invalid {}", key)))
2596    };
2597    let get_bool = |key: &str| -> io::Result<bool> {
2598        value
2599            .get(key)
2600            .and_then(|v| v.as_bool())
2601            .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, format!("invalid {}", key)))
2602    };
2603
2604    let mut dest_hash = [0u8; 16];
2605    dest_hash.copy_from_slice(&get_bytes("dest_hash", 16)?);
2606    let mut identity_hash = [0u8; 16];
2607    identity_hash.copy_from_slice(&get_bytes("identity_hash", 16)?);
2608    let mut public_key = [0u8; 64];
2609    public_key.copy_from_slice(&get_bytes("public_key", 64)?);
2610    let app_data = value
2611        .get("app_data")
2612        .and_then(|v| v.as_bytes())
2613        .map(|bytes| bytes.to_vec());
2614    let last_used_at = value.get("last_used_at").and_then(|v| v.as_float());
2615
2616    Ok(KnownDestinationEntry {
2617        dest_hash,
2618        identity_hash,
2619        public_key,
2620        app_data,
2621        hops: get_int("hops")? as u8,
2622        received_at: get_float("received_at")?,
2623        receiving_interface: rns_core::transport::types::InterfaceId(
2624            get_int("receiving_interface")? as u64,
2625        ),
2626        was_used: get_bool("was_used")?,
2627        last_used_at,
2628        retained: get_bool("retained")?,
2629    })
2630}
2631
2632fn parse_known_destination_list(value: &PickleValue) -> io::Result<Vec<KnownDestinationEntry>> {
2633    let list = value
2634        .as_list()
2635        .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "expected list"))?;
2636    list.iter().map(parse_known_destination_entry).collect()
2637}
2638
2639// --- RPC Client ---
2640
2641/// RPC client for connecting to a running daemon.
2642pub struct RpcClient {
2643    stream: TcpStream,
2644}
2645
2646impl RpcClient {
2647    /// Connect to an RPC server and authenticate.
2648    pub fn connect(addr: &RpcAddr, auth_key: &[u8; 32]) -> io::Result<Self> {
2649        let mut stream = match addr {
2650            RpcAddr::Tcp(host, port) => TcpStream::connect((host.as_str(), *port))?,
2651        };
2652
2653        stream.set_read_timeout(Some(std::time::Duration::from_secs(10)))?;
2654        stream.set_write_timeout(Some(std::time::Duration::from_secs(10)))?;
2655
2656        // Client-side authentication
2657        client_auth(&mut stream, auth_key)?;
2658
2659        Ok(RpcClient { stream })
2660    }
2661
2662    /// Send an RPC request and receive a response.
2663    pub fn call(&mut self, request: &PickleValue) -> io::Result<PickleValue> {
2664        let request_bytes = encode_rpc_payload(request, RpcCodec::Msgpack);
2665        send_bytes(&mut self.stream, &request_bytes)?;
2666
2667        let response_bytes = recv_bytes(&mut self.stream)?;
2668        decode_rpc_payload(&response_bytes).map(|(value, _)| value)
2669    }
2670
2671    pub fn list_hooks(&mut self) -> io::Result<Vec<HookInfo>> {
2672        let response = self.call(&PickleValue::Dict(vec![(
2673            PickleValue::String("get".into()),
2674            PickleValue::String("hooks".into()),
2675        )]))?;
2676        parse_hook_list(&response)
2677    }
2678
2679    pub fn begin_drain(&mut self, timeout: Duration) -> io::Result<bool> {
2680        let response = self.call(&PickleValue::Dict(vec![(
2681            PickleValue::String("begin_drain".into()),
2682            PickleValue::Float(timeout.as_secs_f64()),
2683        )]))?;
2684        Ok(response.as_bool().unwrap_or(false))
2685    }
2686
2687    pub fn drain_status(&mut self) -> io::Result<Option<DrainStatus>> {
2688        let response = self.call(&PickleValue::Dict(vec![(
2689            PickleValue::String("get".into()),
2690            PickleValue::String("drain_status".into()),
2691        )]))?;
2692        parse_drain_status(&response)
2693    }
2694
2695    pub fn provider_bridge_stats(&mut self) -> io::Result<PickleValue> {
2696        self.call(&PickleValue::Dict(vec![(
2697            PickleValue::String("get".into()),
2698            PickleValue::String("provider_bridge_stats".into()),
2699        )]))
2700    }
2701
2702    pub fn load_hook(
2703        &mut self,
2704        name: &str,
2705        attach_point: &str,
2706        priority: i32,
2707        wasm: &[u8],
2708    ) -> io::Result<Result<(), String>> {
2709        let response = self.call(&PickleValue::Dict(vec![
2710            (
2711                PickleValue::String("hook".into()),
2712                PickleValue::String("load".into()),
2713            ),
2714            (
2715                PickleValue::String("name".into()),
2716                PickleValue::String(name.to_string()),
2717            ),
2718            (
2719                PickleValue::String("attach_point".into()),
2720                PickleValue::String(attach_point.to_string()),
2721            ),
2722            (
2723                PickleValue::String("priority".into()),
2724                PickleValue::Int(priority as i64),
2725            ),
2726            (
2727                PickleValue::String("wasm".into()),
2728                PickleValue::Bytes(wasm.to_vec()),
2729            ),
2730        ]))?;
2731        parse_hook_result(&response)
2732    }
2733
2734    pub fn load_builtin_hook(
2735        &mut self,
2736        name: &str,
2737        attach_point: &str,
2738        priority: i32,
2739        builtin_id: &str,
2740    ) -> io::Result<Result<(), String>> {
2741        let response = self.call(&PickleValue::Dict(vec![
2742            (
2743                PickleValue::String("hook".into()),
2744                PickleValue::String("load_builtin".into()),
2745            ),
2746            (
2747                PickleValue::String("name".into()),
2748                PickleValue::String(name.to_string()),
2749            ),
2750            (
2751                PickleValue::String("attach_point".into()),
2752                PickleValue::String(attach_point.to_string()),
2753            ),
2754            (
2755                PickleValue::String("priority".into()),
2756                PickleValue::Int(priority as i64),
2757            ),
2758            (
2759                PickleValue::String("builtin_id".into()),
2760                PickleValue::String(builtin_id.to_string()),
2761            ),
2762        ]))?;
2763        parse_hook_result(&response)
2764    }
2765
2766    pub fn unload_hook(
2767        &mut self,
2768        name: &str,
2769        attach_point: &str,
2770    ) -> io::Result<Result<(), String>> {
2771        let response = self.call(&PickleValue::Dict(vec![
2772            (
2773                PickleValue::String("hook".into()),
2774                PickleValue::String("unload".into()),
2775            ),
2776            (
2777                PickleValue::String("name".into()),
2778                PickleValue::String(name.to_string()),
2779            ),
2780            (
2781                PickleValue::String("attach_point".into()),
2782                PickleValue::String(attach_point.to_string()),
2783            ),
2784        ]))?;
2785        parse_hook_result(&response)
2786    }
2787
2788    pub fn set_hook_enabled(
2789        &mut self,
2790        name: &str,
2791        attach_point: &str,
2792        enabled: bool,
2793    ) -> io::Result<Result<(), String>> {
2794        let op = if enabled { "enable" } else { "disable" };
2795        let response = self.call(&PickleValue::Dict(vec![
2796            (
2797                PickleValue::String("hook".into()),
2798                PickleValue::String(op.into()),
2799            ),
2800            (
2801                PickleValue::String("name".into()),
2802                PickleValue::String(name.to_string()),
2803            ),
2804            (
2805                PickleValue::String("attach_point".into()),
2806                PickleValue::String(attach_point.to_string()),
2807            ),
2808        ]))?;
2809        parse_hook_result(&response)
2810    }
2811
2812    pub fn set_hook_priority(
2813        &mut self,
2814        name: &str,
2815        attach_point: &str,
2816        priority: i32,
2817    ) -> io::Result<Result<(), String>> {
2818        let response = self.call(&PickleValue::Dict(vec![
2819            (
2820                PickleValue::String("hook".into()),
2821                PickleValue::String("set_priority".into()),
2822            ),
2823            (
2824                PickleValue::String("name".into()),
2825                PickleValue::String(name.to_string()),
2826            ),
2827            (
2828                PickleValue::String("attach_point".into()),
2829                PickleValue::String(attach_point.to_string()),
2830            ),
2831            (
2832                PickleValue::String("priority".into()),
2833                PickleValue::Int(priority as i64),
2834            ),
2835        ]))?;
2836        parse_hook_result(&response)
2837    }
2838
2839    pub fn blacklist_backbone_peer(
2840        &mut self,
2841        interface: &str,
2842        ip: &str,
2843        duration_secs: u64,
2844        reason: Option<&str>,
2845        penalty_level: Option<u8>,
2846    ) -> io::Result<bool> {
2847        let mut request = vec![
2848            (
2849                PickleValue::String("set".into()),
2850                PickleValue::String("backbone_peer_blacklist".into()),
2851            ),
2852            (
2853                PickleValue::String("interface".into()),
2854                PickleValue::String(interface.to_string()),
2855            ),
2856            (
2857                PickleValue::String("ip".into()),
2858                PickleValue::String(ip.to_string()),
2859            ),
2860            (
2861                PickleValue::String("duration_secs".into()),
2862                PickleValue::Int(duration_secs as i64),
2863            ),
2864        ];
2865        if let Some(reason) = reason {
2866            request.push((
2867                PickleValue::String("reason".into()),
2868                PickleValue::String(reason.to_string()),
2869            ));
2870        }
2871        if let Some(level) = penalty_level {
2872            request.push((
2873                PickleValue::String("penalty_level".into()),
2874                PickleValue::Int(level as i64),
2875            ));
2876        }
2877        let response = self.call(&PickleValue::Dict(request))?;
2878        Ok(response.as_bool().unwrap_or(false))
2879    }
2880
2881    pub fn known_destinations(&mut self) -> io::Result<Vec<KnownDestinationEntry>> {
2882        let response = self.call(&PickleValue::Dict(vec![(
2883            PickleValue::String("get".into()),
2884            PickleValue::String("known_destinations".into()),
2885        )]))?;
2886        parse_known_destination_list(&response)
2887    }
2888
2889    pub fn retain_known_destination(&mut self, dest_hash: [u8; 16]) -> io::Result<bool> {
2890        let response = self.call(&PickleValue::Dict(vec![
2891            (
2892                PickleValue::String("set".into()),
2893                PickleValue::String("known_destination_retained".into()),
2894            ),
2895            (
2896                PickleValue::String("dest_hash".into()),
2897                PickleValue::Bytes(dest_hash.to_vec()),
2898            ),
2899        ]))?;
2900        Ok(response.as_bool().unwrap_or(false))
2901    }
2902
2903    pub fn retain_identity(&mut self, identity_hash: [u8; 16]) -> io::Result<bool> {
2904        let response = self.call(&PickleValue::Dict(vec![
2905            (
2906                PickleValue::String("set".into()),
2907                PickleValue::String("identity_retained".into()),
2908            ),
2909            (
2910                PickleValue::String("identity_hash".into()),
2911                PickleValue::Bytes(identity_hash.to_vec()),
2912            ),
2913        ]))?;
2914        Ok(response.as_bool().unwrap_or(false))
2915    }
2916
2917    pub fn unretain_known_destination(&mut self, dest_hash: [u8; 16]) -> io::Result<bool> {
2918        let response = self.call(&PickleValue::Dict(vec![
2919            (
2920                PickleValue::String("clear".into()),
2921                PickleValue::String("known_destination_retained".into()),
2922            ),
2923            (
2924                PickleValue::String("dest_hash".into()),
2925                PickleValue::Bytes(dest_hash.to_vec()),
2926            ),
2927        ]))?;
2928        Ok(response.as_bool().unwrap_or(false))
2929    }
2930
2931    pub fn mark_known_destination_used(&mut self, dest_hash: [u8; 16]) -> io::Result<bool> {
2932        let response = self.call(&PickleValue::Dict(vec![
2933            (
2934                PickleValue::String("set".into()),
2935                PickleValue::String("known_destination_used".into()),
2936            ),
2937            (
2938                PickleValue::String("dest_hash".into()),
2939                PickleValue::Bytes(dest_hash.to_vec()),
2940            ),
2941        ]))?;
2942        Ok(response.as_bool().unwrap_or(false))
2943    }
2944}
2945
2946fn parse_lifecycle_state(value: &str) -> Option<LifecycleState> {
2947    match value {
2948        "active" => Some(LifecycleState::Active),
2949        "draining" => Some(LifecycleState::Draining),
2950        "stopping" => Some(LifecycleState::Stopping),
2951        "stopped" => Some(LifecycleState::Stopped),
2952        _ => None,
2953    }
2954}
2955
2956fn parse_drain_status(value: &PickleValue) -> io::Result<Option<DrainStatus>> {
2957    if !matches!(value, PickleValue::Dict(_)) {
2958        return Ok(None);
2959    }
2960    let state = value
2961        .get("state")
2962        .and_then(|entry| entry.as_str())
2963        .and_then(parse_lifecycle_state)
2964        .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "missing drain state"))?;
2965    let drain_age_seconds = value.get("drain_age_seconds").and_then(|entry| {
2966        entry
2967            .as_float()
2968            .or_else(|| entry.as_int().map(|v| v as f64))
2969    });
2970    let deadline_remaining_seconds = value.get("deadline_remaining_seconds").and_then(|entry| {
2971        entry
2972            .as_float()
2973            .or_else(|| entry.as_int().map(|v| v as f64))
2974    });
2975    let drain_complete = value
2976        .get("drain_complete")
2977        .and_then(|entry| entry.as_bool())
2978        .unwrap_or(false);
2979    let interface_writer_queued_frames = value
2980        .get("interface_writer_queued_frames")
2981        .and_then(|entry| entry.as_int())
2982        .unwrap_or(0)
2983        .max(0) as usize;
2984    let provider_backlog_events = value
2985        .get("provider_backlog_events")
2986        .and_then(|entry| entry.as_int())
2987        .unwrap_or(0)
2988        .max(0) as usize;
2989    let provider_consumer_queued_events = value
2990        .get("provider_consumer_queued_events")
2991        .and_then(|entry| entry.as_int())
2992        .unwrap_or(0)
2993        .max(0) as usize;
2994    let detail = value
2995        .get("detail")
2996        .and_then(|entry| entry.as_str().map(|v| v.to_string()));
2997    Ok(Some(DrainStatus {
2998        state,
2999        drain_age_seconds,
3000        deadline_remaining_seconds,
3001        drain_complete,
3002        interface_writer_queued_frames,
3003        provider_backlog_events,
3004        provider_consumer_queued_events,
3005        detail,
3006    }))
3007}
3008
3009fn parse_hook_result(response: &PickleValue) -> io::Result<Result<(), String>> {
3010    let ok = response
3011        .get("ok")
3012        .and_then(|v| v.as_bool())
3013        .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "invalid hook response"))?;
3014    if ok {
3015        Ok(Ok(()))
3016    } else {
3017        Ok(Err(response
3018            .get("error")
3019            .and_then(|v| v.as_str())
3020            .unwrap_or("unknown hook error")
3021            .to_string()))
3022    }
3023}
3024
3025fn parse_hook_list(response: &PickleValue) -> io::Result<Vec<HookInfo>> {
3026    let list = response
3027        .as_list()
3028        .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "invalid hooks response"))?;
3029    let mut hooks = Vec::with_capacity(list.len());
3030    for item in list {
3031        hooks.push(HookInfo {
3032            name: item
3033                .get("name")
3034                .and_then(|v| v.as_str())
3035                .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "missing hook name"))?
3036                .to_string(),
3037            hook_type: item
3038                .get("type")
3039                .and_then(|v| v.as_str())
3040                .unwrap_or(default_hook_type())
3041                .to_string(),
3042            attach_point: item
3043                .get("attach_point")
3044                .and_then(|v| v.as_str())
3045                .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "missing attach_point"))?
3046                .to_string(),
3047            priority: item
3048                .get("priority")
3049                .and_then(|v| v.as_int())
3050                .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "missing priority"))?
3051                as i32,
3052            enabled: item
3053                .get("enabled")
3054                .and_then(|v| v.as_bool())
3055                .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "missing enabled"))?,
3056            consecutive_traps: item
3057                .get("consecutive_traps")
3058                .and_then(|v| v.as_int())
3059                .ok_or_else(|| {
3060                    io::Error::new(io::ErrorKind::InvalidData, "missing consecutive_traps")
3061                })? as u32,
3062        });
3063    }
3064    Ok(hooks)
3065}
3066
3067fn default_hook_type() -> &'static str {
3068    #[cfg(feature = "rns-hooks-native")]
3069    {
3070        "native"
3071    }
3072    #[cfg(all(not(feature = "rns-hooks-native"), feature = "rns-hooks-wasm"))]
3073    {
3074        return "wasm";
3075    }
3076    #[cfg(all(not(feature = "rns-hooks-native"), not(feature = "rns-hooks-wasm")))]
3077    {
3078        "wasm"
3079    }
3080}
3081
3082/// Client-side authentication: answer the server's challenge.
3083fn client_auth(stream: &mut TcpStream, auth_key: &[u8; 32]) -> io::Result<()> {
3084    answer_challenge(stream, auth_key)?;
3085    server_auth(stream, auth_key)
3086}
3087
3088fn answer_challenge<S: Read + Write>(stream: &mut S, auth_key: &[u8; 32]) -> io::Result<()> {
3089    // Read challenge
3090    let challenge = recv_bytes(stream)?;
3091
3092    if !challenge.starts_with(CHALLENGE_PREFIX) {
3093        return Err(io::Error::new(
3094            io::ErrorKind::InvalidData,
3095            "expected challenge",
3096        ));
3097    }
3098
3099    let message = &challenge[CHALLENGE_PREFIX.len()..];
3100
3101    // Create HMAC response
3102    let response = create_response(auth_key, message);
3103    send_bytes(stream, &response)?;
3104
3105    // Read welcome/failure
3106    let result = recv_bytes(stream)?;
3107    if result == WELCOME {
3108        Ok(())
3109    } else {
3110        Err(io::Error::new(
3111            io::ErrorKind::PermissionDenied,
3112            "authentication failed",
3113        ))
3114    }
3115}
3116
3117/// Create an HMAC response to a challenge message.
3118fn create_response(auth_key: &[u8; 32], message: &[u8]) -> Vec<u8> {
3119    // Check if message has {sha256} prefix (modern protocol)
3120    if message.starts_with(b"{sha256}") || message.len() > 20 {
3121        // Modern protocol: use HMAC-SHA256 with {sha256} prefix
3122        let digest = hmac_sha256(auth_key, message);
3123        let mut response = Vec::with_capacity(8 + 32);
3124        response.extend_from_slice(b"{sha256}");
3125        response.extend_from_slice(&digest);
3126        response
3127    } else {
3128        // Legacy protocol: raw HMAC-MD5
3129        let digest = hmac_md5(auth_key, message);
3130        digest.to_vec()
3131    }
3132}
3133
3134/// Derive the RPC auth key from transport identity private key.
3135pub fn derive_auth_key(private_key: &[u8]) -> [u8; 32] {
3136    sha256(private_key)
3137}
3138
3139#[cfg(test)]
3140mod tests {
3141    use super::*;
3142
3143    #[test]
3144    fn discovered_interface_rpc_includes_operator_lxmf_address() {
3145        let interface = crate::discovery::DiscoveredInterface {
3146            interface_type: "BackboneInterface".into(),
3147            impl_name: Some("Reticulum".into()),
3148            impl_version: Some("1.5.5".into()),
3149            transport: true,
3150            name: "Operator Test".into(),
3151            discovered: 1.0,
3152            last_heard: 2.0,
3153            heard_count: 1,
3154            status: crate::discovery::DiscoveredStatus::Available,
3155            stamp: vec![0; crate::discovery::STAMP_SIZE],
3156            stamp_value: 16,
3157            transport_id: [0x11; 16],
3158            network_id: [0x22; 16],
3159            hops: 1,
3160            latitude: None,
3161            longitude: None,
3162            height: None,
3163            operator_lxmf_address: Some([0xa5; 16]),
3164            reachable_on: Some("example.com".into()),
3165            port: Some(4242),
3166            frequency: None,
3167            bandwidth: None,
3168            spreading_factor: None,
3169            coding_rate: None,
3170            modulation: None,
3171            channel: None,
3172            ifac_netname: None,
3173            ifac_netkey: None,
3174            config_entry: None,
3175            discovery_hash: [0x33; 32],
3176        };
3177
3178        let encoded = discovered_interfaces_to_pickle(&[interface]);
3179        let first = &encoded.as_list().unwrap()[0];
3180
3181        assert_eq!(
3182            first.get("impl_name").and_then(PickleValue::as_str),
3183            Some("Reticulum")
3184        );
3185        assert_eq!(
3186            first.get("version").and_then(PickleValue::as_str),
3187            Some("1.5.5")
3188        );
3189
3190        assert_eq!(
3191            first
3192                .get("operator_lxmf_address")
3193                .and_then(PickleValue::as_str),
3194            Some("a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5")
3195        );
3196    }
3197
3198    #[test]
3199    fn send_recv_bytes_roundtrip() {
3200        let (mut c1, mut c2) = tcp_pair();
3201        let data = b"hello world";
3202        send_bytes(&mut c1, data).unwrap();
3203        let received = recv_bytes(&mut c2).unwrap();
3204        assert_eq!(&received, data);
3205    }
3206
3207    #[test]
3208    fn send_recv_empty() {
3209        let (mut c1, mut c2) = tcp_pair();
3210        send_bytes(&mut c1, b"").unwrap();
3211        let received = recv_bytes(&mut c2).unwrap();
3212        assert!(received.is_empty());
3213    }
3214
3215    #[test]
3216    fn auth_success() {
3217        let key = derive_auth_key(b"test-private-key");
3218        let (mut server, mut client) = tcp_pair();
3219
3220        let key2 = key;
3221        let t = thread::spawn(move || {
3222            client_auth(&mut client, &key2).unwrap();
3223        });
3224
3225        server_auth(&mut server, &key).unwrap();
3226        answer_challenge(&mut server, &key).unwrap();
3227        t.join().unwrap();
3228    }
3229
3230    #[test]
3231    fn auth_failure_wrong_key() {
3232        let server_key = derive_auth_key(b"server-key");
3233        let client_key = derive_auth_key(b"wrong-key");
3234        let (mut server, mut client) = tcp_pair();
3235
3236        let t = thread::spawn(move || {
3237            let result = client_auth(&mut client, &client_key);
3238            assert!(result.is_err());
3239        });
3240
3241        let result = server_auth(&mut server, &server_key);
3242        assert!(result.is_err());
3243        t.join().unwrap();
3244    }
3245
3246    #[test]
3247    fn verify_sha256_response() {
3248        let key = derive_auth_key(b"mykey");
3249        let message = b"{sha256}abcdefghijklmnopqrstuvwxyz0123456789ABCD";
3250        let response = create_response(&key, message);
3251        assert!(response.starts_with(b"{sha256}"));
3252        assert!(verify_response(&key, message, &response));
3253    }
3254
3255    #[test]
3256    fn verify_legacy_md5_response() {
3257        let key = derive_auth_key(b"mykey");
3258        // Legacy message: 20 bytes, no prefix
3259        let message = b"01234567890123456789";
3260        // Create legacy response (raw HMAC-MD5)
3261        let digest = hmac_md5(&key, message);
3262        assert!(verify_response(&key, message, &digest));
3263    }
3264
3265    #[test]
3266    fn constant_time_eq_works() {
3267        assert!(constant_time_eq(b"hello", b"hello"));
3268        assert!(!constant_time_eq(b"hello", b"world"));
3269        assert!(!constant_time_eq(b"hello", b"hell"));
3270    }
3271
3272    #[test]
3273    fn rpc_payload_accepts_msgpack_and_legacy_pickle() {
3274        let request = PickleValue::Dict(vec![
3275            (
3276                PickleValue::String("get".into()),
3277                PickleValue::String("link_count".into()),
3278            ),
3279            (PickleValue::String("max_hops".into()), PickleValue::Int(3)),
3280        ]);
3281
3282        let msgpack_bytes = encode_rpc_payload(&request, RpcCodec::Msgpack);
3283        let (decoded_msgpack, msgpack_codec) = decode_rpc_payload(&msgpack_bytes).unwrap();
3284        assert_eq!(decoded_msgpack, request);
3285        assert_eq!(msgpack_codec, RpcCodec::Msgpack);
3286
3287        let pickle_bytes = encode_rpc_payload(&request, RpcCodec::Pickle);
3288        let (decoded_pickle, pickle_codec) = decode_rpc_payload(&pickle_bytes).unwrap();
3289        assert_eq!(decoded_pickle, request);
3290        assert_eq!(pickle_codec, RpcCodec::Pickle);
3291    }
3292
3293    #[test]
3294    fn rpc_roundtrip() {
3295        let key = derive_auth_key(b"test-key");
3296        let (event_tx, event_rx) = crate::event::channel();
3297
3298        // Start server
3299        // Bind manually to get the actual port
3300        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
3301        let port = listener.local_addr().unwrap().port();
3302        listener.set_nonblocking(true).unwrap();
3303
3304        let shutdown = Arc::new(AtomicBool::new(false));
3305        let shutdown2 = shutdown.clone();
3306
3307        // Driver thread that handles queries
3308        let driver_thread = thread::spawn(move || loop {
3309            match event_rx.recv_timeout(std::time::Duration::from_secs(5)) {
3310                Ok(Event::Query(request, resp_tx))
3311                    if matches!(*request, QueryRequest::LinkCount) =>
3312                {
3313                    let _ = resp_tx.send(QueryResponse::LinkCount(42));
3314                }
3315                Ok(Event::Query(request, resp_tx))
3316                    if matches!(*request, QueryRequest::ActiveLinkCount) =>
3317                {
3318                    let _ = resp_tx.send(QueryResponse::ActiveLinkCount(7));
3319                }
3320                Ok(Event::Query(request, resp_tx))
3321                    if matches!(*request, QueryRequest::LowestInterfaceBitrate) =>
3322                {
3323                    let _ = resp_tx.send(QueryResponse::LowestInterfaceBitrate(Some(400)));
3324                }
3325                Ok(Event::Query(request, resp_tx))
3326                    if matches!(*request, QueryRequest::MediumPathTimeout) =>
3327                {
3328                    let _ = resp_tx.send(QueryResponse::MediumPathTimeout(26.0));
3329                }
3330                Ok(Event::Query(request, resp_tx))
3331                    if matches!(*request, QueryRequest::FirstHopTimeout { .. }) =>
3332                {
3333                    let QueryRequest::FirstHopTimeout { dest_hash } = *request else {
3334                        unreachable!()
3335                    };
3336                    assert_eq!(dest_hash, [0x77; 16]);
3337                    let _ = resp_tx.send(QueryResponse::FirstHopTimeout(6.4));
3338                }
3339                Ok(Event::Query(request, resp_tx))
3340                    if matches!(*request, QueryRequest::InterfaceStats) =>
3341                {
3342                    let _ = resp_tx.send(QueryResponse::InterfaceStats(InterfaceStatsResponse {
3343                        interfaces: vec![SingleInterfaceStat {
3344                            id: 7,
3345                            name: "TestInterface".into(),
3346                            status: true,
3347                            mode: 1,
3348                            gravity: -2,
3349                            announces_to_internal: Some(true),
3350                            rxb: 1000,
3351                            txb: 2000,
3352                            tx_drops: 0,
3353                            tx_dropped_bytes: 0,
3354                            tx_stalled: false,
3355                            tx_buffered: 0,
3356                            traffic: TrafficDetail::default(),
3357                            protocol_violations: 0,
3358                            ifac_violations: 0,
3359                            packet_filter_hits: 0,
3360                            rx_packets: 10,
3361                            tx_packets: 20,
3362                            cpu_load: None,
3363                            mem_load: None,
3364                            switch_id: None,
3365                            endpoint_id: None,
3366                            via_switch_id: None,
3367                            peers: None,
3368                            bitrate: Some(10_000_000),
3369                            mtu: Some(262_144),
3370                            ifac_size: None,
3371                            started: 1000.0,
3372                            ia_freq: 0.0,
3373                            ip_freq: 0.0,
3374                            op_freq: 0.0,
3375                            op_samples: 0,
3376                            burst_active: false,
3377                            burst_activated: 0.0,
3378                            pr_burst_active: false,
3379                            pr_burst_activated: 0.0,
3380                            oa_freq: 0.0,
3381                            clients: Some(2),
3382                            blocked_ips: None,
3383                            blocked_ip_list: None,
3384                            announce_rate_target: Some(3600.0),
3385                            announce_rate_grace: 5,
3386                            announce_rate_penalty: 0.0,
3387                            interface_type: "TestInterface".into(),
3388                        }],
3389                        transport_id: None,
3390                        transport_enabled: true,
3391                        transport_uptime: 3600.0,
3392                        total_rxb: 1000,
3393                        total_txb: 2000,
3394                        rx_pps: 3.0,
3395                        tx_pps: 4.0,
3396                        traffic: TrafficDetail::default(),
3397                        probe_responder: None,
3398                        backbone_peer_pool: None,
3399                    }));
3400                }
3401                _ => break,
3402            }
3403        });
3404
3405        let key2 = key;
3406        let shutdown3 = shutdown2.clone();
3407        let server_thread = thread::spawn(move || {
3408            rpc_server_loop(listener, key2, event_tx, shutdown3);
3409        });
3410
3411        // Give server time to start
3412        thread::sleep(std::time::Duration::from_millis(50));
3413
3414        // Client: connect and query link count
3415        let server_addr = RpcAddr::Tcp("127.0.0.1".into(), port);
3416        let mut client = RpcClient::connect(&server_addr, &key).unwrap();
3417        let response = client
3418            .call(&PickleValue::Dict(vec![(
3419                PickleValue::String("get".into()),
3420                PickleValue::String("link_count".into()),
3421            )]))
3422            .unwrap();
3423        assert_eq!(response.as_int().unwrap(), 42);
3424        drop(client);
3425
3426        let mut active_client = RpcClient::connect(&server_addr, &key).unwrap();
3427        let active_response = active_client
3428            .call(&PickleValue::Dict(vec![(
3429                PickleValue::String("get".into()),
3430                PickleValue::String("active_link_count".into()),
3431            )]))
3432            .unwrap();
3433        assert_eq!(active_response.as_int().unwrap(), 7);
3434        drop(active_client);
3435
3436        let mut timeout_client = RpcClient::connect(&server_addr, &key).unwrap();
3437        let bitrate = timeout_client
3438            .call(&PickleValue::Dict(vec![(
3439                PickleValue::String("get".into()),
3440                PickleValue::String("lowest_interface_bitrate".into()),
3441            )]))
3442            .unwrap();
3443        assert_eq!(bitrate.as_int(), Some(400));
3444        drop(timeout_client);
3445        let mut timeout_client = RpcClient::connect(&server_addr, &key).unwrap();
3446        let timeout = timeout_client
3447            .call(&PickleValue::Dict(vec![(
3448                PickleValue::String("get".into()),
3449                PickleValue::String("medium_path_timeout".into()),
3450            )]))
3451            .unwrap();
3452        assert_eq!(timeout.as_float(), Some(26.0));
3453        drop(timeout_client);
3454        let mut timeout_client = RpcClient::connect(&server_addr, &key).unwrap();
3455        let first_hop = timeout_client
3456            .call(&PickleValue::Dict(vec![
3457                (
3458                    PickleValue::String("get".into()),
3459                    PickleValue::String("first_hop_timeout".into()),
3460                ),
3461                (
3462                    PickleValue::String("destination_hash".into()),
3463                    PickleValue::Bytes(vec![0x77; 16]),
3464                ),
3465            ]))
3466            .unwrap();
3467        assert_eq!(first_hop.as_float(), Some(6.4));
3468        drop(timeout_client);
3469
3470        // Client: query interface stats
3471        let mut client2 = RpcClient::connect(&server_addr, &key).unwrap();
3472        let response2 = client2
3473            .call(&PickleValue::Dict(vec![(
3474                PickleValue::String("get".into()),
3475                PickleValue::String("interface_stats".into()),
3476            )]))
3477            .unwrap();
3478        let ifaces = response2.get("interfaces").unwrap().as_list().unwrap();
3479        assert_eq!(ifaces.len(), 1);
3480        let iface = &ifaces[0];
3481        assert_eq!(
3482            iface.get("name").unwrap().as_str().unwrap(),
3483            "TestInterface"
3484        );
3485        assert_eq!(iface.get("rxb").unwrap().as_int().unwrap(), 1000);
3486        assert_eq!(iface.get("mtu").unwrap().as_int(), Some(262_144));
3487        drop(client2);
3488
3489        // Shutdown
3490        shutdown2.store(true, Ordering::Relaxed);
3491        server_thread.join().unwrap();
3492        driver_thread.join().unwrap();
3493    }
3494
3495    #[test]
3496    fn full_event_queue_fails_rpc_query_without_blocking() {
3497        let (event_tx, event_rx) = crate::event::channel_with_capacity(1);
3498        event_tx.send(Event::Tick).unwrap();
3499
3500        let (result_tx, result_rx) = mpsc::channel();
3501        let query_thread = thread::spawn(move || {
3502            let _ = result_tx.send(send_query(&event_tx, QueryRequest::LinkCount));
3503        });
3504
3505        let completed_promptly = match result_rx.recv_timeout(Duration::from_millis(250)) {
3506            Ok(result) => {
3507                assert_eq!(result.unwrap_err().kind(), io::ErrorKind::WouldBlock);
3508                true
3509            }
3510            Err(mpsc::RecvTimeoutError::Timeout) => false,
3511            Err(mpsc::RecvTimeoutError::Disconnected) => {
3512                panic!("RPC query thread disconnected without a result")
3513            }
3514        };
3515
3516        // Let the old blocking implementation finish so a regression failure does
3517        // not leak a thread into the rest of the test process.
3518        assert!(matches!(event_rx.recv().unwrap(), Event::Tick));
3519        if !completed_promptly {
3520            match event_rx.recv_timeout(Duration::from_secs(1)).unwrap() {
3521                Event::Query(request, response_tx)
3522                    if matches!(*request, QueryRequest::LinkCount) =>
3523                {
3524                    response_tx.send(QueryResponse::LinkCount(1)).unwrap();
3525                }
3526                other => panic!("unexpected event after releasing queue: {other:?}"),
3527            }
3528            let _ = result_rx.recv_timeout(Duration::from_secs(1)).unwrap();
3529        }
3530        query_thread.join().unwrap();
3531
3532        assert!(
3533            completed_promptly,
3534            "RPC query blocked while the driver event queue was full"
3535        );
3536    }
3537
3538    #[test]
3539    fn stalled_rpc_request_does_not_block_later_clients() {
3540        let key = derive_auth_key(b"concurrent-rpc-test");
3541        let (event_tx, event_rx) = crate::event::channel();
3542
3543        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
3544        let port = listener.local_addr().unwrap().port();
3545        listener.set_nonblocking(true).unwrap();
3546
3547        let shutdown = Arc::new(AtomicBool::new(false));
3548        let server_shutdown = shutdown.clone();
3549        let server_thread = thread::spawn(move || {
3550            rpc_server_loop(listener, key, event_tx, server_shutdown);
3551        });
3552
3553        let (first_seen_tx, first_seen_rx) = mpsc::channel();
3554        let (concurrent_tx, concurrent_rx) = mpsc::channel();
3555        let driver_thread = thread::spawn(move || {
3556            let first_response = match event_rx.recv_timeout(Duration::from_secs(1)).unwrap() {
3557                Event::Query(request, response_tx)
3558                    if matches!(*request, QueryRequest::LinkCount) =>
3559                {
3560                    response_tx
3561                }
3562                other => panic!("unexpected first RPC event: {other:?}"),
3563            };
3564            first_seen_tx.send(()).unwrap();
3565
3566            let second = event_rx.recv_timeout(Duration::from_millis(500));
3567            let handled_concurrently = match second {
3568                Ok(Event::Query(request, response_tx))
3569                    if matches!(*request, QueryRequest::TransportIdentity) =>
3570                {
3571                    response_tx
3572                        .send(QueryResponse::TransportIdentity(Some([7; 16])))
3573                        .unwrap();
3574                    true
3575                }
3576                Ok(other) => panic!("unexpected second RPC event: {other:?}"),
3577                Err(mpsc::RecvTimeoutError::Timeout) => false,
3578                Err(mpsc::RecvTimeoutError::Disconnected) => {
3579                    panic!("RPC event channel disconnected")
3580                }
3581            };
3582
3583            first_response.send(QueryResponse::LinkCount(1)).unwrap();
3584
3585            if !handled_concurrently {
3586                match event_rx.recv_timeout(Duration::from_secs(1)).unwrap() {
3587                    Event::Query(request, response_tx)
3588                        if matches!(*request, QueryRequest::TransportIdentity) =>
3589                    {
3590                        response_tx
3591                            .send(QueryResponse::TransportIdentity(Some([7; 16])))
3592                            .unwrap();
3593                    }
3594                    other => panic!("unexpected delayed RPC event: {other:?}"),
3595                }
3596            }
3597            concurrent_tx.send(handled_concurrently).unwrap();
3598        });
3599
3600        let first_addr = RpcAddr::Tcp("127.0.0.1".into(), port);
3601        let first_client = thread::spawn(move || {
3602            let mut client = RpcClient::connect(&first_addr, &key).unwrap();
3603            client
3604                .call(&PickleValue::Dict(vec![(
3605                    PickleValue::String("get".into()),
3606                    PickleValue::String("link_count".into()),
3607                )]))
3608                .unwrap()
3609        });
3610
3611        first_seen_rx.recv_timeout(Duration::from_secs(1)).unwrap();
3612
3613        let second_addr = RpcAddr::Tcp("127.0.0.1".into(), port);
3614        let second_client = thread::spawn(move || {
3615            let mut client = RpcClient::connect(&second_addr, &key).unwrap();
3616            client
3617                .call(&PickleValue::Dict(vec![(
3618                    PickleValue::String("get".into()),
3619                    PickleValue::String("transport_identity".into()),
3620                )]))
3621                .unwrap()
3622        });
3623
3624        let handled_concurrently = concurrent_rx.recv_timeout(Duration::from_secs(2)).unwrap();
3625        assert_eq!(first_client.join().unwrap().as_int(), Some(1));
3626        assert_eq!(
3627            second_client.join().unwrap().as_bytes(),
3628            Some([7; 16].as_slice())
3629        );
3630
3631        shutdown.store(true, Ordering::Relaxed);
3632        server_thread.join().unwrap();
3633        driver_thread.join().unwrap();
3634
3635        assert!(
3636            handled_concurrently,
3637            "a stalled RPC request prevented the server from handling a later client"
3638        );
3639    }
3640
3641    #[test]
3642    fn derive_auth_key_deterministic() {
3643        let key1 = derive_auth_key(b"test");
3644        let key2 = derive_auth_key(b"test");
3645        assert_eq!(key1, key2);
3646        // Different input → different key
3647        let key3 = derive_auth_key(b"other");
3648        assert_ne!(key1, key3);
3649    }
3650
3651    #[test]
3652    fn pickle_request_handling() {
3653        // Test the request → query translation without networking
3654        let (event_tx, event_rx) = crate::event::channel();
3655
3656        let driver = thread::spawn(move || {
3657            if let Some((QueryRequest::DropPath { dest_hash }, resp_tx)) = event_rx
3658                .recv()
3659                .ok()
3660                .and_then(|event| event.into_query().ok())
3661            {
3662                assert_eq!(dest_hash, [1u8; 16]);
3663                let _ = resp_tx.send(QueryResponse::DropPath(true));
3664            }
3665        });
3666
3667        let request = PickleValue::Dict(vec![
3668            (
3669                PickleValue::String("drop".into()),
3670                PickleValue::String("path".into()),
3671            ),
3672            (
3673                PickleValue::String("destination_hash".into()),
3674                PickleValue::Bytes(vec![1u8; 16]),
3675            ),
3676        ]);
3677
3678        let response = handle_rpc_request(&request, &event_tx).unwrap();
3679        assert_eq!(response, PickleValue::Bool(true));
3680        driver.join().unwrap();
3681    }
3682
3683    #[test]
3684    fn hook_list_request_handling() {
3685        let (event_tx, event_rx) = crate::event::channel();
3686
3687        let driver = thread::spawn(move || {
3688            if let Ok(Event::ListHooks { response_tx }) = event_rx.recv() {
3689                let _ = response_tx.send(vec![HookInfo {
3690                    name: "stats".into(),
3691                    hook_type: "wasm".into(),
3692                    attach_point: "PreIngress".into(),
3693                    priority: 7,
3694                    enabled: true,
3695                    consecutive_traps: 0,
3696                }]);
3697            }
3698        });
3699
3700        let request = PickleValue::Dict(vec![(
3701            PickleValue::String("get".into()),
3702            PickleValue::String("hooks".into()),
3703        )]);
3704        let response = handle_rpc_request(&request, &event_tx).unwrap();
3705        let hooks = parse_hook_list(&response).unwrap();
3706        assert_eq!(hooks.len(), 1);
3707        assert_eq!(hooks[0].name, "stats");
3708        driver.join().unwrap();
3709    }
3710
3711    #[test]
3712    fn hook_load_request_handling() {
3713        let (event_tx, event_rx) = crate::event::channel();
3714
3715        let driver = thread::spawn(move || {
3716            if let Ok(Event::LoadHook {
3717                name,
3718                wasm_bytes,
3719                attach_point,
3720                priority,
3721                response_tx,
3722            }) = event_rx.recv()
3723            {
3724                assert_eq!(name, "stats");
3725                assert_eq!(attach_point, "PreIngress");
3726                assert_eq!(priority, 11);
3727                assert_eq!(wasm_bytes, vec![1, 2, 3]);
3728                let _ = response_tx.send(Ok(()));
3729            }
3730        });
3731
3732        let request = PickleValue::Dict(vec![
3733            (
3734                PickleValue::String("hook".into()),
3735                PickleValue::String("load".into()),
3736            ),
3737            (
3738                PickleValue::String("name".into()),
3739                PickleValue::String("stats".into()),
3740            ),
3741            (
3742                PickleValue::String("attach_point".into()),
3743                PickleValue::String("PreIngress".into()),
3744            ),
3745            (PickleValue::String("priority".into()), PickleValue::Int(11)),
3746            (
3747                PickleValue::String("wasm".into()),
3748                PickleValue::Bytes(vec![1, 2, 3]),
3749            ),
3750        ]);
3751        let response = handle_rpc_request(&request, &event_tx).unwrap();
3752        assert_eq!(parse_hook_result(&response).unwrap(), Ok(()));
3753        driver.join().unwrap();
3754    }
3755
3756    #[test]
3757    fn interface_stats_pickle_format() {
3758        let traffic = TrafficDetail {
3759            rxs: 800.0,
3760            txs: 1600.0,
3761            arxb: 11,
3762            atxb: 22,
3763            arxc: 1,
3764            atxc: 2,
3765            arxs: 88.0,
3766            atxs: 176.0,
3767            prxb: 33,
3768            ptxb: 44,
3769            prxc: 3,
3770            ptxc: 4,
3771            prxs: 264.0,
3772            ptxs: 352.0,
3773            arxf: 1.0,
3774            atxf: 2.0,
3775            prxf: 3.0,
3776            ptxf: 4.0,
3777        };
3778        let mut stats = InterfaceStatsResponse {
3779            interfaces: vec![SingleInterfaceStat {
3780                id: 1,
3781                name: "TCP".into(),
3782                status: true,
3783                mode: 1,
3784                gravity: -2,
3785                announces_to_internal: Some(true),
3786                rxb: 100,
3787                txb: 200,
3788                tx_drops: 3,
3789                tx_dropped_bytes: 4096,
3790                tx_stalled: true,
3791                tx_buffered: 8192,
3792                traffic,
3793                protocol_violations: 11,
3794                ifac_violations: 12,
3795                packet_filter_hits: 13,
3796                rx_packets: 5,
3797                tx_packets: 10,
3798                cpu_load: Some(12.0),
3799                mem_load: Some(34.5),
3800                switch_id: Some([1, 2, 3, 4]),
3801                endpoint_id: Some([5; 8]),
3802                via_switch_id: Some([6; 4]),
3803                peers: Some(2),
3804                bitrate: Some(1000000),
3805                mtu: Some(131_072),
3806                ifac_size: Some(16),
3807                started: 1000.0,
3808                ia_freq: 1.0,
3809                ip_freq: 2.0,
3810                op_freq: 3.0,
3811                op_samples: 0,
3812                burst_active: true,
3813                burst_activated: 1200.0,
3814                pr_burst_active: true,
3815                pr_burst_activated: 1300.0,
3816                oa_freq: 4.0,
3817                clients: Some(3),
3818                blocked_ips: Some(2),
3819                blocked_ip_list: Some(vec!["192.0.2.1".into(), "203.0.113.2".into()]),
3820                announce_rate_target: Some(3600.0),
3821                announce_rate_grace: 5,
3822                announce_rate_penalty: 0.0,
3823                interface_type: "TCPClientInterface".into(),
3824            }],
3825            transport_id: Some([0xAB; 16]),
3826            transport_enabled: true,
3827            transport_uptime: 3600.0,
3828            total_rxb: 100,
3829            total_txb: 200,
3830            rx_pps: 3.0,
3831            tx_pps: 4.0,
3832            traffic,
3833            probe_responder: None,
3834            backbone_peer_pool: None,
3835        };
3836
3837        let pickle = interface_stats_to_pickle(&stats);
3838
3839        // Verify it round-trips through encode/decode
3840        let encoded = pickle::encode(&pickle);
3841        let decoded = pickle::decode(&encoded).unwrap();
3842        assert!(decoded.get("transport_enabled").unwrap().as_bool().unwrap());
3843        let ifaces = decoded.get("interfaces").unwrap().as_list().unwrap();
3844        assert_eq!(decoded.get("arxf").unwrap().as_float(), Some(1.0));
3845        assert_eq!(decoded.get("atxf").unwrap().as_float(), Some(2.0));
3846        assert_eq!(decoded.get("prxf").unwrap().as_float(), Some(3.0));
3847        assert_eq!(decoded.get("ptxf").unwrap().as_float(), Some(4.0));
3848        assert_eq!(decoded.get("rxpps").unwrap().as_float(), Some(3.0));
3849        assert_eq!(decoded.get("txpps").unwrap().as_float(), Some(4.0));
3850        assert_eq!(ifaces[0].get("id").unwrap().as_int().unwrap(), 1);
3851        assert_eq!(ifaces[0].get("name").unwrap().as_str().unwrap(), "TCP");
3852        assert_eq!(ifaces[0].get("mtu").unwrap().as_int(), Some(131_072));
3853        assert_eq!(ifaces[0].get("gravity").unwrap().as_int().unwrap(), -2);
3854        assert_eq!(ifaces[0].get("txdrp").unwrap().as_int(), Some(3));
3855        assert_eq!(ifaces[0].get("txdrb").unwrap().as_int(), Some(4096));
3856        assert_eq!(ifaces[0].get("txstalled").unwrap().as_bool(), Some(true));
3857        assert_eq!(ifaces[0].get("txbuffered").unwrap().as_int(), Some(8192));
3858        assert!(ifaces[0]
3859            .get("announces_to_internal")
3860            .unwrap()
3861            .as_bool()
3862            .unwrap());
3863        assert_eq!(ifaces[0].get("ia_freq").unwrap().as_float().unwrap(), 1.0);
3864        assert_eq!(ifaces[0].get("ip_freq").unwrap().as_float().unwrap(), 2.0);
3865        assert_eq!(ifaces[0].get("op_freq").unwrap().as_float().unwrap(), 3.0);
3866        assert_eq!(ifaces[0].get("oa_freq").unwrap().as_float().unwrap(), 4.0);
3867        assert_eq!(ifaces[0].get("arxb").unwrap().as_int().unwrap(), 11);
3868        assert_eq!(ifaces[0].get("atxb").unwrap().as_int().unwrap(), 22);
3869        assert_eq!(ifaces[0].get("prxb").unwrap().as_int().unwrap(), 33);
3870        assert_eq!(ifaces[0].get("ptxb").unwrap().as_int().unwrap(), 44);
3871        assert_eq!(ifaces[0].get("arxs").unwrap().as_float().unwrap(), 88.0);
3872        assert_eq!(ifaces[0].get("ptxs").unwrap().as_float().unwrap(), 352.0);
3873        assert_eq!(
3874            ifaces[0]
3875                .get("protocol_violations")
3876                .unwrap()
3877                .as_int()
3878                .unwrap(),
3879            11
3880        );
3881        assert_eq!(ifaces[0].get("ifac_violations").unwrap().as_int(), Some(12));
3882        assert_eq!(
3883            ifaces[0].get("packet_filter_hits").unwrap().as_int(),
3884            Some(13)
3885        );
3886        assert_eq!(decoded.get("arxb").unwrap().as_int().unwrap(), 11);
3887        assert_eq!(decoded.get("prxb").unwrap().as_int().unwrap(), 33);
3888        assert!(ifaces[0].get("burst_active").unwrap().as_bool().unwrap());
3889        assert_eq!(
3890            ifaces[0]
3891                .get("burst_activated")
3892                .unwrap()
3893                .as_float()
3894                .unwrap(),
3895            1200.0
3896        );
3897        assert!(ifaces[0].get("pr_burst_active").unwrap().as_bool().unwrap());
3898        assert_eq!(
3899            ifaces[0]
3900                .get("pr_burst_activated")
3901                .unwrap()
3902                .as_float()
3903                .unwrap(),
3904            1300.0
3905        );
3906        assert_eq!(
3907            ifaces[0]
3908                .get("announce_rate_target")
3909                .unwrap()
3910                .as_float()
3911                .unwrap(),
3912            3600.0
3913        );
3914        assert_eq!(
3915            ifaces[0]
3916                .get("announce_rate_grace")
3917                .unwrap()
3918                .as_int()
3919                .unwrap(),
3920            5
3921        );
3922        assert_eq!(
3923            ifaces[0]
3924                .get("announce_rate_penalty")
3925                .unwrap()
3926                .as_float()
3927                .unwrap(),
3928            0.0
3929        );
3930        assert_eq!(ifaces[0].get("clients").unwrap().as_int().unwrap(), 3);
3931        assert_eq!(ifaces[0].get("blocked_ips").unwrap().as_int().unwrap(), 2);
3932        assert_eq!(
3933            ifaces[0]
3934                .get("blocked_ip_list")
3935                .unwrap()
3936                .as_list()
3937                .unwrap()
3938                .iter()
3939                .map(|value| value.as_str().unwrap())
3940                .collect::<Vec<_>>(),
3941            vec!["192.0.2.1", "203.0.113.2"]
3942        );
3943
3944        stats.interfaces[0].interface_type = "BackboneInterface".into();
3945        let (event_tx, _event_rx) = crate::event::channel_with_capacity(1);
3946        event_tx.register_dynamic_parent(InterfaceId(2), InterfaceId(1));
3947        event_tx.set_ingress_bursts(InterfaceId(2), Some(10.0), Some(20.0));
3948        event_tx
3949            .try_send(Event::Frame {
3950                interface_id: InterfaceId(2),
3951                data: vec![0],
3952                rssi: None,
3953                snr: None,
3954            })
3955            .unwrap();
3956        assert!(event_tx
3957            .try_send(Event::Frame {
3958                interface_id: InterfaceId(2),
3959                data: vec![0],
3960                rssi: None,
3961                snr: None,
3962            })
3963            .is_err());
3964        // Reader `send` now waits for space in the data class instead of
3965        // dropping, so a second refused `try_send` provides the second drop.
3966        assert!(event_tx
3967            .try_send(Event::Frame {
3968                interface_id: InterfaceId(2),
3969                data: vec![0],
3970                rssi: None,
3971                snr: None,
3972            })
3973            .is_err());
3974        let mut runtime_pickle = interface_stats_to_pickle(&stats);
3975        add_inbound_runtime_stats(&mut runtime_pickle, &stats, &event_tx);
3976        let runtime_ifaces = runtime_pickle.get("interfaces").unwrap().as_list().unwrap();
3977        assert_eq!(
3978            runtime_ifaces[0]
3979                .get("burst_count")
3980                .unwrap()
3981                .as_int()
3982                .unwrap(),
3983            1
3984        );
3985        assert_eq!(
3986            runtime_ifaces[0]
3987                .get("pr_burst_count")
3988                .unwrap()
3989                .as_int()
3990                .unwrap(),
3991            1
3992        );
3993        assert!(runtime_ifaces[0]
3994            .get("burst_active")
3995            .unwrap()
3996            .as_bool()
3997            .unwrap());
3998        assert_eq!(
3999            runtime_ifaces[0]
4000                .get("burst_activated")
4001                .unwrap()
4002                .as_float()
4003                .unwrap(),
4004            10.0
4005        );
4006        assert!(runtime_ifaces[0]
4007            .get("pr_burst_active")
4008            .unwrap()
4009            .as_bool()
4010            .unwrap());
4011        assert_eq!(
4012            runtime_ifaces[0]
4013                .get("pr_burst_activated")
4014                .unwrap()
4015                .as_float()
4016                .unwrap(),
4017            20.0
4018        );
4019        assert_eq!(runtime_pickle.get("rxqt").unwrap().as_int().unwrap(), 1);
4020        assert_eq!(runtime_pickle.get("rxqd").unwrap().as_int().unwrap(), 1);
4021        assert_eq!(runtime_pickle.get("rxqtd").unwrap().as_int().unwrap(), 2);
4022        assert_eq!(runtime_pickle.get("rxqdd").unwrap().as_int().unwrap(), 2);
4023        assert_eq!(runtime_pickle.get("rxqad").unwrap().as_int().unwrap(), 0);
4024        assert_eq!(runtime_pickle.get("rxqpd").unwrap().as_int().unwrap(), 0);
4025        assert_eq!(runtime_pickle.get("rxqild").unwrap().as_int().unwrap(), 0);
4026        assert_eq!(
4027            runtime_pickle
4028                .get("dqpressure")
4029                .unwrap()
4030                .as_float()
4031                .unwrap(),
4032            1.0
4033        );
4034        assert_eq!(
4035            runtime_pickle
4036                .get("tqpressure")
4037                .unwrap()
4038                .as_float()
4039                .unwrap(),
4040            0.25
4041        );
4042    }
4043
4044    #[test]
4045    fn profiling_results_rpc_exposes_completed_samples() {
4046        {
4047            let _sample = crate::profiling::profile("entry110.rpc");
4048        }
4049        let (event_tx, _event_rx) = crate::event::channel();
4050        let request = PickleValue::Dict(vec![(
4051            PickleValue::String("get".into()),
4052            PickleValue::String("profiling_results".into()),
4053        )]);
4054        let response = handle_rpc_request(&request, &event_tx).unwrap();
4055        let sample = response.get("entry110.rpc").unwrap();
4056        assert_eq!(
4057            sample.get("name").and_then(PickleValue::as_str),
4058            Some("entry110.rpc")
4059        );
4060        assert_eq!(
4061            sample
4062                .get("stats_all")
4063                .and_then(|stats| stats.get("count"))
4064                .and_then(PickleValue::as_int),
4065            Some(1)
4066        );
4067    }
4068
4069    #[test]
4070    fn interface_stats_pickle_includes_backbone_peer_pool_priority() {
4071        let stats = InterfaceStatsResponse {
4072            interfaces: vec![],
4073            transport_id: None,
4074            transport_enabled: true,
4075            transport_uptime: 1.0,
4076            total_rxb: 0,
4077            total_txb: 0,
4078            rx_pps: 0.0,
4079            tx_pps: 0.0,
4080            traffic: TrafficDetail::default(),
4081            probe_responder: None,
4082            backbone_peer_pool: Some(crate::event::BackbonePeerPoolStatus {
4083                max_connected: 1,
4084                active_count: 1,
4085                standby_count: 0,
4086                cooldown_count: 0,
4087                members: vec![crate::event::BackbonePeerPoolMemberStatus {
4088                    name: "peer".into(),
4089                    remote: "127.0.0.1:4242".into(),
4090                    source: "configured".into(),
4091                    priority: 77,
4092                    state: "active".into(),
4093                    interface_id: Some(42),
4094                    failure_count: 0,
4095                    last_error: None,
4096                    cooldown_remaining_seconds: None,
4097                }],
4098            }),
4099        };
4100
4101        let pickle = interface_stats_to_pickle(&stats);
4102        let encoded = pickle::encode(&pickle);
4103        let decoded = pickle::decode(&encoded).unwrap();
4104        let pool = decoded.get("backbone_peer_pool").unwrap();
4105        let members = pool.get("members").unwrap().as_list().unwrap();
4106        assert_eq!(members[0].get("priority").unwrap().as_int().unwrap(), 77);
4107    }
4108
4109    #[test]
4110    fn send_probe_rpc_unknown_dest() {
4111        let (event_tx, event_rx) = crate::event::channel();
4112
4113        let driver = thread::spawn(move || {
4114            if let Some((
4115                QueryRequest::SendProbe {
4116                    dest_hash,
4117                    payload_size,
4118                },
4119                resp_tx,
4120            )) = event_rx
4121                .recv()
4122                .ok()
4123                .and_then(|event| event.into_query().ok())
4124            {
4125                assert_eq!(dest_hash, [0xAA; 16]);
4126                assert_eq!(payload_size, 16); // default
4127                let _ = resp_tx.send(QueryResponse::SendProbe(None));
4128            }
4129        });
4130
4131        let request = PickleValue::Dict(vec![(
4132            PickleValue::String("send_probe".into()),
4133            PickleValue::Bytes(vec![0xAA; 16]),
4134        )]);
4135
4136        let response = handle_rpc_request(&request, &event_tx).unwrap();
4137        assert_eq!(response, PickleValue::None);
4138        driver.join().unwrap();
4139    }
4140
4141    #[test]
4142    fn send_probe_rpc_with_result() {
4143        let (event_tx, event_rx) = crate::event::channel();
4144
4145        let packet_hash = [0xBB; 32];
4146        let driver = thread::spawn(move || {
4147            if let Some((
4148                QueryRequest::SendProbe {
4149                    dest_hash,
4150                    payload_size,
4151                },
4152                resp_tx,
4153            )) = event_rx
4154                .recv()
4155                .ok()
4156                .and_then(|event| event.into_query().ok())
4157            {
4158                assert_eq!(dest_hash, [0xCC; 16]);
4159                assert_eq!(payload_size, 32);
4160                let _ = resp_tx.send(QueryResponse::SendProbe(Some((packet_hash, 3))));
4161            }
4162        });
4163
4164        let request = PickleValue::Dict(vec![
4165            (
4166                PickleValue::String("send_probe".into()),
4167                PickleValue::Bytes(vec![0xCC; 16]),
4168            ),
4169            (PickleValue::String("size".into()), PickleValue::Int(32)),
4170        ]);
4171
4172        let response = handle_rpc_request(&request, &event_tx).unwrap();
4173        let ph = response.get("packet_hash").unwrap().as_bytes().unwrap();
4174        assert_eq!(ph, &[0xBB; 32]);
4175        assert_eq!(response.get("hops").unwrap().as_int().unwrap(), 3);
4176        driver.join().unwrap();
4177    }
4178
4179    #[test]
4180    fn retain_identity_rpc_sends_identity_hash_query() {
4181        let (event_tx, event_rx) = crate::event::channel();
4182        let identity_hash = [0x44; 16];
4183
4184        let driver = thread::spawn(move || {
4185            if let Some((
4186                QueryRequest::RetainIdentity {
4187                    identity_hash: hash,
4188                },
4189                resp_tx,
4190            )) = event_rx
4191                .recv()
4192                .ok()
4193                .and_then(|event| event.into_query().ok())
4194            {
4195                assert_eq!(hash, identity_hash);
4196                let _ = resp_tx.send(QueryResponse::RetainIdentity(true));
4197            }
4198        });
4199
4200        let request = PickleValue::Dict(vec![
4201            (
4202                PickleValue::String("set".into()),
4203                PickleValue::String("identity_retained".into()),
4204            ),
4205            (
4206                PickleValue::String("identity_hash".into()),
4207                PickleValue::Bytes(identity_hash.to_vec()),
4208            ),
4209        ]);
4210
4211        let response = handle_rpc_request(&request, &event_tx).unwrap();
4212        assert_eq!(response, PickleValue::Bool(true));
4213        driver.join().unwrap();
4214    }
4215
4216    #[test]
4217    fn send_probe_rpc_size_validation() {
4218        let (event_tx, event_rx) = crate::event::channel();
4219
4220        // Negative size should be clamped to default (16)
4221        let driver = thread::spawn(move || {
4222            if let Some((QueryRequest::SendProbe { payload_size, .. }, resp_tx)) = event_rx
4223                .recv()
4224                .ok()
4225                .and_then(|event| event.into_query().ok())
4226            {
4227                assert_eq!(payload_size, 16); // default, not negative
4228                let _ = resp_tx.send(QueryResponse::SendProbe(None));
4229            }
4230        });
4231
4232        let request = PickleValue::Dict(vec![
4233            (
4234                PickleValue::String("send_probe".into()),
4235                PickleValue::Bytes(vec![0xDD; 16]),
4236            ),
4237            (PickleValue::String("size".into()), PickleValue::Int(-1)),
4238        ]);
4239
4240        let response = handle_rpc_request(&request, &event_tx).unwrap();
4241        assert_eq!(response, PickleValue::None);
4242        driver.join().unwrap();
4243    }
4244
4245    #[test]
4246    fn send_probe_rpc_size_too_large() {
4247        let (event_tx, event_rx) = crate::event::channel();
4248
4249        // Size > 400 should be clamped to default (16)
4250        let driver = thread::spawn(move || {
4251            if let Some((QueryRequest::SendProbe { payload_size, .. }, resp_tx)) = event_rx
4252                .recv()
4253                .ok()
4254                .and_then(|event| event.into_query().ok())
4255            {
4256                assert_eq!(payload_size, 16); // default, not 999
4257                let _ = resp_tx.send(QueryResponse::SendProbe(None));
4258            }
4259        });
4260
4261        let request = PickleValue::Dict(vec![
4262            (
4263                PickleValue::String("send_probe".into()),
4264                PickleValue::Bytes(vec![0xDD; 16]),
4265            ),
4266            (PickleValue::String("size".into()), PickleValue::Int(999)),
4267        ]);
4268
4269        let response = handle_rpc_request(&request, &event_tx).unwrap();
4270        assert_eq!(response, PickleValue::None);
4271        driver.join().unwrap();
4272    }
4273
4274    #[test]
4275    fn check_proof_rpc_not_found() {
4276        let (event_tx, event_rx) = crate::event::channel();
4277
4278        let driver = thread::spawn(move || {
4279            if let Some((QueryRequest::CheckProof { packet_hash }, resp_tx)) = event_rx
4280                .recv()
4281                .ok()
4282                .and_then(|event| event.into_query().ok())
4283            {
4284                assert_eq!(packet_hash, [0xEE; 32]);
4285                let _ = resp_tx.send(QueryResponse::CheckProof(None));
4286            }
4287        });
4288
4289        let request = PickleValue::Dict(vec![(
4290            PickleValue::String("check_proof".into()),
4291            PickleValue::Bytes(vec![0xEE; 32]),
4292        )]);
4293
4294        let response = handle_rpc_request(&request, &event_tx).unwrap();
4295        assert_eq!(response, PickleValue::None);
4296        driver.join().unwrap();
4297    }
4298
4299    #[test]
4300    fn check_proof_rpc_found() {
4301        let (event_tx, event_rx) = crate::event::channel();
4302
4303        let driver = thread::spawn(move || {
4304            if let Some((QueryRequest::CheckProof { packet_hash }, resp_tx)) = event_rx
4305                .recv()
4306                .ok()
4307                .and_then(|event| event.into_query().ok())
4308            {
4309                assert_eq!(packet_hash, [0xFF; 32]);
4310                let _ = resp_tx.send(QueryResponse::CheckProof(Some(0.352)));
4311            }
4312        });
4313
4314        let request = PickleValue::Dict(vec![(
4315            PickleValue::String("check_proof".into()),
4316            PickleValue::Bytes(vec![0xFF; 32]),
4317        )]);
4318
4319        let response = handle_rpc_request(&request, &event_tx).unwrap();
4320        if let PickleValue::Float(rtt) = response {
4321            assert!((rtt - 0.352).abs() < 0.001);
4322        } else {
4323            panic!("Expected Float, got {:?}", response);
4324        }
4325        driver.join().unwrap();
4326    }
4327
4328    #[test]
4329    fn request_path_rpc() {
4330        let (event_tx, event_rx) = crate::event::channel();
4331
4332        let driver =
4333            thread::spawn(
4334                move || match event_rx.recv_timeout(std::time::Duration::from_secs(5)) {
4335                    Ok(Event::RequestPath { dest_hash }) => {
4336                        assert_eq!(dest_hash, [0x11; 16]);
4337                    }
4338                    other => panic!("Expected RequestPath event, got {:?}", other),
4339                },
4340            );
4341
4342        let request = PickleValue::Dict(vec![(
4343            PickleValue::String("request_path".into()),
4344            PickleValue::Bytes(vec![0x11; 16]),
4345        )]);
4346
4347        let response = handle_rpc_request(&request, &event_tx).unwrap();
4348        assert_eq!(response, PickleValue::Bool(true));
4349        driver.join().unwrap();
4350    }
4351
4352    #[test]
4353    fn begin_drain_rpc_emits_event() {
4354        let (event_tx, event_rx) = crate::event::channel();
4355
4356        let driver = thread::spawn(
4357            move || match event_rx.recv_timeout(Duration::from_secs(5)) {
4358                Ok(Event::BeginDrain { timeout }) => {
4359                    assert!((timeout.as_secs_f64() - 1.5).abs() < 0.001);
4360                }
4361                other => panic!("Expected BeginDrain event, got {:?}", other),
4362            },
4363        );
4364
4365        let request = PickleValue::Dict(vec![(
4366            PickleValue::String("begin_drain".into()),
4367            PickleValue::Float(1.5),
4368        )]);
4369
4370        let response = handle_rpc_request(&request, &event_tx).unwrap();
4371        assert_eq!(response, PickleValue::Bool(true));
4372        driver.join().unwrap();
4373    }
4374
4375    #[test]
4376    fn drain_status_rpc_roundtrips_fields() {
4377        let (event_tx, event_rx) = crate::event::channel();
4378
4379        let driver = thread::spawn(move || {
4380            if let Some((QueryRequest::DrainStatus, resp_tx)) = event_rx
4381                .recv()
4382                .ok()
4383                .and_then(|event| event.into_query().ok())
4384            {
4385                let _ = resp_tx.send(QueryResponse::DrainStatus(DrainStatus {
4386                    state: LifecycleState::Draining,
4387                    drain_age_seconds: Some(0.75),
4388                    deadline_remaining_seconds: Some(2.25),
4389                    drain_complete: false,
4390                    interface_writer_queued_frames: 3,
4391                    provider_backlog_events: 4,
4392                    provider_consumer_queued_events: 5,
4393                    detail: Some("node is draining existing work".into()),
4394                }));
4395            }
4396        });
4397
4398        let request = PickleValue::Dict(vec![(
4399            PickleValue::String("get".into()),
4400            PickleValue::String("drain_status".into()),
4401        )]);
4402
4403        let response = handle_rpc_request(&request, &event_tx).unwrap();
4404        assert_eq!(response.get("state").unwrap().as_str(), Some("draining"));
4405        assert_eq!(
4406            response.get("drain_complete").unwrap().as_bool(),
4407            Some(false)
4408        );
4409        assert_eq!(
4410            response
4411                .get("deadline_remaining_seconds")
4412                .unwrap()
4413                .as_float(),
4414            Some(2.25)
4415        );
4416        assert_eq!(
4417            response
4418                .get("interface_writer_queued_frames")
4419                .unwrap()
4420                .as_int(),
4421            Some(3)
4422        );
4423        assert_eq!(
4424            response.get("provider_backlog_events").unwrap().as_int(),
4425            Some(4)
4426        );
4427        assert_eq!(
4428            response
4429                .get("provider_consumer_queued_events")
4430                .unwrap()
4431                .as_int(),
4432            Some(5)
4433        );
4434        assert_eq!(
4435            response.get("detail").unwrap().as_str(),
4436            Some("node is draining existing work")
4437        );
4438        driver.join().unwrap();
4439    }
4440
4441    #[test]
4442    fn interface_stats_with_probe_responder() {
4443        let probe_hash = [0x42; 16];
4444        let stats = InterfaceStatsResponse {
4445            interfaces: vec![],
4446            transport_id: None,
4447            transport_enabled: true,
4448            transport_uptime: 100.0,
4449            total_rxb: 0,
4450            total_txb: 0,
4451            rx_pps: 0.0,
4452            tx_pps: 0.0,
4453            traffic: TrafficDetail::default(),
4454            probe_responder: Some(probe_hash),
4455            backbone_peer_pool: None,
4456        };
4457
4458        let pickle = interface_stats_to_pickle(&stats);
4459        let encoded = pickle::encode(&pickle);
4460        let decoded = pickle::decode(&encoded).unwrap();
4461
4462        let pr = decoded.get("probe_responder").unwrap().as_bytes().unwrap();
4463        assert_eq!(pr, &probe_hash);
4464    }
4465
4466    #[test]
4467    fn runtime_config_get_and_set_rpc() {
4468        let (event_tx, event_rx) = crate::event::channel();
4469
4470        let driver = thread::spawn(move || {
4471            if let Some((QueryRequest::GetRuntimeConfig { key }, resp_tx)) = event_rx
4472                .recv()
4473                .ok()
4474                .and_then(|event| event.into_query().ok())
4475            {
4476                assert_eq!(key, "global.tick_interval_ms");
4477                let _ = resp_tx.send(QueryResponse::RuntimeConfigEntry(Some(
4478                    RuntimeConfigEntry {
4479                        key,
4480                        value: RuntimeConfigValue::Int(1000),
4481                        default: RuntimeConfigValue::Int(1000),
4482                        source: RuntimeConfigSource::Startup,
4483                        apply_mode: RuntimeConfigApplyMode::Immediate,
4484                        description: Some("tick".into()),
4485                    },
4486                )));
4487            } else {
4488                panic!("expected GetRuntimeConfig query");
4489            }
4490
4491            if let Some((QueryRequest::SetRuntimeConfig { key, value }, resp_tx)) = event_rx
4492                .recv()
4493                .ok()
4494                .and_then(|event| event.into_query().ok())
4495            {
4496                assert_eq!(key, "global.tick_interval_ms");
4497                assert_eq!(value, RuntimeConfigValue::Int(250));
4498                let _ = resp_tx.send(QueryResponse::RuntimeConfigSet(Ok(RuntimeConfigEntry {
4499                    key,
4500                    value: RuntimeConfigValue::Int(250),
4501                    default: RuntimeConfigValue::Int(1000),
4502                    source: RuntimeConfigSource::RuntimeOverride,
4503                    apply_mode: RuntimeConfigApplyMode::Immediate,
4504                    description: Some("tick".into()),
4505                })));
4506            } else {
4507                panic!("expected SetRuntimeConfig query");
4508            }
4509        });
4510
4511        let get_request = PickleValue::Dict(vec![
4512            (
4513                PickleValue::String("get".into()),
4514                PickleValue::String("runtime_config_entry".into()),
4515            ),
4516            (
4517                PickleValue::String("key".into()),
4518                PickleValue::String("global.tick_interval_ms".into()),
4519            ),
4520        ]);
4521        let get_response = handle_rpc_request(&get_request, &event_tx).unwrap();
4522        assert_eq!(
4523            get_response.get("key").and_then(|v| v.as_str()),
4524            Some("global.tick_interval_ms")
4525        );
4526
4527        let set_request = PickleValue::Dict(vec![
4528            (
4529                PickleValue::String("set".into()),
4530                PickleValue::String("runtime_config".into()),
4531            ),
4532            (
4533                PickleValue::String("key".into()),
4534                PickleValue::String("global.tick_interval_ms".into()),
4535            ),
4536            (PickleValue::String("value".into()), PickleValue::Int(250)),
4537        ]);
4538        let set_response = handle_rpc_request(&set_request, &event_tx).unwrap();
4539        assert_eq!(
4540            set_response.get("value").and_then(|v| v.as_int()),
4541            Some(250)
4542        );
4543
4544        driver.join().unwrap();
4545    }
4546
4547    // Helper: create a connected TCP pair
4548    fn tcp_pair() -> (TcpStream, TcpStream) {
4549        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
4550        let port = listener.local_addr().unwrap().port();
4551        let client = TcpStream::connect(("127.0.0.1", port)).unwrap();
4552        let (server, _) = listener.accept().unwrap();
4553        client
4554            .set_read_timeout(Some(std::time::Duration::from_secs(5)))
4555            .unwrap();
4556        server
4557            .set_read_timeout(Some(std::time::Duration::from_secs(5)))
4558            .unwrap();
4559        (server, client)
4560    }
4561}
rns_net/rpc.rs

rns_net/
rpc.rs
