#![allow(clippy::too_many_arguments)]

use core::panic;
use std::{
    future::Future,
    sync::{Arc, atomic::AtomicU32},
    time::Duration,
};

use easytier_core::{
    connectivity::protocol::raw::TunnelDialer,
    foundation::stats::{LabelSet, LabelType, MetricName, MetricSnapshot},
    process_runtime::CoreProcessRuntime,
    socket::SocketListener,
    tunnel::Tunnel,
};
use rand::{Rng, rngs::OsRng};
use tokio::{net::UdpSocket, task::JoinSet};
use x25519_dalek::StaticSecret;

use super::*;

// TODO: 需要加一个单测，确保 socks5 + exit node == self || proxy_cidr == 0.0.0.0/0 时，可以实现出口节点的能力。

use crate::{
    common::{
        config::{ConfigLoader, NetworkIdentity, PortForwardConfig, TomlConfigLoader},
        netns::{NetNS, ROOT_NETNS_NAME},
    },
    instance::config::test_runtime_instance_config,
    instance::test_instance::TestInstance as Instance,
    proto::{
        api::instance::TcpProxyEntryTransportType,
        common::{CompressionAlgoPb, SecureModeConfig},
        rpc::standalone::{
            RuntimeRpcDialer, RuntimeRpcListener, runtime_rpc_dialer, runtime_rpc_listener,
            runtime_udp_tunnel_dialer, runtime_udp_tunnel_listener,
        },
    },
    tunnel::common::tests::{
        _tunnel_bench_netns, _tunnel_pingpong_netns_with_timeout, wait_for_condition,
    },
};

fn metric_value(metrics: &[MetricSnapshot], name: MetricName, labels: &LabelSet) -> Option<u64> {
    metrics
        .iter()
        .find(|metric| metric.name == name && metric.labels == *labels)
        .map(|metric| metric.value)
}

fn core_tcp_listener(url: url::Url) -> RuntimeRpcListener {
    let addr = url
        .socket_addrs(|| Some(11010))
        .expect("test TCP listener URL should resolve")
        .into_iter()
        .next()
        .expect("test TCP listener URL should have an address");
    runtime_rpc_listener(addr)
}

fn core_tcp_dialer(url: url::Url) -> RuntimeRpcDialer {
    runtime_rpc_dialer(url)
}

fn core_udp_listener(url: url::Url) -> impl SocketListener<Accepted = Box<dyn Tunnel>> + Sync {
    let addr = url
        .socket_addrs(|| Some(11010))
        .expect("test UDP listener URL should resolve")
        .into_iter()
        .next()
        .expect("test UDP listener URL should have an address");
    runtime_udp_tunnel_listener(url, addr)
}

fn core_udp_dialer(url: url::Url) -> impl TunnelDialer {
    runtime_udp_tunnel_dialer(url)
}

async fn reload_instance_acl(inst: &Instance, acl: Option<&crate::proto::acl::Acl>) {
    let mut config = test_runtime_instance_config(&inst.get_global_ctx());
    config.services.acl = easytier_core::config::peers::AclRuleConfig {
        acl: acl.cloned(),
        ..Default::default()
    };
    inst.get_core_instance()
        .update_runtime_config(config)
        .await
        .unwrap();
}

async fn set_foreign_network_refresh_interval(inst: &Instance, seconds: u64) {
    let mut config = test_runtime_instance_config(&inst.get_global_ctx());
    Arc::make_mut(&mut config.peer).ospf_update_my_foreign_network_interval_sec = seconds;
    inst.get_core_instance()
        .update_runtime_config(config)
        .await
        .unwrap();
}

#[cfg(feature = "wireguard")]
use crate::{common::config::VpnPortalConfig, vpn_portal::wireguard::get_wg_config_for_portal};

pub fn prepare_linux_namespaces() {
    del_netns("net_a");
    del_netns("net_b");
    del_netns("net_c");
    del_netns("net_d");
    del_netns("net_e");

    create_netns("net_a", "10.1.1.1/24", "fd11::1/64");
    create_netns("net_b", "10.1.1.2/24", "fd11::2/64");
    create_netns("net_c", "10.1.2.3/24", "fd12::3/64");
    create_netns("net_d", "10.1.2.4/24", "fd12::4/64");
    create_netns("net_e", "10.1.1.3/24", "fd11::3/64");

    prepare_bridge("br_a");
    prepare_bridge("br_b");

    add_ns_to_bridge("br_a", "net_a");
    add_ns_to_bridge("br_a", "net_b");
    add_ns_to_bridge("br_a", "net_e");
    add_ns_to_bridge("br_b", "net_c");
    add_ns_to_bridge("br_b", "net_d");
}

pub fn get_inst_config(
    inst_name: &str,
    ns: Option<&str>,
    ipv4: &str,
    ipv6: &str,
) -> TomlConfigLoader {
    let config = TomlConfigLoader::default();
    config.set_inst_name(inst_name.to_owned());
    config.set_netns(ns.map(|s| s.to_owned()));
    config.set_ipv4(Some(ipv4.parse().unwrap()));
    config.set_ipv6(Some(ipv6.parse().unwrap()));
    config.set_listeners(vec![
        "tcp://0.0.0.0:11010".parse().unwrap(),
        "udp://0.0.0.0:11010".parse().unwrap(),
        "wg://0.0.0.0:11011".parse().unwrap(),
        "ws://0.0.0.0:11011".parse().unwrap(),
        "wss://0.0.0.0:11012".parse().unwrap(),
    ]);
    config.set_socks5_portal(Some("socks5://0.0.0.0:12345".parse().unwrap()));
    config
}

pub async fn init_three_node(proto: &str) -> Vec<Instance> {
    init_three_node_with_process_runtime(proto, CoreProcessRuntime::new()).await
}

async fn init_three_node_with_process_runtime(
    proto: &str,
    process_runtime: Arc<CoreProcessRuntime>,
) -> Vec<Instance> {
    init_three_node_ex_with_inst3(
        proto,
        |cfg| cfg,
        false,
        "net_c",
        "10.144.144.3",
        "fd00::3/64",
        process_runtime,
    )
    .await
}

async fn init_three_node_ex_with_inst3<F: Fn(TomlConfigLoader) -> TomlConfigLoader>(
    proto: &str,
    cfg_cb: F,
    use_public_server: bool,
    inst3_ns: &str,
    inst3_ipv4: &str,
    inst3_ipv6: &str,
    process_runtime: Arc<CoreProcessRuntime>,
) -> Vec<Instance> {
    prepare_linux_namespaces();

    let mut inst1 = Instance::new_with_process_runtime(
        cfg_cb(get_inst_config(
            "inst1",
            Some("net_a"),
            "10.144.144.1",
            "fd00::1/64",
        )),
        process_runtime.clone(),
    );
    let mut inst2 = Instance::new_with_process_runtime(
        cfg_cb(get_inst_config(
            "inst2",
            Some("net_b"),
            "10.144.144.2",
            "fd00::2/64",
        )),
        process_runtime.clone(),
    );
    let mut inst3 = Instance::new_with_process_runtime(
        cfg_cb(get_inst_config(
            "inst3",
            Some(inst3_ns),
            inst3_ipv4,
            inst3_ipv6,
        )),
        process_runtime.clone(),
    );

    inst1.run().await.unwrap();
    inst2.run().await.unwrap();
    inst3.run().await.unwrap();

    if proto == "tcp" {
        inst1.add_connector_url("tcp://10.1.1.2:11010".parse().unwrap());
    } else if proto == "udp" {
        inst1.add_connector_url("udp://10.1.1.2:11010".parse().unwrap());
    } else if proto == "wg" {
        #[cfg(feature = "wireguard")]
        inst1.add_connector_url("wg://10.1.1.2:11011".parse().unwrap());
    } else if proto == "ws" {
        #[cfg(feature = "websocket")]
        inst1.add_connector_url("ws://10.1.1.2:11011".parse().unwrap());
    } else if proto == "wss" {
        #[cfg(feature = "websocket")]
        inst1.add_connector_url("wss://10.1.1.2:11012".parse().unwrap());
    }

    inst3.add_connector_url(inst2.ring_listener_url());

    // wait inst2 have two route.
    wait_for_condition(
        || async {
            if !use_public_server {
                inst2.get_core_instance().route_snapshots().await.len() == 2
            } else {
                inst2
                    .get_core_instance()
                    .foreign_network_snapshots(false)
                    .await
                    .len()
                    == 1
            }
        },
        Duration::from_secs(5),
    )
    .await;

    wait_for_condition(
        || async {
            let routes = inst1.get_core_instance().route_snapshots().await;
            println!("routes: {:?}", routes);
            routes.len() == 2
        },
        Duration::from_secs(5),
    )
    .await;

    wait_for_condition(
        || async {
            let routes = inst3.get_core_instance().route_snapshots().await;
            println!("routes: {:?}", routes);
            routes.len() == 2
        },
        Duration::from_secs(5),
    )
    .await;

    vec![inst1, inst2, inst3]
}

pub async fn init_three_node_ex<F: Fn(TomlConfigLoader) -> TomlConfigLoader>(
    proto: &str,
    cfg_cb: F,
    use_public_server: bool,
) -> Vec<Instance> {
    init_three_node_ex_with_inst3(
        proto,
        cfg_cb,
        use_public_server,
        "net_c",
        "10.144.144.3",
        "fd00::3/64",
        CoreProcessRuntime::new(),
    )
    .await
}

async fn init_lazy_p2p_three_node_ex<F: Fn(TomlConfigLoader) -> TomlConfigLoader>(
    proto: &str,
    cfg_cb: F,
) -> Vec<Instance> {
    init_three_node_ex_with_inst3(
        proto,
        cfg_cb,
        false,
        "net_e",
        "10.144.144.3",
        "fd00::3/64",
        CoreProcessRuntime::new(),
    )
    .await
}

pub async fn drop_insts(insts: Vec<Instance>) {
    let mut set = JoinSet::new();
    for mut inst in insts {
        set.spawn(async move {
            inst.clear_resources().await;
            let core = Arc::downgrade(&inst.get_core_instance());
            drop(inst);
            let now = std::time::Instant::now();
            while now.elapsed().as_secs() < 5 && core.strong_count() > 0 {
                tokio::time::sleep(std::time::Duration::from_millis(50)).await;
            }

            debug_assert_eq!(core.strong_count(), 0, "CoreInstance should be dropped");
        });
    }
    while set.join_next().await.is_some() {}
}

async fn ping_test(from_netns: &str, target_ip: &str, payload_size: Option<usize>) -> bool {
    let _g = NetNS::new(Some(ROOT_NETNS_NAME.to_owned())).guard();
    let code = tokio::process::Command::new("ip")
        .args([
            "netns",
            "exec",
            from_netns,
            "ping",
            "-c",
            "1",
            "-s",
            payload_size.unwrap_or(56).to_string().as_str(),
            "-W",
            "1",
            target_ip.to_string().as_str(),
        ])
        .stdout(std::process::Stdio::null())
        .stderr(std::process::Stdio::null())
        .status()
        .await
        .unwrap();
    code.code().unwrap() == 0
}

async fn ping6_test(from_netns: &str, target_ip: &str, payload_size: Option<usize>) -> bool {
    let _g = NetNS::new(Some(ROOT_NETNS_NAME.to_owned())).guard();
    let code = tokio::process::Command::new("ip")
        .args([
            "netns",
            "exec",
            from_netns,
            "ping6",
            "-c",
            "1",
            "-s",
            payload_size.unwrap_or(56).to_string().as_str(),
            "-W",
            "1",
            target_ip.to_string().as_str(),
        ])
        .stdout(std::process::Stdio::null())
        .stderr(std::process::Stdio::null())
        .status()
        .await
        .unwrap();
    code.code().unwrap() == 0
}

fn run_cmd(program: &str, args: &[&str]) {
    let output = std::process::Command::new(program)
        .args(args)
        .output()
        .unwrap();
    assert!(
        output.status.success(),
        "{} {:?} failed: stdout={}, stderr={}",
        program,
        args,
        String::from_utf8_lossy(&output.stdout),
        String::from_utf8_lossy(&output.stderr)
    );
}

fn run_cmd_output(program: &str, args: &[&str]) -> String {
    let output = std::process::Command::new(program)
        .args(args)
        .output()
        .unwrap();
    assert!(
        output.status.success(),
        "{} {:?} failed: stdout={}, stderr={}",
        program,
        args,
        String::from_utf8_lossy(&output.stdout),
        String::from_utf8_lossy(&output.stderr)
    );
    String::from_utf8(output.stdout).unwrap()
}

fn run_ip(args: &[&str]) {
    run_cmd("ip", args);
}

fn run_ip_in_ns(ns: &str, args: &[&str]) {
    let mut cmd = vec!["netns", "exec", ns, "ip"];
    cmd.extend_from_slice(args);
    run_cmd("ip", &cmd);
}

fn run_ip_in_ns_output(ns: &str, args: &[&str]) -> String {
    let mut cmd = vec!["netns", "exec", ns, "ip"];
    cmd.extend_from_slice(args);
    run_cmd_output("ip", &cmd)
}

fn run_sysctl_in_ns(ns: &str, assignment: &str) {
    run_cmd("ip", &["netns", "exec", ns, "sysctl", "-qw", assignment]);
}

fn create_empty_netns(name: &str) {
    del_netns(name);
    run_ip(&["netns", "add", name]);
    run_ip(&["netns", "exec", name, "ip", "link", "set", "lo", "up"]);
}

fn connect_ns_to_bridge(ns: &str, guest_if: &str, host_if: &str, bridge: &str) {
    let _ = std::process::Command::new("ip")
        .args(["link", "del", host_if])
        .status();
    run_ip(&[
        "link", "add", host_if, "type", "veth", "peer", "name", guest_if,
    ]);
    run_ip(&["link", "set", guest_if, "netns", ns]);
    run_ip(&["link", "set", host_if, "up"]);
    run_cmd("brctl", &["addif", bridge, host_if]);
    run_ip(&["netns", "exec", ns, "ip", "link", "set", guest_if, "up"]);
}

