rns_net/

rpc.rs

//! RPC server and client for cross-process daemon communication.
//!
//! Implements Python `multiprocessing.connection` wire protocol:
//! - 4-byte big-endian signed i32 length prefix + payload
//! - HMAC-SHA256 challenge-response authentication
//! - Pickle serialization for request/response dictionaries
//!
//! Server translates pickle dicts into [`QueryRequest`] events, sends
//! them through the driver event channel, and returns pickle responses.

use std::io::{self, Read, Write};
use std::net::{TcpListener, TcpStream};
use std::sync::mpsc;
use std::sync::{Arc, atomic::{AtomicBool, Ordering}};
use std::thread;

use rns_crypto::sha256::sha256;
use rns_crypto::hmac::hmac_sha256;

use crate::event::{
    BlackholeInfo, Event, EventSender, QueryRequest, QueryResponse,
    InterfaceStatsResponse, SingleInterfaceStat,
    PathTableEntry, RateTableEntry,
};
use crate::md5::hmac_md5;
use crate::pickle::{self, PickleValue};

const CHALLENGE_PREFIX: &[u8] = b"#CHALLENGE#";
const WELCOME: &[u8] = b"#WELCOME#";
const FAILURE: &[u8] = b"#FAILURE#";
const CHALLENGE_LEN: usize = 40;

/// RPC address types.
#[derive(Debug, Clone)]
pub enum RpcAddr {
    Tcp(String, u16),
}

/// RPC server that listens for incoming connections and handles queries.
pub struct RpcServer {
    shutdown: Arc<AtomicBool>,
    thread: Option<thread::JoinHandle<()>>,
}

impl RpcServer {
    /// Start the RPC server on the given address.
    pub fn start(
        addr: &RpcAddr,
        auth_key: [u8; 32],
        event_tx: EventSender,
    ) -> io::Result<Self> {
        let shutdown = Arc::new(AtomicBool::new(false));
        let shutdown2 = shutdown.clone();

        let listener = match addr {
            RpcAddr::Tcp(host, port) => {
                let l = TcpListener::bind((host.as_str(), *port))?;
                // Non-blocking so we can check shutdown flag
                l.set_nonblocking(true)?;
                l
            }
        };

        let thread = thread::Builder::new()
            .name("rpc-server".into())
            .spawn(move || {
                rpc_server_loop(listener, auth_key, event_tx, shutdown2);
            })
            .map_err(|e| io::Error::new(io::ErrorKind::Other, e))?;

        Ok(RpcServer {
            shutdown,
            thread: Some(thread),
        })
    }

    /// Stop the RPC server.
    pub fn stop(&mut self) {
        self.shutdown.store(true, Ordering::Relaxed);
        if let Some(handle) = self.thread.take() {
            let _ = handle.join();
        }
    }
}

impl Drop for RpcServer {
    fn drop(&mut self) {
        self.stop();
    }
}

fn rpc_server_loop(
    listener: TcpListener,
    auth_key: [u8; 32],
    event_tx: EventSender,
    shutdown: Arc<AtomicBool>,
) {
    loop {
        if shutdown.load(Ordering::Relaxed) {
            break;
        }

        match listener.accept() {
            Ok((stream, _addr)) => {
                // Set blocking for this connection
                let _ = stream.set_nonblocking(false);
                let _ = stream.set_read_timeout(Some(std::time::Duration::from_secs(10)));
                let _ = stream.set_write_timeout(Some(std::time::Duration::from_secs(10)));

                if let Err(e) = handle_connection(stream, &auth_key, &event_tx) {
                    log::debug!("RPC connection error: {}", e);
                }
            }
            Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
                // No pending connection, sleep briefly and retry
                thread::sleep(std::time::Duration::from_millis(100));
            }
            Err(e) => {
                log::error!("RPC accept error: {}", e);
                thread::sleep(std::time::Duration::from_millis(100));
            }
        }
    }
}

