Compare commits

..

4 Commits

11 changed files with 964 additions and 17 deletions

View File

@@ -1,5 +1,24 @@
# Changelog
## 2026-02-18 - 3.3.0 - feat(readme)
document dynamic port assignment and runtime port updates; clarify TLS multiplexing, frame format, and handshake sequence
- Adds documentation for dynamic port configuration: hub-assigned listen ports, hot-reloadable via FRAME_CONFIG frames
- Introduces new FRAME type CONFIG (0x06) and describes payload as JSON; notes immediate push of port changes to connected edges
- Clarifies that the tunnel is a single encrypted TLS multiplexed connection to the hub (preserves PROXY v1 behavior)
- Specifies frame integer fields are big-endian and that stream IDs are 32-bit unsigned integers
- Adds new events: portsAssigned and portsUpdated, and updates examples showing updateAllowedEdges usage and live port changes
## 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

View File

@@ -1,6 +1,6 @@
{
"name": "@serve.zone/remoteingress",
"version": "3.2.0",
"version": "3.3.0",
"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",
@@ -9,7 +9,7 @@
"author": "Task Venture Capital GmbH",
"license": "MIT",
"scripts": {
"test": "(tstest test/ --web)",
"test": "(tstest test/ --verbose --logfile --timeout 60)",
"build": "(tsbuild tsfolders --allowimplicitany && tsrust)",
"buildDocs": "(tsdoc)"
},

View File

@@ -1,6 +1,6 @@
# @serve.zone/remoteingress
Edge ingress tunnel for DcRouter — accepts incoming TCP connections at the network edge and tunnels them to a DcRouter SmartProxy instance, preserving the original client IP via PROXY protocol v1.
Edge ingress tunnel for DcRouter — accepts incoming TCP connections at the network edge and tunnels them over a single encrypted TLS connection to a DcRouter SmartProxy instance, preserving the original client IP via PROXY protocol v1.
## Issue Reporting and Security
@@ -22,8 +22,8 @@ pnpm install @serve.zone/remoteingress
│ │ (multiplexed frames + │ │
│ RemoteIngressEdge │ shared-secret auth) │ RemoteIngressHub │
│ Accepts client TCP │ │ Forwards to │
│ connections │ │ SmartProxy on │
│ │ local ports │
│ connections on │ │ SmartProxy on │
hub-assigned ports │ │ local ports │
└─────────────────────┘ └─────────────────────┘
▲ │
│ TCP from end users ▼
@@ -32,8 +32,8 @@ pnpm install @serve.zone/remoteingress
| Component | Role |
|-----------|------|
| **RemoteIngressEdge** | Deployed at the network edge (e.g. a VPS or cloud instance). Accepts raw TCP connections and multiplexes them over a single TLS tunnel to the hub. |
| **RemoteIngressHub** | Deployed alongside DcRouter/SmartProxy in a private cluster. Accepts edge connections, demuxes streams, and forwards each to SmartProxy with a PROXY protocol v1 header so the real client IP is preserved. |
| **RemoteIngressEdge** | Deployed at the network edge (e.g. a VPS or cloud instance). Listens on ports assigned by the hub, accepts raw TCP connections, and multiplexes them over a single TLS tunnel to the hub. Ports are hot-reloadable — the hub can change them at runtime. |
| **RemoteIngressHub** | Deployed alongside DcRouter/SmartProxy in a private cluster. Accepts edge connections, demuxes streams, and forwards each to SmartProxy with a PROXY protocol v1 header so the real client IP is preserved. Controls which ports each edge listens on. |
| **Rust Binary** (`remoteingress-bin`) | The performance-critical networking core. Managed via `@push.rocks/smartrust` RustBridge IPC — you never interact with it directly. Cross-compiled for `linux/amd64` and `linux/arm64`. |
### ✨ Key Features
@@ -45,6 +45,7 @@ pnpm install @serve.zone/remoteingress
- 🎫 **Connection tokens** — encode all connection details into a single opaque string
- 📡 **STUN-based public IP discovery** — the edge automatically discovers its public IP via Cloudflare STUN
- 🔄 **Auto-reconnect** with exponential backoff if the tunnel drops
- 🎛️ **Dynamic port configuration** — the hub assigns listen ports per edge and can hot-reload them at runtime via `FRAME_CONFIG` frames
- 📣 **Event-driven** — both Hub and Edge extend `EventEmitter` for real-time monitoring
-**Rust core** — all frame encoding, TLS, and TCP proxying happen in native code for maximum throughput
@@ -79,10 +80,28 @@ await hub.start({
targetHost: '127.0.0.1', // SmartProxy host to forward streams to (default: 127.0.0.1)
});
// Register which edges are allowed to connect
// Register which edges are allowed to connect, including their listen ports
await hub.updateAllowedEdges([
{ id: 'edge-nyc-01', secret: 'supersecrettoken1' },
{ id: 'edge-fra-02', secret: 'supersecrettoken2' },
{
id: 'edge-nyc-01',
secret: 'supersecrettoken1',
listenPorts: [80, 443], // ports the edge should listen on
stunIntervalSecs: 300, // STUN discovery interval (default: 300)
},
{
id: 'edge-fra-02',
secret: 'supersecrettoken2',
listenPorts: [443, 8080],
},
]);
// Dynamically update ports for a connected edge — changes are pushed instantly
await hub.updateAllowedEdges([
{
id: 'edge-nyc-01',
secret: 'supersecrettoken1',
listenPorts: [80, 443, 8443], // added port 8443 — edge picks it up in real time
},
]);
// Check status at any time
@@ -116,6 +135,8 @@ const edge = new RemoteIngressEdge();
edge.on('tunnelConnected', () => console.log('Tunnel established'));
edge.on('tunnelDisconnected', () => console.log('Tunnel lost — will auto-reconnect'));
edge.on('publicIpDiscovered', ({ ip }) => console.log(`Public IP: ${ip}`));
edge.on('portsAssigned', ({ listenPorts }) => console.log(`Listening on ports: ${listenPorts}`));
edge.on('portsUpdated', ({ listenPorts }) => console.log(`Ports updated: ${listenPorts}`));
// Single token contains hubHost, hubPort, edgeId, and secret
await edge.start({
@@ -133,6 +154,8 @@ const edge = new RemoteIngressEdge();
edge.on('tunnelConnected', () => console.log('Tunnel established'));
edge.on('tunnelDisconnected', () => console.log('Tunnel lost — will auto-reconnect'));
edge.on('publicIpDiscovered', ({ ip }) => console.log(`Public IP: ${ip}`));
edge.on('portsAssigned', ({ listenPorts }) => console.log(`Listening on ports: ${listenPorts}`));
edge.on('portsUpdated', ({ listenPorts }) => console.log(`Ports updated: ${listenPorts}`));
await edge.start({
hubHost: 'hub.example.com', // hostname or IP of the hub
@@ -194,7 +217,7 @@ Tokens are base64url-encoded (URL-safe, no padding) — safe to pass as environm
|-------------------|-------------|
| `start(config?)` | Spawns the Rust binary and starts the tunnel listener. Config: `{ tunnelPort?: number, targetHost?: string }` |
| `stop()` | Gracefully shuts down the hub and kills the Rust process. |
| `updateAllowedEdges(edges)` | Dynamically update which edges are authorized. Each edge: `{ id: string, secret: string }` |
| `updateAllowedEdges(edges)` | Dynamically update which edges are authorized and what ports they listen on. Each edge: `{ id: string, secret: string, listenPorts?: number[], stunIntervalSecs?: number }`. If ports change for a connected edge, the update is pushed immediately via a `FRAME_CONFIG` frame. |
| `getStatus()` | Returns current hub status including connected edges and active stream counts. |
| `running` | `boolean` — whether the Rust binary is alive. |
@@ -204,12 +227,12 @@ Tokens are base64url-encoded (URL-safe, no padding) — safe to pass as environm
| Method / Property | Description |
|-------------------|-------------|
| `start(config)` | Spawns the Rust binary and connects to the hub. Accepts `{ token: string }` or `IEdgeConfig`. |
| `start(config)` | Spawns the Rust binary and connects to the hub. Accepts `{ token: string }` or `IEdgeConfig`. Listen ports are received from the hub during handshake. |
| `stop()` | Gracefully shuts down the edge and kills the Rust process. |
| `getStatus()` | Returns current edge status including connection state, public IP, and active streams. |
| `getStatus()` | Returns current edge status including connection state, public IP, listen ports, and active streams. |
| `running` | `boolean` — whether the Rust binary is alive. |
**Events:** `tunnelConnected`, `tunnelDisconnected`, `publicIpDiscovered`
**Events:** `tunnelConnected`, `tunnelDisconnected`, `publicIpDiscovered`, `portsAssigned`, `portsUpdated`
### Token Utilities
@@ -246,7 +269,7 @@ interface IConnectionTokenData {
The tunnel uses a custom binary frame protocol over TLS:
```
[stream_id: 4 bytes][type: 1 byte][length: 4 bytes][payload: N bytes]
[stream_id: 4 bytes BE][type: 1 byte][length: 4 bytes BE][payload: N bytes]
```
| Frame Type | Value | Direction | Purpose |
@@ -256,8 +279,18 @@ The tunnel uses a custom binary frame protocol over TLS:
| `CLOSE` | `0x03` | Edge → Hub | Client closed the connection |
| `DATA_BACK` | `0x04` | Hub → Edge | Response data flowing downstream |
| `CLOSE_BACK` | `0x05` | Hub → Edge | Upstream (SmartProxy) closed the connection |
| `CONFIG` | `0x06` | Hub → Edge | Runtime configuration update (e.g. port changes); payload is JSON |
Max payload size per frame: **16 MB**.
Max payload size per frame: **16 MB**. Stream IDs are 32-bit unsigned integers.
### Handshake Sequence
1. Edge opens a TLS connection to the hub
2. Edge sends: `EDGE <edgeId> <secret>\n`
3. Hub verifies credentials (constant-time comparison) and responds with JSON: `{"listenPorts":[...],"stunIntervalSecs":300}\n`
4. Edge starts TCP listeners on the assigned ports
5. Frame protocol begins — `OPEN`/`DATA`/`CLOSE` frames flow in both directions
6. Hub can push `CONFIG` frames at any time to update the edge's listen ports
## 💡 Example Scenarios
@@ -292,6 +325,22 @@ const edge = new RemoteIngressEdge();
await edge.start({ token });
```
### 5. Dynamic Port Management
The hub controls which ports each edge listens on. Ports can be changed at runtime without restarting the edge — the hub pushes a `CONFIG` frame and the edge hot-reloads its TCP listeners.
```typescript
// Initially assign ports 80 and 443
await hub.updateAllowedEdges([
{ id: 'edge-nyc-01', secret: 'secret', listenPorts: [80, 443] },
]);
// Later, add port 8080 — the connected edge picks it up instantly
await hub.updateAllowedEdges([
{ id: 'edge-nyc-01', secret: 'secret', listenPorts: [80, 443, 8080] },
]);
```
## License and Legal Information
This repository contains open-source code licensed under the MIT License. A copy of the license can be found in the [LICENSE](./LICENSE) file.

View File

@@ -369,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");
}
}

View File

@@ -544,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;

View File

@@ -549,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);
}
}

View File

@@ -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];

View File

@@ -170,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());
}
}

35
test/test.classes.node.ts Normal file
View File

@@ -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();

152
test/test.token.node.ts Normal file
View File

@@ -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();

View File

@@ -3,6 +3,6 @@
*/
export const commitinfo = {
name: '@serve.zone/remoteingress',
version: '3.2.0',
version: '3.3.0',
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.'
}