struct PublicIpv6Lab {
    extra_namespaces: [&'static str; 2],
    extra_bridges: [&'static str; 2],
}

#[derive(Clone, Copy)]
enum PublicIpv6LabTopology {
    DelegatedPrefix,
    OnLinkPrefix,
}

impl PublicIpv6Lab {
    const PROVIDER_NS: &'static str = "net_a";
    const CLIENT_NS: &'static str = "net_b";
    const UPSTREAM_NS: &'static str = "net_pubgw";
    const SERVER_NS: &'static str = "net_pubsrv";
    const WAN_BRIDGE: &'static str = "br_pubwan";
    const SERVER_BRIDGE: &'static str = "br_pubsrv";
    const PROVIDER_TUN: &'static str = "etpubv6p";
    const CLIENT_TUN: &'static str = "etpubv6c";
    const PROVIDER_PREFIX: &'static str = "2001:db8:100::/64";
    const PROVIDER_DEFAULT_FROM: &'static str = "2001:db8:100::/64";
    const PROVIDER_WAN_ADDR: &'static str = "2001:db8:ffff:1::2/64";
    const UPSTREAM_WAN_ADDR: &'static str = "2001:db8:ffff:1::1/64";
    const ON_LINK_PROVIDER_WAN_ADDR: &'static str = "2001:db8:100::2/64";
    const ON_LINK_UPSTREAM_WAN_ADDR: &'static str = "2001:db8:100::1/64";
    const UPSTREAM_SERVER_ADDR: &'static str = "2001:db8:ffff:2::1/64";
    const SERVER_ADDR: &'static str = "2001:db8:ffff:2::100/64";
    const SERVER_IP: &'static str = "2001:db8:ffff:2::100";

    fn setup_with_topology(topology: PublicIpv6LabTopology) -> Self {
        prepare_linux_namespaces();

        del_netns(Self::UPSTREAM_NS);
        del_netns(Self::SERVER_NS);
        let _ = std::process::Command::new("ip")
            .args(["link", "del", Self::WAN_BRIDGE])
            .status();
        let _ = std::process::Command::new("ip")
            .args(["link", "del", Self::SERVER_BRIDGE])
            .status();
        let _ = std::process::Command::new("brctl")
            .args(["delbr", Self::WAN_BRIDGE])
            .status();
        let _ = std::process::Command::new("brctl")
            .args(["delbr", Self::SERVER_BRIDGE])
            .status();

        create_empty_netns(Self::UPSTREAM_NS);
        create_empty_netns(Self::SERVER_NS);
        prepare_bridge(Self::WAN_BRIDGE);
        prepare_bridge(Self::SERVER_BRIDGE);
        run_ip(&["link", "set", Self::WAN_BRIDGE, "up"]);
        run_ip(&["link", "set", Self::SERVER_BRIDGE, "up"]);

        connect_ns_to_bridge(
            Self::PROVIDER_NS,
            "pubwan0",
            "veth_pubwan_p",
            Self::WAN_BRIDGE,
        );
        connect_ns_to_bridge(
            Self::UPSTREAM_NS,
            "upwan0",
            "veth_pubwan_u",
            Self::WAN_BRIDGE,
        );
        connect_ns_to_bridge(
            Self::UPSTREAM_NS,
            "upsrv0",
            "veth_pubsrv_u",
            Self::SERVER_BRIDGE,
        );
        connect_ns_to_bridge(
            Self::SERVER_NS,
            "srv0",
            "veth_pubsrv_s",
            Self::SERVER_BRIDGE,
        );

        let (provider_wan_addr, upstream_wan_addr) = match topology {
            PublicIpv6LabTopology::DelegatedPrefix => {
                (Self::PROVIDER_WAN_ADDR, Self::UPSTREAM_WAN_ADDR)
            }
            PublicIpv6LabTopology::OnLinkPrefix => (
                Self::ON_LINK_PROVIDER_WAN_ADDR,
                Self::ON_LINK_UPSTREAM_WAN_ADDR,
            ),
        };

        run_ip_in_ns(
            Self::PROVIDER_NS,
            &["addr", "add", provider_wan_addr, "dev", "pubwan0"],
        );
        run_ip_in_ns(
            Self::UPSTREAM_NS,
            &["addr", "add", upstream_wan_addr, "dev", "upwan0"],
        );
        run_ip_in_ns(
            Self::UPSTREAM_NS,
            &["addr", "add", Self::UPSTREAM_SERVER_ADDR, "dev", "upsrv0"],
        );
        run_ip_in_ns(
            Self::SERVER_NS,
            &["addr", "add", Self::SERVER_ADDR, "dev", "srv0"],
        );

        match topology {
            PublicIpv6LabTopology::DelegatedPrefix => {
                run_ip_in_ns(
                    Self::PROVIDER_NS,
                    &["link", "add", "pubprefix0", "type", "dummy"],
                );
                run_ip_in_ns(Self::PROVIDER_NS, &["link", "set", "pubprefix0", "up"]);
                run_ip_in_ns(
                    Self::PROVIDER_NS,
                    &[
                        "-6",
                        "route",
                        "add",
                        Self::PROVIDER_PREFIX,
                        "dev",
                        "pubprefix0",
                    ],
                );
                run_ip_in_ns(
                    Self::PROVIDER_NS,
                    &[
                        "-6",
                        "route",
                        "add",
                        "default",
                        "from",
                        Self::PROVIDER_DEFAULT_FROM,
                        "via",
                        "2001:db8:ffff:1::1",
                        "dev",
                        "pubwan0",
                    ],
                );
            }
            PublicIpv6LabTopology::OnLinkPrefix => {
                run_ip_in_ns(
                    Self::PROVIDER_NS,
                    &[
                        "-6",
                        "route",
                        "add",
                        "default",
                        "via",
                        "2001:db8:100::1",
                        "dev",
                        "pubwan0",
                    ],
                );
            }
        }

        run_ip_in_ns(
            Self::SERVER_NS,
            &[
                "-6",
                "route",
                "add",
                "default",
                "via",
                "2001:db8:ffff:2::1",
                "dev",
                "srv0",
            ],
        );
        if matches!(topology, PublicIpv6LabTopology::DelegatedPrefix) {
            run_ip_in_ns(
                Self::UPSTREAM_NS,
                &[
                    "-6",
                    "route",
                    "add",
                    Self::PROVIDER_PREFIX,
                    "via",
                    "2001:db8:ffff:1::2",
                    "dev",
                    "upwan0",
                ],
            );
        }

        run_sysctl_in_ns(Self::PROVIDER_NS, "net.ipv6.conf.all.forwarding=1");
        run_sysctl_in_ns(Self::UPSTREAM_NS, "net.ipv6.conf.all.forwarding=1");

        Self {
            extra_namespaces: [Self::UPSTREAM_NS, Self::SERVER_NS],
            extra_bridges: [Self::WAN_BRIDGE, Self::SERVER_BRIDGE],
        }
    }
}

impl Drop for PublicIpv6Lab {
    fn drop(&mut self) {
        for ns in self.extra_namespaces {
            del_netns(ns);
        }
        for bridge in self.extra_bridges {
            let _ = std::process::Command::new("ip")
                .args(["link", "del", bridge])
                .status();
            let _ = std::process::Command::new("brctl")
                .args(["delbr", bridge])
                .status();
        }
    }
}

fn get_public_ipv6_config(
    inst_name: &str,
    netns: &str,
    ipv4: &str,
    dev_name: &str,
    inst_id: uuid::Uuid,
) -> TomlConfigLoader {
    let config = get_inst_config(inst_name, Some(netns), ipv4, "fd00::1/64");
    config.set_id(inst_id);
    config.set_ipv6(None);
    config.set_socks5_portal(None);
    config.set_network_identity(NetworkIdentity {
        network_name: "public_ipv6_auto_addr_test".to_string(),
        network_secret: Some("public_ipv6_auto_addr_secret".to_string()),
        network_secret_digest: None,
    });
    config.set_listeners(vec!["tcp://0.0.0.0:11010".parse().unwrap()]);
    let mut flags = config.get_flags();
    flags.dev_name = dev_name.to_string();
    config.set_flags(flags);
    config
}

async fn init_public_ipv6_two_node(
    client_inst_id: uuid::Uuid,
) -> (PublicIpv6Lab, Arc<CoreProcessRuntime>, Instance, Instance) {
    init_public_ipv6_two_node_with_topology(client_inst_id, PublicIpv6LabTopology::DelegatedPrefix)
        .await
}

async fn init_public_ipv6_two_node_with_topology(
    client_inst_id: uuid::Uuid,
    topology: PublicIpv6LabTopology,
) -> (PublicIpv6Lab, Arc<CoreProcessRuntime>, Instance, Instance) {
    let lab = PublicIpv6Lab::setup_with_topology(topology);
    let process_runtime = CoreProcessRuntime::new();

    let provider_cfg = get_public_ipv6_config(
        "provider_public_ipv6",
        PublicIpv6Lab::PROVIDER_NS,
        "10.144.144.1",
        PublicIpv6Lab::PROVIDER_TUN,
        uuid::Uuid::parse_str("11111111-1111-1111-1111-111111111111").unwrap(),
    );
    provider_cfg.set_ipv6_public_addr_provider(true);

    let client_cfg = get_public_ipv6_config(
        "client_public_ipv6",
        PublicIpv6Lab::CLIENT_NS,
        "10.144.144.2",
        PublicIpv6Lab::CLIENT_TUN,
        client_inst_id,
    );
    client_cfg.set_ipv6_public_addr_auto(true);

    let mut provider = Instance::new_with_process_runtime(provider_cfg, process_runtime.clone());
    let mut client = Instance::new_with_process_runtime(client_cfg, process_runtime.clone());

    provider.run().await.unwrap();
    client.run().await.unwrap();

    provider.add_connector_url("tcp://10.1.1.2:11010".parse().unwrap());

    wait_for_condition(
        || async {
            provider.get_core_instance().route_snapshots().await.len() == 1
                && client.get_core_instance().route_snapshots().await.len() == 1
        },
        Duration::from_secs(8),
    )
    .await;

    (lab, process_runtime, provider, client)
}

async fn wait_for_public_ipv6_addr(inst: &Instance) -> cidr::Ipv6Inet {
    wait_for_condition(
        || async {
            inst.get_core_instance()
                .packet_plane()
                .public_ipv6_addr()
                .await
                .is_some()
        },
        Duration::from_secs(10),
    )
    .await;
    inst.get_core_instance()
        .packet_plane()
        .public_ipv6_addr()
        .await
        .unwrap()
}

async fn wait_for_public_ipv6_route(inst: &Instance, target: cidr::Ipv6Inet) {
    wait_for_condition(
        || async {
            inst.get_core_instance()
                .packet_plane()
                .public_ipv6_routes()
                .await
                .contains(&target)
        },
        Duration::from_secs(10),
    )
    .await;
}

fn route_exists_in_ns(ns: &str, needle: &str) -> bool {
    run_ip_in_ns_output(ns, &["-6", "route", "show"])
        .lines()
        .any(|line| line.contains(needle))
}

fn addr_exists_in_ns(ns: &str, dev: &str, needle: &str) -> bool {
    run_ip_in_ns_output(ns, &["-6", "addr", "show", "dev", dev]).contains(needle)
}

fn ndp_proxy_exists_in_ns(ns: &str, dev: &str, addr: std::net::Ipv6Addr) -> bool {
    let addr = addr.to_string();
    run_ip_in_ns_output(ns, &["-6", "neigh", "show", "proxy", "dev", dev])
        .lines()
        .any(|line| line.split_whitespace().next() == Some(addr.as_str()))
}

#[tokio::test]
#[serial_test::serial]
pub async fn public_ipv6_auto_addr_end_to_end() {
    let client_id = uuid::Uuid::parse_str("22222222-2222-2222-2222-222222222222").unwrap();
    let (_lab, _process_runtime, provider, client) = init_public_ipv6_two_node(client_id).await;

    wait_for_condition(
        || async {
            provider
                .get_core_instance()
                .node_snapshot()
                .await
                .ipv6_public_addr_prefix
                == Some(PublicIpv6Lab::PROVIDER_PREFIX.parse().unwrap())
        },
        Duration::from_secs(10),
    )
    .await;

    let leased = wait_for_public_ipv6_addr(&client).await;
    wait_for_public_ipv6_route(&provider, leased).await;

    assert_eq!(
        provider
            .get_global_ctx()
            .config
            .get_ipv6_public_addr_prefix(),
        None
    );
    assert_eq!(
        provider
            .get_core_instance()
            .node_snapshot()
            .await
            .ipv6_public_addr_prefix,
        Some(PublicIpv6Lab::PROVIDER_PREFIX.parse().unwrap())
    );
    let provider_prefix = PublicIpv6Lab::PROVIDER_PREFIX
        .parse::<cidr::Ipv6Cidr>()
        .unwrap();
    assert_eq!(
        provider
            .get_core_instance()
            .node_snapshot()
            .await
            .ipv6_public_addr_prefix,
        Some(
            cidr::Ipv6Inet::new(
                provider_prefix.first_address(),
                provider_prefix.network_length()
            )
            .unwrap()
        )
    );
    let provider_info = provider.get_core_instance().local_public_ipv6_info().await;
    let client_peer_id = client.get_core_instance().node_snapshot().await.peer_id;
    assert_eq!(
        provider_info.provider_prefix,
        Some(
            cidr::Ipv6Inet::new(
                provider_prefix.first_address(),
                provider_prefix.network_length()
            )
            .unwrap()
            .into()
        )
    );
    assert_eq!(provider_info.provider_leases.len(), 1);
    assert_eq!(provider_info.provider_leases[0].peer_id, client_peer_id);
    assert_eq!(
        provider_info.provider_leases[0].inst_id,
        client_id.to_string()
    );
    assert_eq!(
        provider_info.provider_leases[0].leased_addr,
        Some(leased.into())
    );
    assert!(
        leased.address().segments()[0] & 0xfe00 != 0xfc00,
        "leased address should not be unique-local: {leased}"
    );

    wait_for_condition(
        || async {
            addr_exists_in_ns(
                PublicIpv6Lab::CLIENT_NS,
                PublicIpv6Lab::CLIENT_TUN,
                &leased.to_string(),
            ) && route_exists_in_ns(
                PublicIpv6Lab::CLIENT_NS,
                &format!("default dev {}", PublicIpv6Lab::CLIENT_TUN),
            ) && route_exists_in_ns(
                PublicIpv6Lab::PROVIDER_NS,
                &format!("{} dev {}", leased.address(), PublicIpv6Lab::PROVIDER_TUN),
            )
        },
        Duration::from_secs(10),
    )
    .await;

    wait_for_condition(
        || async { ping6_test(PublicIpv6Lab::CLIENT_NS, PublicIpv6Lab::SERVER_IP, None).await },
        Duration::from_secs(10),
    )
    .await;

    wait_for_condition(
        || async {
            ping6_test(
                PublicIpv6Lab::SERVER_NS,
                leased.address().to_string().as_str(),
                None,
            )
            .await
        },
        Duration::from_secs(10),
    )
    .await;

    drop_insts(vec![provider, client]).await;
}

#[tokio::test]
#[serial_test::serial]
pub async fn public_ipv6_auto_addr_on_link_ndp_proxy_end_to_end() {
    let client_id = uuid::Uuid::parse_str("44444444-4444-4444-4444-444444444444").unwrap();
    let (_lab, _process_runtime, provider, client) =
        init_public_ipv6_two_node_with_topology(client_id, PublicIpv6LabTopology::OnLinkPrefix)
            .await;

    wait_for_condition(
        || async {
            provider
                .get_core_instance()
                .node_snapshot()
                .await
                .ipv6_public_addr_prefix
                == Some(PublicIpv6Lab::PROVIDER_PREFIX.parse().unwrap())
        },
        Duration::from_secs(10),
    )
    .await;

    let leased = wait_for_public_ipv6_addr(&client).await;
    wait_for_public_ipv6_route(&provider, leased).await;

    wait_for_condition(
        || async {
            addr_exists_in_ns(
                PublicIpv6Lab::CLIENT_NS,
                PublicIpv6Lab::CLIENT_TUN,
                &leased.to_string(),
            ) && route_exists_in_ns(
                PublicIpv6Lab::PROVIDER_NS,
                &format!("{} dev {}", leased.address(), PublicIpv6Lab::PROVIDER_TUN),
            )
        },
        Duration::from_secs(10),
    )
    .await;

    wait_for_condition(
        || async {
            ndp_proxy_exists_in_ns(PublicIpv6Lab::PROVIDER_NS, "pubwan0", leased.address())
        },
        Duration::from_secs(20),
    )
    .await;

    wait_for_condition(
        || async {
            ping6_test(
                PublicIpv6Lab::SERVER_NS,
                leased.address().to_string().as_str(),
                None,
            )
            .await
        },
        Duration::from_secs(20),
    )
    .await;

    drop_insts(vec![provider, client]).await;
}

#[tokio::test]
#[serial_test::serial]
pub async fn public_ipv6_auto_addr_reconnect_reuses_same_address() {
    let client_id = uuid::Uuid::parse_str("33333333-3333-3333-3333-333333333333").unwrap();
    let (_lab, process_runtime, provider, client) = init_public_ipv6_two_node(client_id).await;
    let first = wait_for_public_ipv6_addr(&client).await;

    drop_insts(vec![client]).await;

    let client_cfg = get_public_ipv6_config(
        "client_public_ipv6_reconnect",
        PublicIpv6Lab::CLIENT_NS,
        "10.144.144.2",
        PublicIpv6Lab::CLIENT_TUN,
        client_id,
    );
    client_cfg.set_ipv6_public_addr_auto(true);
    let mut client = Instance::new_with_process_runtime(client_cfg, process_runtime.clone());
    client.run().await.unwrap();
    provider.add_connector_url("tcp://10.1.1.2:11010".parse().unwrap());

    wait_for_condition(
        || async {
            provider.get_core_instance().route_snapshots().await.len() == 1
                && client.get_core_instance().route_snapshots().await.len() == 1
        },
        Duration::from_secs(8),
    )
    .await;

    let second = wait_for_public_ipv6_addr(&client).await;
    assert_eq!(first, second);

    wait_for_condition(
        || async { ping6_test(PublicIpv6Lab::CLIENT_NS, PublicIpv6Lab::SERVER_IP, None).await },
        Duration::from_secs(10),
    )
    .await;

    drop_insts(vec![provider, client]).await;
}

#[cfg(feature = "tun")]
#[tokio::test]
#[serial_test::serial]
pub async fn shared_tun_public_ipv6_auto_addr_end_to_end() {
    let lab = PublicIpv6Lab::setup_with_topology(PublicIpv6LabTopology::DelegatedPrefix);
    let provider_dev = format!("st{:08x}", rand::random::<u32>());
    let client_dev = format!("st{:08x}", rand::random::<u32>());
    let process_runtime = CoreProcessRuntime::new();
    let shared_virtual_nic_registry = Instance::new_shared_virtual_nic_registry();

    let provider_cfg = get_public_ipv6_config(
        "provider_shared_public_ipv6",
        PublicIpv6Lab::PROVIDER_NS,
        "10.144.144.1",
        &provider_dev,
        uuid::Uuid::parse_str("44444444-4444-4444-4444-444444444444").unwrap(),
    );
    provider_cfg.set_ipv6_public_addr_provider(true);

    let client_cfg = get_public_ipv6_config(
        "client_shared_public_ipv6",
        PublicIpv6Lab::CLIENT_NS,
        "10.144.144.2",
        &client_dev,
        uuid::Uuid::parse_str("55555555-5555-5555-5555-555555555555").unwrap(),
    );
    client_cfg.set_ipv6_public_addr_auto(true);

    let client_peer_cfg = get_public_ipv6_config(
        "client_shared_public_ipv6_peer",
        PublicIpv6Lab::CLIENT_NS,
        "10.144.144.3",
        &client_dev,
        uuid::Uuid::parse_str("66666666-6666-6666-6666-666666666666").unwrap(),
    );
    client_peer_cfg.set_listeners(vec![]);

    let mut provider = Instance::new_with_process_runtime_and_shared_virtual_nic_registry(
        provider_cfg,
        process_runtime.clone(),
        shared_virtual_nic_registry.clone(),
    );
    let mut client = Instance::new_with_process_runtime_and_shared_virtual_nic_registry(
        client_cfg,
        process_runtime.clone(),
        shared_virtual_nic_registry.clone(),
    );
    let mut client_peer = Instance::new_with_process_runtime_and_shared_virtual_nic_registry(
        client_peer_cfg,
        process_runtime,
        shared_virtual_nic_registry,
    );
    let mut client_events = client.get_global_ctx().subscribe();
    let mut client_peer_events = client_peer.get_global_ctx().subscribe();

    provider.run().await.unwrap();
    client.run().await.unwrap();
    client_peer.run().await.unwrap();

    let shared_ifname = wait_for_tun_ready(&mut client_events).await;
    assert_eq!(
        shared_ifname,
        wait_for_tun_ready(&mut client_peer_events).await
    );
    assert_eq!(shared_ifname, client_dev);

    provider.add_connector_url("tcp://10.1.1.2:11010".parse().unwrap());

    wait_for_condition(
        || async {
            provider.get_core_instance().route_snapshots().await.len() == 1
                && client.get_core_instance().route_snapshots().await.len() == 1
        },
        Duration::from_secs(8),
    )
    .await;

    wait_for_condition(
        || async {
            provider
                .get_core_instance()
                .node_snapshot()
                .await
                .ipv6_public_addr_prefix
                == Some(PublicIpv6Lab::PROVIDER_PREFIX.parse().unwrap())
        },
        Duration::from_secs(10),
    )
    .await;

    let leased = wait_for_public_ipv6_addr(&client).await;
    wait_for_public_ipv6_route(&provider, leased).await;

    wait_for_condition(
        || async {
            addr_exists_in_ns(PublicIpv6Lab::CLIENT_NS, &client_dev, &leased.to_string())
                && route_exists_in_ns(
                    PublicIpv6Lab::CLIENT_NS,
                    &format!("default dev {client_dev}"),
                )
                && route_exists_in_ns(
                    PublicIpv6Lab::PROVIDER_NS,
                    &format!("{} dev {provider_dev}", leased.address()),
                )
        },
        Duration::from_secs(10),
    )
    .await;

    wait_for_condition(
        || async { ping6_test(PublicIpv6Lab::CLIENT_NS, PublicIpv6Lab::SERVER_IP, None).await },
        Duration::from_secs(10),
    )
    .await;

    wait_for_condition(
        || async {
            ping6_test(
                PublicIpv6Lab::SERVER_NS,
                leased.address().to_string().as_str(),
                None,
            )
            .await
        },
        Duration::from_secs(10),
    )
    .await;

    drop_insts(vec![provider, client, client_peer]).await;
    drop(lab);
}

#[cfg(feature = "tun")]
async fn wait_for_tun_ready(
    receiver: &mut tokio::sync::broadcast::Receiver<crate::common::global_ctx::GlobalCtxEvent>,
) -> String {
    tokio::time::timeout(Duration::from_secs(5), async {
        loop {
            match receiver.recv().await.unwrap() {
                crate::common::global_ctx::GlobalCtxEvent::TunDeviceReady(ifname) => return ifname,
                crate::common::global_ctx::GlobalCtxEvent::TunDeviceError(error) => {
                    panic!("tun device error: {error}")
                }
                _ => {}
            }
        }
    })
    .await
    .expect("timed out waiting for tun ready")
}

#[rstest::rstest]
#[tokio::test]
#[serial_test::serial]
pub async fn basic_three_node_test(
    #[values("tcp", "udp", "wg", "ws", "wss")] proto: &str,
    #[values(
        ["aes-gcm", "aes-gcm"],
        ["aes-256-gcm", "aes-256-gcm"],
        ["chacha20", "chacha20"],
        ["xor", "xor"],
        ["openssl-chacha20", "openssl-chacha20"],
        ["openssl-aes-gcm", "openssl-aes-gcm"],
        ["openssl-aes-256-gcm", "openssl-aes-256-gcm"],
        ["aes-gcm", "openssl-aes-gcm"],
        ["openssl-aes-gcm", "aes-gcm"],
        ["aes-256-gcm", "openssl-aes-256-gcm"],
        ["openssl-aes-256-gcm", "aes-256-gcm"],
        ["chacha20", "openssl-chacha20"],
        ["openssl-chacha20", "chacha20"],
    )]
    encrypt_algorithm_pair: [&str; 2],
) {
    let insts = init_three_node_ex(
        proto,
        |cfg| {
            let mut flags = cfg.get_flags();
            if cfg.get_inst_name() == "inst0" {
                flags.encryption_algorithm = encrypt_algorithm_pair[0].to_string();
            } else {
                flags.encryption_algorithm = encrypt_algorithm_pair[1].to_string();
            }
            cfg.set_flags(flags);
            cfg
        },
        false,
    )
    .await;

    check_route(
        "10.144.144.2/24",
        insts[1].peer_id(),
        insts[0].get_core_instance().route_snapshots().await,
    );

    check_route(
        "10.144.144.3/24",
        insts[2].peer_id(),
        insts[0].get_core_instance().route_snapshots().await,
    );

    // Test IPv4 connectivity
    wait_for_condition(
        || async { ping_test("net_c", "10.144.144.1", None).await },
        Duration::from_secs(5000),
    )
    .await;

    // Test IPv6 connectivity
    wait_for_condition(
        || async { ping6_test("net_c", "fd00::1", None).await },
        Duration::from_secs(5),
    )
    .await;

    wait_for_condition(
        || async { ping6_test("net_a", "fd00::3", None).await },
        Duration::from_secs(5),
    )
    .await;

    drop_insts(insts).await;
}

