Add ckb-onion integration test

Signed-off-by: Eval EXEC <[email protected]>

ckb-bin/src/subcommand/run.rs

@@ -7,6 +7,7 @@ use ckb_build_info::Version;
 use ckb_launcher::Launcher;
 use ckb_logger::info;
 use ckb_logger::warn;
+use ckb_network::multiaddr::Multiaddr;
 use ckb_resource::{Resource, TemplateContext};
 
 use ckb_stop_handler::{broadcast_exit_signals, wait_all_ckb_services_exit};

network/src/network.rs

@@ -122,7 +122,8 @@ impl NetworkState {
             config.peer_store_path(),
         ));
         info!("Loaded the peer store.");
-        if let Some(ref proxy_url) = config.proxy_config.proxy_url {
+
+        if let Some(ref proxy_url) = config.proxy.proxy_url {
             proxy::check_proxy_url(proxy_url).map_err(|reason| Error::Config(reason))?;
         }
 
@@ -348,6 +349,12 @@ impl NetworkState {
             .collect()
     }
 
+    /// After onion service created,
+    /// ckb use this method to add onion address to public_addr
+    pub fn add_public_addr(&self, addr: Multiaddr) {
+        self.public_addrs.write().insert(addr);
+    }
+
     pub(crate) fn connection_status(&self) -> ConnectionStatus {
         self.peer_registry.read().connection_status()
     }
@@ -954,7 +961,7 @@ impl NetworkService {
             .max_connection_number(1024)
             .set_send_buffer_size(config.max_send_buffer())
             .set_channel_size(config.channel_size())
-            .timeout(Duration::from_secs(5));
+            .timeout(Duration::from_secs(50));
 
         #[cfg(not(target_family = "wasm"))]
         {
@@ -996,10 +1003,11 @@ impl NetworkService {
                     if init.is_ready() {
                         break;
                     }
-                    let proxy_config_enable = config.proxy_config.proxy_url.is_some();
+                    let proxy_config_enable =
+                        config.proxy.proxy_url.is_some() || config.onion.onion_server.is_some();
                     service_builder = service_builder
-                        .tcp_proxy_config(config.proxy_config.proxy_url.clone())
-                        .tcp_onion_config(config.onion_config.onion_server.clone());
+                        .tcp_proxy_config(config.proxy.proxy_url.clone())
+                        .tcp_onion_config(config.onion.onion_server.clone());
 
                     match find_type(multi_addr) {
                         TransportType::Tcp => {
@@ -1329,6 +1337,11 @@ impl NetworkController {
         self.network_state.add_node(&self.p2p_control, address)
     }
 
+    /// Add a public_addr to NetworkState.public_addrs
+    pub fn add_public_addr(&self, public_addr: Multiaddr) {
+        self.network_state.add_public_addr(public_addr)
+    }
+
     /// Disconnect session with peer id
     pub fn remove_node(&self, peer_id: &PeerId) {
         if let Some(session_id) = self

test/Cargo.toml

@@ -33,7 +33,7 @@ ckb-db = { path = "../db", version = "= 0.121.0-pre" }
 ckb-store = { path = "../store", version = "= 0.121.0-pre" }
 ckb-shared = { path = "../shared", version = "= 0.121.0-pre" }
 tempfile = "3"
-reqwest = { version = "0.12", features = ["blocking", "json"] }
+reqwest = { version = "0.12", features = ["blocking", "json", "socks"] }
 rand = "0.8"
 ckb-systemtime = { path = "../util/systemtime", version = "= 0.121.0-pre" }
 serde_json = "1.0"

test/src/main.rs

@@ -606,6 +606,8 @@ fn all_specs() -> Vec<Box<dyn Spec>> {
         Box::new(CheckVmBExtension),
         Box::new(RandomlyKill),
         Box::new(SyncChurn),
+        Box::new(TorServiceContainsPublicAddr),
+        Box::new(TorConnect::new()),
     ];
     specs.shuffle(&mut thread_rng());
     specs

test/src/node.rs

@@ -216,6 +216,19 @@ impl Node {
         self.inner.rpc_listen.clone()
     }
 
+    pub fn get_onion_public_addr(&self) -> Option<String> {
+        let onion_public_addr = self
+            .rpc_client()
+            .local_node_info()
+            .addresses
+            .iter()
+            .filter(|addr| addr.address.contains("/onion3/"))
+            .collect::<Vec<_>>()
+            .first()
+            .map(|addr| addr.address.clone());
+        onion_public_addr
+    }
+
     pub fn p2p_address(&self) -> String {
         format!("{}/p2p/{}", self.p2p_listen(), self.node_id())
     }
@@ -290,6 +303,32 @@ impl Node {
         }
     }
 
+    pub fn connect_onion(&self, peer: &Self) {
+        wait_until(30, || peer.get_onion_public_addr().is_some());
+
+        let onion_pub_address = peer
+            .get_onion_public_addr()
+            .expect("peer onion address is not found");
+
+        info!(
+            "got peer:{}'s onion address: {}",
+            peer.node_id(),
+            onion_pub_address
+        );
+
+        self.rpc_client()
+            .add_node(peer.node_id(), onion_pub_address);
+        let connected = wait_until(6000, || {
+            self.rpc_client()
+                .get_peers()
+                .iter()
+                .any(|p| p.node_id == peer.node_id())
+        });
+        if !connected {
+            panic!("Connect outbound peer timeout, node id: {}", peer.node_id());
+        }
+    }
+
     pub fn connect_uncheck(&self, peer: &Self) {
         self.rpc_client()
             .add_node(peer.node_id(), peer.p2p_listen());

test/src/specs/mod.rs

@@ -8,6 +8,7 @@ mod p2p;
 mod relay;
 mod rpc;
 mod sync;
+mod tor;
 mod tx_pool;
 
 pub use alert::*;
@@ -20,6 +21,7 @@ pub use p2p::*;
 pub use relay::*;
 pub use rpc::*;
 pub use sync::*;
+pub use tor::*;
 pub use tx_pool::*;
 
 use crate::Node;

test/src/specs/tor/mod.rs

@@ -0,0 +1,46 @@
+mod tor_basic;
+mod tor_connect;
+
+use std::process::Child;
+
+use ckb_logger::info;
+pub use tor_basic::*;
+pub use tor_connect::*;
+
+use crate::utils::find_available_port;
+
+#[derive(Clone, Debug)]
+struct TorServer {
+    tor_command_path: String,
+    socks_port: u16,
+    control_port: u16,
+}
+
+impl TorServer {
+    pub fn new() -> Self {
+        TorServer {
+            tor_command_path: std::option_env!("TOR_COMMAND_PATH")
+                .unwrap_or("tor")
+                .to_string(),
+            socks_port: find_available_port(),
+            control_port: find_available_port(),
+        }
+    }
+
+    fn build_tor_args(&self) -> Vec<String> {
+        vec![
+            "--SocksPort".to_string(),
+            self.socks_port.to_string(),
+            "--ControlPort".to_string(),
+            self.control_port.to_string(),
+        ]
+    }
+
+    fn tor_start(&self) -> Child {
+        let mut cmd = std::process::Command::new(&self.tor_command_path);
+        let cmd = cmd.args(self.build_tor_args().clone());
+        let child = cmd.spawn().unwrap();
+        info!("tor started:({:?}) ; pid: {}", &self, child.id());
+        child
+    }
+}

test/src/specs/tor/tor_basic.rs

@@ -0,0 +1,86 @@
+use crate::utils::find_available_port;
+use crate::{Node, Spec};
+use ckb_logger::{error, info};
+use std::process::Child;
+
+use super::TorServer;
+
+// create a sender and receiver for tor_server signal
+static TOR_SERVER_PROCESS: std::sync::LazyLock<std::sync::Mutex<Option<Child>>> =
+    std::sync::LazyLock::new(|| std::sync::Mutex::new(None));
+
+static TOR_SERVER: std::sync::OnceLock<TorServer> = std::sync::OnceLock::new();
+
+struct TorServerGuard {}
+
+impl Drop for TorServerGuard {
+    fn drop(&mut self) {
+        let mut child = TOR_SERVER_PROCESS.lock().unwrap();
+        let child = child.as_mut().unwrap();
+        info!("killing tor server... {}", child.id());
+        match child.kill() {
+            Ok(_) => {
+                info!("tor server exit success");
+            }
+            Err(e) => {
+                error!("tor server exit failed: {:?}", e);
+            }
+        };
+    }
+}
+
+pub struct TorServiceContainsPublicAddr;
+
+impl Spec for TorServiceContainsPublicAddr {
+    crate::setup!(num_nodes: 1);
+
+    fn before_run(&self) -> Vec<Node> {
+        let tor_server = TorServer::new();
+
+        TOR_SERVER.set(tor_server.clone());
+
+        let tor_server_process = tor_server.tor_start();
+        *TOR_SERVER_PROCESS.lock().unwrap() = Some(tor_server_process);
+
+        std::thread::sleep(std::time::Duration::from_secs(5));
+
+        let mut node0 = Node::new(self.name(), "node0");
+        node0.modify_app_config(|config: &mut ckb_app_config::CKBAppConfig| {
+            config.network.onion.listen_on_onion = true;
+            config.network.onion.onion_server =
+                Some(format!("127.0.0.1:{}", tor_server.socks_port));
+            config.network.onion.tor_controller = format!("127.0.0.1:{}", tor_server.control_port);
+        });
+
+        node0.start();
+
+        vec![node0]
+    }
+
+    fn run(&self, nodes: &mut Vec<Node>) {
+        // when _tor_server_guard dropped, the tor server will be killed by Drop
+        let _tor_server_guard = TorServerGuard {};
+
+        let node = &nodes[0];
+
+        let rpc_client = node.rpc_client();
+        let node_info = rpc_client.local_node_info();
+
+        let node_onion_addrs: Vec<_> = node_info
+            .addresses
+            .iter()
+            .filter(|addr| {
+                // check contains the onion address
+                info!("addr: {:?}", addr.address);
+                addr.address.contains("/onion3")
+            })
+            .collect();
+        assert!(
+            !node_onion_addrs.is_empty(),
+            "node should contains onion address"
+        );
+
+        let node_onion_p2p_addr: String = node_onion_addrs.first().unwrap().address.clone();
+        info!("node_onion_p2p_addr: {}", node_onion_p2p_addr);
+    }
+}

test/src/specs/tor/tor_connect.rs

@@ -0,0 +1,110 @@
+use ckb_logger::{error, info};
+
+use crate::{utils::wait_until, Node, Spec};
+
+use super::TorServer;
+
+pub struct TorConnect {
+    tor_server: super::TorServer,
+    tor_server_process: std::process::Child,
+}
+
+impl TorConnect {
+    pub fn new() -> Self {
+        let tor_server = TorServer::new();
+        let tor_server_process = tor_server.tor_start();
+        TorConnect {
+            tor_server,
+            tor_server_process,
+        }
+    }
+}
+
+impl Drop for TorConnect {
+    fn drop(&mut self) {
+        match self.tor_server_process.kill() {
+            Ok(_) => info!("tor server process killed"),
+            Err(e) => error!("tor server process kill failed: {:?}", e),
+        }
+    }
+}
+
+impl Spec for TorConnect {
+    crate::setup!(num_nodes: 3);
+
+    fn before_run(&self) -> Vec<Node> {
+        let mut nodes = (0..self.setup().num_nodes)
+            .map(|i| Node::new(self.name(), &format!("node{i}")))
+            .collect::<Vec<_>>();
+        nodes.iter_mut().for_each(|node| {
+            node.modify_app_config(|config: &mut ckb_app_config::CKBAppConfig| {
+                config.logger.filter =
+                    Some("ckb-network=trace,info".to_string());
+                config.network.onion.listen_on_onion = true;
+
+                // config.network.onion.onion_server = Some(format!("socks5://127.0.0.1:9050"));
+                // config.network.onion.tor_controller = format!("127.0.0.1:9051");
+
+                config.network.onion.onion_server =
+                    Some(format!("socks5://127.0.0.1:{}", self.tor_server.socks_port));
+
+                config.network.onion.tor_controller =
+                    format!("127.0.0.1:{}", self.tor_server.control_port);
+
+                let p2p_addr = config.network.listen_addresses.first().unwrap().to_string();
+
+                let p2p_port: u16 = p2p_addr.split("/tcp/").last().unwrap().parse().unwrap();
+                info!("node p2p listen port: {}", p2p_port);
+
+                config.network.onion.onion_service_target = Some(format!("127.0.0.1:{}", p2p_port));
+            });
+
+            node.start();
+        });
+        nodes
+    }
+
+    fn run(&self, nodes: &mut Vec<crate::Node>) {
+        let node0 = &nodes[0];
+        let node1 = &nodes[1];
+        let node2 = &nodes[2];
+
+        node0.mine_until_out_bootstrap_period();
+        info!("node0 tip: {}", node0.get_tip_block_number());
+        nodes.iter().for_each(|node| node.mine_until_out_ibd_mode());
+
+        info!(
+            "node0 {} connecting to node1 {}",
+            node0.node_id(),
+            node1.node_id()
+        );
+        node1.connect_onion(node0);
+        info!(
+            "node0 {} connecting to node2 {}",
+            node0.node_id(),
+            node2.node_id()
+        );
+        node2.connect_onion(node0);
+
+        info!("node0 and node1 connected, node0 and node2 conencted");
+
+        if wait_until(30, || {
+            let node0_peers = node0.rpc_client().get_peers();
+            let node1_peers = node1.rpc_client().get_peers();
+            let node2_peers = node2.rpc_client().get_peers();
+
+            info!("node0_peers: {:?}", node0_peers);
+            info!("node1_peers: {:?}", node1_peers);
+            info!("node2_peers: {:?}", node2_peers);
+            node1_peers
+                .iter()
+                .map(|peer| peer.node_id.clone())
+                .collect::<Vec<_>>()
+                .contains(&node2.node_id())
+        }) {
+            info!("node1 and node2 connected");
+        } else {
+            panic!("node1 and node2 not connected");
+        }
+    }
+}