Skip to main content

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