#[tokio::test]
#[serial_test::serial]
pub async fn ping_own_virtual_ip_should_work() {
    let insts = init_three_node("udp").await;

    wait_for_condition(
        || async { ping_test("net_a", "10.144.144.1", None).await },
        Duration::from_secs(5),
    )
    .await;

    wait_for_condition(
        || async { ping6_test("net_a", "fd00::1", None).await },
        Duration::from_secs(5),
    )
    .await;

    drop_insts(insts).await;
}

#[tokio::test]
#[serial_test::serial]
pub async fn subnet_proxy_loop_prevention_test() {
    // 测试场景：inst1 和 inst2 都代理了 10.1.2.0/24 网段，
    // inst1 发起对 10.1.2.5 的 ping，不应该出现环路
    let insts = init_three_node_ex(
        "udp",
        |cfg| {
            if cfg.get_inst_name() == "inst1" {
                // inst1 代理 10.1.2.0/24 网段
                cfg.add_proxy_cidr("10.1.2.0/24".parse().unwrap(), None)
                    .unwrap();
            } else if cfg.get_inst_name() == "inst2" {
                // inst2 也代理相同的 10.1.2.0/24 网段
                cfg.add_proxy_cidr("10.1.2.0/24".parse().unwrap(), None)
                    .unwrap();
            }
            cfg
        },
        false,
    )
    .await;

    // 等待代理路由出现 - inst1 应该看到 inst2 的代理路由
    wait_proxy_route_appear(
        &insts[0].get_core_instance(),
        "10.144.144.2/24",
        insts[1].peer_id(),
        "10.1.2.0/24",
    )
    .await;

    // 等待代理路由出现 - inst2 应该看到 inst1 的代理路由
    wait_proxy_route_appear(
        &insts[1].get_core_instance(),
        "10.144.144.1/24",
        insts[0].peer_id(),
        "10.1.2.0/24",
    )
    .await;

    // 从 inst1 (net_a) 发起对 10.1.2.5 的 ping 测试
    // 这应该失败，并且不会产生环路
    let now = std::time::Instant::now();
    while now.elapsed().as_secs() < 10 {
        ping_test("net_a", "10.1.2.5", None).await;
        tokio::time::sleep(Duration::from_secs(1)).await;
    }

    println!(
        "inst0 metrics: {:?}",
        insts[0].get_core_instance().prometheus_metrics()
    );

    let all_metrics = insts[0].get_core_instance().metric_snapshots();
    for metric in all_metrics {
        if metric.name == MetricName::TrafficPacketsSelfTx {
            assert!(metric.value < 40);
        }
    }

    drop_insts(insts).await;
}

async fn subnet_proxy_test_udp(listen_ip: &str, target_ip: &str, timeout: Duration) {
    use crate::tunnel::common::tests::_tunnel_pingpong_netns_with_timeout;
    use rand::Rng;

    let udp_listener = core_udp_listener(format!("udp://{}:22233", listen_ip).parse().unwrap());
    let udp_connector = core_udp_dialer(format!("udp://{}:22233", target_ip).parse().unwrap());

    // NOTE: this should not excced udp tunnel max buffer size
    let mut buf = vec![0; 7 * 1024];
    rand::thread_rng().fill(&mut buf[..]);

    let ns_name = if target_ip == "10.144.144.3" {
        "net_c"
    } else {
        "net_d"
    };

    let result = _tunnel_pingpong_netns_with_timeout(
        udp_listener,
        udp_connector,
        NetNS::new(Some(ns_name.into())),
        NetNS::new(Some("net_a".into())),
        buf,
        timeout,
    )
    .await;
    assert!(result.is_ok(), "{}", result.unwrap_err());

    // no fragment
    let udp_listener = core_udp_listener(format!("udp://{}:22233", listen_ip).parse().unwrap());
    let udp_connector = core_udp_dialer(format!("udp://{}:22233", target_ip).parse().unwrap());

    let mut buf = vec![0; 1024];
    rand::thread_rng().fill(&mut buf[..]);

    let result = _tunnel_pingpong_netns_with_timeout(
        udp_listener,
        udp_connector,
        NetNS::new(Some(ns_name.into())),
        NetNS::new(Some("net_a".into())),
        buf,
        timeout,
    )
    .await;
    assert!(result.is_ok(), "{}", result.unwrap_err());
}

async fn subnet_proxy_test_tcp(listen_ip: &str, connect_ip: &str, timeout: Duration) {
    use crate::tunnel::common::tests::_tunnel_pingpong_netns_with_timeout;
    use rand::Rng;

    let tcp_listener = core_tcp_listener(format!("tcp://{listen_ip}:22223").parse().unwrap());
    let tcp_connector = core_tcp_dialer(format!("tcp://{}:22223", connect_ip).parse().unwrap());

    let mut buf = vec![0; 32];
    rand::thread_rng().fill(&mut buf[..]);

    let ns_name = if connect_ip == "10.144.144.3" {
        "net_c"
    } else {
        "net_d"
    };

    let result = _tunnel_pingpong_netns_with_timeout(
        tcp_listener,
        tcp_connector,
        NetNS::new(Some(ns_name.into())),
        NetNS::new(Some("net_a".into())),
        buf,
        timeout,
    )
    .await;
    assert!(result.is_ok(), "{}", result.unwrap_err());
}

async fn subnet_proxy_test_icmp(target_ip: &str, timeout: Duration) {
    wait_for_condition(
        || async { ping_test("net_a", target_ip, None).await },
        timeout,
    )
    .await;

    wait_for_condition(
        || async { ping_test("net_a", target_ip, Some(5 * 1024)).await },
        timeout,
    )
    .await;
}

#[tokio::test]
pub async fn quic_proxy() {
    let insts = init_three_node_ex(
        "udp",
        |cfg| {
            if cfg.get_inst_name() == "inst3" {
                cfg.add_proxy_cidr("10.1.2.0/24".parse().unwrap(), None)
                    .unwrap();
            } else if cfg.get_inst_name() == "inst1" {
                let mut flags = cfg.get_flags();
                flags.enable_quic_proxy = true;
                cfg.set_flags(flags);
            }
            cfg
        },
        false,
    )
    .await;

    assert_eq!(insts[2].get_global_ctx().config.get_proxy_cidrs().len(), 1);

    wait_proxy_route_appear(
        &insts[0].get_core_instance(),
        "10.144.144.3/24",
        insts[2].peer_id(),
        "10.1.2.0/24",
    )
    .await;

    let target_ip = "10.1.2.4";

    subnet_proxy_test_icmp(target_ip, Duration::from_secs(5)).await;
    subnet_proxy_test_icmp("10.144.144.3", Duration::from_secs(5)).await;
    subnet_proxy_test_tcp(target_ip, target_ip, Duration::from_secs(5)).await;
    subnet_proxy_test_tcp("0.0.0.0", "10.144.144.3", Duration::from_secs(5)).await;

    let metrics = insts[0]
        .get_core_instance()
        .metric_snapshots()
        .into_iter()
        .filter(|metric| metric.name == MetricName::TcpProxyConnect)
        .collect::<Vec<_>>();
    assert_eq!(metrics.len(), 2);
    assert_eq!(1, metrics[0].value);
    assert_eq!(1, metrics[1].value);

    drop_insts(insts).await;
}

#[rstest::rstest]
#[serial_test::serial]
#[tokio::test]
pub async fn subnet_proxy_three_node_test(
    #[values(true, false)] no_tun: bool,
    #[values(true, false)] relay_by_public_server: bool,
    #[values(true, false)] enable_kcp_proxy: bool,
    #[values(true, false)] enable_quic_proxy: bool,
    #[values(true, false)] disable_kcp_input: bool,
    #[values(true, false)] disable_quic_input: bool,
    #[values(true, false)] dst_enable_kcp_proxy: bool,
    #[values(true, false)] dst_enable_quic_proxy: bool,
) {
    let insts = init_three_node_ex(
        "udp",
        |cfg| {
            if cfg.get_inst_name() == "inst3" {
                let mut flags = cfg.get_flags();
                flags.no_tun = no_tun;
                flags.disable_kcp_input = disable_kcp_input;
                flags.enable_kcp_proxy = dst_enable_kcp_proxy;
                flags.disable_quic_input = disable_quic_input;
                flags.enable_quic_proxy = dst_enable_quic_proxy;
                cfg.set_flags(flags);
                cfg.add_proxy_cidr("10.1.2.0/24".parse().unwrap(), None)
                    .unwrap();
                cfg.add_proxy_cidr(
                    "10.1.2.0/24".parse().unwrap(),
                    Some("10.1.3.0/24".parse().unwrap()),
                )
                .unwrap();
            }

            if cfg.get_inst_name() == "inst2" && relay_by_public_server {
                cfg.set_network_identity(NetworkIdentity::new(
                    "public".to_string(),
                    "public".to_string(),
                ));
            }

            if cfg.get_inst_name() == "inst1" {
                let mut flags = cfg.get_flags();
                if enable_kcp_proxy {
                    flags.enable_kcp_proxy = true;
                }
                if enable_quic_proxy {
                    flags.enable_quic_proxy = true;
                }
                cfg.set_flags(flags);
            }

            cfg
        },
        relay_by_public_server,
    )
    .await;

    assert_eq!(insts[2].get_global_ctx().config.get_proxy_cidrs().len(), 2);

    wait_proxy_route_appear(
        &insts[0].get_core_instance(),
        "10.144.144.3/24",
        insts[2].peer_id(),
        "10.1.2.0/24",
    )
    .await;
    wait_proxy_route_appear(
        &insts[0].get_core_instance(),
        "10.144.144.3/24",
        insts[2].peer_id(),
        "10.1.3.0/24",
    )
    .await;

    for target_ip in ["10.1.3.4", "10.1.2.4", "10.144.144.3"] {
        subnet_proxy_test_icmp(target_ip, Duration::from_secs(5)).await;
        let listen_ip = if target_ip == "10.144.144.3" {
            "0.0.0.0"
        } else {
            "10.1.2.4"
        };
        subnet_proxy_test_tcp(listen_ip, target_ip, Duration::from_secs(5)).await;
        subnet_proxy_test_udp(listen_ip, target_ip, Duration::from_secs(5)).await;
    }
    if enable_quic_proxy && !disable_quic_input {
        let metrics = insts[0]
            .get_core_instance()
            .metric_snapshots()
            .into_iter()
            .filter(|metric| metric.name == MetricName::TcpProxyConnect)
            .collect::<Vec<_>>();
        assert_eq!(metrics.len(), 3);
        for metric in metrics {
            assert_eq!(1, metric.value);
            assert!(metric.labels.labels().iter().any(|l| {
                let t =
                    LabelType::Protocol(TcpProxyEntryTransportType::Quic.as_str_name().to_string());
                t.key() == l.key && t.value() == l.value
            }));
        }
    } else if enable_kcp_proxy && !disable_kcp_input {
        let metrics = insts[0]
            .get_core_instance()
            .metric_snapshots()
            .into_iter()
            .filter(|metric| metric.name == MetricName::TcpProxyConnect)
            .collect::<Vec<_>>();
        assert_eq!(metrics.len(), 3);
        for metric in metrics {
            assert_eq!(1, metric.value);
            assert!(metric.labels.labels().iter().any(|l| {
                let t =
                    LabelType::Protocol(TcpProxyEntryTransportType::Kcp.as_str_name().to_string());
                t.key() == l.key && t.value() == l.value
            }));
        }
    } else {
        // tcp subnet proxy
        let metrics = insts[2]
            .get_core_instance()
            .metric_snapshots()
            .into_iter()
            .filter(|metric| metric.name == MetricName::TcpProxyConnect)
            .collect::<Vec<_>>();
        if no_tun {
            assert_eq!(metrics.len(), 3);
        } else {
            assert_eq!(metrics.len(), 2);
        }
        for metric in metrics {
            assert_eq!(1, metric.value);
            assert!(metric.labels.labels().iter().any(|l| {
                let t =
                    LabelType::Protocol(TcpProxyEntryTransportType::Tcp.as_str_name().to_string());
                t.key() == l.key && t.value() == l.value
            }));
        }
    }

    drop_insts(insts).await;
}

#[rstest::rstest]
#[tokio::test]
#[serial_test::serial]
pub async fn data_compress(
    #[values(true, false)] inst1_compress: bool,
    #[values(true, false)] inst2_compress: bool,
) {
    let _insts = init_three_node_ex(
        "udp",
        |cfg| {
            if cfg.get_inst_name() == "inst1" && inst1_compress {
                let mut flags = cfg.get_flags();
                flags.data_compress_algo = CompressionAlgoPb::Zstd.into();
                cfg.set_flags(flags);
            }

            if cfg.get_inst_name() == "inst3" && inst2_compress {
                let mut flags = cfg.get_flags();
                flags.data_compress_algo = CompressionAlgoPb::Zstd.into();
                cfg.set_flags(flags);
            }

            cfg
        },
        false,
    )
    .await;

    wait_for_condition(
        || async { ping_test("net_a", "10.144.144.3", None).await },
        Duration::from_secs(5),
    )
    .await;

    wait_for_condition(
        || async { ping_test("net_a", "10.144.144.3", Some(5 * 1024)).await },
        Duration::from_secs(5),
    )
    .await;

    drop_insts(_insts).await;
}