fn handle_connection(
    mut stream: TcpStream,
    auth_key: &[u8; 32],
    event_tx: &EventSender,
) -> io::Result<()> {
    // Authentication: send challenge, verify response
    server_auth(&mut stream, auth_key)?;

    // Read request (pickle dict)
    let request_bytes = recv_bytes(&mut stream)?;
    let request = pickle::decode(&request_bytes)
        .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e.to_string()))?;

    // Translate pickle dict to query, send to driver, get response
    let response = handle_rpc_request(&request, event_tx)?;

    // Encode response and send
    let response_bytes = pickle::encode(&response);
    send_bytes(&mut stream, &response_bytes)?;

    Ok(())
}

/// Server-side authentication: challenge-response.
fn server_auth(stream: &mut TcpStream, auth_key: &[u8; 32]) -> io::Result<()> {
    // Generate challenge: #CHALLENGE#{sha256}<40 random bytes>
    let mut random_bytes = [0u8; CHALLENGE_LEN];
    // Use /dev/urandom for randomness
    {
        let mut f = std::fs::File::open("/dev/urandom")?;
        f.read_exact(&mut random_bytes)?;
    }

    let mut challenge_message = Vec::with_capacity(CHALLENGE_PREFIX.len() + 8 + CHALLENGE_LEN);
    challenge_message.extend_from_slice(CHALLENGE_PREFIX);
    challenge_message.extend_from_slice(b"{sha256}");
    challenge_message.extend_from_slice(&random_bytes);

    send_bytes(stream, &challenge_message)?;

    // Read response (max 256 bytes)
    let response = recv_bytes(stream)?;

    // Verify response
    // The message to HMAC is everything after #CHALLENGE# (i.e. {sha256}<random>)
    let message = &challenge_message[CHALLENGE_PREFIX.len()..];

    if verify_response(auth_key, message, &response) {
        send_bytes(stream, WELCOME)?;
        Ok(())
    } else {
        send_bytes(stream, FAILURE)?;
        Err(io::Error::new(io::ErrorKind::PermissionDenied, "auth failed"))
    }
}

/// Verify a client's HMAC response.
fn verify_response(auth_key: &[u8; 32], message: &[u8], response: &[u8]) -> bool {
    // Modern protocol: response = {sha256}<hmac-sha256 digest>
    if response.starts_with(b"{sha256}") {
        let digest = &response[8..];
        let expected = hmac_sha256(auth_key, message);
        constant_time_eq(digest, &expected)
    }
    // Legacy protocol: response = raw 16-byte HMAC-MD5 digest
    else if response.len() == 16 {
        let expected = hmac_md5(auth_key, message);
        constant_time_eq(response, &expected)
    }
    // Legacy with {md5} prefix
    else if response.starts_with(b"{md5}") {
        let digest = &response[5..];
        let expected = hmac_md5(auth_key, message);
        constant_time_eq(digest, &expected)
    } else {
        false
    }
}

/// Constant-time byte comparison.
fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
    if a.len() != b.len() {
        return false;
    }
    let mut diff = 0u8;
    for (x, y) in a.iter().zip(b.iter()) {
        diff |= x ^ y;
    }
    diff == 0
}

/// Send bytes with 4-byte big-endian length prefix.
fn send_bytes(stream: &mut TcpStream, data: &[u8]) -> io::Result<()> {
    let len = data.len() as i32;
    stream.write_all(&len.to_be_bytes())?;
    stream.write_all(data)?;
    stream.flush()
}

/// Receive bytes with 4-byte big-endian length prefix.
fn recv_bytes(stream: &mut TcpStream) -> io::Result<Vec<u8>> {
    let mut len_buf = [0u8; 4];
    stream.read_exact(&mut len_buf)?;
    let len = i32::from_be_bytes(len_buf);

    if len < 0 {
        // Extended format: 8-byte length
        let mut len8_buf = [0u8; 8];
        stream.read_exact(&mut len8_buf)?;
        let len = u64::from_be_bytes(len8_buf) as usize;
        if len > 64 * 1024 * 1024 {
            return Err(io::Error::new(io::ErrorKind::InvalidData, "message too large"));
        }
        let mut buf = vec![0u8; len];
        stream.read_exact(&mut buf)?;
        Ok(buf)
    } else {
        let len = len as usize;
        if len > 64 * 1024 * 1024 {
            return Err(io::Error::new(io::ErrorKind::InvalidData, "message too large"));
        }
        let mut buf = vec![0u8; len];
        stream.read_exact(&mut buf)?;
        Ok(buf)
    }
}

