Emit 'connecting' event from PQ handshake

Signed-off-by: Mateusz Szczygieł <[email protected]>

.unreleased/LLT-5814

@@ -0,0 +1 @@
+Emit 'CONNECTING' node event when starting the PQ handshake

crates/telio-pq/src/conn.rs

@@ -27,6 +27,9 @@ impl ConnKeyRotation {
         let request_retry = Duration::from_secs(features.handshake_retry_interval_s as _);
 
         let task = async move {
+            #[allow(mpsc_blocking_send)]
+            let _ = chan.send(super::Event::Connecting(peer)).await;
+
             let mut retry_interval = telio_utils::interval(request_retry);
 
             let proto::KeySet {

crates/telio-pq/src/entity.rs

@@ -71,6 +71,7 @@ impl Entity {
                     }
                 }
             }
+            _ => (),
         }
     }
 

crates/telio-pq/src/lib.rs

@@ -24,6 +24,7 @@ pub trait PostQuantum {
 
 #[derive(Copy, Clone)]
 pub enum Event {
+    Connecting(telio_crypto::PublicKey),
     Handshake(SocketAddr, Keys),
     Rekey(Keys),
 }

nat-lab/tests/test_pq.py

@@ -4,7 +4,7 @@
 from helpers import SetupParameters, setup_environment
 from telio import Client
 from utils import stun
-from utils.bindings import TelioAdapterType
+from utils.bindings import TelioAdapterType, NodeState
 from utils.connection import Connection
 from utils.connection_util import (
     generate_connection_tracker_config,
@@ -388,8 +388,8 @@ async def test_dns_with_pq(
                 adapter_type_override=TelioAdapterType.NEP_TUN,
                 connection_tracker_config=generate_connection_tracker_config(
                     ConnectionTag.DOCKER_CONE_CLIENT_1,
-                    vpn_1_limits=ConnectionLimits(1, None),
-                    nlx_1_limits=ConnectionLimits(1, 2),
+                    vpn_1_limits=(1, None),
+                    nlx_1_limits=(1, 2),
                 ),
                 is_meshnet=False,
             ),
@@ -410,12 +410,30 @@ async def test_pq_vpn_silent_pq_upgrader(
         client, *_ = env.clients
 
         wg_server = config.WG_SERVER  # use non PQ server
+        pubkey = str(wg_server["public_key"])
+        port = int(wg_server["port"])
+
+        await client.restart_interface()
+        await client.get_router().create_vpn_route()
+        await client.get_proxy().connect_to_exit_node_pq(
+            public_key=pubkey,
+            allowed_ips=None,
+            endpoint=f"{pubkey}:{port}",
+        )
+
+        await client.wait_for_event_peer(
+            pubkey,
+            [NodeState.CONNECTING],
+            is_exit=True,
+            is_vpn=True,
+        )
+
         try:
-            await client.connect_to_vpn(
-                str(wg_server["ipv4"]),
-                int(wg_server["port"]),
-                str(wg_server["public_key"]),
-                pq=True,
+            await client.wait_for_event_peer(
+                pubkey,
+                [NodeState.CONNECTED],
+                is_exit=True,
+                is_vpn=True,
                 timeout=4,
             )
             raise Exception("This shouldn't connect succesfully")
@@ -438,7 +456,7 @@ async def test_pq_vpn_silent_pq_upgrader(
                 adapter_type_override=TelioAdapterType.NEP_TUN,
                 connection_tracker_config=generate_connection_tracker_config(
                     ConnectionTag.DOCKER_CONE_CLIENT_1,
-                    nlx_1_limits=ConnectionLimits(1, 2),
+                    nlx_1_limits=(1, 2),
                 ),
                 is_meshnet=False,
             ),

src/device.rs

@@ -81,6 +81,7 @@ use telio_utils::{
 
 use telio_model::{
     config::{Config, Peer, PeerBase, Server as DerpServer},
+    constants::{VPN_EXTERNAL_IPV4, VPN_INTERNAL_IPV4},
     event::{Event, Set},
     features::{FeaturePersistentKeepalive, Features, PathType},
     mesh::{ExitNode, LinkState, Node},
@@ -2251,25 +2252,11 @@ impl Runtime {
             (None, Some(exit_node)) => {
                 // Exit node
                 Some(Node {
-                    identifier: exit_node.identifier.clone(),
-                    public_key: exit_node.public_key,
-                    nickname: None,
                     state: state.unwrap_or_else(|| peer.state()),
                     link_state,
-                    is_exit: true,
-                    is_vpn: exit_node.endpoint.is_some(),
-                    ip_addresses: vec![
-                        IpAddr::V4(Ipv4Addr::new(10, 5, 0, 1)),
-                        IpAddr::V4(Ipv4Addr::new(100, 64, 0, 1)),
-                    ],
                     allowed_ips: peer.allowed_ips.clone(),
-                    endpoint,
-                    hostname: None,
-                    allow_incoming_connections: false,
-                    allow_peer_send_files: false,
                     path: path_type,
-                    allow_multicast: false,
-                    peer_allows_multicast: false,
+                    ..node_from_exit_node(exit_node)
                 })
             }
             _ => None,
@@ -2447,6 +2434,22 @@ impl TaskRuntime for Runtime {
             Some(pq_event) = self.event_listeners.post_quantum_subscriber.recv() => {
                 telio_log_debug!("WG consolidation triggered by PQ event");
 
+                if let (telio_pq::Event::Connecting(pubkey), Some(exit)) =
+                    (&pq_event, &self.requested_state.exit_node) {
+
+                    if exit.public_key == *pubkey {
+                        let body = Node {
+                            state: PeerState::Connecting,
+                            link_state: Some(LinkState::Down),
+                            ..node_from_exit_node(exit)
+                        };
+
+                        let _ = self.event_publishers.libtelio_event_publisher.send(
+                            Box::new(Event::Node { body  })
+                        );
+                    }
+                }
+
                 self.entities.postquantum_wg.on_event(pq_event);
 
                 if let Err(err) = wg_controller::consolidate_wg_state(&self.requested_state, &self.entities, &self.features)
@@ -2563,6 +2566,26 @@ fn set_tunnel_interface(socket_pool: &Arc<SocketPool>, config: &DeviceConfig) {
     }
 }
 
+fn node_from_exit_node(exit_node: &ExitNode) -> Node {
+    Node {
+        identifier: exit_node.identifier.clone(),
+        public_key: exit_node.public_key,
+        nickname: None,
+        is_exit: true,
+        is_vpn: exit_node.endpoint.is_some(),
+        ip_addresses: vec![IpAddr::V4(VPN_EXTERNAL_IPV4), IpAddr::V4(VPN_INTERNAL_IPV4)],
+        allowed_ips: exit_node.allowed_ips.clone().unwrap_or_default(),
+        endpoint: exit_node.endpoint,
+        hostname: None,
+        allow_incoming_connections: false,
+        allow_peer_send_files: false,
+        path: PathType::Direct,
+        allow_multicast: false,
+        peer_allows_multicast: false,
+        ..Default::default()
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;