#[cfg(feature = "wireguard")]
#[rstest::rstest]
#[tokio::test]
#[serial_test::serial]
pub async fn proxy_three_node_disconnect_test(#[values("tcp", "wg")] proto: &str) {
    use tokio_util::task::AbortOnDropHandle;

    let process_runtime = CoreProcessRuntime::new();
    let insts = init_three_node_with_process_runtime(proto, process_runtime.clone()).await;
    let mut inst4 = Instance::new_with_process_runtime(
        get_inst_config("inst4", Some("net_d"), "10.144.144.4", "fd00::4/64"),
        process_runtime.clone(),
    );
    if proto == "tcp" {
        inst4.add_connector_url("tcp://10.1.2.3:11010".parse().unwrap());
    } else if proto == "wg" {
        inst4.add_connector_url("wg://10.1.2.3:11011".parse().unwrap());
    } else {
        unreachable!("not support");
    }
    inst4.run().await.unwrap();

    tracing::info!("inst1 peer id: {:?}", insts[0].peer_id());
    tracing::info!("inst2 peer id: {:?}", insts[1].peer_id());
    tracing::info!("inst3 peer id: {:?}", insts[2].peer_id());
    tracing::info!("inst4 peer id: {:?}", inst4.peer_id());

    let task = tokio::spawn(async move {
        for _ in 1..=2 {
            // inst4 should be in inst1's route list
            wait_for_condition(
                || async {
                    insts[0]
                        .get_core_instance()
                        .route_snapshots()
                        .await
                        .iter()
                        .any(|r| r.peer_id == inst4.peer_id())
                },
                Duration::from_secs(8),
            )
            .await;

            set_link_status("net_d", false);
            let _t = AbortOnDropHandle::new(tokio::spawn(async move {
                // do some ping in net_a to trigger net_c pingpong
                loop {
                    ping_test("net_a", "10.144.144.4", Some(1)).await;
                }
            }));
            wait_for_condition(
                || async {
                    !insts[2]
                        .get_core_instance()
                        .connected_peers()
                        .await
                        .iter()
                        .any(|r| *r == inst4.peer_id())
                },
                // 0 down, assume last packet is recv in -0.01
                // [2, 7) send ping
                // [4, 9) ping fail and close connection
                Duration::from_secs(11),
            )
            .await;

            wait_for_condition(
                || async {
                    !insts[0]
                        .get_core_instance()
                        .route_snapshots()
                        .await
                        .iter()
                        .any(|r| r.peer_id == inst4.peer_id())
                },
                Duration::from_secs(7),
            )
            .await;

            set_link_status("net_d", true);
        }

        drop_insts(insts).await;
    });

    let (ret,) = tokio::join!(task);
    assert!(ret.is_ok());
}

#[tokio::test]
#[serial_test::serial]
pub async fn udp_broadcast_test() {
    let _insts = init_three_node("tcp").await;

    let udp_broadcast_responder = |net_ns: NetNS, counter: Arc<AtomicU32>| async move {
        let _g = net_ns.guard();
        let socket: UdpSocket = UdpSocket::bind("0.0.0.0:22111").await.unwrap();
        socket.set_broadcast(true).unwrap();

        println!("Awaiting responses..."); // self.recv_buff is a [u8; 8092]
        let mut recv_buff = [0; 8092];
        while let Ok((n, addr)) = socket.recv_from(&mut recv_buff).await {
            println!("{} bytes response from {:?}", n, addr);
            counter.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
            // Remaining code not directly relevant to the question
        }
    };

    let mut tasks = JoinSet::new();
    let counter = Arc::new(AtomicU32::new(0));
    tasks.spawn(udp_broadcast_responder(
        NetNS::new(Some("net_b".into())),
        counter.clone(),
    ));
    tasks.spawn(udp_broadcast_responder(
        NetNS::new(Some("net_c".into())),
        counter.clone(),
    ));

    tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;

    // send broadcast
    let net_ns = NetNS::new(Some("net_a".into()));
    let _g = net_ns.guard();
    let socket: UdpSocket = UdpSocket::bind("0.0.0.0:0").await.unwrap();
    socket.set_broadcast(true).unwrap();
    // socket.connect(("10.144.144.255", 22111)).await.unwrap();
    let call: Vec<u8> = vec![1; 1024];
    println!("Sending call, {} bytes", call.len());
    if let Err(e) = socket.send_to(&call, "10.144.144.255:22111").await {
        panic!("Error sending call: {:?}", e)
    }

    tokio::time::sleep(tokio::time::Duration::from_secs(2)).await;
    assert_eq!(counter.load(std::sync::atomic::Ordering::Relaxed), 2);

    drop_insts(_insts).await;
}

#[tokio::test]
#[serial_test::serial]
pub async fn foreign_network_forward_nic_data() {
    prepare_linux_namespaces();
    let process_runtime = CoreProcessRuntime::new();

    let center_node_config = get_inst_config("inst1", Some("net_a"), "10.144.144.1", "fd00::1/64");
    center_node_config
        .set_network_identity(NetworkIdentity::new("center".to_string(), "".to_string()));
    let mut center_inst =
        Instance::new_with_process_runtime(center_node_config, process_runtime.clone());

    let mut inst1 = Instance::new_with_process_runtime(
        get_inst_config("inst1", Some("net_b"), "10.144.145.1", "fd00:1::1/64"),
        process_runtime.clone(),
    );
    let mut inst2 = Instance::new_with_process_runtime(
        get_inst_config("inst2", Some("net_c"), "10.144.145.2", "fd00:1::2/64"),
        process_runtime,
    );

    center_inst.run().await.unwrap();
    inst1.run().await.unwrap();
    inst2.run().await.unwrap();

    assert_ne!(inst1.ring_listener_url(), center_inst.ring_listener_url());
    assert_ne!(inst2.ring_listener_url(), center_inst.ring_listener_url());

    inst1.add_connector_url(center_inst.ring_listener_url());

    inst2.add_connector_url(center_inst.ring_listener_url());

    wait_for_condition(
        || async {
            inst1.get_core_instance().route_snapshots().await.len() == 2
                && inst2.get_core_instance().route_snapshots().await.len() == 2
        },
        Duration::from_secs(5),
    )
    .await;

    wait_for_condition(
        || async { ping_test("net_b", "10.144.145.2", None).await },
        Duration::from_secs(5),
    )
    .await;

    drop_insts(vec![center_inst, inst1, inst2]).await;
}

use std::{net::SocketAddr, str::FromStr};

use defguard_wireguard_rs::{
    InterfaceConfiguration, WGApi, WireguardInterfaceApi, host::Peer, key::Key, net::IpAddrMask,
};

fn run_wireguard_client(
    endpoint: SocketAddr,
    peer_public_key: Key,
    client_private_key: Key,
    allowed_ips: Vec<String>,
    client_ip: String,
) -> Result<(), Box<dyn std::error::Error>> {
    // Create new API object for interface
    let ifname: String = if cfg!(target_os = "linux") || cfg!(target_os = "freebsd") {
        "wg0".into()
    } else {
        "utun3".into()
    };
    let wgapi = WGApi::new(ifname.clone(), false)?;

    // create interface
    wgapi.create_interface()?;

    // Peer secret key
    let mut peer = Peer::new(peer_public_key.clone());

    tracing::info!("endpoint");
    // Peer endpoint and interval
    peer.endpoint = Some(endpoint);
    peer.persistent_keepalive_interval = Some(1);
    for ip in allowed_ips {
        peer.allowed_ips.push(IpAddrMask::from_str(ip.as_str())?);
    }

    // interface configuration
    let interface_config = InterfaceConfiguration {
        name: ifname.clone(),
        prvkey: client_private_key.to_string(),
        address: client_ip,
        port: 12345,
        peers: vec![peer],
    };

    #[cfg(not(windows))]
    wgapi.configure_interface(&interface_config)?;
    #[cfg(windows)]
    wgapi.configure_interface(&interface_config, &[])?;
    wgapi.configure_peer_routing(&interface_config.peers)?;
    Ok(())
}

#[cfg(feature = "wireguard")]
#[rstest::rstest]
#[tokio::test]
#[serial_test::serial]
pub async fn wireguard_vpn_portal(#[values(true, false)] test_v6: bool) {
    let insts = init_three_node_ex(
        "tcp",
        |config| {
            if config.get_inst_name() == "inst3" {
                config.set_vpn_portal_config(VpnPortalConfig {
                    wireguard_listen: "0.0.0.0:22121".parse().unwrap(),
                    client_cidr: "10.14.14.0/24".parse().unwrap(),
                });
            }
            config
        },
        false,
    )
    .await;

    if test_v6 {
        ping6_test("net_d", "fd12::3", None).await;
    } else {
        ping_test("net_d", "10.1.2.3", None).await;
    }

    let dst_socket_addr = if test_v6 {
        "[fd12::3]:22121".parse().unwrap()
    } else {
        "10.1.2.3:22121".parse().unwrap()
    };

    let net_ns = NetNS::new(Some("net_d".into()));
    let _g = net_ns.guard();
    let wg_cfg = get_wg_config_for_portal(&insts[2].get_global_ctx().get_network_identity());
    run_wireguard_client(
        dst_socket_addr,
        Key::try_from(wg_cfg.my_public_key()).unwrap(),
        Key::try_from(wg_cfg.peer_secret_key()).unwrap(),
        vec!["10.14.14.0/24".to_string(), "10.144.144.0/24".to_string()],
        "10.14.14.2".to_string(),
    )
    .unwrap();

    // ping other node in network
    wait_for_condition(
        || async { ping_test("net_d", "10.144.144.1", None).await },
        Duration::from_secs(5),
    )
    .await;
    wait_for_condition(
        || async { ping_test("net_d", "10.144.144.2", None).await },
        Duration::from_secs(5),
    )
    .await;

    // ping portal node
    wait_for_condition(
        || async { ping_test("net_d", "10.144.144.3", None).await },
        Duration::from_secs(5),
    )
    .await;

    drop_insts(insts).await;
}

#[cfg(feature = "wireguard")]
#[rstest::rstest]
#[tokio::test]
#[serial_test::serial]
pub async fn socks5_vpn_portal(
    #[values("10.144.144.1", "10.144.144.3", "10.1.2.4")] dst_addr: &str,
) {
    use rand::Rng as _;
    use tokio::{
        io::{AsyncReadExt, AsyncWriteExt},
        net::{TcpListener, TcpStream},
    };
    use tokio_socks::tcp::socks5::Socks5Stream;

    let _insts = init_three_node_ex(
        "tcp",
        |cfg| {
            if cfg.get_inst_name() == "inst3" {
                // 添加子网代理配置
                cfg.add_proxy_cidr("10.1.2.0/24".parse().unwrap(), None)
                    .unwrap();
            }
            cfg
        },
        false,
    )
    .await;

    let mut buf = vec![0u8; 1024];
    rand::thread_rng().fill(&mut buf[..]);

    let buf_clone = buf.clone();
    let dst_addr_clone = dst_addr.to_owned();
    let task = tokio::spawn(async move {
        let net_ns = if dst_addr_clone == "10.144.144.1" {
            NetNS::new(Some("net_a".into()))
        } else if dst_addr_clone == "10.144.144.3" {
            NetNS::new(Some("net_c".into()))
        } else {
            NetNS::new(Some("net_d".into()))
        };

        let _g = net_ns.guard();

        let socket = TcpListener::bind("0.0.0.0:22222").await.unwrap();
        let (mut st, addr) = socket.accept().await.unwrap();

        if dst_addr_clone == "10.144.144.1" {
            assert_eq!(addr.ip().to_string(), "127.0.0.1".to_string());
        } else if dst_addr_clone == "10.144.144.3" {
            assert_eq!(addr.ip().to_string(), "10.144.144.1".to_string());
        } else {
            assert_eq!(addr.ip().to_string(), "10.1.2.3".to_string());
        }

        let rbuf = &mut [0u8; 1024];
        st.read_exact(rbuf).await.unwrap();
        assert_eq!(rbuf, buf_clone.as_slice());
    });

    let net_ns = if dst_addr == "10.1.2.4" {
        NetNS::new(Some("net_c".into()))
    } else {
        NetNS::new(Some("net_a".into()))
    };
    let _g = net_ns.guard();

    println!("connect to socks5 portal");
    let stream = TcpStream::connect("127.0.0.1:12345").await.unwrap();
    println!("connect to socks5 portal done");

    stream.set_nodelay(true).unwrap();
    let mut conn = Socks5Stream::connect_with_socket(stream, format!("{}:22222", dst_addr))
        .await
        .unwrap();

    conn.write_all(&buf).await.unwrap();
    drop(conn);

    tokio::join!(task).0.unwrap();

    drop_insts(_insts).await;
}

#[tokio::test]
#[serial_test::serial]
pub async fn foreign_network_functional_cluster() {
    prepare_linux_namespaces();
    let process_runtime = CoreProcessRuntime::new();

    let center_node_config1 = get_inst_config("inst1", Some("net_a"), "10.144.144.1", "fd00::1/64");
    center_node_config1
        .set_network_identity(NetworkIdentity::new("center".to_string(), "".to_string()));
    let mut center_inst1 =
        Instance::new_with_process_runtime(center_node_config1, process_runtime.clone());

    let center_node_config2 = get_inst_config("inst2", Some("net_b"), "10.144.144.2", "fd00::2/64");
    center_node_config2
        .set_network_identity(NetworkIdentity::new("center".to_string(), "".to_string()));
    let mut center_inst2 =
        Instance::new_with_process_runtime(center_node_config2, process_runtime.clone());

    let inst1_config = get_inst_config("inst1", Some("net_c"), "10.144.145.1", "fd00:2::1/64");
    inst1_config.set_listeners(vec![]);
    let mut inst1 = Instance::new_with_process_runtime(inst1_config, process_runtime.clone());

    let mut inst2 = Instance::new_with_process_runtime(
        get_inst_config("inst2", Some("net_d"), "10.144.145.2", "fd00:2::2/64"),
        process_runtime,
    );

    center_inst1.run().await.unwrap();
    center_inst2.run().await.unwrap();
    inst1.run().await.unwrap();
    inst2.run().await.unwrap();

    for instance in [&center_inst1, &center_inst2, &inst1, &inst2] {
        set_foreign_network_refresh_interval(instance, 1).await;
    }

    center_inst1.add_connector_url(center_inst2.ring_listener_url());

    inst1.add_connector_url(center_inst1.ring_listener_url());

    inst2.add_connector_url(center_inst2.ring_listener_url());

    println!(
        "inst1 peer map: {:?}",
        inst1.get_core_instance().route_snapshots().await
    );

    wait_for_condition(
        || async { ping_test("net_c", "10.144.145.2", None).await },
        Duration::from_secs(5),
    )
    .await;

    // connect to two centers, ping should work
    inst1.add_connector_url(center_inst2.ring_listener_url());
    tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
    wait_for_condition(
        || async { ping_test("net_c", "10.144.145.2", None).await },
        Duration::from_secs(5),
    )
    .await;

    drop_insts(vec![center_inst1, center_inst2, inst1, inst2]).await;
}

#[rstest::rstest]
#[tokio::test]
#[serial_test::serial]
pub async fn manual_reconnector(#[values(true, false)] is_foreign: bool) {
    prepare_linux_namespaces();
    let process_runtime = CoreProcessRuntime::new();

    let center_node_config = get_inst_config("inst1", Some("net_a"), "10.144.144.1", "fd00::1/64");
    if is_foreign {
        center_node_config
            .set_network_identity(NetworkIdentity::new("center".to_string(), "".to_string()));
    }
    let mut center_inst =
        Instance::new_with_process_runtime(center_node_config, process_runtime.clone());

    let inst1_config = get_inst_config("inst1", Some("net_b"), "10.144.145.1", "fd00:1::1/64");
    inst1_config.set_listeners(vec![]);
    let mut inst1 = Instance::new_with_process_runtime(inst1_config, process_runtime.clone());

    let mut inst2 = Instance::new_with_process_runtime(
        get_inst_config("inst2", Some("net_c"), "10.144.145.2", "fd00:1::2/64"),
        process_runtime,
    );

    center_inst.run().await.unwrap();
    inst1.run().await.unwrap();
    inst2.run().await.unwrap();

    assert_ne!(inst1.ring_listener_url(), center_inst.ring_listener_url());
    assert_ne!(inst2.ring_listener_url(), center_inst.ring_listener_url());

    inst1.add_connector_url(center_inst.ring_listener_url());

    inst2.add_connector_url(center_inst.ring_listener_url());

    tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;

    let center_inst_peer_id = if !is_foreign {
        center_inst.peer_id()
    } else {
        let network_name = inst1.get_global_ctx().get_network_identity().network_name;
        center_inst
            .get_core_instance()
            .foreign_network_snapshots(false)
            .await
            .get(&network_name)
            .map(|network| network.my_peer_id_for_this_network)
            .unwrap()
    };

    let conns_len = inst1
        .get_core_instance()
        .peer_snapshots()
        .await
        .into_iter()
        .find(|peer| peer.peer_id == center_inst_peer_id)
        .map_or(0, |peer| peer.conns.len());

    assert!(conns_len > 0);

    wait_for_condition(
        || async { ping_test("net_b", "10.144.145.2", None).await },
        Duration::from_secs(5),
    )
    .await;

    drop_insts(vec![center_inst, inst1, inst2]).await;
}