/// Translate a pickle request dict to a query event and get response.
fn handle_rpc_request(
    request: &PickleValue,
    event_tx: &EventSender,
) -> io::Result<PickleValue> {
    // Handle "get" requests
    if let Some(get_val) = request.get("get") {
        if let Some(path) = get_val.as_str() {
            return match path {
                "interface_stats" => {
                    let resp = send_query(event_tx, QueryRequest::InterfaceStats)?;
                    if let QueryResponse::InterfaceStats(stats) = resp {
                        Ok(interface_stats_to_pickle(&stats))
                    } else {
                        Ok(PickleValue::None)
                    }
                }
                "path_table" => {
                    let max_hops = request.get("max_hops").and_then(|v| {
                        v.as_int().map(|n| n as u8)
                    });
                    let resp = send_query(event_tx, QueryRequest::PathTable { max_hops })?;
                    if let QueryResponse::PathTable(entries) = resp {
                        Ok(path_table_to_pickle(&entries))
                    } else {
                        Ok(PickleValue::None)
                    }
                }
                "rate_table" => {
                    let resp = send_query(event_tx, QueryRequest::RateTable)?;
                    if let QueryResponse::RateTable(entries) = resp {
                        Ok(rate_table_to_pickle(&entries))
                    } else {
                        Ok(PickleValue::None)
                    }
                }
                "next_hop" => {
                    let hash = extract_dest_hash(request, "destination_hash")?;
                    let resp = send_query(event_tx, QueryRequest::NextHop { dest_hash: hash })?;
                    if let QueryResponse::NextHop(Some(nh)) = resp {
                        Ok(PickleValue::Bytes(nh.next_hop.to_vec()))
                    } else {
                        Ok(PickleValue::None)
                    }
                }
                "next_hop_if_name" => {
                    let hash = extract_dest_hash(request, "destination_hash")?;
                    let resp = send_query(event_tx, QueryRequest::NextHopIfName { dest_hash: hash })?;
                    if let QueryResponse::NextHopIfName(Some(name)) = resp {
                        Ok(PickleValue::String(name))
                    } else {
                        Ok(PickleValue::None)
                    }
                }
                "link_count" => {
                    let resp = send_query(event_tx, QueryRequest::LinkCount)?;
                    if let QueryResponse::LinkCount(n) = resp {
                        Ok(PickleValue::Int(n as i64))
                    } else {
                        Ok(PickleValue::None)
                    }
                }
                "transport_identity" => {
                    let resp = send_query(event_tx, QueryRequest::TransportIdentity)?;
                    if let QueryResponse::TransportIdentity(Some(hash)) = resp {
                        Ok(PickleValue::Bytes(hash.to_vec()))
                    } else {
                        Ok(PickleValue::None)
                    }
                }
                "blackholed" => {
                    let resp = send_query(event_tx, QueryRequest::GetBlackholed)?;
                    if let QueryResponse::Blackholed(entries) = resp {
                        Ok(blackholed_to_pickle(&entries))
                    } else {
                        Ok(PickleValue::None)
                    }
                }
                _ => Ok(PickleValue::None),
            };
        }
    }

    // Handle "blackhole" requests
    if let Some(hash_val) = request.get("blackhole") {
        if let Some(hash_bytes) = hash_val.as_bytes() {
            if hash_bytes.len() >= 16 {
                let mut identity_hash = [0u8; 16];
                identity_hash.copy_from_slice(&hash_bytes[..16]);
                let duration_hours = request.get("duration").and_then(|v| v.as_float());
                let reason = request.get("reason").and_then(|v| v.as_str()).map(|s| s.to_string());
                let resp = send_query(event_tx, QueryRequest::BlackholeIdentity {
                    identity_hash,
                    duration_hours,
                    reason,
                })?;
                return Ok(PickleValue::Bool(matches!(resp, QueryResponse::BlackholeResult(true))));
            }
        }
    }

    // Handle "unblackhole" requests
    if let Some(hash_val) = request.get("unblackhole") {
        if let Some(hash_bytes) = hash_val.as_bytes() {
            if hash_bytes.len() >= 16 {
                let mut identity_hash = [0u8; 16];
                identity_hash.copy_from_slice(&hash_bytes[..16]);
                let resp = send_query(event_tx, QueryRequest::UnblackholeIdentity {
                    identity_hash,
                })?;
                return Ok(PickleValue::Bool(matches!(resp, QueryResponse::UnblackholeResult(true))));
            }
        }
    }

    // Handle "drop" requests
    if let Some(drop_val) = request.get("drop") {
        if let Some(path) = drop_val.as_str() {
            return match path {
                "path" => {
                    let hash = extract_dest_hash(request, "destination_hash")?;
                    let resp = send_query(event_tx, QueryRequest::DropPath { dest_hash: hash })?;
                    if let QueryResponse::DropPath(ok) = resp {
                        Ok(PickleValue::Bool(ok))
                    } else {
                        Ok(PickleValue::None)
                    }
                }
                "all_via" => {
                    let hash = extract_dest_hash(request, "destination_hash")?;
                    let resp = send_query(event_tx, QueryRequest::DropAllVia { transport_hash: hash })?;
                    if let QueryResponse::DropAllVia(n) = resp {
                        Ok(PickleValue::Int(n as i64))
                    } else {
                        Ok(PickleValue::None)
                    }
                }
                "announce_queues" => {
                    let resp = send_query(event_tx, QueryRequest::DropAnnounceQueues)?;
                    if let QueryResponse::DropAnnounceQueues = resp {
                        Ok(PickleValue::Bool(true))
                    } else {
                        Ok(PickleValue::None)
                    }
                }
                _ => Ok(PickleValue::None),
            };
        }
    }

    Ok(PickleValue::None)
}

