fix: protect self peer during credential refresh and allow need-p2p peers through public server (#2192)

* fix: protect self peer during credential refresh

* fix: allow need-p2p peers through public server

easytier/src/peers/peer_manager.rs

@@ -2283,8 +2283,6 @@ mod tests {
             PacketType::QuicDst,
             PacketType::DataWithKcpSrcModified,
             PacketType::DataWithQuicSrcModified,
-            PacketType::RelayHandshake,
-            PacketType::RelayHandshakeAck,
             PacketType::ForeignNetworkPacket,
         ] {
             assert!(PeerManager::is_relay_data_packet(packet_type as u8));
@@ -2299,6 +2297,8 @@ mod tests {
             PacketType::NoiseHandshakeMsg1,
             PacketType::NoiseHandshakeMsg2,
             PacketType::NoiseHandshakeMsg3,
+            PacketType::RelayHandshake,
+            PacketType::RelayHandshakeAck,
         ] {
             assert!(!PeerManager::is_relay_data_packet(packet_type as u8));
         }

easytier/src/peers/peer_ospf_route.rs

@@ -1228,6 +1228,25 @@ impl SyncedRouteInfo {
         Vec<PeerId>,
         HashMap<Vec<u8>, crate::common::global_ctx::TrustedKeyMetadata>,
     )
+    where
+        F: FnMut(PeerId) -> bool,
+    {
+        self.verify_and_update_credential_trusts_with_active_peers_protecting(
+            network_secret,
+            is_peer_active,
+            None,
+        )
+    }
+
+    fn verify_and_update_credential_trusts_with_active_peers_protecting<F>(
+        &self,
+        network_secret: Option<&str>,
+        is_peer_active: F,
+        protected_peer_id: Option<PeerId>,
+    ) -> (
+        Vec<PeerId>,
+        HashMap<Vec<u8>, crate::common::global_ctx::TrustedKeyMetadata>,
+    )
     where
         F: FnMut(PeerId) -> bool,
     {
@@ -1248,6 +1267,9 @@ impl SyncedRouteInfo {
         let mut untrusted_peers =
             Self::collect_revoked_credential_peers(&peer_infos, &prev_trusted, &all_trusted);
         untrusted_peers.extend(duplicate_untrusted_peers);
+        if let Some(protected_peer_id) = protected_peer_id {
+            untrusted_peers.remove(&protected_peer_id);
+        }
 
         // Remove untrusted peers from peer_infos so they won't appear in route graph
         if !untrusted_peers.is_empty() {
@@ -2735,7 +2757,11 @@ impl PeerRouteServiceImpl {
         let network_identity = self.global_ctx.get_network_identity();
         let (untrusted, global_trusted_keys) = self
             .synced_route_info
-            .verify_and_update_credential_trusts(network_identity.network_secret.as_deref());
+            .verify_and_update_credential_trusts_with_active_peers_protecting(
+                network_identity.network_secret.as_deref(),
+                |_| true,
+                Some(self.my_peer_id),
+            );
         self.global_ctx
             .update_trusted_keys(global_trusted_keys, &network_identity.network_name);
 
@@ -2751,9 +2777,10 @@ impl PeerRouteServiceImpl {
 
         let (untrusted, global_trusted_keys) = self
             .synced_route_info
-            .verify_and_update_credential_trusts_with_active_peers(
+            .verify_and_update_credential_trusts_with_active_peers_protecting(
                 network_identity.network_secret.as_deref(),
                 |peer_id| self.is_active_non_reusable_credential_peer(peer_id),
+                Some(self.my_peer_id),
             );
         self.global_ctx
             .update_trusted_keys(global_trusted_keys, &network_identity.network_name);
@@ -5047,6 +5074,58 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn credential_trust_refresh_does_not_remove_self_peer() {
+        let my_peer_id = 11;
+        let remote_peer_id = 12;
+        let credential_key = vec![8; 32];
+        let service_impl = PeerRouteServiceImpl::new(my_peer_id, get_mock_global_ctx());
+
+        let self_info = make_credential_route_peer_info(my_peer_id, &credential_key);
+        let remote_info = make_credential_route_peer_info(remote_peer_id, &credential_key);
+
+        {
+            let mut guard = service_impl.synced_route_info.peer_infos.write();
+            guard.insert(self_info.peer_id, self_info);
+            guard.insert(remote_info.peer_id, remote_info);
+        }
+        service_impl
+            .synced_route_info
+            .trusted_credential_pubkeys
+            .insert(
+                credential_key.clone(),
+                TrustedCredentialPubkey {
+                    pubkey: credential_key,
+                    expiry_unix: i64::MAX,
+                    ..Default::default()
+                },
+            );
+
+        let (untrusted_peers, _) = service_impl
+            .synced_route_info
+            .verify_and_update_credential_trusts_with_active_peers_protecting(
+                None,
+                |_| true,
+                Some(my_peer_id),
+            );
+
+        assert_eq!(untrusted_peers, vec![remote_peer_id]);
+        assert!(
+            service_impl
+                .synced_route_info
+                .peer_infos
+                .read()
+                .contains_key(&my_peer_id)
+        );
+        assert!(
+            !service_impl
+                .synced_route_info
+                .peer_infos
+                .read()
+                .contains_key(&remote_peer_id)
+        );
+    }
+
     #[tokio::test]
     async fn credential_refresh_rebuilds_reachability_before_owner_election() {
         const NETWORK_SECRET: &str = "sec1";

easytier/src/peers/traffic_metrics.rs

@@ -242,9 +242,9 @@ pub(crate) fn traffic_kind(packet_type: u8) -> TrafficKind {
 }
 
 pub(crate) fn is_relay_data_packet_type(packet_type: u8) -> bool {
+    // Relay handshakes are control-plane setup; payload data is blocked by its
+    // original packet type after the session exists.
     traffic_kind(packet_type) == TrafficKind::Data
-        || packet_type == PacketType::RelayHandshake as u8
-        || packet_type == PacketType::RelayHandshakeAck as u8
         || packet_type == PacketType::ForeignNetworkPacket as u8
 }
 

easytier/src/tests/credential_tests.rs

@@ -14,13 +14,17 @@ use crate::{
     },
     instance::instance::Instance,
     tests::three_node::{generate_secure_mode_config, generate_secure_mode_config_with_key},
-    tunnel::{common::tests::wait_for_condition, tcp::TcpTunnelConnector},
+    tunnel::{common::tests::wait_for_condition, tcp::TcpTunnelConnector, udp::UdpTunnelConnector},
 };
 
 use super::{add_ns_to_bridge, create_netns, del_netns, drop_insts, ping_test};
 
 use rstest::rstest;
 
+const PUBLIC_SERVER_NETWORK_NAME: &str = "__public_server__";
+const PUBLIC_SERVER_SHARED_SECRET: &str = "public-server-shared-secret";
+const NEED_P2P_ADMIN_NETWORK_NAME: &str = "need_p2p_credential_test_network";
+
 /// Prepare network namespaces for credential tests
 /// Topology:
 ///   br_a (10.1.1.0/24): ns_adm (10.1.1.1), ns_c1 (10.1.1.2), ns_c2 (10.1.1.3), ns_c3 (10.1.1.4), ns_c4 (10.1.1.5)
@@ -221,6 +225,297 @@ fn create_shared_config(
     config
 }
 
+fn create_public_server_config() -> TomlConfigLoader {
+    let config = TomlConfigLoader::default();
+    config.set_inst_name(PUBLIC_SERVER_NETWORK_NAME.to_string());
+    config.set_hostname(Some("public-server".to_string()));
+    config.set_netns(Some("ns_adm".to_string()));
+    config.set_listeners(vec!["udp://0.0.0.0:11010".parse().unwrap()]);
+    config.set_network_identity(NetworkIdentity::new(
+        PUBLIC_SERVER_NETWORK_NAME.to_string(),
+        PUBLIC_SERVER_SHARED_SECRET.to_string(),
+    ));
+    config.set_secure_mode(Some(generate_secure_mode_config()));
+
+    let mut flags = config.get_flags();
+    flags.no_tun = true;
+    flags.private_mode = true;
+    flags.relay_all_peer_rpc = true;
+    flags.relay_network_whitelist = "".to_string();
+    config.set_flags(flags);
+
+    config
+}
+
+fn create_need_p2p_admin_config() -> TomlConfigLoader {
+    let config = TomlConfigLoader::default();
+    config.set_inst_name(NEED_P2P_ADMIN_NETWORK_NAME.to_string());
+    config.set_hostname(Some("need-p2p-admin".to_string()));
+    config.set_netns(Some("ns_c3".to_string()));
+    config.set_listeners(vec!["tcp://0.0.0.0:11020".parse().unwrap()]);
+    config.set_network_identity(NetworkIdentity::new(
+        NEED_P2P_ADMIN_NETWORK_NAME.to_string(),
+        PUBLIC_SERVER_SHARED_SECRET.to_string(),
+    ));
+    config.set_secure_mode(Some(generate_secure_mode_config()));
+
+    let mut flags = config.get_flags();
+    flags.no_tun = true;
+    flags.relay_all_peer_rpc = true;
+    flags.need_p2p = true;
+    flags.disable_udp_hole_punching = true;
+    flags.disable_tcp_hole_punching = true;
+    flags.disable_sym_hole_punching = true;
+    config.set_flags(flags);
+
+    config
+}
+
+#[allow(clippy::too_many_arguments)]
+fn create_public_server_credential_config(
+    credential_secret: &str,
+    inst_name: &str,
+    hostname: &str,
+    ns: &str,
+    ipv4: &str,
+    ipv6: &str,
+    tcp_listener_port: u16,
+    udp_listener_port: u16,
+    proxy_cidrs: &[&str],
+) -> TomlConfigLoader {
+    let config = create_credential_config_from_secret(
+        NEED_P2P_ADMIN_NETWORK_NAME.to_string(),
+        credential_secret,
+        inst_name,
+        Some(ns),
+        ipv4,
+        ipv6,
+    );
+    config.set_hostname(Some(hostname.to_string()));
+    config.set_listeners(vec![
+        format!("tcp://0.0.0.0:{tcp_listener_port}")
+            .parse()
+            .unwrap(),
+        format!("udp://0.0.0.0:{udp_listener_port}")
+            .parse()
+            .unwrap(),
+    ]);
+    for cidr in proxy_cidrs {
+        config
+            .add_proxy_cidr((*cidr).parse().unwrap(), None)
+            .unwrap();
+    }
+
+    let mut flags = config.get_flags();
+    flags.disable_p2p = true;
+    config.set_flags(flags);
+
+    config
+}
+
+async fn wait_direct_peer(inst: &Instance, peer_id: u32, timeout: Duration, label: &str) {
+    wait_for_condition(
+        || async {
+            let peers = inst.get_peer_manager().get_peer_map().list_peers();
+            let connected = peers.contains(&peer_id);
+            println!("{label}: direct peers={:?}, target={}", peers, peer_id);
+            connected
+        },
+        timeout,
+    )
+    .await;
+}
+
+async fn wait_route_cost(inst: &Instance, peer_id: u32, cost: i32, timeout: Duration, label: &str) {
+    wait_for_condition(
+        || async {
+            let routes = inst.get_peer_manager().list_routes().await;
+            let matched = routes
+                .iter()
+                .any(|route| route.peer_id == peer_id && route.cost == cost);
+            println!(
+                "{label}: routes={:?}, target={}, cost={}",
+                routes
+                    .iter()
+                    .map(|route| (route.peer_id, route.cost))
+                    .collect::<Vec<_>>(),
+                peer_id,
+                cost
+            );
+            matched
+        },
+        timeout,
+    )
+    .await;
+}
+
+async fn wait_foreign_network_count(inst: &Instance, expected: usize, timeout: Duration) {
+    wait_for_condition(
+        || async {
+            let foreign_networks = inst
+                .get_peer_manager()
+                .get_foreign_network_manager()
+                .list_foreign_networks()
+                .await
+                .foreign_networks;
+            println!("foreign networks: {:?}", foreign_networks);
+            foreign_networks.len() == expected
+        },
+        timeout,
+    )
+    .await;
+}
+
+/// Regression coverage for a public-server-mediated credential topology:
+/// Public server <- admin peer (need_p2p) <- two credential peers.
+///
+/// Credential peers set `disable_p2p=true`, while the admin peer advertises `need_p2p=true`.
+/// The credential peers should still proactively build direct TCP peers with the admin peer
+/// through peer RPC forwarded by the public server.
+#[tokio::test]
+#[serial_test::serial]
+async fn credential_peers_p2p_to_need_p2p_admin_through_public_server() {
+    prepare_credential_network();
+
+    let mut public_server_inst = Instance::new(create_public_server_config());
+    public_server_inst.run().await.unwrap();
+
+    let mut admin_inst = Instance::new(create_need_p2p_admin_config());
+    admin_inst.run().await.unwrap();
+    admin_inst
+        .get_conn_manager()
+        .add_connector(UdpTunnelConnector::new(
+            "udp://10.1.1.1:11010".parse().unwrap(),
+        ));
+
+    wait_foreign_network_count(&public_server_inst, 1, Duration::from_secs(10)).await;
+
+    let (_credential_a_id, credential_a_secret) = admin_inst
+        .get_global_ctx()
+        .get_credential_manager()
+        .generate_credential_with_options(
+            vec![],
+            false,
+            vec!["10.1.0.0/24".to_string()],
+            Duration::from_secs(3600),
+            Some("credential-peer-a".to_string()),
+            false,
+        );
+    let (_credential_b_id, credential_b_secret) = admin_inst
+        .get_global_ctx()
+        .get_credential_manager()
+        .generate_credential_with_options(
+            vec![],
+            false,
+            vec![],
+            Duration::from_secs(3600),
+            Some("credential-peer-b".to_string()),
+            false,
+        );
+    admin_inst
+        .get_global_ctx()
+        .issue_event(GlobalCtxEvent::CredentialChanged);
+
+    wait_foreign_network_count(&public_server_inst, 1, Duration::from_secs(10)).await;
+
+    let mut credential_a_inst = Instance::new(create_public_server_credential_config(
+        &credential_a_secret,
+        "credential-peer-a",
+        "credential-a",
+        "ns_c1",
+        "10.154.0.1",
+        "fd00::1/64",
+        11030,
+        11031,
+        &["10.1.0.0/24"],
+    ));
+    let mut credential_b_inst = Instance::new(create_public_server_credential_config(
+        &credential_b_secret,
+        "credential-peer-b",
+        "credential-b",
+        "ns_c2",
+        "10.154.0.2",
+        "fd00::2/64",
+        11040,
+        11041,
+        &[],
+    ));
+    credential_a_inst.run().await.unwrap();
+    credential_b_inst.run().await.unwrap();
+
+    credential_a_inst
+        .get_conn_manager()
+        .add_connector(UdpTunnelConnector::new(
+            "udp://10.1.1.1:11010".parse().unwrap(),
+        ));
+    credential_b_inst
+        .get_conn_manager()
+        .add_connector(UdpTunnelConnector::new(
+            "udp://10.1.1.1:11010".parse().unwrap(),
+        ));
+
+    let admin_peer_id = admin_inst.peer_id();
+    let credential_a_peer_id = credential_a_inst.peer_id();
+    let credential_b_peer_id = credential_b_inst.peer_id();
+    println!(
+        "admin={}, credential_a={}, credential_b={}",
+        admin_peer_id, credential_a_peer_id, credential_b_peer_id
+    );
+
+    wait_direct_peer(
+        &credential_a_inst,
+        admin_peer_id,
+        Duration::from_secs(30),
+        "credential_a -> admin",
+    )
+    .await;
+    wait_direct_peer(
+        &credential_b_inst,
+        admin_peer_id,
+        Duration::from_secs(30),
+        "credential_b -> admin",
+    )
+    .await;
+    wait_direct_peer(
+        &admin_inst,
+        credential_a_peer_id,
+        Duration::from_secs(10),
+        "admin -> credential_a",
+    )
+    .await;
+    wait_direct_peer(
+        &admin_inst,
+        credential_b_peer_id,
+        Duration::from_secs(10),
+        "admin -> credential_b",
+    )
+    .await;
+    wait_route_cost(
+        &credential_a_inst,
+        admin_peer_id,
+        1,
+        Duration::from_secs(10),
+        "credential_a route to admin",
+    )
+    .await;
+    wait_route_cost(
+        &credential_b_inst,
+        admin_peer_id,
+        1,
+        Duration::from_secs(10),
+        "credential_b route to admin",
+    )
+    .await;
+
+    drop_insts(vec![
+        public_server_inst,
+        admin_inst,
+        credential_a_inst,
+        credential_b_inst,
+    ])
+    .await;
+}
+
 fn create_generated_credential_config(
     admin_inst: &Instance,
     inst_name: &str,