#[rstest::rstest]
#[tokio::test]
#[serial_test::serial]
pub async fn port_forward_test(
    #[values(true, false)] no_tun: bool,
    #[values(64, 1900)] buf_size: u64,
    #[values(true, false)] enable_kcp: bool,
    #[values(true, false)] dst_disable_kcp_input: bool,
    #[values(true, false)] disable_relay_kcp: bool,
) {
    prepare_linux_namespaces();

    let _insts = init_three_node_ex(
        "udp",
        |cfg| {
            if cfg.get_inst_name() == "inst1" {
                cfg.set_port_forwards(vec![
                    // test port forward to other virtual node
                    PortForwardConfig {
                        bind_addr: "0.0.0.0:23456".parse().unwrap(),
                        dst_addr: "10.144.144.3:23456".parse().unwrap(),
                        proto: "tcp".to_string(),
                    },
                    // test port forward to subnet proxy
                    PortForwardConfig {
                        bind_addr: "0.0.0.0:23457".parse().unwrap(),
                        dst_addr: "10.1.2.4:23457".parse().unwrap(),
                        proto: "tcp".to_string(),
                    },
                    // test udp port forward to other virtual node
                    PortForwardConfig {
                        bind_addr: "0.0.0.0:23458".parse().unwrap(),
                        dst_addr: "10.144.144.3:23458".parse().unwrap(),
                        proto: "udp".to_string(),
                    },
                    // test udp port forward to subnet proxy
                    PortForwardConfig {
                        bind_addr: "0.0.0.0:23459".parse().unwrap(),
                        dst_addr: "10.1.2.4:23459".parse().unwrap(),
                        proto: "udp".to_string(),
                    },
                ]);

                let mut flags = cfg.get_flags();
                flags.no_tun = no_tun;
                flags.enable_kcp_proxy = enable_kcp;
                cfg.set_flags(flags);
            } else if cfg.get_inst_name() == "inst3" {
                cfg.add_proxy_cidr("10.1.2.0/24".parse().unwrap(), None)
                    .unwrap();
                let mut flags = cfg.get_flags();
                flags.disable_kcp_input = dst_disable_kcp_input;
                cfg.set_flags(flags);
            } else if cfg.get_inst_name() == "inst2" {
                let mut flags = cfg.get_flags();
                flags.disable_relay_kcp = disable_relay_kcp;
                cfg.set_flags(flags);
            }

            cfg
        },
        false,
    )
    .await;

    let tcp_listener = core_tcp_listener("tcp://0.0.0.0:23456".parse().unwrap());
    let tcp_connector = core_tcp_dialer("tcp://127.0.0.1:23456".parse().unwrap());

    let mut buf = vec![0; buf_size as usize];
    rand::thread_rng().fill(&mut buf[..]);

    _tunnel_pingpong_netns_with_timeout(
        tcp_listener,
        tcp_connector,
        NetNS::new(Some("net_c".into())),
        NetNS::new(Some("net_a".into())),
        buf,
        Duration::from_secs(1),
    )
    .await
    .unwrap();

    let tcp_listener = core_tcp_listener("tcp://0.0.0.0:23457".parse().unwrap());
    let tcp_connector = core_tcp_dialer("tcp://127.0.0.1:23457".parse().unwrap());

    let mut buf = vec![0; buf_size as usize];
    rand::thread_rng().fill(&mut buf[..]);

    _tunnel_pingpong_netns_with_timeout(
        tcp_listener,
        tcp_connector,
        NetNS::new(Some("net_d".into())),
        NetNS::new(Some("net_a".into())),
        buf,
        Duration::from_secs(1),
    )
    .await
    .unwrap();

    let udp_listener = core_udp_listener("udp://0.0.0.0:23458".parse().unwrap());
    let udp_connector = core_udp_dialer("udp://127.0.0.1:23458".parse().unwrap());

    let mut buf = vec![0; buf_size as usize];
    rand::thread_rng().fill(&mut buf[..]);

    _tunnel_pingpong_netns_with_timeout(
        udp_listener,
        udp_connector,
        NetNS::new(Some("net_c".into())),
        NetNS::new(Some("net_a".into())),
        buf,
        Duration::from_secs(1),
    )
    .await
    .unwrap();

    let udp_listener = core_udp_listener("udp://0.0.0.0:23459".parse().unwrap());
    let udp_connector = core_udp_dialer("udp://127.0.0.1:23459".parse().unwrap());

    let mut buf = vec![0; buf_size as usize];
    rand::thread_rng().fill(&mut buf[..]);

    _tunnel_pingpong_netns_with_timeout(
        udp_listener,
        udp_connector,
        NetNS::new(Some("net_d".into())),
        NetNS::new(Some("net_a".into())),
        buf,
        Duration::from_secs(1),
    )
    .await
    .unwrap();

    drop_insts(_insts).await;
}

#[rstest::rstest]
#[case(false, false)]
#[case(true, false)]
#[case(true, true)]
#[serial_test::serial]
#[tokio::test]
pub async fn port_forward_with_inbound_default_drop_acl_test(
    #[case] dhcp: bool,
    #[case] enable_quic_proxy: bool,
) {
    use crate::proto::acl::*;

    let acl = Acl {
        acl_v1: Some(AclV1 {
            chains: vec![Chain {
                name: "drop_unsolicited_inbound".to_string(),
                chain_type: ChainType::Inbound as i32,
                enabled: true,
                default_action: Action::Drop as i32,
                ..Default::default()
            }],
            ..Default::default()
        }),
    };

    let insts = init_three_node_ex(
        "udp",
        |cfg| {
            if cfg.get_inst_name() == "inst1" {
                if dhcp {
                    cfg.set_ipv4(None);
                    cfg.set_dhcp(true);
                }
                cfg.set_acl(Some(acl.clone()));
                cfg.set_port_forwards(vec![
                    PortForwardConfig {
                        bind_addr: "0.0.0.0:23456".parse().unwrap(),
                        dst_addr: "10.144.144.3:23456".parse().unwrap(),
                        proto: "tcp".to_string(),
                    },
                    PortForwardConfig {
                        bind_addr: "0.0.0.0:23457".parse().unwrap(),
                        dst_addr: "10.1.2.4:23457".parse().unwrap(),
                        proto: "tcp".to_string(),
                    },
                ]);

                let mut flags = cfg.get_flags();
                flags.no_tun = true;
                flags.enable_kcp_proxy = false;
                flags.enable_quic_proxy = enable_quic_proxy;
                cfg.set_flags(flags);
            } else if cfg.get_inst_name() == "inst3" {
                cfg.add_proxy_cidr("10.1.2.0/24".parse().unwrap(), None)
                    .unwrap();

                let mut flags = cfg.get_flags();
                flags.disable_kcp_input = true;
                flags.disable_quic_input = !enable_quic_proxy;
                cfg.set_flags(flags);
            } else if cfg.get_inst_name() == "inst2" {
                let mut flags = cfg.get_flags();
                flags.disable_relay_kcp = true;
                cfg.set_flags(flags);
            }

            cfg
        },
        false,
    )
    .await;

    if dhcp {
        wait_for_condition(
            || async { insts[0].get_global_ctx().get_ipv4().is_some() },
            Duration::from_secs(5),
        )
        .await;
    }

    for (bind_port, server_ns) in [(23456, "net_c"), (23457, "net_d")] {
        let tcp_listener = core_tcp_listener(format!("tcp://0.0.0.0:{bind_port}").parse().unwrap());
        let tcp_connector =
            core_tcp_dialer(format!("tcp://127.0.0.1:{bind_port}").parse().unwrap());

        let mut buf = vec![0; 64];
        rand::thread_rng().fill(&mut buf[..]);

        let result = _tunnel_pingpong_netns_with_timeout(
            tcp_listener,
            tcp_connector,
            NetNS::new(Some(server_ns.into())),
            NetNS::new(Some("net_a".into())),
            buf,
            Duration::from_secs(1),
        )
        .await;

        let stats = insts[0].get_core_instance().acl_stats();
        println!(
            "port forward source bind_port={} dhcp={} enable_quic_proxy={} ACL stats: {}",
            bind_port, dhcp, enable_quic_proxy, stats
        );

        assert!(
            result.is_ok(),
            "port-forward TCP should complete through outbound ACL state, bind_port={}, dhcp={}, enable_quic_proxy={}; stats: {}",
            bind_port,
            dhcp,
            enable_quic_proxy,
            stats,
        );
    }

    drop_insts(insts).await;
}

#[rstest::rstest]
#[serial_test::serial]
#[tokio::test]
pub async fn relay_bps_limit_test(#[values(100, 200, 400, 800)] bps_limit: u64) {
    let insts = init_three_node_ex(
        "udp",
        |cfg| {
            if cfg.get_inst_name() == "inst2" {
                cfg.set_network_identity(NetworkIdentity::new(
                    "public".to_string(),
                    "public".to_string(),
                ));
                let mut f = cfg.get_flags();
                f.foreign_relay_bps_limit = bps_limit * 1024;
                cfg.set_flags(f);
            }
            cfg
        },
        true,
    )
    .await;

    // connect to virtual ip (no tun mode)
    let tcp_listener = core_tcp_listener("tcp://0.0.0.0:22223".parse().unwrap());
    let tcp_connector = core_tcp_dialer("tcp://10.144.144.3:22223".parse().unwrap());

    let bps = _tunnel_bench_netns(
        tcp_listener,
        tcp_connector,
        NetNS::new(Some("net_c".into())),
        NetNS::new(Some("net_a".into())),
    )
    .await;

    println!("bps: {}", bps);

    let bps = bps as u64 / 1024;
    assert_limited_payload_bps(bps, bps_limit);

    drop_insts(insts).await;
}

#[rstest::rstest]
#[serial_test::serial]
#[tokio::test]
pub async fn instance_recv_bps_limit_test(#[values(100, 800)] bps_limit: u64) {
    let insts = init_three_node_ex(
        "tcp",
        |cfg| {
            if cfg.get_inst_name() == "inst2" {
                let mut f = cfg.get_flags();
                f.instance_recv_bps_limit = bps_limit * 1024;
                cfg.set_flags(f);
            }
            cfg
        },
        false,
    )
    .await;

    let tcp_listener = core_tcp_listener("tcp://0.0.0.0:22223".parse().unwrap());
    let tcp_connector = core_tcp_dialer("tcp://10.144.144.3:22223".parse().unwrap());

    let bps = _tunnel_bench_netns(
        tcp_listener,
        tcp_connector,
        NetNS::new(Some("net_c".into())),
        NetNS::new(Some("net_a".into())),
    )
    .await;

    println!("bps: {}", bps);

    let bps = bps as u64 / 1024;
    assert_limited_payload_bps(bps, bps_limit);

    drop_insts(insts).await;
}

fn assert_limited_payload_bps(bps: u64, bps_limit: u64) {
    // The benchmark measures TCP application payload while the limiter counts
    // EasyTier data payload, including the inner IP and transport headers.
    let min_bps = bps_limit.saturating_sub((bps_limit / 10).max(50));
    let max_bps = bps_limit + 50;
    assert!(
        bps >= min_bps && bps <= max_bps,
        "bps: {}, expected: {}..={}",
        bps,
        min_bps,
        max_bps
    );
}

async fn assert_peer_admission_blocked(inst: &Instance, url: url::Url) {
    let ip = url
        .host_str()
        .expect("test URL should have a host")
        .parse()
        .expect("test URL should have a literal IP");
    let target = std::net::SocketAddr::new(ip, url.port().expect("test URL should have a port"));
    let host = crate::instance::host::native_instance_host(inst.get_global_ctx());
    let protocol = crate::tunnel::protocol::runtime_client_protocol_upgrader(inst.get_global_ctx());
    let core = inst.get_core_instance();
    let connect = async {
        let connected = match url.scheme() {
            "tcp" => easytier_core::connectivity::transport::ConnectedTransport::Tcp(
                easytier_core::socket::tcp::VirtualTcpSocketFactory::connect_tcp(
                    host.as_ref(),
                    easytier_core::socket::tcp::TcpConnectOptions::direct_connect(target),
                )
                .await?,
            ),
            "udp" => easytier_core::connectivity::transport::ConnectedTransport::Udp(
                easytier_core::connectivity::transport::connect_udp(
                    host,
                    target,
                    Vec::new(),
                    easytier_core::socket::udp::UdpBindOptions::direct_connect(),
                    easytier_core::connectivity::transport::UdpSessionMode::EasyTierMux,
                )
                .await?,
            ),
            scheme => panic!("unsupported test scheme: {scheme}"),
        };
        let tunnel = easytier_core::connectivity::protocol::ClientProtocolUpgrader::upgrade_client(
            protocol.as_ref(),
            connected,
            url,
        )
        .await?;
        core.admit_client_tunnel_for_test(tunnel, true)
            .await
            .map(|_| ())
            .map_err(anyhow::Error::from)
    };
    let result = tokio::time::timeout(Duration::from_millis(100), connect).await;
    assert!(matches!(result, Err(_) | Ok(Err(_))));
}

use std::fs;
use std::io;

fn print_all_fds() -> io::Result<()> {
    let fd_dir = "/proc/self/fd";

    // 读取 /proc/self/fd 目录中的所有条目
    for entry in fs::read_dir(fd_dir)? {
        let entry = entry?;
        let file_name = entry.file_name();
        let fd_str = file_name.to_string_lossy();

        // 尝试解析为数字（跳过 . 和 ..）
        if let Ok(fd_num) = fd_str.parse::<i32>() {
            // 获取文件描述符指向的文件路径（如果可能）
            let target_path = format!("{}/{}", fd_dir, fd_num);
            match fs::read_link(&target_path) {
                Ok(target) => {
                    println!("FD {}: {}", fd_num, target.to_string_lossy());
                }
                Err(e) => {
                    println!("FD {}: (unreadable: {})", fd_num, e);
                }
            }
        }
    }
    Ok(())
}

#[rstest::rstest]
#[serial_test::serial]
#[tokio::test]
async fn avoid_tunnel_loop_back_to_virtual_network(
    #[values(true, false)] no_tun: bool,
    #[values(true, false)] enable_kcp_proxy: bool,
    #[values(true, false)] enable_quic_proxy: bool,
) {
    if enable_kcp_proxy && enable_quic_proxy {
        return;
    }

    let insts = init_three_node_ex(
        "udp",
        |cfg| {
            if matches!(cfg.get_inst_name().as_str(), "inst2" | "inst3") {
                let mut flags = cfg.get_flags();
                flags.no_tun = no_tun;
                cfg.set_flags(flags);
            }

            if cfg.get_inst_name().as_str() == "inst1" {
                let mut flags = cfg.get_flags();
                flags.enable_kcp_proxy = enable_kcp_proxy;
                flags.enable_quic_proxy = enable_quic_proxy;
                cfg.set_flags(flags);
            }

            if cfg.get_inst_name().as_str() == "inst3" {
                cfg.add_proxy_cidr("10.1.2.0/24".parse().unwrap(), None)
                    .unwrap();
            }

            cfg
        },
        false,
    )
    .await;

    assert_peer_admission_blocked(&insts[0], "tcp://10.144.144.2:11010".parse().unwrap()).await;

    assert_peer_admission_blocked(&insts[0], "udp://10.144.144.3:11010".parse().unwrap()).await;

    assert_peer_admission_blocked(&insts[0], "tcp://10.1.2.3:11010".parse().unwrap()).await;

    assert_peer_admission_blocked(&insts[0], "udp://10.1.2.3:11010".parse().unwrap()).await;

    drop_insts(insts).await;

    let _ = print_all_fds();
}