/// Send a query to the driver and wait for the response.
fn send_query(event_tx: &EventSender, request: QueryRequest) -> io::Result<QueryResponse> {
    let (resp_tx, resp_rx) = mpsc::channel();
    event_tx
        .send(Event::Query(request, resp_tx))
        .map_err(|_| io::Error::new(io::ErrorKind::BrokenPipe, "driver shut down"))?;
    resp_rx
        .recv_timeout(std::time::Duration::from_secs(5))
        .map_err(|_| io::Error::new(io::ErrorKind::TimedOut, "query timed out"))
}

/// Extract a 16-byte destination hash from a pickle dict field.
fn extract_dest_hash(request: &PickleValue, key: &str) -> io::Result<[u8; 16]> {
    let bytes = request
        .get(key)
        .and_then(|v| v.as_bytes())
        .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "missing destination_hash"))?;
    if bytes.len() < 16 {
        return Err(io::Error::new(io::ErrorKind::InvalidData, "hash too short"));
    }
    let mut hash = [0u8; 16];
    hash.copy_from_slice(&bytes[..16]);
    Ok(hash)
}

// --- Pickle response builders ---

fn interface_stats_to_pickle(stats: &InterfaceStatsResponse) -> PickleValue {
    let mut ifaces = Vec::new();
    for iface in &stats.interfaces {
        ifaces.push(single_iface_to_pickle(iface));
    }

    let mut dict = vec![
        (
            PickleValue::String("interfaces".into()),
            PickleValue::List(ifaces),
        ),
        (
            PickleValue::String("transport_enabled".into()),
            PickleValue::Bool(stats.transport_enabled),
        ),
        (
            PickleValue::String("transport_uptime".into()),
            PickleValue::Float(stats.transport_uptime),
        ),
        (
            PickleValue::String("rxb".into()),
            PickleValue::Int(stats.total_rxb as i64),
        ),
        (
            PickleValue::String("txb".into()),
            PickleValue::Int(stats.total_txb as i64),
        ),
    ];

    if let Some(tid) = stats.transport_id {
        dict.push((
            PickleValue::String("transport_id".into()),
            PickleValue::Bytes(tid.to_vec()),
        ));
    } else {
        dict.push((
            PickleValue::String("transport_id".into()),
            PickleValue::None,
        ));
    }

    PickleValue::Dict(dict)
}

