v3.2.1

changelog.md

@@ -1,5 +1,25 @@
 # Changelog
 
+## 2026-02-18 - 3.2.1 - fix(tests)
+add comprehensive unit and async tests across Rust crates and TypeScript runtime
+
+- Added IPC serialization tests in remoteingress-bin (IPC request/response/event)
+- Added serde and async tests for Edge and Handshake configs and EdgeEvent/EdgeStatus in remoteingress-core (edge.rs)
+- Added extensive Hub tests: constant_time_eq, PROXY header port parsing, serde/camelCase checks, Hub events and async TunnelHub behavior (hub.rs)
+- Added STUN parser unit tests including XOR_MAPPED_ADDRESS, MAPPED_ADDRESS fallback, truncated attribute handling and other edge cases (stun.rs)
+- Added protocol frame encoding and FrameReader tests covering all frame types, payload limits and EOF conditions (remoteingress-protocol)
+- Added TypeScript Node tests for token encode/decode edge cases and RemoteIngressHub/RemoteIngressEdge class basics (test/*.node.ts)
+
+## 2026-02-18 - 3.2.0 - feat(remoteingress (edge/hub/protocol))
+add dynamic port configuration: handshake, FRAME_CONFIG frames, and hot-reloadable listeners
+
+- Introduce a JSON handshake from hub -> edge with initial listen ports and stun interval so edges can configure listeners at connect time.
+- Add FRAME_CONFIG (0x06) to the protocol and implement runtime config updates pushed from hub to connected edges.
+- Edge now applies initial ports and supports hot-reloading: spawn/abort listeners when ports change, and emit PortsAssigned / PortsUpdated events.
+- Hub now stores allowed edge metadata (listen_ports, stun_interval_secs), sends handshake responses on auth, and forwards config updates to connected edges.
+- TypeScript bridge/client updated to emit new port events and periodically log status; updateAllowedEdges API accepts listenPorts and stunIntervalSecs.
+- Stun interval handling moved to use handshake-provided/stored value instead of config.listen_ports being static.
+
 ## 2026-02-18 - 3.1.1 - fix(readme)
 update README: add issue reporting/security section, document connection tokens and token utilities, clarify architecture/API and improve examples/formatting
 

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@serve.zone/remoteingress",
-  "version": "3.1.1",
+  "version": "3.2.1",
   "private": false,
   "description": "Edge ingress tunnel for DcRouter - accepts incoming TCP connections at network edge and tunnels them to DcRouter SmartProxy preserving client IP via PROXY protocol v1.",
   "main": "dist_ts/index.js",

rust/crates/remoteingress-bin/src/main.rs

@@ -301,6 +301,18 @@ async fn handle_request(
                                     serde_json::json!({ "ip": ip }),
                                 );
                             }
+                            EdgeEvent::PortsAssigned { listen_ports } => {
+                                send_event(
+                                    "portsAssigned",
+                                    serde_json::json!({ "listenPorts": listen_ports }),
+                                );
+                            }
+                            EdgeEvent::PortsUpdated { listen_ports } => {
+                                send_event(
+                                    "portsUpdated",
+                                    serde_json::json!({ "listenPorts": listen_ports }),
+                                );
+                            }
                         }
                     }
                 });
@@ -357,3 +369,58 @@ async fn handle_request(
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_ipc_request_deserialize() {
+        let json = r#"{"id": "1", "method": "ping", "params": {}}"#;
+        let req: IpcRequest = serde_json::from_str(json).unwrap();
+        assert_eq!(req.id, "1");
+        assert_eq!(req.method, "ping");
+        assert!(req.params.is_object());
+    }
+
+    #[test]
+    fn test_ipc_response_skip_error_when_none() {
+        let resp = IpcResponse {
+            id: "1".to_string(),
+            success: true,
+            result: Some(serde_json::json!({"pong": true})),
+            error: None,
+        };
+        let json = serde_json::to_value(&resp).unwrap();
+        assert_eq!(json["id"], "1");
+        assert_eq!(json["success"], true);
+        assert_eq!(json["result"]["pong"], true);
+        assert!(json.get("error").is_none());
+    }
+
+    #[test]
+    fn test_ipc_response_skip_result_when_none() {
+        let resp = IpcResponse {
+            id: "2".to_string(),
+            success: false,
+            result: None,
+            error: Some("something failed".to_string()),
+        };
+        let json = serde_json::to_value(&resp).unwrap();
+        assert_eq!(json["id"], "2");
+        assert_eq!(json["success"], false);
+        assert_eq!(json["error"], "something failed");
+        assert!(json.get("result").is_none());
+    }
+
+    #[test]
+    fn test_ipc_event_serialize() {
+        let evt = IpcEvent {
+            event: "ready".to_string(),
+            data: serde_json::json!({"version": "2.0.0"}),
+        };
+        let json = serde_json::to_value(&evt).unwrap();
+        assert_eq!(json["event"], "ready");
+        assert_eq!(json["data"]["version"], "2.0.0");
+    }
+}

rust/crates/remoteingress-core/src/edge.rs

@@ -1,15 +1,16 @@
 use std::collections::HashMap;
 use std::sync::atomic::{AtomicU32, Ordering};
 use std::sync::Arc;
-use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
 use tokio::net::{TcpListener, TcpStream};
 use tokio::sync::{mpsc, Mutex, RwLock};
+use tokio::task::JoinHandle;
 use tokio_rustls::TlsConnector;
 use serde::{Deserialize, Serialize};
 
 use remoteingress_protocol::*;
 
-/// Edge configuration.
+/// Edge configuration (hub-host + credentials only; ports come from hub).
 #[derive(Debug, Clone, Deserialize, Serialize)]
 #[serde(rename_all = "camelCase")]
 pub struct EdgeConfig {
@@ -17,8 +18,26 @@ pub struct EdgeConfig {
     pub hub_port: u16,
     pub edge_id: String,
     pub secret: String,
-    pub listen_ports: Vec<u16>,
-    pub stun_interval_secs: Option<u64>,
+}
+
+/// Handshake config received from hub after authentication.
+#[derive(Debug, Clone, Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct HandshakeConfig {
+    listen_ports: Vec<u16>,
+    #[serde(default = "default_stun_interval")]
+    stun_interval_secs: u64,
+}
+
+fn default_stun_interval() -> u64 {
+    300
+}
+
+/// Runtime config update received from hub via FRAME_CONFIG.
+#[derive(Debug, Clone, Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct ConfigUpdate {
+    listen_ports: Vec<u16>,
 }
 
 /// Events emitted by the edge.
@@ -30,6 +49,10 @@ pub enum EdgeEvent {
     TunnelDisconnected,
     #[serde(rename_all = "camelCase")]
     PublicIpDiscovered { ip: String },
+    #[serde(rename_all = "camelCase")]
+    PortsAssigned { listen_ports: Vec<u16> },
+    #[serde(rename_all = "camelCase")]
+    PortsUpdated { listen_ports: Vec<u16> },
 }
 
 /// Edge status response.
@@ -54,6 +77,7 @@ pub struct TunnelEdge {
     public_ip: Arc<RwLock<Option<String>>>,
     active_streams: Arc<AtomicU32>,
     next_stream_id: Arc<AtomicU32>,
+    listen_ports: Arc<RwLock<Vec<u16>>>,
 }
 
 impl TunnelEdge {
@@ -69,6 +93,7 @@ impl TunnelEdge {
             public_ip: Arc::new(RwLock::new(None)),
             active_streams: Arc::new(AtomicU32::new(0)),
             next_stream_id: Arc::new(AtomicU32::new(1)),
+            listen_ports: Arc::new(RwLock::new(Vec::new())),
         }
     }
 
@@ -84,7 +109,7 @@ impl TunnelEdge {
             connected: *self.connected.read().await,
             public_ip: self.public_ip.read().await.clone(),
             active_streams: self.active_streams.load(Ordering::Relaxed) as usize,
-            listen_ports: self.config.read().await.listen_ports.clone(),
+            listen_ports: self.listen_ports.read().await.clone(),
         }
     }
 
@@ -100,6 +125,7 @@ impl TunnelEdge {
         let active_streams = self.active_streams.clone();
         let next_stream_id = self.next_stream_id.clone();
         let event_tx = self.event_tx.clone();
+        let listen_ports = self.listen_ports.clone();
 
         tokio::spawn(async move {
             edge_main_loop(
@@ -109,6 +135,7 @@ impl TunnelEdge {
                 active_streams,
                 next_stream_id,
                 event_tx,
+                listen_ports,
                 shutdown_rx,
             )
             .await;
@@ -124,6 +151,7 @@ impl TunnelEdge {
         }
         *self.running.write().await = false;
         *self.connected.write().await = false;
+        self.listen_ports.write().await.clear();
     }
 }
 
@@ -134,6 +162,7 @@ async fn edge_main_loop(
     active_streams: Arc<AtomicU32>,
     next_stream_id: Arc<AtomicU32>,
     event_tx: mpsc::UnboundedSender<EdgeEvent>,
+    listen_ports: Arc<RwLock<Vec<u16>>>,
     mut shutdown_rx: mpsc::Receiver<()>,
 ) {
     let mut backoff_ms: u64 = 1000;
@@ -148,6 +177,7 @@ async fn edge_main_loop(
             &active_streams,
             &next_stream_id,
             &event_tx,
+            &listen_ports,
             &mut shutdown_rx,
         )
         .await;
@@ -155,6 +185,7 @@ async fn edge_main_loop(
         *connected.write().await = false;
         let _ = event_tx.send(EdgeEvent::TunnelDisconnected);
         active_streams.store(0, Ordering::Relaxed);
+        listen_ports.write().await.clear();
 
         match result {
             EdgeLoopResult::Shutdown => break,
@@ -182,6 +213,7 @@ async fn connect_to_hub_and_run(
     active_streams: &Arc<AtomicU32>,
     next_stream_id: &Arc<AtomicU32>,
     event_tx: &mpsc::UnboundedSender<EdgeEvent>,
+    listen_ports: &Arc<RwLock<Vec<u16>>>,
     shutdown_rx: &mut mpsc::Receiver<()>,
 ) -> EdgeLoopResult {
     // Build TLS connector that skips cert verification (auth is via secret)
@@ -220,12 +252,47 @@ async fn connect_to_hub_and_run(
         return EdgeLoopResult::Reconnect;
     }
 
+    // Read handshake response line from hub (JSON with initial config)
+    let mut buf_reader = BufReader::new(read_half);
+    let mut handshake_line = String::new();
+    match buf_reader.read_line(&mut handshake_line).await {
+        Ok(0) => {
+            log::error!("Hub rejected connection (EOF before handshake)");
+            return EdgeLoopResult::Reconnect;
+        }
+        Ok(_) => {}
+        Err(e) => {
+            log::error!("Failed to read handshake response: {}", e);
+            return EdgeLoopResult::Reconnect;
+        }
+    }
+
+    let handshake: HandshakeConfig = match serde_json::from_str(handshake_line.trim()) {
+        Ok(h) => h,
+        Err(e) => {
+            log::error!("Invalid handshake response: {}", e);
+            return EdgeLoopResult::Reconnect;
+        }
+    };
+
+    log::info!(
+        "Handshake from hub: ports {:?}, stun_interval {}s",
+        handshake.listen_ports,
+        handshake.stun_interval_secs
+    );
+
     *connected.write().await = true;
     let _ = event_tx.send(EdgeEvent::TunnelConnected);
     log::info!("Connected to hub at {}", addr);
 
+    // Store initial ports and emit event
+    *listen_ports.write().await = handshake.listen_ports.clone();
+    let _ = event_tx.send(EdgeEvent::PortsAssigned {
+        listen_ports: handshake.listen_ports.clone(),
+    });
+
     // Start STUN discovery
-    let stun_interval = config.stun_interval_secs.unwrap_or(300);
+    let stun_interval = handshake.stun_interval_secs;
     let public_ip_clone = public_ip.clone();
     let event_tx_clone = event_tx.clone();
     let stun_handle = tokio::spawn(async move {
@@ -249,14 +316,112 @@ async fn connect_to_hub_and_run(
     // Shared tunnel writer
     let tunnel_writer = Arc::new(Mutex::new(write_half));
 
-    // Start TCP listeners for each port
-    let mut listener_handles = Vec::new();
-    for &port in &config.listen_ports {
+    // Start TCP listeners for initial ports (hot-reloadable)
+    let mut port_listeners: HashMap<u16, JoinHandle<()>> = HashMap::new();
+    apply_port_config(
+        &handshake.listen_ports,
+        &mut port_listeners,
+        &tunnel_writer,
+        &client_writers,
+        active_streams,
+        next_stream_id,
+        &config.edge_id,
+    );
+
+    // Read frames from hub
+    let mut frame_reader = FrameReader::new(buf_reader);
+    let result = loop {
+        tokio::select! {
+            frame_result = frame_reader.next_frame() => {
+                match frame_result {
+                    Ok(Some(frame)) => {
+                        match frame.frame_type {
+                            FRAME_DATA_BACK => {
+                                let writers = client_writers.lock().await;
+                                if let Some(tx) = writers.get(&frame.stream_id) {
+                                    let _ = tx.send(frame.payload).await;
+                                }
+                            }
+                            FRAME_CLOSE_BACK => {
+                                let mut writers = client_writers.lock().await;
+                                writers.remove(&frame.stream_id);
+                            }
+                            FRAME_CONFIG => {
+                                if let Ok(update) = serde_json::from_slice::<ConfigUpdate>(&frame.payload) {
+                                    log::info!("Config update from hub: ports {:?}", update.listen_ports);
+                                    *listen_ports.write().await = update.listen_ports.clone();
+                                    let _ = event_tx.send(EdgeEvent::PortsUpdated {
+                                        listen_ports: update.listen_ports.clone(),
+                                    });
+                                    apply_port_config(
+                                        &update.listen_ports,
+                                        &mut port_listeners,
+                                        &tunnel_writer,
+                                        &client_writers,
+                                        active_streams,
+                                        next_stream_id,
+                                        &config.edge_id,
+                                    );
+                                }
+                            }
+                            _ => {
+                                log::warn!("Unexpected frame type {} from hub", frame.frame_type);
+                            }
+                        }
+                    }
+                    Ok(None) => {
+                        log::info!("Hub disconnected (EOF)");
+                        break EdgeLoopResult::Reconnect;
+                    }
+                    Err(e) => {
+                        log::error!("Hub frame error: {}", e);
+                        break EdgeLoopResult::Reconnect;
+                    }
+                }
+            }
+            _ = shutdown_rx.recv() => {
+                break EdgeLoopResult::Shutdown;
+            }
+        }
+    };
+
+    // Cleanup
+    stun_handle.abort();
+    for (_, h) in port_listeners.drain() {
+        h.abort();
+    }
+
+    result
+}
+
+/// Apply a new port configuration: spawn listeners for added ports, abort removed ports.
+fn apply_port_config(
+    new_ports: &[u16],
+    port_listeners: &mut HashMap<u16, JoinHandle<()>>,
+    tunnel_writer: &Arc<Mutex<tokio::io::WriteHalf<tokio_rustls::client::TlsStream<TcpStream>>>>,
+    client_writers: &Arc<Mutex<HashMap<u32, mpsc::Sender<Vec<u8>>>>>,
+    active_streams: &Arc<AtomicU32>,
+    next_stream_id: &Arc<AtomicU32>,
+    edge_id: &str,
+) {
+    let new_set: std::collections::HashSet<u16> = new_ports.iter().copied().collect();
+    let old_set: std::collections::HashSet<u16> = port_listeners.keys().copied().collect();
+
+    // Remove ports no longer needed
+    for &port in old_set.difference(&new_set) {
+        if let Some(handle) = port_listeners.remove(&port) {
+            log::info!("Stopping listener on port {}", port);
+            handle.abort();
+        }
+    }
+
+    // Add new ports
+    for &port in new_set.difference(&old_set) {
         let tunnel_writer = tunnel_writer.clone();
         let client_writers = client_writers.clone();
         let active_streams = active_streams.clone();
         let next_stream_id = next_stream_id.clone();
-        let edge_id = config.edge_id.clone();
+        let edge_id = edge_id.to_string();
 
         let handle = tokio::spawn(async move {
             let listener = match TcpListener::bind(("0.0.0.0", port)).await {
@@ -299,55 +464,8 @@ async fn connect_to_hub_and_run(
                 }
             }
         });
-        listener_handles.push(handle);
+        port_listeners.insert(port, handle);
     }
-
-    // Read frames from hub
-    let mut frame_reader = FrameReader::new(read_half);
-    let result = loop {
-        tokio::select! {
-            frame_result = frame_reader.next_frame() => {
-                match frame_result {
-                    Ok(Some(frame)) => {
-                        match frame.frame_type {
-                            FRAME_DATA_BACK => {
-                                let writers = client_writers.lock().await;
-                                if let Some(tx) = writers.get(&frame.stream_id) {
-                                    let _ = tx.send(frame.payload).await;
-                                }
-                            }
-                            FRAME_CLOSE_BACK => {
-                                let mut writers = client_writers.lock().await;
-                                writers.remove(&frame.stream_id);
-                            }
-                            _ => {
-                                log::warn!("Unexpected frame type {} from hub", frame.frame_type);
-                            }
-                        }
-                    }
-                    Ok(None) => {
-                        log::info!("Hub disconnected (EOF)");
-                        break EdgeLoopResult::Reconnect;
-                    }
-                    Err(e) => {
-                        log::error!("Hub frame error: {}", e);
-                        break EdgeLoopResult::Reconnect;
-                    }
-                }
-            }
-            _ = shutdown_rx.recv() => {
-                break EdgeLoopResult::Shutdown;
-            }
-        }
-    };
-
-    // Cleanup
-    stun_handle.abort();
-    for h in listener_handles {
-        h.abort();
-    }
-
-    result
 }
 
 async fn handle_client_connection(
@@ -426,6 +544,186 @@ async fn handle_client_connection(
     let _ = edge_id; // used for logging context
 }
 
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    // --- Serde tests ---
+
+    #[test]
+    fn test_edge_config_deserialize_camel_case() {
+        let json = r#"{
+            "hubHost": "hub.example.com",
+            "hubPort": 8443,
+            "edgeId": "edge-1",
+            "secret": "my-secret"
+        }"#;
+        let config: EdgeConfig = serde_json::from_str(json).unwrap();
+        assert_eq!(config.hub_host, "hub.example.com");
+        assert_eq!(config.hub_port, 8443);
+        assert_eq!(config.edge_id, "edge-1");
+        assert_eq!(config.secret, "my-secret");
+    }
+
+    #[test]
+    fn test_edge_config_serialize_roundtrip() {
+        let config = EdgeConfig {
+            hub_host: "host.test".to_string(),
+            hub_port: 9999,
+            edge_id: "e1".to_string(),
+            secret: "sec".to_string(),
+        };
+        let json = serde_json::to_string(&config).unwrap();
+        let back: EdgeConfig = serde_json::from_str(&json).unwrap();
+        assert_eq!(back.hub_host, config.hub_host);
+        assert_eq!(back.hub_port, config.hub_port);
+        assert_eq!(back.edge_id, config.edge_id);
+        assert_eq!(back.secret, config.secret);
+    }
+
+    #[test]
+    fn test_handshake_config_deserialize_all_fields() {
+        let json = r#"{"listenPorts": [80, 443], "stunIntervalSecs": 120}"#;
+        let hc: HandshakeConfig = serde_json::from_str(json).unwrap();
+        assert_eq!(hc.listen_ports, vec![80, 443]);
+        assert_eq!(hc.stun_interval_secs, 120);
+    }
+
+    #[test]
+    fn test_handshake_config_default_stun_interval() {
+        let json = r#"{"listenPorts": [443]}"#;
+        let hc: HandshakeConfig = serde_json::from_str(json).unwrap();
+        assert_eq!(hc.listen_ports, vec![443]);
+        assert_eq!(hc.stun_interval_secs, 300);
+    }
+
+    #[test]
+    fn test_config_update_deserialize() {
+        let json = r#"{"listenPorts": [8080, 9090]}"#;
+        let update: ConfigUpdate = serde_json::from_str(json).unwrap();
+        assert_eq!(update.listen_ports, vec![8080, 9090]);
+    }
+
+    #[test]
+    fn test_edge_status_serialize() {
+        let status = EdgeStatus {
+            running: true,
+            connected: true,
+            public_ip: Some("1.2.3.4".to_string()),
+            active_streams: 5,
+            listen_ports: vec![443],
+        };
+        let json = serde_json::to_value(&status).unwrap();
+        assert_eq!(json["running"], true);
+        assert_eq!(json["connected"], true);
+        assert_eq!(json["publicIp"], "1.2.3.4");
+        assert_eq!(json["activeStreams"], 5);
+        assert_eq!(json["listenPorts"], serde_json::json!([443]));
+    }
+
+    #[test]
+    fn test_edge_status_serialize_none_ip() {
+        let status = EdgeStatus {
+            running: false,
+            connected: false,
+            public_ip: None,
+            active_streams: 0,
+            listen_ports: vec![],
+        };
+        let json = serde_json::to_value(&status).unwrap();
+        assert!(json["publicIp"].is_null());
+    }
+
+    #[test]
+    fn test_edge_event_tunnel_connected() {
+        let event = EdgeEvent::TunnelConnected;
+        let json = serde_json::to_value(&event).unwrap();
+        assert_eq!(json["type"], "tunnelConnected");
+    }
+
+    #[test]
+    fn test_edge_event_tunnel_disconnected() {
+        let event = EdgeEvent::TunnelDisconnected;
+        let json = serde_json::to_value(&event).unwrap();
+        assert_eq!(json["type"], "tunnelDisconnected");
+    }
+
+    #[test]
+    fn test_edge_event_public_ip_discovered() {
+        let event = EdgeEvent::PublicIpDiscovered {
+            ip: "203.0.113.1".to_string(),
+        };
+        let json = serde_json::to_value(&event).unwrap();
+        assert_eq!(json["type"], "publicIpDiscovered");
+        assert_eq!(json["ip"], "203.0.113.1");
+    }
+
+    #[test]
+    fn test_edge_event_ports_assigned() {
+        let event = EdgeEvent::PortsAssigned {
+            listen_ports: vec![443, 8080],
+        };
+        let json = serde_json::to_value(&event).unwrap();
+        assert_eq!(json["type"], "portsAssigned");
+        assert_eq!(json["listenPorts"], serde_json::json!([443, 8080]));
+    }
+
+    #[test]
+    fn test_edge_event_ports_updated() {
+        let event = EdgeEvent::PortsUpdated {
+            listen_ports: vec![9090],
+        };
+        let json = serde_json::to_value(&event).unwrap();
+        assert_eq!(json["type"], "portsUpdated");
+        assert_eq!(json["listenPorts"], serde_json::json!([9090]));
+    }
+
+    // --- Async tests ---
+
+    #[tokio::test]
+    async fn test_tunnel_edge_new_get_status() {
+        let edge = TunnelEdge::new(EdgeConfig {
+            hub_host: "localhost".to_string(),
+            hub_port: 8443,
+            edge_id: "test-edge".to_string(),
+            secret: "test-secret".to_string(),
+        });
+        let status = edge.get_status().await;
+        assert!(!status.running);
+        assert!(!status.connected);
+        assert!(status.public_ip.is_none());
+        assert_eq!(status.active_streams, 0);
+        assert!(status.listen_ports.is_empty());
+    }
+
+    #[tokio::test]
+    async fn test_tunnel_edge_take_event_rx() {
+        let edge = TunnelEdge::new(EdgeConfig {
+            hub_host: "localhost".to_string(),
+            hub_port: 8443,
+            edge_id: "e".to_string(),
+            secret: "s".to_string(),
+        });
+        let rx1 = edge.take_event_rx().await;
+        assert!(rx1.is_some());
+        let rx2 = edge.take_event_rx().await;
+        assert!(rx2.is_none());
+    }
+
+    #[tokio::test]
+    async fn test_tunnel_edge_stop_without_start() {
+        let edge = TunnelEdge::new(EdgeConfig {
+            hub_host: "localhost".to_string(),
+            hub_port: 8443,
+            edge_id: "e".to_string(),
+            secret: "s".to_string(),
+        });
+        edge.stop().await; // should not panic
+        let status = edge.get_status().await;
+        assert!(!status.running);
+    }
+}
+
 /// TLS certificate verifier that accepts any certificate (auth is via shared secret).
 #[derive(Debug)]
 struct NoCertVerifier;

rust/crates/remoteingress-core/src/hub.rs

@@ -37,6 +37,24 @@ impl Default for HubConfig {
 pub struct AllowedEdge {
     pub id: String,
     pub secret: String,
+    #[serde(default)]
+    pub listen_ports: Vec<u16>,
+    pub stun_interval_secs: Option<u64>,
+}
+
+/// Handshake response sent to edge after authentication.
+#[derive(Debug, Clone, Serialize)]
+#[serde(rename_all = "camelCase")]
+struct HandshakeResponse {
+    listen_ports: Vec<u16>,
+    stun_interval_secs: u64,
+}
+
+/// Configuration update pushed to a connected edge at runtime.
+#[derive(Debug, Clone, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct EdgeConfigUpdate {
+    pub listen_ports: Vec<u16>,
 }
 
 /// Runtime status of a connected edge.
@@ -75,7 +93,7 @@ pub struct HubStatus {
 /// The tunnel hub that accepts edge connections and demuxes streams to SmartProxy.
 pub struct TunnelHub {
     config: RwLock<HubConfig>,
-    allowed_edges: Arc<RwLock<HashMap<String, String>>>, // id -> secret
+    allowed_edges: Arc<RwLock<HashMap<String, AllowedEdge>>>,
     connected_edges: Arc<Mutex<HashMap<String, ConnectedEdgeInfo>>>,
     event_tx: mpsc::UnboundedSender<HubEvent>,
     event_rx: Mutex<Option<mpsc::UnboundedReceiver<HubEvent>>>,
@@ -86,6 +104,7 @@ pub struct TunnelHub {
 struct ConnectedEdgeInfo {
     connected_at: u64,
     active_streams: Arc<Mutex<HashMap<u32, mpsc::Sender<Vec<u8>>>>>,
+    config_tx: mpsc::Sender<EdgeConfigUpdate>,
 }
 
 impl TunnelHub {
@@ -108,12 +127,35 @@ impl TunnelHub {
     }
 
     /// Update the list of allowed edges.
+    /// For any currently-connected edge whose ports changed, push a config update.
     pub async fn update_allowed_edges(&self, edges: Vec<AllowedEdge>) {
         let mut map = self.allowed_edges.write().await;
-        map.clear();
-        for edge in edges {
-            map.insert(edge.id, edge.secret);
+
+        // Build new map
+        let mut new_map = HashMap::new();
+        for edge in &edges {
+            new_map.insert(edge.id.clone(), edge.clone());
         }
+
+        // Push config updates to connected edges whose ports changed
+        let connected = self.connected_edges.lock().await;
+        for edge in &edges {
+            if let Some(info) = connected.get(&edge.id) {
+                // Check if ports changed compared to old config
+                let ports_changed = match map.get(&edge.id) {
+                    Some(old) => old.listen_ports != edge.listen_ports,
+                    None => true, // newly allowed edge that's already connected
+                };
+                if ports_changed {
+                    let update = EdgeConfigUpdate {
+                        listen_ports: edge.listen_ports.clone(),
+                    };
+                    let _ = info.config_tx.try_send(update);
+                }
+            }
+        }
+
+        *map = new_map;
     }
 
     /// Get the current hub status.
@@ -208,13 +250,13 @@ impl TunnelHub {
 async fn handle_edge_connection(
     stream: TcpStream,
     acceptor: TlsAcceptor,
-    allowed: Arc<RwLock<HashMap<String, String>>>,
+    allowed: Arc<RwLock<HashMap<String, AllowedEdge>>>,
     connected: Arc<Mutex<HashMap<String, ConnectedEdgeInfo>>>,
     event_tx: mpsc::UnboundedSender<HubEvent>,
     target_host: String,
 ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     let tls_stream = acceptor.accept(stream).await?;
-    let (read_half, write_half) = tokio::io::split(tls_stream);
+    let (read_half, mut write_half) = tokio::io::split(tls_stream);
     let mut buf_reader = BufReader::new(read_half);
 
     // Read auth line: "EDGE <edgeId> <secret>\n"
@@ -230,26 +272,36 @@ async fn handle_edge_connection(
     let edge_id = parts[1].to_string();
     let secret = parts[2];
 
-    // Verify credentials
-    {
+    // Verify credentials and extract edge config
+    let (listen_ports, stun_interval_secs) = {
         let edges = allowed.read().await;
         match edges.get(&edge_id) {
-            Some(expected) => {
-                if !constant_time_eq(secret.as_bytes(), expected.as_bytes()) {
+            Some(edge) => {
+                if !constant_time_eq(secret.as_bytes(), edge.secret.as_bytes()) {
                     return Err(format!("invalid secret for edge {}", edge_id).into());
                 }
+                (edge.listen_ports.clone(), edge.stun_interval_secs.unwrap_or(300))
             }
             None => {
                 return Err(format!("unknown edge {}", edge_id).into());
             }
         }
-    }
+    };
 
     log::info!("Edge {} authenticated", edge_id);
     let _ = event_tx.send(HubEvent::EdgeConnected {
         edge_id: edge_id.clone(),
     });
 
+    // Send handshake response with initial config before frame protocol begins
+    let handshake = HandshakeResponse {
+        listen_ports: listen_ports.clone(),
+        stun_interval_secs,
+    };
+    let mut handshake_json = serde_json::to_string(&handshake)?;
+    handshake_json.push('\n');
+    write_half.write_all(handshake_json.as_bytes()).await?;
+
     // Track this edge
     let streams: Arc<Mutex<HashMap<u32, mpsc::Sender<Vec<u8>>>>> =
         Arc::new(Mutex::new(HashMap::new()));
@@ -258,6 +310,9 @@ async fn handle_edge_connection(
         .unwrap_or_default()
         .as_secs();
 
+    // Create config update channel
+    let (config_tx, mut config_rx) = mpsc::channel::<EdgeConfigUpdate>(16);
+
     {
         let mut edges = connected.lock().await;
         edges.insert(
@@ -265,6 +320,7 @@ async fn handle_edge_connection(
             ConnectedEdgeInfo {
                 connected_at: now,
                 active_streams: streams.clone(),
+                config_tx,
             },
         );
     }
@@ -272,6 +328,23 @@ async fn handle_edge_connection(
     // Shared writer for sending frames back to edge
     let write_half = Arc::new(Mutex::new(write_half));
 
+    // Spawn task to forward config updates as FRAME_CONFIG frames
+    let config_writer = write_half.clone();
+    let config_edge_id = edge_id.clone();
+    let config_handle = tokio::spawn(async move {
+        while let Some(update) = config_rx.recv().await {
+            if let Ok(payload) = serde_json::to_vec(&update) {
+                let frame = encode_frame(0, FRAME_CONFIG, &payload);
+                let mut w = config_writer.lock().await;
+                if w.write_all(&frame).await.is_err() {
+                    log::error!("Failed to send config update to edge {}", config_edge_id);
+                    break;
+                }
+                log::info!("Sent config update to edge {}: ports {:?}", config_edge_id, update.listen_ports);
+            }
+        }
+    });
+
     // Frame reading loop
     let mut frame_reader = FrameReader::new(buf_reader);
 
@@ -398,6 +471,7 @@ async fn handle_edge_connection(
     }
 
     // Cleanup
+    config_handle.abort();
     {
         let mut edges = connected.lock().await;
         edges.remove(&edge_id);
@@ -475,3 +549,210 @@ fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
     }
     diff == 0
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    // --- constant_time_eq tests ---
+
+    #[test]
+    fn test_constant_time_eq_equal() {
+        assert!(constant_time_eq(b"hello", b"hello"));
+    }
+
+    #[test]
+    fn test_constant_time_eq_different_content() {
+        assert!(!constant_time_eq(b"hello", b"world"));
+    }
+
+    #[test]
+    fn test_constant_time_eq_different_lengths() {
+        assert!(!constant_time_eq(b"short", b"longer"));
+    }
+
+    #[test]
+    fn test_constant_time_eq_both_empty() {
+        assert!(constant_time_eq(b"", b""));
+    }
+
+    #[test]
+    fn test_constant_time_eq_one_empty() {
+        assert!(!constant_time_eq(b"", b"notempty"));
+    }
+
+    #[test]
+    fn test_constant_time_eq_single_bit_difference() {
+        // 'A' = 0x41, 'a' = 0x61 — differ by one bit
+        assert!(!constant_time_eq(b"A", b"a"));
+    }
+
+    // --- parse_dest_port_from_proxy tests ---
+
+    #[test]
+    fn test_parse_dest_port_443() {
+        let header = "PROXY TCP4 1.2.3.4 5.6.7.8 12345 443\r\n";
+        assert_eq!(parse_dest_port_from_proxy(header), Some(443));
+    }
+
+    #[test]
+    fn test_parse_dest_port_80() {
+        let header = "PROXY TCP4 10.0.0.1 10.0.0.2 54321 80\r\n";
+        assert_eq!(parse_dest_port_from_proxy(header), Some(80));
+    }
+
+    #[test]
+    fn test_parse_dest_port_65535() {
+        let header = "PROXY TCP4 10.0.0.1 10.0.0.2 1 65535\r\n";
+        assert_eq!(parse_dest_port_from_proxy(header), Some(65535));
+    }
+
+    #[test]
+    fn test_parse_dest_port_too_few_fields() {
+        let header = "PROXY TCP4 1.2.3.4";
+        assert_eq!(parse_dest_port_from_proxy(header), None);
+    }
+
+    #[test]
+    fn test_parse_dest_port_empty_string() {
+        assert_eq!(parse_dest_port_from_proxy(""), None);
+    }
+
+    #[test]
+    fn test_parse_dest_port_non_numeric() {
+        let header = "PROXY TCP4 1.2.3.4 5.6.7.8 12345 abc\r\n";
+        assert_eq!(parse_dest_port_from_proxy(header), None);
+    }
+
+    // --- Serde tests ---
+
+    #[test]
+    fn test_allowed_edge_deserialize_all_fields() {
+        let json = r#"{
+            "id": "edge-1",
+            "secret": "s3cret",
+            "listenPorts": [443, 8080],
+            "stunIntervalSecs": 120
+        }"#;
+        let edge: AllowedEdge = serde_json::from_str(json).unwrap();
+        assert_eq!(edge.id, "edge-1");
+        assert_eq!(edge.secret, "s3cret");
+        assert_eq!(edge.listen_ports, vec![443, 8080]);
+        assert_eq!(edge.stun_interval_secs, Some(120));
+    }
+
+    #[test]
+    fn test_allowed_edge_deserialize_with_defaults() {
+        let json = r#"{"id": "edge-2", "secret": "key"}"#;
+        let edge: AllowedEdge = serde_json::from_str(json).unwrap();
+        assert_eq!(edge.id, "edge-2");
+        assert_eq!(edge.secret, "key");
+        assert!(edge.listen_ports.is_empty());
+        assert_eq!(edge.stun_interval_secs, None);
+    }
+
+    #[test]
+    fn test_handshake_response_serializes_camel_case() {
+        let resp = HandshakeResponse {
+            listen_ports: vec![443, 8080],
+            stun_interval_secs: 300,
+        };
+        let json = serde_json::to_value(&resp).unwrap();
+        assert_eq!(json["listenPorts"], serde_json::json!([443, 8080]));
+        assert_eq!(json["stunIntervalSecs"], 300);
+        // Ensure snake_case keys are NOT present
+        assert!(json.get("listen_ports").is_none());
+        assert!(json.get("stun_interval_secs").is_none());
+    }
+
+    #[test]
+    fn test_edge_config_update_serializes_camel_case() {
+        let update = EdgeConfigUpdate {
+            listen_ports: vec![80, 443],
+        };
+        let json = serde_json::to_value(&update).unwrap();
+        assert_eq!(json["listenPorts"], serde_json::json!([80, 443]));
+        assert!(json.get("listen_ports").is_none());
+    }
+
+    #[test]
+    fn test_hub_config_default() {
+        let config = HubConfig::default();
+        assert_eq!(config.tunnel_port, 8443);
+        assert_eq!(config.target_host, Some("127.0.0.1".to_string()));
+        assert!(config.tls_cert_pem.is_none());
+        assert!(config.tls_key_pem.is_none());
+    }
+
+    #[test]
+    fn test_hub_event_edge_connected_serialize() {
+        let event = HubEvent::EdgeConnected {
+            edge_id: "edge-1".to_string(),
+        };
+        let json = serde_json::to_value(&event).unwrap();
+        assert_eq!(json["type"], "edgeConnected");
+        assert_eq!(json["edgeId"], "edge-1");
+    }
+
+    #[test]
+    fn test_hub_event_edge_disconnected_serialize() {
+        let event = HubEvent::EdgeDisconnected {
+            edge_id: "edge-2".to_string(),
+        };
+        let json = serde_json::to_value(&event).unwrap();
+        assert_eq!(json["type"], "edgeDisconnected");
+        assert_eq!(json["edgeId"], "edge-2");
+    }
+
+    #[test]
+    fn test_hub_event_stream_opened_serialize() {
+        let event = HubEvent::StreamOpened {
+            edge_id: "e".to_string(),
+            stream_id: 42,
+        };
+        let json = serde_json::to_value(&event).unwrap();
+        assert_eq!(json["type"], "streamOpened");
+        assert_eq!(json["edgeId"], "e");
+        assert_eq!(json["streamId"], 42);
+    }
+
+    #[test]
+    fn test_hub_event_stream_closed_serialize() {
+        let event = HubEvent::StreamClosed {
+            edge_id: "e".to_string(),
+            stream_id: 7,
+        };
+        let json = serde_json::to_value(&event).unwrap();
+        assert_eq!(json["type"], "streamClosed");
+        assert_eq!(json["edgeId"], "e");
+        assert_eq!(json["streamId"], 7);
+    }
+
+    // --- Async tests ---
+
+    #[tokio::test]
+    async fn test_tunnel_hub_new_get_status() {
+        let hub = TunnelHub::new(HubConfig::default());
+        let status = hub.get_status().await;
+        assert!(!status.running);
+        assert!(status.connected_edges.is_empty());
+        assert_eq!(status.tunnel_port, 8443);
+    }
+
+    #[tokio::test]
+    async fn test_tunnel_hub_take_event_rx() {
+        let hub = TunnelHub::new(HubConfig::default());
+        let rx1 = hub.take_event_rx().await;
+        assert!(rx1.is_some());
+        let rx2 = hub.take_event_rx().await;
+        assert!(rx2.is_none());
+    }
+
+    #[tokio::test]
+    async fn test_tunnel_hub_stop_without_start() {
+        let hub = TunnelHub::new(HubConfig::default());
+        hub.stop().await; // should not panic
+        let status = hub.get_status().await;
+        assert!(!status.running);
+    }
+}

rust/crates/remoteingress-core/src/stun.rs

@@ -121,6 +121,133 @@ fn parse_stun_response(data: &[u8], _txn_id: &[u8; 12]) -> Option<String> {
     None
 }
 
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// Build a synthetic STUN Binding Response with given attributes.
+    fn build_stun_response(attrs: &[(u16, &[u8])]) -> Vec<u8> {
+        let mut attrs_bytes = Vec::new();
+        for &(attr_type, attr_data) in attrs {
+            attrs_bytes.extend_from_slice(&attr_type.to_be_bytes());
+            attrs_bytes.extend_from_slice(&(attr_data.len() as u16).to_be_bytes());
+            attrs_bytes.extend_from_slice(attr_data);
+            // Pad to 4-byte boundary
+            let pad = (4 - (attr_data.len() % 4)) % 4;
+            attrs_bytes.extend(std::iter::repeat(0u8).take(pad));
+        }
+
+        let mut response = Vec::new();
+        // msg_type = 0x0101 (Binding Response)
+        response.extend_from_slice(&0x0101u16.to_be_bytes());
+        // message length
+        response.extend_from_slice(&(attrs_bytes.len() as u16).to_be_bytes());
+        // magic cookie
+        response.extend_from_slice(&STUN_MAGIC_COOKIE.to_be_bytes());
+        // transaction ID (12 bytes)
+        response.extend_from_slice(&[0u8; 12]);
+        // attributes
+        response.extend_from_slice(&attrs_bytes);
+        response
+    }
+
+    #[test]
+    fn test_xor_mapped_address_ipv4() {
+        // IP 203.0.113.1 = 0xCB007101, XOR'd with magic 0x2112A442 = 0xEA12D543
+        let attr_data: [u8; 8] = [
+            0x00, 0x01, // reserved + family (IPv4)
+            0x11, 0x2B, // port XOR'd with 0x2112 (port 0x3039 = 12345)
+            0xEA, 0x12, 0xD5, 0x43, // IP XOR'd
+        ];
+        let data = build_stun_response(&[(ATTR_XOR_MAPPED_ADDRESS, &attr_data)]);
+        let txn_id = [0u8; 12];
+        let result = parse_stun_response(&data, &txn_id);
+        assert_eq!(result, Some("203.0.113.1".to_string()));
+    }
+
+    #[test]
+    fn test_mapped_address_fallback_ipv4() {
+        // IP 192.168.1.1 = 0xC0A80101 (no XOR)
+        let attr_data: [u8; 8] = [
+            0x00, 0x01, // reserved + family (IPv4)
+            0x00, 0x50, // port 80
+            0xC0, 0xA8, 0x01, 0x01, // IP
+        ];
+        let data = build_stun_response(&[(ATTR_MAPPED_ADDRESS, &attr_data)]);
+        let txn_id = [0u8; 12];
+        let result = parse_stun_response(&data, &txn_id);
+        assert_eq!(result, Some("192.168.1.1".to_string()));
+    }
+
+    #[test]
+    fn test_response_too_short() {
+        let data = vec![0u8; 19]; // < 20 bytes
+        let txn_id = [0u8; 12];
+        assert_eq!(parse_stun_response(&data, &txn_id), None);
+    }
+
+    #[test]
+    fn test_wrong_msg_type() {
+        // Build with correct helper then overwrite msg_type to 0x0001 (Binding Request)
+        let mut data = build_stun_response(&[]);
+        data[0] = 0x00;
+        data[1] = 0x01;
+        let txn_id = [0u8; 12];
+        assert_eq!(parse_stun_response(&data, &txn_id), None);
+    }
+
+    #[test]
+    fn test_no_mapped_address_attributes() {
+        // Valid response with no attributes
+        let data = build_stun_response(&[]);
+        let txn_id = [0u8; 12];
+        assert_eq!(parse_stun_response(&data, &txn_id), None);
+    }
+
+    #[test]
+    fn test_xor_preferred_over_mapped() {
+        // XOR gives 203.0.113.1, MAPPED gives 192.168.1.1
+        let xor_data: [u8; 8] = [
+            0x00, 0x01,
+            0x11, 0x2B,
+            0xEA, 0x12, 0xD5, 0x43,
+        ];
+        let mapped_data: [u8; 8] = [
+            0x00, 0x01,
+            0x00, 0x50,
+            0xC0, 0xA8, 0x01, 0x01,
+        ];
+        // XOR listed first — should be preferred
+        let data = build_stun_response(&[
+            (ATTR_XOR_MAPPED_ADDRESS, &xor_data),
+            (ATTR_MAPPED_ADDRESS, &mapped_data),
+        ]);
+        let txn_id = [0u8; 12];
+        let result = parse_stun_response(&data, &txn_id);
+        assert_eq!(result, Some("203.0.113.1".to_string()));
+    }
+
+    #[test]
+    fn test_truncated_attribute_data() {
+        // Attribute claims 8 bytes but only 4 are present
+        let mut data = build_stun_response(&[]);
+        // Manually append a truncated XOR_MAPPED_ADDRESS attribute
+        let attr_type = ATTR_XOR_MAPPED_ADDRESS.to_be_bytes();
+        let attr_len = 8u16.to_be_bytes(); // claims 8 bytes
+        let truncated = [0x00, 0x01, 0x11, 0x2B]; // only 4 bytes
+        // Update message length
+        let new_msg_len = (attr_type.len() + attr_len.len() + truncated.len()) as u16;
+        data[2..4].copy_from_slice(&new_msg_len.to_be_bytes());
+        data.extend_from_slice(&attr_type);
+        data.extend_from_slice(&attr_len);
+        data.extend_from_slice(&truncated);
+
+        let txn_id = [0u8; 12];
+        // Should return None, not panic
+        assert_eq!(parse_stun_response(&data, &txn_id), None);
+    }
+}
+
 /// Generate 12 random bytes for transaction ID.
 fn rand_bytes() -> [u8; 12] {
     let mut bytes = [0u8; 12];

rust/crates/remoteingress-protocol/src/lib.rs

@@ -6,6 +6,7 @@ pub const FRAME_DATA: u8 = 0x02;
 pub const FRAME_CLOSE: u8 = 0x03;
 pub const FRAME_DATA_BACK: u8 = 0x04;
 pub const FRAME_CLOSE_BACK: u8 = 0x05;
+pub const FRAME_CONFIG: u8 = 0x06;     // Hub -> Edge: configuration update
 
 // Frame header size: 4 (stream_id) + 1 (type) + 4 (length) = 9 bytes
 pub const FRAME_HEADER_SIZE: usize = 9;
@@ -169,4 +170,127 @@ mod tests {
         // EOF
         assert!(reader.next_frame().await.unwrap().is_none());
     }
+
+    #[test]
+    fn test_encode_frame_config_type() {
+        let payload = b"{\"listenPorts\":[443]}";
+        let encoded = encode_frame(0, FRAME_CONFIG, payload);
+        assert_eq!(encoded[4], FRAME_CONFIG);
+        assert_eq!(&encoded[0..4], &0u32.to_be_bytes());
+        assert_eq!(&encoded[9..], payload.as_slice());
+    }
+
+    #[test]
+    fn test_encode_frame_data_back_type() {
+        let payload = b"response data";
+        let encoded = encode_frame(7, FRAME_DATA_BACK, payload);
+        assert_eq!(encoded[4], FRAME_DATA_BACK);
+        assert_eq!(&encoded[0..4], &7u32.to_be_bytes());
+        assert_eq!(&encoded[5..9], &(payload.len() as u32).to_be_bytes());
+        assert_eq!(&encoded[9..], payload.as_slice());
+    }
+
+    #[test]
+    fn test_encode_frame_close_back_type() {
+        let encoded = encode_frame(99, FRAME_CLOSE_BACK, &[]);
+        assert_eq!(encoded[4], FRAME_CLOSE_BACK);
+        assert_eq!(&encoded[0..4], &99u32.to_be_bytes());
+        assert_eq!(&encoded[5..9], &0u32.to_be_bytes());
+        assert_eq!(encoded.len(), FRAME_HEADER_SIZE);
+    }
+
+    #[test]
+    fn test_encode_frame_large_stream_id() {
+        let encoded = encode_frame(u32::MAX, FRAME_DATA, b"x");
+        assert_eq!(&encoded[0..4], &u32::MAX.to_be_bytes());
+        assert_eq!(encoded[4], FRAME_DATA);
+        assert_eq!(&encoded[5..9], &1u32.to_be_bytes());
+        assert_eq!(encoded[9], b'x');
+    }
+
+    #[tokio::test]
+    async fn test_frame_reader_max_payload_rejection() {
+        let mut data = Vec::new();
+        data.extend_from_slice(&1u32.to_be_bytes());
+        data.push(FRAME_DATA);
+        data.extend_from_slice(&(MAX_PAYLOAD_SIZE + 1).to_be_bytes());
+
+        let cursor = std::io::Cursor::new(data);
+        let mut reader = FrameReader::new(cursor);
+
+        let result = reader.next_frame().await;
+        assert!(result.is_err());
+        let err = result.unwrap_err();
+        assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
+    }
+
+    #[tokio::test]
+    async fn test_frame_reader_eof_mid_header() {
+        // Only 5 bytes — not enough for a 9-byte header
+        let data = vec![0u8; 5];
+        let cursor = std::io::Cursor::new(data);
+        let mut reader = FrameReader::new(cursor);
+
+        // Should return Ok(None) on partial header EOF
+        let result = reader.next_frame().await;
+        assert!(result.unwrap().is_none());
+    }
+
+    #[tokio::test]
+    async fn test_frame_reader_eof_mid_payload() {
+        // Full header claiming 100 bytes of payload, but only 10 bytes present
+        let mut data = Vec::new();
+        data.extend_from_slice(&1u32.to_be_bytes());
+        data.push(FRAME_DATA);
+        data.extend_from_slice(&100u32.to_be_bytes());
+        data.extend_from_slice(&[0xAB; 10]);
+
+        let cursor = std::io::Cursor::new(data);
+        let mut reader = FrameReader::new(cursor);
+
+        let result = reader.next_frame().await;
+        assert!(result.is_err());
+    }
+
+    #[tokio::test]
+    async fn test_frame_reader_all_frame_types() {
+        let types = [
+            FRAME_OPEN,
+            FRAME_DATA,
+            FRAME_CLOSE,
+            FRAME_DATA_BACK,
+            FRAME_CLOSE_BACK,
+            FRAME_CONFIG,
+        ];
+
+        let mut data = Vec::new();
+        for (i, &ft) in types.iter().enumerate() {
+            let payload = format!("payload_{}", i);
+            data.extend_from_slice(&encode_frame(i as u32, ft, payload.as_bytes()));
+        }
+
+        let cursor = std::io::Cursor::new(data);
+        let mut reader = FrameReader::new(cursor);
+
+        for (i, &ft) in types.iter().enumerate() {
+            let frame = reader.next_frame().await.unwrap().unwrap();
+            assert_eq!(frame.stream_id, i as u32);
+            assert_eq!(frame.frame_type, ft);
+            assert_eq!(frame.payload, format!("payload_{}", i).as_bytes());
+        }
+
+        assert!(reader.next_frame().await.unwrap().is_none());
+    }
+
+    #[tokio::test]
+    async fn test_frame_reader_zero_length_payload() {
+        let data = encode_frame(42, FRAME_CLOSE, &[]);
+        let cursor = std::io::Cursor::new(data);
+        let mut reader = FrameReader::new(cursor);
+
+        let frame = reader.next_frame().await.unwrap().unwrap();
+        assert_eq!(frame.stream_id, 42);
+        assert_eq!(frame.frame_type, FRAME_CLOSE);
+        assert!(frame.payload.is_empty());
+    }
 }

test/test.classes.node.ts

@@ -0,0 +1,35 @@
+import { expect, tap } from '@push.rocks/tapbundle';
+import { EventEmitter } from 'events';
+import { RemoteIngressHub, RemoteIngressEdge } from '../ts/index.js';
+
+tap.test('RemoteIngressHub constructor does not throw', async () => {
+  const hub = new RemoteIngressHub();
+  expect(hub).toBeTruthy();
+});
+
+tap.test('RemoteIngressHub is instanceof EventEmitter', async () => {
+  const hub = new RemoteIngressHub();
+  expect(hub).toBeInstanceOf(EventEmitter);
+});
+
+tap.test('RemoteIngressHub.running is false before start', async () => {
+  const hub = new RemoteIngressHub();
+  expect(hub.running).toBeFalse();
+});
+
+tap.test('RemoteIngressEdge constructor does not throw', async () => {
+  const edge = new RemoteIngressEdge();
+  expect(edge).toBeTruthy();
+});
+
+tap.test('RemoteIngressEdge is instanceof EventEmitter', async () => {
+  const edge = new RemoteIngressEdge();
+  expect(edge).toBeInstanceOf(EventEmitter);
+});
+
+tap.test('RemoteIngressEdge.running is false before start', async () => {
+  const edge = new RemoteIngressEdge();
+  expect(edge.running).toBeFalse();
+});
+
+export default tap.start();

test/test.token.node.ts

@@ -0,0 +1,152 @@
+import { expect, tap } from '@push.rocks/tapbundle';
+import { encodeConnectionToken, decodeConnectionToken, type IConnectionTokenData } from '../ts/classes.token.js';
+
+tap.test('token roundtrip with unicode chars in secret', async () => {
+  const data: IConnectionTokenData = {
+    hubHost: 'hub.example.com',
+    hubPort: 8443,
+    edgeId: 'edge-1',
+    secret: 'sécret-with-ünïcödé-日本語',
+  };
+  const token = encodeConnectionToken(data);
+  const decoded = decodeConnectionToken(token);
+  expect(decoded.secret).toEqual(data.secret);
+});
+
+tap.test('token roundtrip with empty edgeId', async () => {
+  const data: IConnectionTokenData = {
+    hubHost: 'hub.test',
+    hubPort: 443,
+    edgeId: '',
+    secret: 'key',
+  };
+  const token = encodeConnectionToken(data);
+  const decoded = decodeConnectionToken(token);
+  expect(decoded.edgeId).toEqual('');
+});
+
+tap.test('token roundtrip with port 0', async () => {
+  const data: IConnectionTokenData = {
+    hubHost: 'h',
+    hubPort: 0,
+    edgeId: 'e',
+    secret: 's',
+  };
+  const token = encodeConnectionToken(data);
+  const decoded = decodeConnectionToken(token);
+  expect(decoded.hubPort).toEqual(0);
+});
+
+tap.test('token roundtrip with port 65535', async () => {
+  const data: IConnectionTokenData = {
+    hubHost: 'h',
+    hubPort: 65535,
+    edgeId: 'e',
+    secret: 's',
+  };
+  const token = encodeConnectionToken(data);
+  const decoded = decodeConnectionToken(token);
+  expect(decoded.hubPort).toEqual(65535);
+});
+
+tap.test('token roundtrip with very long secret (10k chars)', async () => {
+  const longSecret = 'x'.repeat(10000);
+  const data: IConnectionTokenData = {
+    hubHost: 'host',
+    hubPort: 1234,
+    edgeId: 'edge',
+    secret: longSecret,
+  };
+  const token = encodeConnectionToken(data);
+  const decoded = decodeConnectionToken(token);
+  expect(decoded.secret).toEqual(longSecret);
+  expect(decoded.secret.length).toEqual(10000);
+});
+
+tap.test('token string is URL-safe', async () => {
+  const data: IConnectionTokenData = {
+    hubHost: 'hub.example.com',
+    hubPort: 8443,
+    edgeId: 'edge-001',
+    secret: 'super+secret/key==with+special/chars',
+  };
+  const token = encodeConnectionToken(data);
+  expect(token).toMatch(/^[A-Za-z0-9_-]+$/);
+});
+
+tap.test('decode empty string throws', async () => {
+  let error: Error | undefined;
+  try {
+    decodeConnectionToken('');
+  } catch (e) {
+    error = e as Error;
+  }
+  expect(error).toBeInstanceOf(Error);
+});
+
+tap.test('decode valid base64 but wrong JSON shape throws missing required fields', async () => {
+  // Encode { "a": 1, "b": 2 } — valid JSON but wrong shape
+  const token = Buffer.from(JSON.stringify({ a: 1, b: 2 }), 'utf-8')
+    .toString('base64')
+    .replace(/\+/g, '-')
+    .replace(/\//g, '_')
+    .replace(/=+$/, '');
+
+  let error: Error | undefined;
+  try {
+    decodeConnectionToken(token);
+  } catch (e) {
+    error = e as Error;
+  }
+  expect(error).toBeInstanceOf(Error);
+  expect(error!.message).toInclude('missing required fields');
+});
+
+tap.test('decode valid JSON but wrong field types throws missing required fields', async () => {
+  // h is number instead of string, p is string instead of number
+  const token = Buffer.from(JSON.stringify({ h: 123, p: 'notnum', e: 'e', s: 's' }), 'utf-8')
+    .toString('base64')
+    .replace(/\+/g, '-')
+    .replace(/\//g, '_')
+    .replace(/=+$/, '');
+
+  let error: Error | undefined;
+  try {
+    decodeConnectionToken(token);
+  } catch (e) {
+    error = e as Error;
+  }
+  expect(error).toBeInstanceOf(Error);
+  expect(error!.message).toInclude('missing required fields');
+});
+
+tap.test('decode with extra fields succeeds', async () => {
+  const token = Buffer.from(
+    JSON.stringify({ h: 'host', p: 443, e: 'edge', s: 'secret', extra: 'ignored' }),
+    'utf-8',
+  )
+    .toString('base64')
+    .replace(/\+/g, '-')
+    .replace(/\//g, '_')
+    .replace(/=+$/, '');
+
+  const decoded = decodeConnectionToken(token);
+  expect(decoded.hubHost).toEqual('host');
+  expect(decoded.hubPort).toEqual(443);
+  expect(decoded.edgeId).toEqual('edge');
+  expect(decoded.secret).toEqual('secret');
+});
+
+tap.test('encode is deterministic', async () => {
+  const data: IConnectionTokenData = {
+    hubHost: 'hub.test',
+    hubPort: 8443,
+    edgeId: 'edge-1',
+    secret: 'deterministic-key',
+  };
+  const token1 = encodeConnectionToken(data);
+  const token2 = encodeConnectionToken(data);
+  expect(token1).toEqual(token2);
+});
+
+export default tap.start();

ts/00_commitinfo_data.ts

@@ -3,6 +3,6 @@
  */
 export const commitinfo = {
   name: '@serve.zone/remoteingress',
-  version: '3.1.1',
+  version: '3.2.1',
   description: 'Edge ingress tunnel for DcRouter - accepts incoming TCP connections at network edge and tunnels them to DcRouter SmartProxy preserving client IP via PROXY protocol v1.'
 }

ts/classes.remoteingressedge.ts

@@ -43,6 +43,7 @@ export interface IEdgeConfig {
 export class RemoteIngressEdge extends EventEmitter {
   private bridge: InstanceType<typeof plugins.smartrust.RustBridge<TEdgeCommands>>;
   private started = false;
+  private statusInterval: ReturnType<typeof setInterval> | undefined;
 
   constructor() {
     super();
@@ -79,6 +80,14 @@ export class RemoteIngressEdge extends EventEmitter {
     this.bridge.on('management:publicIpDiscovered', (data: { ip: string }) => {
       this.emit('publicIpDiscovered', data);
     });
+    this.bridge.on('management:portsAssigned', (data: { listenPorts: number[] }) => {
+      console.log(`[RemoteIngressEdge] Ports assigned by hub: ${data.listenPorts.join(', ')}`);
+      this.emit('portsAssigned', data);
+    });
+    this.bridge.on('management:portsUpdated', (data: { listenPorts: number[] }) => {
+      console.log(`[RemoteIngressEdge] Ports updated by hub: ${data.listenPorts.join(', ')}`);
+      this.emit('portsUpdated', data);
+    });
   }
 
   /**
@@ -113,12 +122,30 @@ export class RemoteIngressEdge extends EventEmitter {
     });
 
     this.started = true;
+
+    // Start periodic status logging
+    this.statusInterval = setInterval(async () => {
+      try {
+        const status = await this.getStatus();
+        console.log(
+          `[RemoteIngressEdge] Status: connected=${status.connected}, ` +
+          `streams=${status.activeStreams}, ports=[${status.listenPorts.join(',')}], ` +
+          `publicIp=${status.publicIp ?? 'unknown'}`
+        );
+      } catch {
+        // Bridge may be shutting down
+      }
+    }, 60_000);
   }
 
   /**
    * Stop the edge and kill the Rust process.
    */
   public async stop(): Promise<void> {
+    if (this.statusInterval) {
+      clearInterval(this.statusInterval);
+      this.statusInterval = undefined;
+    }
     if (this.started) {
       try {
         await this.bridge.sendCommand('stopEdge', {} as Record<string, never>);

ts/classes.remoteingresshub.ts

@@ -20,7 +20,7 @@ type THubCommands = {
   };
   updateAllowedEdges: {
     params: {
-      edges: Array<{ id: string; secret: string }>;
+      edges: Array<{ id: string; secret: string; listenPorts?: number[]; stunIntervalSecs?: number }>;
     };
     result: { updated: boolean };
   };
@@ -122,7 +122,7 @@ export class RemoteIngressHub extends EventEmitter {
   /**
    * Update the list of allowed edges that can connect to this hub.
    */
-  public async updateAllowedEdges(edges: Array<{ id: string; secret: string }>): Promise<void> {
+  public async updateAllowedEdges(edges: Array<{ id: string; secret: string; listenPorts?: number[]; stunIntervalSecs?: number }>): Promise<void> {
     await this.bridge.sendCommand('updateAllowedEdges', { edges });
   }