#[rstest::rstest]
#[tokio::test]
#[serial_test::serial]
pub async fn acl_rule_test_inbound(
    #[values(true, false)] enable_kcp_proxy: bool,
    #[values(true, false)] enable_quic_proxy: bool,
) {
    use crate::tunnel::common::tests::_tunnel_pingpong_netns_with_timeout;
    use rand::Rng;
    let insts = init_three_node_ex(
        "udp",
        |cfg| {
            if cfg.get_inst_name() == "inst1" {
                let mut flags = cfg.get_flags();
                flags.enable_kcp_proxy = enable_kcp_proxy;
                flags.enable_quic_proxy = enable_quic_proxy;
                cfg.set_flags(flags);
            }
            cfg
        },
        false,
    )
    .await;

    // 构造 ACL 配置
    use crate::proto::acl::*;
    let mut acl = Acl::default();
    let mut acl_v1 = AclV1::default();

    let mut chain = Chain {
        name: "test_inbound".to_string(),
        chain_type: ChainType::Inbound as i32,
        enabled: true,
        ..Default::default()
    };

    // 禁止 8080
    let deny_rule = Rule {
        name: "deny_8080".to_string(),
        priority: 200,
        enabled: true,
        action: Action::Drop as i32,
        protocol: Protocol::Any as i32,
        ports: vec!["8080".to_string()],
        ..Default::default()
    };
    chain.rules.push(deny_rule);

    // 允许其他
    let allow_rule = Rule {
        name: "allow_all".to_string(),
        priority: 100,
        enabled: true,
        action: Action::Allow as i32,
        protocol: Protocol::Any as i32,
        stateful: true,
        ..Default::default()
    };
    chain.rules.push(allow_rule);

    // 禁止 src ip 为 10.144.144.2 的流量
    let deny_rule = Rule {
        name: "deny_10.144.144.2".to_string(),
        priority: 200,
        enabled: true,
        action: Action::Drop as i32,
        protocol: Protocol::Any as i32,
        source_ips: vec!["10.144.144.2/32".to_string()],
        ..Default::default()
    };
    chain.rules.push(deny_rule);

    acl_v1.chains.push(chain);
    acl.acl_v1 = Some(acl_v1);

    // convert acl to to toml
    let acl_toml = toml::to_string(&acl).unwrap();
    println!("ACL TOML: {}", acl_toml);

    reload_instance_acl(&insts[2], Some(&acl)).await;

    // TCP 测试部分
    {
        // 2. 在 inst2 上监听 8080 和 8081
        let listener_8080 = core_tcp_listener("tcp://0.0.0.0:8080".parse().unwrap());
        let listener_8081 = core_tcp_listener("tcp://0.0.0.0:8081".parse().unwrap());
        let listener_8082 = core_tcp_listener("tcp://0.0.0.0:8082".parse().unwrap());

        // 3. inst1 作为客户端，尝试连接 inst2 的 8080（应被拒绝）和 8081（应被允许）
        let connector_8080 =
            core_tcp_dialer(format!("tcp://{}:8080", "10.144.144.3").parse().unwrap());
        let connector_8081 =
            core_tcp_dialer(format!("tcp://{}:8081", "10.144.144.3").parse().unwrap());
        let connector_8082 =
            core_tcp_dialer(format!("tcp://{}:8082", "10.144.144.3").parse().unwrap());

        // 4. 构造测试数据
        let mut buf = vec![0; 32];
        rand::thread_rng().fill(&mut buf[..]);

        // 5. 8081 应该可以 pingpong 成功
        let result = _tunnel_pingpong_netns_with_timeout(
            listener_8081,
            connector_8081,
            NetNS::new(Some("net_c".into())),
            NetNS::new(Some("net_a".into())),
            buf.clone(),
            Duration::from_secs(5),
        )
        .await;
        assert!(result.is_ok(), "{}", result.unwrap_err());

        // 6. 8080 应该连接失败（被 ACL 拦截）
        let result = _tunnel_pingpong_netns_with_timeout(
            listener_8080,
            connector_8080,
            NetNS::new(Some("net_c".into())),
            NetNS::new(Some("net_a".into())),
            buf.clone(),
            Duration::from_millis(500),
        )
        .await;

        assert!(result.is_err(), "TCP 连接 8080 应被 ACL 拦截，不能成功");

        // 7. 从 10.144.144.2 连接 8082 应该连接失败（被 ACL 拦截）
        let result = _tunnel_pingpong_netns_with_timeout(
            listener_8082,
            connector_8082,
            NetNS::new(Some("net_c".into())),
            NetNS::new(Some("net_b".into())),
            buf.clone(),
            Duration::from_millis(500),
        )
        .await;

        assert!(result.is_err(), "TCP 连接 8082 应被 ACL 拦截，不能成功");

        let stats = insts[2].get_core_instance().acl_stats();
        println!("stats: {:?}", stats);
    }

    // UDP 测试部分
    {
        // 1. 在 inst2 上监听 UDP 8080 和 8081
        let listener_8080 = core_udp_listener("udp://0.0.0.0:8080".parse().unwrap());
        let listener_8081 = core_udp_listener("udp://0.0.0.0:8081".parse().unwrap());

        // 2. inst1 作为客户端，尝试连接 inst2 的 8080（应被拒绝）和 8081（应被允许）
        let connector_8080 =
            core_udp_dialer(format!("udp://{}:8080", "10.144.144.3").parse().unwrap());
        let connector_8081 =
            core_udp_dialer(format!("udp://{}:8081", "10.144.144.3").parse().unwrap());

        // 3. 构造测试数据
        let mut buf = vec![0; 32];
        rand::thread_rng().fill(&mut buf[..]);

        // 4. 8081 应该可以 pingpong 成功
        let result = _tunnel_pingpong_netns_with_timeout(
            listener_8081,
            connector_8081,
            NetNS::new(Some("net_c".into())),
            NetNS::new(Some("net_a".into())),
            buf.clone(),
            Duration::from_secs(5),
        )
        .await;
        assert!(result.is_ok(), "{}", result.unwrap_err());

        // 5. 8080 应该连接失败（被 ACL 拦截）
        let result = _tunnel_pingpong_netns_with_timeout(
            listener_8080,
            connector_8080,
            NetNS::new(Some("net_c".into())),
            NetNS::new(Some("net_a".into())),
            buf.clone(),
            Duration::from_millis(500),
        )
        .await;

        assert!(result.is_err(), "UDP 连接 8080 应被 ACL 拦截，不能成功");

        let stats = insts[2].get_core_instance().acl_stats();
        println!("stats: {}", stats);
    }

    // remove acl, 8080 should succ
    reload_instance_acl(&insts[2], None).await;

    drop_insts(insts).await;
}

#[rstest::rstest]
#[tokio::test]
#[serial_test::serial]
pub async fn acl_rule_test_subnet_proxy(
    #[values(true, false)] enable_kcp_proxy: bool,
    #[values(true, false)] enable_quic_proxy: bool,
) {
    use crate::tunnel::common::tests::_tunnel_pingpong_netns_with_timeout;
    use rand::Rng;

    let insts = init_three_node_ex(
        "udp",
        |cfg| {
            if cfg.get_inst_name() == "inst1" {
                let mut flags = cfg.get_flags();
                flags.enable_kcp_proxy = enable_kcp_proxy;
                flags.enable_quic_proxy = enable_quic_proxy;
                cfg.set_flags(flags);
            } else if cfg.get_inst_name() == "inst3" {
                // 添加子网代理配置
                cfg.add_proxy_cidr("10.1.2.0/24".parse().unwrap(), None)
                    .unwrap();
            }
            cfg
        },
        false,
    )
    .await;

    // 等待代理路由出现
    wait_proxy_route_appear(
        &insts[0].get_core_instance(),
        "10.144.144.3/24",
        insts[2].peer_id(),
        "10.1.2.0/24",
    )
    .await;

    // Test IPv4 connectivity
    wait_for_condition(
        || async { ping_test("net_a", "10.1.2.4", None).await },
        Duration::from_secs(5),
    )
    .await;

    // 构造 ACL 配置 - 针对子网代理流量
    use crate::proto::acl::*;
    let mut acl = Acl::default();
    let mut acl_v1 = AclV1::default();

    let mut chain = Chain {
        name: "test_subnet_proxy_inbound".to_string(),
        chain_type: ChainType::Forward as i32,
        enabled: true,
        ..Default::default()
    };

    // 禁止访问子网代理中的 8080 端口
    let deny_rule = Rule {
        name: "deny_subnet_8080".to_string(),
        priority: 200,
        enabled: true,
        action: Action::Drop as i32,
        protocol: Protocol::Any as i32,
        ports: vec!["8080".to_string()],
        destination_ips: vec!["10.1.2.0/24".to_string()],
        ..Default::default()
    };
    chain.rules.push(deny_rule);

    // 禁止来自 inst1 (10.144.144.1) 访问子网代理中的 8081 端口
    let deny_src_rule = Rule {
        name: "deny_inst1_to_subnet_8081".to_string(),
        priority: 200,
        enabled: true,
        action: Action::Drop as i32,
        protocol: Protocol::Any as i32,
        ports: vec!["8081".to_string()],
        source_ips: vec!["10.144.144.1/32".to_string()],
        destination_ips: vec!["10.1.2.0/24".to_string()],
        ..Default::default()
    };
    chain.rules.push(deny_src_rule);

    // 允许其他流量
    let allow_rule = Rule {
        name: "allow_all".to_string(),
        priority: 100,
        enabled: true,
        action: Action::Allow as i32,
        protocol: Protocol::Any as i32,
        stateful: true,
        ..Default::default()
    };
    chain.rules.push(allow_rule);

    acl_v1.chains.push(chain);
    acl.acl_v1 = Some(acl_v1);

    // 在 inst3 上应用 ACL 规则
    reload_instance_acl(&insts[2], Some(&acl)).await;

    // TCP 测试部分 - 测试子网代理的 ACL 规则
    {
        // 在 net_d (10.1.2.4) 上监听多个端口
        let listener_8080 = core_tcp_listener("tcp://0.0.0.0:8080".parse().unwrap());
        let listener_8081 = core_tcp_listener("tcp://0.0.0.0:8081".parse().unwrap());
        let listener_8082 = core_tcp_listener("tcp://0.0.0.0:8082".parse().unwrap());

        // 从 inst1 (net_a) 连接到子网代理
        let connector_8080 = core_tcp_dialer("tcp://10.1.2.4:8080".parse().unwrap());
        let connector_8081 = core_tcp_dialer("tcp://10.1.2.4:8081".parse().unwrap());
        let connector_8082 = core_tcp_dialer("tcp://10.1.2.4:8082".parse().unwrap());

        let mut buf = vec![0; 32];
        rand::thread_rng().fill(&mut buf[..]);

        // 8082 应该可以连接成功（不被 ACL 拦截）
        let result = _tunnel_pingpong_netns_with_timeout(
            listener_8082,
            connector_8082,
            NetNS::new(Some("net_d".into())),
            NetNS::new(Some("net_a".into())),
            buf.clone(),
            Duration::from_secs(5),
        )
        .await;
        assert!(result.is_ok(), "{}", result.unwrap_err());

        // 8080 应该连接失败（被 ACL 拦截 - 禁止访问子网代理的 8080）
        let result = _tunnel_pingpong_netns_with_timeout(
            listener_8080,
            connector_8080,
            NetNS::new(Some("net_d".into())),
            NetNS::new(Some("net_a".into())),
            buf.clone(),
            Duration::from_millis(500),
        )
        .await;

        assert!(
            result.is_err(),
            "TCP 连接子网代理 8080 应被 ACL 拦截，不能成功"
        );

        // 8081 应该连接失败（被 ACL 拦截 - 禁止 inst1 访问子网代理的 8081）
        let result = _tunnel_pingpong_netns_with_timeout(
            listener_8081,
            connector_8081,
            NetNS::new(Some("net_d".into())),
            NetNS::new(Some("net_a".into())),
            buf.clone(),
            Duration::from_millis(500),
        )
        .await;

        assert!(
            result.is_err(),
            "TCP 连接子网代理 8081 应被 ACL 拦截，不能成功"
        );

        let stats = insts[2].get_core_instance().acl_stats();
        println!("ACL stats after TCP tests: {:?}", stats);
    }

    // UDP 测试部分 - 测试子网代理的 ACL 规则
    {
        let listener_8080 = core_udp_listener("udp://0.0.0.0:8080".parse().unwrap());
        let listener_8082 = core_udp_listener("udp://0.0.0.0:8082".parse().unwrap());

        let connector_8080 = core_udp_dialer("udp://10.1.2.4:8080".parse().unwrap());
        let connector_8082 = core_udp_dialer("udp://10.1.2.4:8082".parse().unwrap());

        let mut buf = vec![0; 32];
        rand::thread_rng().fill(&mut buf[..]);

        // 8082 应该可以连接成功
        let result = _tunnel_pingpong_netns_with_timeout(
            listener_8082,
            connector_8082,
            NetNS::new(Some("net_d".into())),
            NetNS::new(Some("net_a".into())),
            buf.clone(),
            Duration::from_secs(5),
        )
        .await;
        assert!(result.is_ok(), "{}", result.unwrap_err());

        // 8080 应该连接失败（被 ACL 拦截）
        let result = _tunnel_pingpong_netns_with_timeout(
            listener_8080,
            connector_8080,
            NetNS::new(Some("net_d".into())),
            NetNS::new(Some("net_a".into())),
            buf.clone(),
            Duration::from_millis(500),
        )
        .await;

        let stats = insts[2].get_core_instance().acl_stats();
        println!("ACL stats after UDP tests: {}", stats);

        assert!(
            result.is_err(),
            "UDP 连接子网代理 8080 应被 ACL 拦截，不能成功"
        );
    }

    // 测试 ICMP 到子网代理（应该被拒绝，因为 Any 协议被拒绝）
    tokio::spawn(wait_for_condition(
        || async { ping_test("net_a", "10.1.2.4", None).await },
        Duration::from_secs(1),
    ))
    .await
    .unwrap_err();

    // 移除 ACL 规则
    reload_instance_acl(&insts[2], None).await;

    // 验证移除 ACL 后，ICMP 可以正常工作
    wait_for_condition(
        || async { ping_test("net_a", "10.1.2.4", None).await },
        Duration::from_secs(5),
    )
    .await;

    drop_insts(insts).await;
}

async fn assert_panics_ext<F, Fut>(f: F, expect_panic: bool)
where
    F: FnOnce() -> Fut + Send + 'static,
    Fut: Future + Send + 'static,
{
    // Run the async function in a separate task so panics surface as JoinError
    let res = tokio::spawn(async move {
        f().await;
    })
    .await;

    if expect_panic {
        assert!(
            res.is_err() && res.as_ref().unwrap_err().is_panic(),
            "Expected function to panic, but it didn't",
        );
    } else {
        assert!(res.is_ok(), "Expected function not to panic, but it did");
    }
}

async fn wait_route_cost(inst: &Instance, peer_id: u32, cost: i32, timeout: Duration) {
    let core = inst.get_core_instance();
    wait_for_condition(
        move || {
            let core = core.clone();
            async move {
                core.route_snapshots()
                    .await
                    .iter()
                    .any(|route| route.peer_id == peer_id && route.cost == cost)
            }
        },
        timeout,
    )
    .await;
}

#[rstest::rstest]
#[tokio::test]
#[serial_test::serial]
pub async fn p2p_only_test(
    #[values(true, false)] has_p2p_conn: bool,
    #[values(true, false)] enable_kcp_proxy: bool,
    #[values(true, false)] enable_quic_proxy: bool,
) {
    let insts = init_three_node_ex(
        "udp",
        |cfg| {
            if cfg.get_inst_name() == "inst1" {
                let mut flags = cfg.get_flags();
                flags.enable_kcp_proxy = enable_kcp_proxy;
                flags.enable_quic_proxy = enable_quic_proxy;
                flags.disable_p2p = true;
                flags.p2p_only = true;
                cfg.set_flags(flags);
            } else if cfg.get_inst_name() == "inst3" {
                // 添加子网代理配置
                cfg.add_proxy_cidr("10.1.2.0/24".parse().unwrap(), None)
                    .unwrap();
            }
            cfg
        },
        false,
    )
    .await;

    if has_p2p_conn {
        insts[2].add_connector_url(insts[0].ring_listener_url());
        wait_route_cost(&insts[2], insts[0].peer_id(), 1, Duration::from_secs(5)).await;
    }

    let target_ip = "10.1.2.4";

    for target_ip in ["10.144.144.3", target_ip] {
        assert_panics_ext(
            || async {
                subnet_proxy_test_icmp(target_ip, Duration::from_millis(100)).await;
            },
            !has_p2p_conn,
        )
        .await;

        let listen_ip = if target_ip == "10.144.144.3" {
            "0.0.0.0"
        } else {
            "10.1.2.4"
        };
        assert_panics_ext(
            || async {
                subnet_proxy_test_tcp(listen_ip, target_ip, Duration::from_millis(100)).await;
            },
            !has_p2p_conn,
        )
        .await;

        assert_panics_ext(
            || async {
                subnet_proxy_test_udp(listen_ip, target_ip, Duration::from_millis(100)).await;
            },
            !has_p2p_conn,
        )
        .await;
    }
}