fn single_iface_to_pickle(s: &SingleInterfaceStat) -> PickleValue {
    let mut dict = vec![
        (PickleValue::String("name".into()), PickleValue::String(s.name.clone())),
        (PickleValue::String("status".into()), PickleValue::Bool(s.status)),
        (PickleValue::String("mode".into()), PickleValue::Int(s.mode as i64)),
        (PickleValue::String("rxb".into()), PickleValue::Int(s.rxb as i64)),
        (PickleValue::String("txb".into()), PickleValue::Int(s.txb as i64)),
        (PickleValue::String("rx_packets".into()), PickleValue::Int(s.rx_packets as i64)),
        (PickleValue::String("tx_packets".into()), PickleValue::Int(s.tx_packets as i64)),
        (PickleValue::String("started".into()), PickleValue::Float(s.started)),
        (PickleValue::String("ia_freq".into()), PickleValue::Float(s.ia_freq)),
        (PickleValue::String("oa_freq".into()), PickleValue::Float(s.oa_freq)),
    ];

    match s.bitrate {
        Some(br) => dict.push((
            PickleValue::String("bitrate".into()),
            PickleValue::Int(br as i64),
        )),
        None => dict.push((
            PickleValue::String("bitrate".into()),
            PickleValue::None,
        )),
    }

    match s.ifac_size {
        Some(sz) => dict.push((
            PickleValue::String("ifac_size".into()),
            PickleValue::Int(sz as i64),
        )),
        None => dict.push((
            PickleValue::String("ifac_size".into()),
            PickleValue::None,
        )),
    }

    PickleValue::Dict(dict)
}

fn path_table_to_pickle(entries: &[PathTableEntry]) -> PickleValue {
    let list: Vec<PickleValue> = entries.iter().map(|e| {
        PickleValue::Dict(vec![
            (PickleValue::String("hash".into()), PickleValue::Bytes(e.hash.to_vec())),
            (PickleValue::String("timestamp".into()), PickleValue::Float(e.timestamp)),
            (PickleValue::String("via".into()), PickleValue::Bytes(e.via.to_vec())),
            (PickleValue::String("hops".into()), PickleValue::Int(e.hops as i64)),
            (PickleValue::String("expires".into()), PickleValue::Float(e.expires)),
            (PickleValue::String("interface".into()), PickleValue::String(e.interface_name.clone())),
        ])
    }).collect();
    PickleValue::List(list)
}

fn rate_table_to_pickle(entries: &[RateTableEntry]) -> PickleValue {
    let list: Vec<PickleValue> = entries.iter().map(|e| {
        PickleValue::Dict(vec![
            (PickleValue::String("hash".into()), PickleValue::Bytes(e.hash.to_vec())),
            (PickleValue::String("last".into()), PickleValue::Float(e.last)),
            (PickleValue::String("rate_violations".into()), PickleValue::Int(e.rate_violations as i64)),
            (PickleValue::String("blocked_until".into()), PickleValue::Float(e.blocked_until)),
            (PickleValue::String("timestamps".into()), PickleValue::List(
                e.timestamps.iter().map(|&t| PickleValue::Float(t)).collect()
            )),
        ])
    }).collect();
    PickleValue::List(list)
}

fn blackholed_to_pickle(entries: &[BlackholeInfo]) -> PickleValue {
    let list: Vec<PickleValue> = entries.iter().map(|e| {
        let mut dict = vec![
            (PickleValue::String("identity_hash".into()), PickleValue::Bytes(e.identity_hash.to_vec())),
            (PickleValue::String("created".into()), PickleValue::Float(e.created)),
            (PickleValue::String("expires".into()), PickleValue::Float(e.expires)),
        ];
        if let Some(ref reason) = e.reason {
            dict.push((PickleValue::String("reason".into()), PickleValue::String(reason.clone())));
        } else {
            dict.push((PickleValue::String("reason".into()), PickleValue::None));
        }
        PickleValue::Dict(dict)
    }).collect();
    PickleValue::List(list)
}

// --- RPC Client ---

/// RPC client for connecting to a running daemon.
pub struct RpcClient {
    stream: TcpStream,
}

impl RpcClient {
    /// Connect to an RPC server and authenticate.
    pub fn connect(addr: &RpcAddr, auth_key: &[u8; 32]) -> io::Result<Self> {
        let mut stream = match addr {
            RpcAddr::Tcp(host, port) => {
                TcpStream::connect((host.as_str(), *port))?
            }
        };

        stream.set_read_timeout(Some(std::time::Duration::from_secs(10)))?;
        stream.set_write_timeout(Some(std::time::Duration::from_secs(10)))?;

        // Client-side authentication
        client_auth(&mut stream, auth_key)?;

        Ok(RpcClient { stream })
    }

    /// Send a pickle request and receive a pickle response.
    pub fn call(&mut self, request: &PickleValue) -> io::Result<PickleValue> {
        let request_bytes = pickle::encode(request);
        send_bytes(&mut self.stream, &request_bytes)?;

        let response_bytes = recv_bytes(&mut self.stream)?;
        pickle::decode(&response_bytes)
            .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e.to_string()))
    }
}

/// Client-side authentication: answer the server's challenge.
fn client_auth(stream: &mut TcpStream, auth_key: &[u8; 32]) -> io::Result<()> {
    // Read challenge
    let challenge = recv_bytes(stream)?;

    if !challenge.starts_with(CHALLENGE_PREFIX) {
        return Err(io::Error::new(
            io::ErrorKind::InvalidData,
            "expected challenge",
        ));
    }

    let message = &challenge[CHALLENGE_PREFIX.len()..];

    // Create HMAC response
    let response = create_response(auth_key, message);
    send_bytes(stream, &response)?;

    // Read welcome/failure
    let result = recv_bytes(stream)?;
    if result == WELCOME {
        Ok(())
    } else {
        Err(io::Error::new(
            io::ErrorKind::PermissionDenied,
            "authentication failed",
        ))
    }
}

/// Create an HMAC response to a challenge message.
fn create_response(auth_key: &[u8; 32], message: &[u8]) -> Vec<u8> {
    // Check if message has {sha256} prefix (modern protocol)
    if message.starts_with(b"{sha256}") || message.len() > 20 {
        // Modern protocol: use HMAC-SHA256 with {sha256} prefix
        let digest = hmac_sha256(auth_key, message);
        let mut response = Vec::with_capacity(8 + 32);
        response.extend_from_slice(b"{sha256}");
        response.extend_from_slice(&digest);
        response
    } else {
        // Legacy protocol: raw HMAC-MD5
        let digest = hmac_md5(auth_key, message);
        digest.to_vec()
    }
}