#[rstest::rstest]
#[tokio::test]
#[serial_test::serial]
pub async fn acl_group_base_test(
    #[values("tcp", "udp")] protocol: &str,
    #[values(true, false)] enable_kcp_proxy: bool,
    #[values(true, false)] enable_quic_proxy: bool,
) {
    use crate::tunnel::common::tests::_tunnel_pingpong_netns_with_timeout;
    use rand::Rng;

    // 构造 ACL 配置，包含组信息
    use crate::proto::acl::*;

    // 设置组信息
    let group_declares = vec![
        GroupIdentity {
            group_name: "admin".to_string(),
            group_secret: "admin-secret".to_string(),
        },
        GroupIdentity {
            group_name: "user".to_string(),
            group_secret: "user-secret".to_string(),
        },
    ];

    let mut chain = Chain {
        name: "group_acl_test".to_string(),
        chain_type: ChainType::Inbound as i32,
        enabled: true,
        default_action: Action::Drop as i32,
        ..Default::default()
    };

    // 规则1: 允许admin组访问所有端口
    let admin_allow_rule = Rule {
        name: "allow_admin_all".to_string(),
        priority: 300,
        enabled: true,
        action: Action::Allow as i32,
        protocol: Protocol::Any as i32,
        source_groups: vec!["admin".to_string()],
        stateful: true,
        ..Default::default()
    };
    chain.rules.push(admin_allow_rule);

    // 规则2: 允许user组访问8080端口
    let user_8080_rule = Rule {
        name: "allow_user_8080".to_string(),
        priority: 200,
        enabled: true,
        action: Action::Allow as i32,
        protocol: Protocol::Any as i32,
        source_groups: vec!["user".to_string()],
        ports: vec!["8080".to_string()],
        stateful: true,
        ..Default::default()
    };
    chain.rules.push(user_8080_rule);

    let acl_admin = Acl {
        acl_v1: Some(AclV1 {
            group: Some(GroupInfo {
                declares: group_declares.clone(),
                members: vec!["admin".to_string()],
            }),
            ..AclV1::default()
        }),
    };

    let acl_user = Acl {
        acl_v1: Some(AclV1 {
            group: Some(GroupInfo {
                declares: group_declares.clone(),
                members: vec!["user".to_string()],
            }),
            ..AclV1::default()
        }),
    };

    let acl_target = Acl {
        acl_v1: Some(AclV1 {
            chains: vec![chain.clone()],
            group: Some(GroupInfo {
                declares: group_declares.clone(),
                members: vec![],
            }),
        }),
    };

    let insts = init_three_node_ex(
        protocol,
        move |cfg| {
            match cfg.get_inst_name().as_str() {
                "inst1" => {
                    cfg.set_acl(Some(acl_admin.clone()));
                }
                "inst2" => {
                    cfg.set_acl(Some(acl_user.clone()));
                }
                "inst3" => {
                    cfg.set_acl(Some(acl_target.clone()));
                }
                _ => {}
            }

            let mut flags = cfg.get_flags();
            flags.enable_kcp_proxy = enable_kcp_proxy;
            flags.enable_quic_proxy = enable_quic_proxy;
            cfg.set_flags(flags);

            cfg
        },
        false,
    )
    .await;

    println!("Testing group-based ACL rules...");

    let make_listener = |port: u16| -> Box<dyn SocketListener<Accepted = Box<dyn Tunnel>> + Sync> {
        match protocol {
            "tcp" => Box::new(core_tcp_listener(
                format!("tcp://0.0.0.0:{}", port).parse().unwrap(),
            )),
            "udp" => Box::new(core_udp_listener(
                format!("udp://0.0.0.0:{}", port).parse().unwrap(),
            )),
            _ => panic!("unsupported protocol: {}", protocol),
        }
    };

    let make_connector = |port: u16| -> Box<dyn TunnelDialer> {
        match protocol {
            "tcp" => Box::new(core_tcp_dialer(
                format!("tcp://10.144.144.3:{}", port).parse().unwrap(),
            )),
            "udp" => Box::new(core_udp_dialer(
                format!("udp://10.144.144.3:{}", port).parse().unwrap(),
            )),
            _ => panic!("unsupported protocol: {}", protocol),
        }
    };

    // 构造测试数据
    let mut buf = vec![0; 32];
    rand::thread_rng().fill(&mut buf[..]);

    // 测试1: inst1 (admin组) 访问8080 - 应该成功
    let result = _tunnel_pingpong_netns_with_timeout(
        make_listener(8080),
        make_connector(8080),
        NetNS::new(Some("net_c".into())),
        NetNS::new(Some("net_a".into())),
        buf.clone(),
        std::time::Duration::from_millis(30000),
    )
    .await;
    assert!(
        result.is_ok(),
        "Admin group access to port 8080 should be allowed (protocol={})",
        protocol
    );
    println!(
        "✓ Admin group access to port 8080 succeeded ({})\n",
        protocol
    );

    // 测试2: inst1 (admin组) 访问8081 - 应该成功
    let result = _tunnel_pingpong_netns_with_timeout(
        make_listener(8081),
        make_connector(8081),
        NetNS::new(Some("net_c".into())),
        NetNS::new(Some("net_a".into())),
        buf.clone(),
        std::time::Duration::from_millis(30000),
    )
    .await;
    assert!(
        result.is_ok(),
        "Admin group access to port 8081 should be allowed (protocol={})",
        protocol
    );
    println!(
        "✓ Admin group access to port 8081 succeeded ({})\n",
        protocol
    );

    // 测试3: inst2 (user组) 访问8080 - 应该成功
    let result = _tunnel_pingpong_netns_with_timeout(
        make_listener(8080),
        make_connector(8080),
        NetNS::new(Some("net_c".into())),
        NetNS::new(Some("net_b".into())),
        buf.clone(),
        std::time::Duration::from_millis(30000),
    )
    .await;
    assert!(
        result.is_ok(),
        "User group access to port 8080 should be allowed (protocol={})",
        protocol
    );
    println!(
        "✓ User group access to port 8080 succeeded ({})\n",
        protocol
    );

    // 测试4: inst2 (user组) 访问8081 - 应该失败
    let result = _tunnel_pingpong_netns_with_timeout(
        make_listener(8081),
        make_connector(8081),
        NetNS::new(Some("net_c".into())),
        NetNS::new(Some("net_b".into())),
        buf.clone(),
        std::time::Duration::from_millis(200),
    )
    .await;
    assert!(
        result.is_err(),
        "User group access to port 8081 should be blocked (protocol={})",
        protocol
    );
    println!(
        "✓ User group access to port 8081 blocked as expected ({})\n",
        protocol
    );

    let stats = insts[2].get_core_instance().acl_stats();
    println!("ACL stats after group {} tests: {:?}", protocol, stats);

    println!("✓ All group-based ACL tests completed successfully");

    drop_insts(insts).await;
}

#[tokio::test]
#[serial_test::serial]
pub async fn lazy_p2p_builds_direct_connection_on_demand() {
    let insts = init_lazy_p2p_three_node_ex("udp", |cfg| {
        if cfg.get_inst_name() == "inst1" {
            let mut flags = cfg.get_flags();
            flags.lazy_p2p = true;
            cfg.set_flags(flags);
        }
        cfg
    })
    .await;

    let inst3_peer_id = insts[2].peer_id();
    assert!(
        !insts[0]
            .get_core_instance()
            .connected_peers()
            .await
            .contains(&inst3_peer_id),
        "inst1 should not proactively connect to inst3 when lazy_p2p is enabled"
    );
    wait_route_cost(&insts[0], inst3_peer_id, 2, Duration::from_secs(5)).await;

    assert!(
        ping_test("net_a", "10.144.144.3", None).await,
        "initial relay traffic should still succeed"
    );

    wait_for_condition(
        || async {
            insts[0]
                .get_core_instance()
                .connected_peers()
                .await
                .contains(&inst3_peer_id)
        },
        Duration::from_secs(10),
    )
    .await;
    wait_route_cost(&insts[0], inst3_peer_id, 1, Duration::from_secs(10)).await;

    drop_insts(insts).await;
}

#[tokio::test]
#[serial_test::serial]
pub async fn need_p2p_overrides_lazy_p2p() {
    let insts = init_lazy_p2p_three_node_ex("udp", |cfg| {
        let mut flags = cfg.get_flags();
        if cfg.get_inst_name() == "inst1" {
            flags.lazy_p2p = true;
        }
        if cfg.get_inst_name() == "inst3" {
            flags.need_p2p = true;
        }
        cfg.set_flags(flags);
        cfg
    })
    .await;

    let inst3_peer_id = insts[2].peer_id();
    wait_route_cost(&insts[0], inst3_peer_id, 2, Duration::from_secs(5)).await;
    wait_for_condition(
        || async {
            insts[0]
                .get_core_instance()
                .connected_peers()
                .await
                .contains(&inst3_peer_id)
        },
        Duration::from_secs(10),
    )
    .await;
    wait_route_cost(&insts[0], inst3_peer_id, 1, Duration::from_secs(10)).await;

    drop_insts(insts).await;
}

#[tokio::test]
#[serial_test::serial]
pub async fn disable_p2p_still_connects_to_need_p2p_peers() {
    let insts = init_lazy_p2p_three_node_ex("udp", |cfg| {
        let mut flags = cfg.get_flags();
        if cfg.get_inst_name() == "inst1" {
            flags.disable_p2p = true;
        }
        if cfg.get_inst_name() == "inst3" {
            flags.need_p2p = true;
        }
        cfg.set_flags(flags);
        cfg
    })
    .await;

    let inst3_peer_id = insts[2].peer_id();
    wait_route_cost(&insts[0], inst3_peer_id, 2, Duration::from_secs(5)).await;
    wait_for_condition(
        || async {
            insts[0]
                .get_core_instance()
                .connected_peers()
                .await
                .contains(&inst3_peer_id)
        },
        Duration::from_secs(10),
    )
    .await;
    wait_route_cost(&insts[0], inst3_peer_id, 1, Duration::from_secs(10)).await;

    drop_insts(insts).await;
}

#[tokio::test]
#[serial_test::serial]
pub async fn ordinary_nodes_do_not_proactively_connect_to_disable_p2p_peers() {
    let insts = init_lazy_p2p_three_node_ex("udp", |cfg| {
        if cfg.get_inst_name() == "inst3" {
            let mut flags = cfg.get_flags();
            flags.disable_p2p = true;
            cfg.set_flags(flags);
        }
        cfg
    })
    .await;

    let inst3_peer_id = insts[2].peer_id();
    wait_route_cost(&insts[0], inst3_peer_id, 2, Duration::from_secs(5)).await;
    assert!(
        ping_test("net_a", "10.144.144.3", None).await,
        "relay traffic to disable-p2p peers should still succeed"
    );

    tokio::time::sleep(Duration::from_secs(3)).await;

    assert!(
        !insts[0]
            .get_core_instance()
            .connected_peers()
            .await
            .contains(&inst3_peer_id),
        "ordinary nodes should not proactively establish p2p with disable-p2p peers"
    );
    wait_route_cost(&insts[0], inst3_peer_id, 2, Duration::from_secs(3)).await;

    drop_insts(insts).await;
}

#[tokio::test]
#[serial_test::serial]
pub async fn lazy_p2p_warms_up_before_p2p_only_send() {
    let insts = init_lazy_p2p_three_node_ex("udp", |cfg| {
        if cfg.get_inst_name() == "inst1" {
            let mut flags = cfg.get_flags();
            flags.lazy_p2p = true;
            flags.p2p_only = true;
            cfg.set_flags(flags);
        }
        cfg
    })
    .await;

    let inst3_peer_id = insts[2].peer_id();
    wait_route_cost(&insts[0], inst3_peer_id, 2, Duration::from_secs(5)).await;
    assert!(
        !ping_test("net_a", "10.144.144.3", None).await,
        "the first send should still fail under p2p_only before direct connectivity exists"
    );

    wait_for_condition(
        || async {
            insts[0]
                .get_core_instance()
                .connected_peers()
                .await
                .contains(&inst3_peer_id)
        },
        Duration::from_secs(10),
    )
    .await;
    wait_route_cost(&insts[0], inst3_peer_id, 1, Duration::from_secs(10)).await;

    wait_for_condition(
        || async { ping_test("net_a", "10.144.144.3", None).await },
        Duration::from_secs(6),
    )
    .await;

    drop_insts(insts).await;
}

#[rstest::rstest]
#[tokio::test]
#[serial_test::serial]
pub async fn acl_group_self_test(
    #[values("tcp", "udp")] protocol: &str,
    #[values(true, false)] enable_kcp_proxy: bool,
    #[values(true, false)] enable_quic_proxy: bool,
) {
    use crate::tunnel::common::tests::_tunnel_pingpong_netns_with_timeout;
    use rand::Rng;

    // 构造 ACL 配置，包含组信息
    use crate::proto::acl::*;

    // 设置组信息
    let group_declares = vec![GroupIdentity {
        group_name: "admin".to_string(),
        group_secret: "admin-secret".to_string(),
    }];

    let mut chain = Chain {
        name: "group_acl_test".to_string(),
        chain_type: ChainType::Inbound as i32,
        enabled: true,
        default_action: Action::Drop as i32,
        ..Default::default()
    };

    // 规则1: 允许admin组访问admin组
    let admin_allow_rule = Rule {
        name: "allow_admin_admin".to_string(),
        priority: 300,
        enabled: true,
        action: Action::Allow as i32,
        protocol: Protocol::Any as i32,
        source_groups: vec!["admin".to_string()],
        destination_groups: vec!["admin".to_string()],
        stateful: true,
        ..Default::default()
    };
    chain.rules.push(admin_allow_rule);

    let acl_admin = Acl {
        acl_v1: Some(AclV1 {
            chains: vec![chain.clone()],
            group: Some(GroupInfo {
                declares: group_declares.clone(),
                members: vec!["admin".to_string()],
            }),
        }),
    };

    let acl_common = Acl {
        acl_v1: Some(AclV1 {
            chains: vec![chain.clone()],
            group: Some(GroupInfo {
                declares: group_declares.clone(),
                members: vec![],
            }),
        }),
    };

    let insts = init_three_node_ex(
        protocol,
        move |cfg| {
            match cfg.get_inst_name().as_str() {
                "inst1" => {
                    cfg.set_acl(Some(acl_admin.clone()));
                }
                "inst2" => {
                    cfg.set_acl(Some(acl_common.clone()));
                }
                "inst3" => {
                    cfg.set_acl(Some(acl_admin.clone()));
                }
                _ => {}
            }

            let mut flags = cfg.get_flags();
            flags.enable_kcp_proxy = enable_kcp_proxy;
            flags.enable_quic_proxy = enable_quic_proxy;
            cfg.set_flags(flags);

            cfg
        },
        false,
    )
    .await;

    println!("Testing group-based ACL rules...");

    let make_listener = |port: u16| -> Box<dyn SocketListener<Accepted = Box<dyn Tunnel>> + Sync> {
        match protocol {
            "tcp" => Box::new(core_tcp_listener(
                format!("tcp://0.0.0.0:{}", port).parse().unwrap(),
            )),
            "udp" => Box::new(core_udp_listener(
                format!("udp://0.0.0.0:{}", port).parse().unwrap(),
            )),
            _ => panic!("unsupported protocol: {}", protocol),
        }
    };

    let make_connector = |port: u16| -> Box<dyn TunnelDialer> {
        match protocol {
            "tcp" => Box::new(core_tcp_dialer(
                format!("tcp://10.144.144.3:{}", port).parse().unwrap(),
            )),
            "udp" => Box::new(core_udp_dialer(
                format!("udp://10.144.144.3:{}", port).parse().unwrap(),
            )),
            _ => panic!("unsupported protocol: {}", protocol),
        }
    };

    // 构造测试数据
    let mut buf = vec![0; 32];
    rand::thread_rng().fill(&mut buf[..]);

    // 测试1: inst1 (admin组) 访问inst3 (admin组) - 应该成功
    let result = _tunnel_pingpong_netns_with_timeout(
        make_listener(8080),
        make_connector(8080),
        NetNS::new(Some("net_c".into())),
        NetNS::new(Some("net_a".into())),
        buf.clone(),
        std::time::Duration::from_millis(30000),
    )
    .await;
    assert!(
        result.is_ok(),
        "Admin group access to Admin group should be allowed (protocol={})",
        protocol
    );
    println!(
        "✓ Admin group access to Admin group succeeded ({})\n",
        protocol
    );

    // 测试2: inst2 (无组) 访问inst3 (admin组) - 应该失败
    let result = _tunnel_pingpong_netns_with_timeout(
        make_listener(8080),
        make_connector(8080),
        NetNS::new(Some("net_c".into())),
        NetNS::new(Some("net_b".into())),
        buf.clone(),
        std::time::Duration::from_millis(200),
    )
    .await;
    assert!(
        result.is_err(),
        "None group access to inst3 (admin group) should be blocked (protocol={})",
        protocol
    );
    println!(
        "✓ None group access to inst3 (admin group) blocked as expected ({})\n",
        protocol
    );

    let stats = insts[2].get_core_instance().acl_stats();
    println!("ACL stats after group {} tests: {:?}", protocol, stats);

    println!("✓ All group-based ACL tests completed successfully");

    drop_insts(insts).await;
}

#[rstest::rstest]
#[tokio::test]
#[serial_test::serial]
pub async fn whitelist_test(
    #[values("tcp", "udp")] protocol: &str,
    #[values(true, false)] test_outbound_allow_list: bool,
) {
    let port = 44553;
    let acl_configured_inst = if test_outbound_allow_list {
        "inst1"
    } else {
        "inst3"
    };
    let insts = init_three_node_ex(
        protocol,
        move |cfg| {
            let port = if test_outbound_allow_list { 0 } else { port };
            if cfg.get_inst_name() == acl_configured_inst {
                if protocol == "tcp" {
                    cfg.set_tcp_whitelist(vec![format!("{}", port)]);
                } else if protocol == "udp" {
                    cfg.set_udp_whitelist(vec![format!("{}", port)]);
                }
            }
            cfg
        },
        false,
    )
    .await;

    use crate::tunnel::common::tests::_tunnel_pingpong_netns_with_timeout;
    use rand::Rng;

    let make_listener =
        |protocol: &str, port: u16| -> Box<dyn SocketListener<Accepted = Box<dyn Tunnel>> + Sync> {
            match protocol {
                "tcp" => Box::new(core_tcp_listener(
                    format!("tcp://0.0.0.0:{}", port).parse().unwrap(),
                )),
                "udp" => Box::new(core_udp_listener(
                    format!("udp://0.0.0.0:{}", port).parse().unwrap(),
                )),
                _ => panic!("unsupported protocol: {}", protocol),
            }
        };

    let make_connector = |protocol: &str, port: u16| -> Box<dyn TunnelDialer> {
        match protocol {
            "tcp" => Box::new(core_tcp_dialer(
                format!("tcp://10.144.144.3:{}", port).parse().unwrap(),
            )),
            "udp" => Box::new(core_udp_dialer(
                format!("udp://10.144.144.3:{}", port).parse().unwrap(),
            )),
            _ => panic!("unsupported protocol: {}", protocol),
        }
    };

    let mut buf = vec![0; 32];
    rand::thread_rng().fill(&mut buf[..]);

    for p in &["tcp", "udp"] {
        _tunnel_pingpong_netns_with_timeout(
            make_listener(p, port),
            make_connector(p, port),
            NetNS::new(Some("net_c".into())),
            NetNS::new(Some("net_a".into())),
            buf.clone(),
            std::time::Duration::from_millis(100),
        )
        .await
        .unwrap_or_else(|_| panic!("{} should be allowed", p));
    }

    if test_outbound_allow_list {
        return;
    }

    // test other port
    let other_port = port + 1;
    for p in ["tcp", "udp"] {
        let r = _tunnel_pingpong_netns_with_timeout(
            make_listener(p, other_port),
            make_connector(p, other_port),
            NetNS::new(Some("net_c".into())),
            NetNS::new(Some("net_a".into())),
            buf.clone(),
            std::time::Duration::from_millis(100),
        )
        .await;

        if p != protocol {
            assert!(r.is_ok(), "{} should be allowed", p);
        } else {
            assert!(r.is_err(), "{} should be blocked", p);
        }
    }

    drop_insts(insts).await;
}