/// Derive the RPC auth key from transport identity private key.
pub fn derive_auth_key(private_key: &[u8]) -> [u8; 32] {
    sha256(private_key)
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn send_recv_bytes_roundtrip() {
        let (mut c1, mut c2) = tcp_pair();
        let data = b"hello world";
        send_bytes(&mut c1, data).unwrap();
        let received = recv_bytes(&mut c2).unwrap();
        assert_eq!(&received, data);
    }

    #[test]
    fn send_recv_empty() {
        let (mut c1, mut c2) = tcp_pair();
        send_bytes(&mut c1, b"").unwrap();
        let received = recv_bytes(&mut c2).unwrap();
        assert!(received.is_empty());
    }

    #[test]
    fn auth_success() {
        let key = derive_auth_key(b"test-private-key");
        let (mut server, mut client) = tcp_pair();

        let key2 = key;
        let t = thread::spawn(move || {
            client_auth(&mut client, &key2).unwrap();
        });

        server_auth(&mut server, &key).unwrap();
        t.join().unwrap();
    }

    #[test]
    fn auth_failure_wrong_key() {
        let server_key = derive_auth_key(b"server-key");
        let client_key = derive_auth_key(b"wrong-key");
        let (mut server, mut client) = tcp_pair();

        let t = thread::spawn(move || {
            let result = client_auth(&mut client, &client_key);
            assert!(result.is_err());
        });

        let result = server_auth(&mut server, &server_key);
        assert!(result.is_err());
        t.join().unwrap();
    }

    #[test]
    fn verify_sha256_response() {
        let key = derive_auth_key(b"mykey");
        let message = b"{sha256}abcdefghijklmnopqrstuvwxyz0123456789ABCD";
        let response = create_response(&key, message);
        assert!(response.starts_with(b"{sha256}"));
        assert!(verify_response(&key, message, &response));
    }

    #[test]
    fn verify_legacy_md5_response() {
        let key = derive_auth_key(b"mykey");
        // Legacy message: 20 bytes, no prefix
        let message = b"01234567890123456789";
        // Create legacy response (raw HMAC-MD5)
        let digest = hmac_md5(&key, message);
        assert!(verify_response(&key, message, &digest));
    }

    #[test]
    fn constant_time_eq_works() {
        assert!(constant_time_eq(b"hello", b"hello"));
        assert!(!constant_time_eq(b"hello", b"world"));
        assert!(!constant_time_eq(b"hello", b"hell"));
    }

    #[test]
    fn rpc_roundtrip() {
        let key = derive_auth_key(b"test-key");
        let (event_tx, event_rx) = crate::event::channel();

        // Start server
        let addr = RpcAddr::Tcp("127.0.0.1".into(), 0);
        // Bind manually to get the actual port
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let port = listener.local_addr().unwrap().port();
        listener.set_nonblocking(true).unwrap();

        let shutdown = Arc::new(AtomicBool::new(false));
        let shutdown2 = shutdown.clone();

        // Driver thread that handles queries
        let driver_thread = thread::spawn(move || {
            loop {
                match event_rx.recv_timeout(std::time::Duration::from_secs(5)) {
                    Ok(Event::Query(QueryRequest::LinkCount, resp_tx)) => {
                        let _ = resp_tx.send(QueryResponse::LinkCount(42));
                    }
                    Ok(Event::Query(QueryRequest::InterfaceStats, resp_tx)) => {
                        let _ = resp_tx.send(QueryResponse::InterfaceStats(InterfaceStatsResponse {
                            interfaces: vec![SingleInterfaceStat {
                                name: "TestInterface".into(),
                                status: true,
                                mode: 1,
                                rxb: 1000,
                                txb: 2000,
                                rx_packets: 10,
                                tx_packets: 20,
                                bitrate: Some(10_000_000),
                                ifac_size: None,
                                started: 1000.0,
                                ia_freq: 0.0,
                                oa_freq: 0.0,
                                interface_type: "TestInterface".into(),
                            }],
                            transport_id: None,
                            transport_enabled: true,
                            transport_uptime: 3600.0,
                            total_rxb: 1000,
                            total_txb: 2000,
                        }));
                    }
                    _ => break,
                }
            }
        });

        let key2 = key;
        let shutdown3 = shutdown2.clone();
        let server_thread = thread::spawn(move || {
            rpc_server_loop(listener, key2, event_tx, shutdown3);
        });

        // Give server time to start
        thread::sleep(std::time::Duration::from_millis(50));

        // Client: connect and query link count
        let server_addr = RpcAddr::Tcp("127.0.0.1".into(), port);
        let mut client = RpcClient::connect(&server_addr, &key).unwrap();
        let response = client.call(&PickleValue::Dict(vec![
            (PickleValue::String("get".into()), PickleValue::String("link_count".into())),
        ])).unwrap();
        assert_eq!(response.as_int().unwrap(), 42);
        drop(client);

        // Client: query interface stats
        let mut client2 = RpcClient::connect(&server_addr, &key).unwrap();
        let response2 = client2.call(&PickleValue::Dict(vec![
            (PickleValue::String("get".into()), PickleValue::String("interface_stats".into())),
        ])).unwrap();
        let ifaces = response2.get("interfaces").unwrap().as_list().unwrap();
        assert_eq!(ifaces.len(), 1);
        let iface = &ifaces[0];
        assert_eq!(iface.get("name").unwrap().as_str().unwrap(), "TestInterface");
        assert_eq!(iface.get("rxb").unwrap().as_int().unwrap(), 1000);
        drop(client2);

        // Shutdown
        shutdown2.store(true, Ordering::Relaxed);
        server_thread.join().unwrap();
        driver_thread.join().unwrap();
    }

    #[test]
    fn derive_auth_key_deterministic() {
        let key1 = derive_auth_key(b"test");
        let key2 = derive_auth_key(b"test");
        assert_eq!(key1, key2);
        // Different input → different key
        let key3 = derive_auth_key(b"other");
        assert_ne!(key1, key3);
    }

    #[test]
    fn pickle_request_handling() {
        // Test the request → query translation without networking
        let (event_tx, event_rx) = crate::event::channel();

        let driver = thread::spawn(move || {
            if let Ok(Event::Query(QueryRequest::DropPath { dest_hash }, resp_tx)) = event_rx.recv() {
                assert_eq!(dest_hash, [1u8; 16]);
                let _ = resp_tx.send(QueryResponse::DropPath(true));
            }
        });

        let request = PickleValue::Dict(vec![
            (PickleValue::String("drop".into()), PickleValue::String("path".into())),
            (PickleValue::String("destination_hash".into()), PickleValue::Bytes(vec![1u8; 16])),
        ]);

        let response = handle_rpc_request(&request, &event_tx).unwrap();
        assert_eq!(response, PickleValue::Bool(true));
        driver.join().unwrap();
    }

    #[test]
    fn interface_stats_pickle_format() {
        let stats = InterfaceStatsResponse {
            interfaces: vec![SingleInterfaceStat {
                name: "TCP".into(),
                status: true,
                mode: 1,
                rxb: 100,
                txb: 200,
                rx_packets: 5,
                tx_packets: 10,
                bitrate: Some(1000000),
                ifac_size: Some(16),
                started: 1000.0,
                ia_freq: 0.0,
                oa_freq: 0.0,
                interface_type: "TCPClientInterface".into(),
            }],
            transport_id: Some([0xAB; 16]),
            transport_enabled: true,
            transport_uptime: 3600.0,
            total_rxb: 100,
            total_txb: 200,
        };

        let pickle = interface_stats_to_pickle(&stats);

        // Verify it round-trips through encode/decode
        let encoded = pickle::encode(&pickle);
        let decoded = pickle::decode(&encoded).unwrap();
        assert_eq!(decoded.get("transport_enabled").unwrap().as_bool().unwrap(), true);
        let ifaces = decoded.get("interfaces").unwrap().as_list().unwrap();
        assert_eq!(ifaces[0].get("name").unwrap().as_str().unwrap(), "TCP");
    }

    // Helper: create a connected TCP pair
    fn tcp_pair() -> (TcpStream, TcpStream) {
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let port = listener.local_addr().unwrap().port();
        let client = TcpStream::connect(("127.0.0.1", port)).unwrap();
        let (server, _) = listener.accept().unwrap();
        client.set_read_timeout(Some(std::time::Duration::from_secs(5))).unwrap();
        server.set_read_timeout(Some(std::time::Duration::from_secs(5))).unwrap();
        (server, client)
    }
}