#[rstest::rstest]
#[tokio::test]
#[serial_test::serial]
pub async fn config_patch_test() {
    use crate::proto::{
        api::config::{
            ConfigPatchAction, InstanceConfigPatch, PortForwardPatch, ProxyNetworkPatch,
        },
        common::{PortForwardConfigPb, SocketType},
    };
    use crate::tunnel::common::tests::_tunnel_pingpong_netns_with_timeout;

    let insts = init_three_node_ex(
        "udp",
        |cfg| {
            cfg.set_ipv6(None);
            cfg
        },
        false,
    )
    .await;

    check_route(
        "10.144.144.2/24",
        insts[1].peer_id(),
        insts[0].get_core_instance().route_snapshots().await,
    );

    check_route(
        "10.144.144.3/24",
        insts[2].peer_id(),
        insts[0].get_core_instance().route_snapshots().await,
    );

    // 测试1： 修改hostname、ip、子网代理
    let patch = InstanceConfigPatch {
        hostname: Some("new_inst1".to_string()),
        ipv4: Some("10.144.144.22/24".parse().unwrap()),
        proxy_networks: vec![ProxyNetworkPatch {
            action: ConfigPatchAction::Add as i32,
            cidr: Some("10.144.145.0/24".parse().unwrap()),
            mapped_cidr: None,
        }],
        ..Default::default()
    };
    insts[1]
        .get_config_patcher()
        .apply_patch(patch)
        .await
        .unwrap();
    assert_eq!(insts[1].get_global_ctx().get_hostname(), "new_inst1");
    assert_eq!(
        insts[1].get_global_ctx().get_ipv4().unwrap(),
        "10.144.144.22/24".parse().unwrap()
    );
    tokio::time::sleep(Duration::from_secs(1)).await;
    check_route_ex(
        insts[0].get_core_instance().route_snapshots().await,
        insts[1].peer_id(),
        |r| {
            assert_eq!(r.hostname, "new_inst1");
            assert_eq!(r.ipv4_addr, Some("10.144.144.22/24".parse().unwrap()));
            assert_eq!(r.proxy_cidrs[0], "10.144.145.0/24");
            true
        },
    );
    let patch = InstanceConfigPatch {
        proxy_networks: vec![ProxyNetworkPatch {
            action: ConfigPatchAction::Clear as i32,
            ..Default::default()
        }],
        ..Default::default()
    };
    insts[1]
        .get_config_patcher()
        .apply_patch(patch)
        .await
        .unwrap();
    assert!(
        insts[1]
            .get_global_ctx()
            .config
            .get_proxy_cidrs()
            .is_empty()
    );

    // 测试1.1：修改公网 IPv6 provider 相关配置
    let public_prefix = "2001:db8:100::/64";
    let patch = InstanceConfigPatch {
        ipv6_public_addr_provider: Some(true),
        ipv6_public_addr_auto: Some(true),
        ipv6_public_addr_prefix: Some(public_prefix.to_string()),
        ..Default::default()
    };
    insts[1]
        .get_config_patcher()
        .apply_patch(patch)
        .await
        .unwrap();
    assert!(
        insts[1]
            .get_global_ctx()
            .config
            .get_ipv6_public_addr_provider()
    );
    assert!(insts[1].get_global_ctx().config.get_ipv6_public_addr_auto());
    assert_eq!(
        insts[1]
            .get_global_ctx()
            .config
            .get_ipv6_public_addr_prefix(),
        Some(public_prefix.parse().unwrap())
    );
    assert!(
        insts[1]
            .get_core_instance()
            .node_snapshot()
            .await
            .feature_flags
            .ipv6_public_addr_provider
    );
    assert_eq!(
        insts[1]
            .get_core_instance()
            .node_snapshot()
            .await
            .ipv6_public_addr_prefix,
        Some(public_prefix.parse().unwrap())
    );

    // 测试2: 端口转发
    let patch = InstanceConfigPatch {
        port_forwards: vec![PortForwardPatch {
            action: ConfigPatchAction::Add as i32,
            cfg: Some(PortForwardConfigPb {
                bind_addr: Some("0.0.0.0:23458".parse::<SocketAddr>().unwrap().into()),
                dst_addr: Some("10.144.144.3:23457".parse::<SocketAddr>().unwrap().into()),
                socket_type: SocketType::Tcp as i32,
            }),
        }],
        ..Default::default()
    };
    insts[0]
        .get_config_patcher()
        .apply_patch(patch)
        .await
        .unwrap();

    let mut buf = vec![0; 32];
    rand::thread_rng().fill(&mut buf[..]);
    let tcp_listener = core_tcp_listener("tcp://0.0.0.0:23457".parse().unwrap());
    let tcp_connector = core_tcp_dialer("tcp://127.0.0.1:23458".parse().unwrap());
    let result = _tunnel_pingpong_netns_with_timeout(
        tcp_listener,
        tcp_connector,
        NetNS::new(Some("net_c".into())),
        NetNS::new(Some("net_a".into())),
        buf.clone(),
        std::time::Duration::from_millis(30000),
    )
    .await;
    assert!(result.is_ok(), "Port forward pingpong should succeed");

    drop_insts(insts).await;
}

#[rstest::rstest]
#[tokio::test]
#[serial_test::serial]
pub async fn config_patch_disable_relay_data_test() {
    use crate::proto::api::config::InstanceConfigPatch;

    let insts = init_three_node_ex(
        "udp",
        |cfg| {
            cfg.set_ipv6(None);
            cfg
        },
        false,
    )
    .await;

    let relay_peer_id = insts[1].peer_id();
    let dst_peer_id = insts[2].peer_id();
    assert!(!insts[1].get_global_ctx().get_flags().disable_relay_data);
    assert!(
        !insts[1]
            .get_core_instance()
            .node_snapshot()
            .await
            .feature_flags
            .avoid_relay_data
    );

    check_route_ex(
        insts[0].get_core_instance().route_snapshots().await,
        dst_peer_id,
        |route| {
            assert_eq!(route.next_hop_peer_id, relay_peer_id);
            true
        },
    );

    wait_for_condition(
        || async { ping_test("net_a", "10.144.144.3", None).await },
        Duration::from_secs(5),
    )
    .await;

    insts[1]
        .get_config_patcher()
        .apply_patch(InstanceConfigPatch {
            disable_relay_data: Some(true),
            ..Default::default()
        })
        .await
        .unwrap();

    assert!(insts[1].get_global_ctx().get_flags().disable_relay_data);
    assert!(
        insts[1]
            .get_global_ctx()
            .config
            .get_flags()
            .disable_relay_data
    );
    assert!(
        insts[1]
            .get_core_instance()
            .node_snapshot()
            .await
            .feature_flags
            .avoid_relay_data
    );

    wait_for_condition(
        || {
            let core = insts[0].get_core_instance();
            async move {
                core.route_snapshots().await.iter().any(|route| {
                    route.peer_id == relay_peer_id
                        && route
                            .feature_flag
                            .as_ref()
                            .map(|flag| flag.avoid_relay_data)
                            .unwrap_or(false)
                })
            }
        },
        Duration::from_secs(5),
    )
    .await;

    check_route_ex(
        insts[0].get_core_instance().route_snapshots().await,
        dst_peer_id,
        |route| {
            assert_eq!(route.next_hop_peer_id, relay_peer_id);
            true
        },
    );
    assert!(
        !ping_test("net_a", "10.144.144.3", None).await,
        "traffic from inst1 to inst3 should be blocked while inst2 relay data is disabled"
    );

    insts[1]
        .get_config_patcher()
        .apply_patch(InstanceConfigPatch {
            disable_relay_data: Some(false),
            ..Default::default()
        })
        .await
        .unwrap();

    assert!(!insts[1].get_global_ctx().get_flags().disable_relay_data);
    assert!(
        !insts[1]
            .get_global_ctx()
            .config
            .get_flags()
            .disable_relay_data
    );
    assert!(
        !insts[1]
            .get_core_instance()
            .node_snapshot()
            .await
            .feature_flags
            .avoid_relay_data
    );

    wait_for_condition(
        || {
            let core = insts[0].get_core_instance();
            async move {
                core.route_snapshots().await.iter().any(|route| {
                    route.peer_id == relay_peer_id
                        && route
                            .feature_flag
                            .as_ref()
                            .map(|flag| !flag.avoid_relay_data)
                            .unwrap_or(false)
                })
            }
        },
        Duration::from_secs(5),
    )
    .await;

    wait_for_condition(
        || async { ping_test("net_a", "10.144.144.3", None).await },
        Duration::from_secs(5),
    )
    .await;

    drop_insts(insts).await;
}

/// Generate SecureModeConfig with specified x25519 private key
pub fn generate_secure_mode_config_with_key(
    private_key: &x25519_dalek::StaticSecret,
) -> SecureModeConfig {
    use base64::{Engine, prelude::BASE64_STANDARD};
    use x25519_dalek::PublicKey;

    let public = PublicKey::from(private_key);

    SecureModeConfig {
        enabled: true,
        local_private_key: Some(BASE64_STANDARD.encode(private_key.as_bytes())),
        local_public_key: Some(BASE64_STANDARD.encode(public.as_bytes())),
    }
}

/// Generate SecureModeConfig with random x25519 keypair
pub fn generate_secure_mode_config() -> SecureModeConfig {
    let private = StaticSecret::random_from_rng(OsRng);
    generate_secure_mode_config_with_key(&private)
}

/// Test relay peer end-to-end encryption with TCP
#[rstest::rstest]
#[tokio::test]
#[serial_test::serial]
pub async fn relay_peer_e2e_encryption(#[values("tcp", "udp")] proto: &str) {
    let insts = init_three_node_ex(
        proto,
        |cfg| {
            cfg.set_secure_mode(Some(generate_secure_mode_config()));
            cfg
        },
        false,
    )
    .await;

    let inst1_peer_id = insts[0].peer_id();
    let inst2_peer_id = insts[1].peer_id();
    let inst3_peer_id = insts[2].peer_id();

    println!(
        "Test topology: inst1({}) <-> inst2({}) <-> inst3({})",
        inst1_peer_id, inst2_peer_id, inst3_peer_id
    );

    // Check secure mode is enabled
    let secure_mode_1 = insts[0].get_global_ctx().config.get_secure_mode();
    let secure_mode_2 = insts[1].get_global_ctx().config.get_secure_mode();
    let secure_mode_3 = insts[2].get_global_ctx().config.get_secure_mode();
    println!(
        "Secure mode enabled: inst1={}, inst2={}, inst3={}",
        secure_mode_1.is_some(),
        secure_mode_2.is_some(),
        secure_mode_3.is_some()
    );

    // Wait for routes to be established
    wait_for_condition(
        || async {
            let routes = insts[0].get_core_instance().route_snapshots().await;
            routes.len() == 2
        },
        Duration::from_secs(10),
    )
    .await;

    // Verify inst1 sees inst3 via inst2 (non-direct path)
    let next_hop_to_inst3 = insts[0]
        .get_core_instance()
        .route_snapshots()
        .await
        .into_iter()
        .find(|route| route.peer_id == inst3_peer_id)
        .map(|route| route.next_hop_peer_id);
    println!("Next hop from inst1 to inst3: {:?}", next_hop_to_inst3);
    assert_eq!(
        next_hop_to_inst3,
        Some(inst2_peer_id),
        "inst1 should reach inst3 via inst2 (relay)"
    );

    // Verify inst1 has no direct connection to inst3
    assert!(
        !insts[0]
            .get_core_instance()
            .connected_peers()
            .await
            .contains(&inst3_peer_id),
        "inst1 should NOT have direct connection to inst3"
    );

    // Check if noise_static_pubkey is available for relay handshake
    let route_has_static_key = insts[0]
        .get_core_instance()
        .relay_route_has_static_key_for_test(inst3_peer_id)
        .await;
    println!(
        "Route info for inst3 on inst1 has a relay static key: {}",
        route_has_static_key
    );

    // Wait until relay route info includes inst3 static pubkey for IK handshake.
    wait_for_condition(
        || async {
            insts[0]
                .get_core_instance()
                .relay_route_has_static_key_for_test(inst3_peer_id)
                .await
        },
        Duration::from_secs(10),
    )
    .await;

    // Test basic connectivity through relay
    println!("Starting ping test from net_a to 10.144.144.3...");

    assert!(
        ping_test("net_a", "10.144.144.3", None).await,
        "Ping from net_a to inst3 should succeed"
    );

    // Verify relay sessions are established
    let relay_1 = insts[0]
        .get_core_instance()
        .relay_session_snapshot_for_test(inst3_peer_id);
    let relay_3 = insts[2]
        .get_core_instance()
        .relay_session_snapshot_for_test(inst1_peer_id);

    println!(
        "Relay states after ping: inst1->inst3: {}, inst3->inst1: {}",
        relay_1.has_state, relay_3.has_state
    );

    // Test bidirectional connectivity
    assert!(
        ping_test("net_a", "10.144.144.3", None).await,
        "Ping from net_a to inst3 should work"
    );
    assert!(
        ping_test("net_c", "10.144.144.1", None).await,
        "Ping from net_c to inst1 should work"
    );

    println!("Test completed successfully!");
    drop_insts(insts).await;
}

#[tokio::test]
#[serial_test::serial]
pub async fn relay_peer_e2e_encryption_udp() {
    let insts = init_three_node_ex(
        "udp",
        |cfg| {
            cfg.set_secure_mode(Some(generate_secure_mode_config()));
            cfg
        },
        false,
    )
    .await;

    let inst1_id = insts[0].get_global_ctx().get_id().to_string();
    let inst3_id = insts[2].get_global_ctx().get_id().to_string();
    let network_name = insts[0].get_global_ctx().get_network_name();
    let total_labels =
        LabelSet::new().with_label_type(LabelType::NetworkName(network_name.clone()));

    wait_for_condition(
        || async {
            let routes = insts[0].get_core_instance().route_snapshots().await;
            routes.len() == 2
        },
        Duration::from_secs(10),
    )
    .await;

    wait_for_condition(
        || async { ping_test("net_a", "10.144.144.3", None).await },
        Duration::from_secs(6),
    )
    .await;

    let tx_labels = LabelSet::new()
        .with_label_type(LabelType::NetworkName(network_name.clone()))
        .with_label_type(LabelType::ToInstanceId(inst3_id.clone()));
    let rx_labels = LabelSet::new()
        .with_label_type(LabelType::NetworkName(network_name.clone()))
        .with_label_type(LabelType::FromInstanceId(inst1_id.clone()));

    wait_for_condition(
        || async {
            let metrics = insts[0].get_core_instance().metric_snapshots();
            metric_value(&metrics, MetricName::TrafficBytesTx, &tx_labels).is_none()
                && metric_value(&metrics, MetricName::TrafficPacketsTx, &tx_labels).is_none()
                && metric_value(&metrics, MetricName::TrafficBytesTx, &total_labels)
                    .is_some_and(|value| value > 0)
                && metric_value(&metrics, MetricName::TrafficPacketsTx, &total_labels)
                    .is_some_and(|value| value > 0)
                && metric_value(&metrics, MetricName::TrafficBytesTxByInstance, &tx_labels)
                    .is_some_and(|value| value > 0)
                && metric_value(&metrics, MetricName::TrafficPacketsTxByInstance, &tx_labels)
                    .is_some_and(|value| value > 0)
        },
        Duration::from_secs(10),
    )
    .await;

    wait_for_condition(
        || async {
            let metrics = insts[2].get_core_instance().metric_snapshots();
            metric_value(&metrics, MetricName::TrafficBytesRx, &rx_labels).is_none()
                && metric_value(&metrics, MetricName::TrafficPacketsRx, &rx_labels).is_none()
                && metric_value(&metrics, MetricName::TrafficBytesRx, &total_labels)
                    .is_some_and(|value| value > 0)
                && metric_value(&metrics, MetricName::TrafficPacketsRx, &total_labels)
                    .is_some_and(|value| value > 0)
                && metric_value(&metrics, MetricName::TrafficBytesRxByInstance, &rx_labels)
                    .is_some_and(|value| value > 0)
                && metric_value(&metrics, MetricName::TrafficPacketsRxByInstance, &rx_labels)
                    .is_some_and(|value| value > 0)
        },
        Duration::from_secs(10),
    )
    .await;

    drop_insts(insts).await;
}

/// Test Relay Peer session cleanup on relay failure - TCP
#[tokio::test]
#[serial_test::serial]
pub async fn relay_peer_session_cleanup() {
    let mut insts = init_three_node_ex(
        "tcp",
        |cfg| {
            cfg.set_secure_mode(Some(generate_secure_mode_config()));
            cfg
        },
        false,
    )
    .await;

    let inst2_peer_id = insts[1].peer_id();
    let inst3_peer_id = insts[2].peer_id();
    let core_1 = insts[0].get_core_instance();

    wait_for_condition(
        || async { ping_test("net_a", "10.144.144.3", None).await },
        Duration::from_secs(6),
    )
    .await;

    wait_for_condition(
        || async {
            let relay = core_1.relay_session_snapshot_for_test(inst3_peer_id);
            relay.has_state && relay.has_session
        },
        Duration::from_secs(3),
    )
    .await;

    let next_hop = insts[0]
        .get_core_instance()
        .route_snapshots()
        .await
        .into_iter()
        .find(|route| route.peer_id == inst3_peer_id)
        .map(|route| route.next_hop_peer_id);
    assert_eq!(next_hop, Some(inst2_peer_id));

    let mut inst2 = insts.remove(1);
    inst2.clear_resources().await;
    drop(inst2);

    wait_for_condition(
        || async {
            let routes = insts[0].get_core_instance().route_snapshots().await;
            !routes.iter().any(|r| r.peer_id == inst3_peer_id)
        },
        Duration::from_secs(6),
    )
    .await;

    core_1.evict_idle_relay_sessions_for_test(Duration::from_millis(0));
    assert!(
        !core_1
            .relay_session_snapshot_for_test(inst3_peer_id)
            .has_state
    );

    core_1.evict_unused_peer_sessions_for_test(Duration::from_millis(0));

    wait_for_condition(
        || async {
            !core_1
                .relay_session_snapshot_for_test(inst3_peer_id)
                .has_session
        },
        Duration::from_secs(1),
    )
    .await;

    drop(core_1);
    drop_insts(insts).await;
}
