diff --git a/docs/design/fips-configuration.md b/docs/design/fips-configuration.md index c7843f9..1ae1c7d 100644 --- a/docs/design/fips-configuration.md +++ b/docs/design/fips-configuration.md @@ -285,6 +285,37 @@ Each named instance operates independently with its own socket and discovery state. The instance name is used in log messages and the `name()` method on the Transport trait. +### TCP (`transports.tcp.*`) + +TCP transport enables firewall traversal on networks that block UDP but +allow TCP (e.g., port 443). Uses FMP header-based framing with zero +overhead. + +| Parameter | Type | Default | Description | +|-----------|------|---------|-------------| +| `transports.tcp.bind_addr` | string | *(none)* | Listen address (e.g., `"0.0.0.0:443"`). If omitted, outbound-only mode. | +| `transports.tcp.mtu` | u16 | `1400` | Default MTU. Per-connection MTU derived from `TCP_MAXSEG` when available. | +| `transports.tcp.connect_timeout_ms` | u64 | `5000` | Outbound connect timeout in milliseconds | +| `transports.tcp.nodelay` | bool | `true` | `TCP_NODELAY` (disable Nagle for low latency) | +| `transports.tcp.keepalive_secs` | u64 | `30` | TCP keepalive interval in seconds (0 = disabled) | +| `transports.tcp.recv_buf_size` | usize | `2097152` | Socket receive buffer size in bytes (2 MB) | +| `transports.tcp.send_buf_size` | usize | `2097152` | Socket send buffer size in bytes (2 MB) | +| `transports.tcp.max_inbound_connections` | usize | `256` | Maximum simultaneous inbound connections | +| `transports.tcp.socks5_proxy` | string | *(none)* | SOCKS5 proxy for outbound connections (implementation deferred) | + +**Named instances.** Like other transports, multiple TCP instances can +be configured with named sub-keys: + +```yaml +transports: + tcp: + public: + bind_addr: "0.0.0.0:443" + tor: + socks5_proxy: "127.0.0.1:9050" + connect_timeout_ms: 30000 +``` + ## Peers (`peers[]`) Static peer list. Each entry defines a peer to connect to. @@ -293,8 +324,8 @@ Static peer list. Each entry defines a peer to connect to. |-----------|------|---------|-------------| | `peers[].npub` | string | *(required)* | Peer's Nostr public key (npub-encoded) | | `peers[].alias` | string | *(none)* | Human-readable name for logging | -| `peers[].addresses[].transport` | string | *(required)* | Transport type: `udp` or `ethernet` | -| `peers[].addresses[].addr` | string | *(required)* | Transport address. UDP: `"ip:port"`. Ethernet: `"interface/mac"` (e.g., `"eth0/aa:bb:cc:dd:ee:ff"`) | +| `peers[].addresses[].transport` | string | *(required)* | Transport type: `udp`, `tcp`, or `ethernet` | +| `peers[].addresses[].addr` | string | *(required)* | Transport address. UDP/TCP: `"ip:port"`. Ethernet: `"interface/mac"` (e.g., `"eth0/aa:bb:cc:dd:ee:ff"`) | | `peers[].addresses[].priority` | u8 | `100` | Address priority (lower = preferred) | | `peers[].connect_policy` | string | `"auto_connect"` | Connection policy: `auto_connect`, `on_demand`, or `manual` | | `peers[].auto_reconnect` | bool | `true` | Automatically reconnect after MMP link-dead removal (exponential backoff, unlimited retries) | @@ -483,6 +514,16 @@ transports: # auto_connect: false # connect to discovered peers # accept_connections: false # accept inbound handshakes # beacon_interval_secs: 30 # beacon interval (min 10) + # tcp: # uncomment to enable TCP transport + # bind_addr: "0.0.0.0:443" # listen address (omit for outbound-only) + # mtu: 1400 # default MTU + # connect_timeout_ms: 5000 # outbound connect timeout + # nodelay: true # TCP_NODELAY + # keepalive_secs: 30 # keepalive interval (0 = disabled) + # recv_buf_size: 2097152 # 2 MB + # send_buf_size: 2097152 # 2 MB + # max_inbound_connections: 256 # resource protection limit + # socks5_proxy: null # SOCKS5 for outbound (deferred) peers: # static peer list # - npub: "npub1..." diff --git a/docs/design/fips-transport-layer.md b/docs/design/fips-transport-layer.md index cf3fe0f..8faafdd 100644 --- a/docs/design/fips-transport-layer.md +++ b/docs/design/fips-transport-layer.md @@ -312,6 +312,98 @@ Startup logging: Ethernet transport started name=eth0 interface=eth0 mac=aa:bb:cc:dd:ee:ff mtu=1499 if_mtu=1500 ``` +## TCP/IP: Firewall Traversal Transport + +For networks where UDP is blocked but TCP port 443 is open, the TCP +transport provides an alternative path. It also serves as the foundation +for the future Tor transport. + +FIPS protocols (FMP, FSP, MMP) are all unreliable datagrams. Running them +over TCP introduces head-of-line blocking, which adds latency jitter. MMP +correctly measures this jitter, and cost-based parent selection naturally +penalizes TCP links (higher SRTT leads to higher link cost). ETX will be +1.0 over TCP since TCP handles retransmission. + +### Architecture + +Unlike UDP (one socket serves all peers), TCP requires one `TcpStream` per +peer. The transport maintains a connection pool (`HashMap`) plus an optional `TcpListener` for inbound connections. + +| Property | Value | +| -------- | ----- | +| Addressing | IP:port (same as UDP) | +| Default MTU | 1400 bytes | +| Per-link MTU | Derived from `TCP_MAXSEG` socket option | +| Framing | FMP header-based (zero overhead) | +| Connection model | Connect-on-send, optional listener | +| Platform | Cross-platform (no `#[cfg]` gates) | + +### FMP Header-Based Framing + +TCP is a byte stream; FIPS packets need delineation. Rather than adding a +separate length-prefix layer, the TCP transport uses the existing 4-byte +FMP common prefix `[ver+phase:1][flags:1][payload_len:2 LE]` to determine +packet boundaries: + +- **Phase 0x0 (established)**: remaining = 12 + payload_len + 16 (header + AEAD tag) +- **Phase 0x1 (msg1)**: remaining = payload_len (fixed at 110, total 114 bytes) +- **Phase 0x2 (msg2)**: remaining = payload_len (fixed at 65, total 69 bytes) +- **Unknown phase**: close connection (protocol error) + +This provides zero framing overhead and built-in phase validation. The +stream reader is implemented in a separate module (`stream.rs`) for reuse +by the future Tor transport. + +### Connect-on-Send + +When `send(addr, data)` is called with no existing connection: + +1. Connect with configurable timeout (default 5s) +2. Configure socket: `TCP_NODELAY`, keepalive, buffer sizes +3. Read `TCP_MAXSEG` for per-connection MTU +4. Split stream into read/write halves +5. Spawn per-connection receive task +6. Store connection in pool +7. Write packet directly to stream + +If connect fails, return error. The node's handshake retry mechanism +handles re-attempts. + +### Session Independence + +TCP connection loss does **not** tear down the FIPS peer. Noise keys, MMP +state, and FSP sessions are bound to the peer's npub, not the TCP +connection. The transport reconnects transparently on the next send via +connect-on-send. MMP liveness timeout is the sole authority for peer death. + +### Connection Deduplication + +Simultaneous outbound connections from both sides are resolved by the +existing cross-connection tie-breaker in `promote_connection`. The losing +TCP connection is closed via `Transport::close_connection(addr)`, which +removes it from the pool and aborts its receive task. + +### Configuration + +```yaml +transports: + tcp: + bind_addr: "0.0.0.0:443" # Listen address (omit for outbound-only) + mtu: 1400 # Default MTU + connect_timeout_ms: 5000 # Outbound connect timeout + nodelay: true # TCP_NODELAY (disable Nagle) + keepalive_secs: 30 # TCP keepalive interval (0 = disabled) + recv_buf_size: 2097152 # SO_RCVBUF (2 MB) + send_buf_size: 2097152 # SO_SNDBUF (2 MB) + max_inbound_connections: 256 # Resource protection limit + socks5_proxy: "127.0.0.1:9050" # SOCKS5 for outbound (deferred) +``` + +If `bind_addr` is configured, the transport accepts inbound connections. +Without it, the transport operates in outbound-only mode (no listener +socket is created). + ## Discovery Discovery determines that a FIPS-capable endpoint is reachable at a given @@ -369,12 +461,13 @@ Key properties: ### Current State -> **Implemented**: UDP peers are configured via YAML. Ethernet peers are -> discovered via beacon broadcast — the `discover()` trait method returns -> newly seen endpoints, and per-transport `auto_connect()` / -> `accept_connections()` policies control whether discovered peers are -> connected automatically or require explicit configuration. Nostr relay -> discovery is not yet implemented. +> **Implemented**: UDP and TCP peers are configured via YAML. Ethernet +> peers are discovered via beacon broadcast — the `discover()` trait +> method returns newly seen endpoints, and per-transport `auto_connect()` +> / `accept_connections()` policies control whether discovered peers are +> connected automatically or require explicit configuration. TCP has no +> discovery mechanism (peers are configured). Nostr relay discovery is +> not yet implemented. ## Transport Interface @@ -392,6 +485,7 @@ link_mtu(addr) → u16 Per-link MTU (defaults to mtu()) start() → lifecycle Bring transport up (bind socket, open device) stop() → lifecycle Bring transport down send(addr, data) → delivery Send datagram to transport address +close_connection(addr)→ () Close a specific connection (no-op for connectionless) discover() → Vec Report discovered FIPS endpoints (optional) auto_connect() → bool Auto-connect discovered peers (default: false) accept_connections() → bool Accept inbound handshakes (default: true) @@ -449,7 +543,7 @@ transitions through `Starting` to `Up` (operational). `stop()` moves to | Transport | Status | Notes | | --------- | ------ | ----- | | UDP/IP | **Implemented** | Primary transport, async send/receive, configurable MTU | -| TCP/IP | Future direction | Requires stream framing, TCP-over-TCP concern | +| TCP/IP | **Implemented** | FMP header-based framing, connect-on-send, per-connection MSS MTU | | Ethernet | **Implemented** | AF_PACKET SOCK_DGRAM, EtherType 0x2121, beacon discovery, Linux only | | WiFi | Future direction | Infrastructure mode = Ethernet driver | | Tor | Future direction | High latency, .onion addressing | diff --git a/src/config/mod.rs b/src/config/mod.rs index c72170f..936c51e 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -33,7 +33,7 @@ pub use node::{ NodeConfig, RateLimitConfig, RetryConfig, SessionConfig, SessionMmpConfig, TreeConfig, }; pub use peer::{ConnectPolicy, PeerAddress, PeerConfig}; -pub use transport::{EthernetConfig, TransportInstances, TransportsConfig, UdpConfig}; +pub use transport::{EthernetConfig, TcpConfig, TransportInstances, TransportsConfig, UdpConfig}; /// Default config filename. const CONFIG_FILENAME: &str = "fips.yaml"; diff --git a/src/config/transport.rs b/src/config/transport.rs index eac13d2..01d786c 100644 --- a/src/config/transport.rs +++ b/src/config/transport.rs @@ -238,6 +238,111 @@ impl EthernetConfig { } } +// ============================================================================ +// TCP Transport Configuration +// ============================================================================ + +/// Default TCP MTU (conservative, matches typical Ethernet MSS minus overhead). +const DEFAULT_TCP_MTU: u16 = 1400; + +/// Default TCP connect timeout in milliseconds. +const DEFAULT_TCP_CONNECT_TIMEOUT_MS: u64 = 5000; + +/// Default TCP keepalive interval in seconds. +const DEFAULT_TCP_KEEPALIVE_SECS: u64 = 30; + +/// Default TCP receive buffer size (2 MB). +const DEFAULT_TCP_RECV_BUF: usize = 2 * 1024 * 1024; + +/// Default TCP send buffer size (2 MB). +const DEFAULT_TCP_SEND_BUF: usize = 2 * 1024 * 1024; + +/// Default maximum inbound TCP connections. +const DEFAULT_TCP_MAX_INBOUND: usize = 256; + +/// TCP transport instance configuration. +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct TcpConfig { + /// Listen address (e.g., "0.0.0.0:443"). If not set, outbound-only. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub bind_addr: Option, + + /// Default MTU for TCP connections. Defaults to 1400. + /// Per-connection MTU is derived from TCP_MAXSEG when available. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub mtu: Option, + + /// Outbound connect timeout in milliseconds. Defaults to 5000. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub connect_timeout_ms: Option, + + /// Enable TCP_NODELAY (disable Nagle). Defaults to true. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub nodelay: Option, + + /// TCP keepalive interval in seconds. 0 = disabled. Defaults to 30. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub keepalive_secs: Option, + + /// TCP receive buffer size in bytes. Defaults to 2 MB. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub recv_buf_size: Option, + + /// TCP send buffer size in bytes. Defaults to 2 MB. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub send_buf_size: Option, + + /// SOCKS5 proxy for outbound connections (placeholder; not yet implemented). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub socks5_proxy: Option, + + /// Maximum simultaneous inbound connections. Defaults to 256. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub max_inbound_connections: Option, +} + +impl TcpConfig { + /// Get the default MTU. + pub fn mtu(&self) -> u16 { + self.mtu.unwrap_or(DEFAULT_TCP_MTU) + } + + /// Get the connect timeout in milliseconds. + pub fn connect_timeout_ms(&self) -> u64 { + self.connect_timeout_ms.unwrap_or(DEFAULT_TCP_CONNECT_TIMEOUT_MS) + } + + /// Whether TCP_NODELAY is enabled. Default: true. + pub fn nodelay(&self) -> bool { + self.nodelay.unwrap_or(true) + } + + /// Get the keepalive interval in seconds. 0 = disabled. Default: 30. + pub fn keepalive_secs(&self) -> u64 { + self.keepalive_secs.unwrap_or(DEFAULT_TCP_KEEPALIVE_SECS) + } + + /// Get the receive buffer size. Default: 2 MB. + pub fn recv_buf_size(&self) -> usize { + self.recv_buf_size.unwrap_or(DEFAULT_TCP_RECV_BUF) + } + + /// Get the send buffer size. Default: 2 MB. + pub fn send_buf_size(&self) -> usize { + self.send_buf_size.unwrap_or(DEFAULT_TCP_SEND_BUF) + } + + /// Get the maximum number of inbound connections. Default: 256. + pub fn max_inbound_connections(&self) -> usize { + self.max_inbound_connections.unwrap_or(DEFAULT_TCP_MAX_INBOUND) + } +} + +// ============================================================================ +// TransportsConfig +// ============================================================================ + /// Transports configuration section. /// /// Each transport type can have either a single instance (config directly @@ -251,6 +356,10 @@ pub struct TransportsConfig { /// Ethernet transport instances. #[serde(default, skip_serializing_if = "is_transport_empty")] pub ethernet: TransportInstances, + + /// TCP transport instances. + #[serde(default, skip_serializing_if = "is_transport_empty")] + pub tcp: TransportInstances, } /// Helper for skip_serializing_if on TransportInstances. @@ -261,7 +370,7 @@ fn is_transport_empty(instances: &TransportInstances) -> bool { impl TransportsConfig { /// Check if any transports are configured. pub fn is_empty(&self) -> bool { - self.udp.is_empty() && self.ethernet.is_empty() + self.udp.is_empty() && self.ethernet.is_empty() && self.tcp.is_empty() } /// Merge another TransportsConfig into this one. @@ -274,5 +383,8 @@ impl TransportsConfig { if !other.ethernet.is_empty() { self.ethernet = other.ethernet; } + if !other.tcp.is_empty() { + self.tcp = other.tcp; + } } } diff --git a/src/node/handlers/handshake.rs b/src/node/handlers/handshake.rs index 51dbe09..d0b65b4 100644 --- a/src/node/handlers/handshake.rs +++ b/src/node/handlers/handshake.rs @@ -280,6 +280,14 @@ impl Node { if let Some(peer) = self.peers.get_mut(&node_addr) { peer.set_handshake_msg2(wire_msg2.clone()); } + // Close the losing TCP connection (no-op for connectionless) + if let Some(loser_link) = self.links.get(&loser_link_id) { + let loser_tid = loser_link.transport_id(); + let loser_addr = loser_link.remote_addr().clone(); + if let Some(transport) = self.transports.get(&loser_tid) { + transport.close_connection(&loser_addr).await; + } + } // Clean up the losing connection's link self.remove_link(&loser_link_id); info!( @@ -295,6 +303,10 @@ impl Node { self.bloom_state.mark_update_needed(node_addr); } PromotionResult::CrossConnectionLost { winner_link_id } => { + // Close the losing TCP connection (no-op for connectionless) + if let Some(transport) = self.transports.get(&packet.transport_id) { + transport.close_connection(&packet.remote_addr).await; + } // This connection lost — clean up its link self.remove_link(&link_id); // Restore addr_to_link for the winner's link @@ -519,6 +531,14 @@ impl Node { // Clean up outbound connection state self.pending_outbound.remove(&key); + // Close the losing TCP connection (no-op for connectionless) + if let Some(link) = self.links.get(&link_id) { + let tid = link.transport_id(); + let addr = link.remote_addr().clone(); + if let Some(transport) = self.transports.get(&tid) { + transport.close_connection(&addr).await; + } + } self.remove_link(&link_id); // Send TreeAnnounce now that sessions are aligned @@ -550,6 +570,14 @@ impl Node { self.bloom_state.mark_update_needed(node_addr); } PromotionResult::CrossConnectionWon { loser_link_id, node_addr } => { + // Close the losing TCP connection (no-op for connectionless) + if let Some(loser_link) = self.links.get(&loser_link_id) { + let loser_tid = loser_link.transport_id(); + let loser_addr = loser_link.remote_addr().clone(); + if let Some(transport) = self.transports.get(&loser_tid) { + transport.close_connection(&loser_addr).await; + } + } // Clean up the losing connection's link self.remove_link(&loser_link_id); // Ensure addr_to_link points to the winning link @@ -570,6 +598,10 @@ impl Node { self.bloom_state.mark_update_needed(node_addr); } PromotionResult::CrossConnectionLost { winner_link_id } => { + // Close the losing TCP connection (no-op for connectionless) + if let Some(transport) = self.transports.get(&packet.transport_id) { + transport.close_connection(&packet.remote_addr).await; + } // This connection lost — clean up its link self.remove_link(&link_id); // Ensure addr_to_link points to the winner's link @@ -756,6 +788,12 @@ impl Node { return Err(NodeError::MaxPeersExceeded { max: self.max_peers }); } + // Preserve tree announce rate-limit state from old peer (if reconnecting). + // Without this, reconnection resets the rate limit window to zero, + // allowing an immediate announce that can feed an announce loop. + let old_announce_ts = self.peers.get(&peer_node_addr) + .map(|p| p.last_tree_announce_sent_ms()); + let mut new_peer = ActivePeer::with_session( verified_identity, link_id, @@ -771,6 +809,9 @@ impl Node { remote_epoch, ); new_peer.set_tree_announce_min_interval_ms(self.config.node.tree.announce_min_interval_ms); + if let Some(ts) = old_announce_ts { + new_peer.set_last_tree_announce_sent_ms(ts); + } self.peers.insert(peer_node_addr, new_peer); self.peers_by_index diff --git a/src/node/lifecycle.rs b/src/node/lifecycle.rs index 5323fa1..7636222 100644 --- a/src/node/lifecycle.rs +++ b/src/node/lifecycle.rs @@ -157,13 +157,28 @@ impl Node { // Allocate link ID and create link let link_id = self.allocate_link_id(); - let link = Link::connectionless( - link_id, - transport_id, - remote_addr.clone(), - LinkDirection::Outbound, - Duration::from_millis(self.config.node.base_rtt_ms), - ); + // Use Link::new() for connection-oriented transports (Connecting state) + // and Link::connectionless() for connectionless transports (Connected state) + let link = if self.transports.get(&transport_id) + .map(|t| t.transport_type().connection_oriented) + .unwrap_or(false) + { + Link::new( + link_id, + transport_id, + remote_addr.clone(), + LinkDirection::Outbound, + Duration::from_millis(self.config.node.base_rtt_ms), + ) + } else { + Link::connectionless( + link_id, + transport_id, + remote_addr.clone(), + LinkDirection::Outbound, + Duration::from_millis(self.config.node.base_rtt_ms), + ) + }; self.links.insert(link_id, link); diff --git a/src/node/mod.rs b/src/node/mod.rs index 2eceb73..1b18a1a 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -28,6 +28,7 @@ use crate::transport::{ Link, LinkId, PacketRx, PacketTx, TransportAddr, TransportError, TransportHandle, TransportId, }; use crate::transport::udp::UdpTransport; +use crate::transport::tcp::TcpTransport; #[cfg(target_os = "linux")] use crate::transport::ethernet::EthernetTransport; use crate::tree::TreeState; @@ -598,6 +599,21 @@ impl Node { } } + // Create TCP transport instances + let tcp_instances: Vec<_> = self + .config + .transports + .tcp + .iter() + .map(|(name, config)| (name.map(|s| s.to_string()), config.clone())) + .collect(); + + for (name, tcp_config) in tcp_instances { + let transport_id = self.allocate_transport_id(); + let tcp = TcpTransport::new(transport_id, name, tcp_config, packet_tx.clone()); + transports.push(TransportHandle::Tcp(tcp)); + } + transports } diff --git a/src/peer/active.rs b/src/peer/active.rs index cbe7fde..8da5024 100644 --- a/src/peer/active.rs +++ b/src/peer/active.rs @@ -622,6 +622,16 @@ impl ActivePeer { self.tree_announce_min_interval_ms = ms; } + /// Get the last tree announce send timestamp (for carrying across reconnection). + pub fn last_tree_announce_sent_ms(&self) -> u64 { + self.last_tree_announce_sent_ms + } + + /// Set the last tree announce send timestamp (to preserve rate limit across reconnection). + pub fn set_last_tree_announce_sent_ms(&mut self, ms: u64) { + self.last_tree_announce_sent_ms = ms; + } + /// Check if we can send a TreeAnnounce now (rate limiting). pub fn can_send_tree_announce(&self, now_ms: u64) -> bool { now_ms.saturating_sub(self.last_tree_announce_sent_ms) >= self.tree_announce_min_interval_ms diff --git a/src/transport/mod.rs b/src/transport/mod.rs index bf127cf..8804a79 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -5,12 +5,14 @@ //! which FIPS links are established. pub mod udp; +pub mod tcp; #[cfg(target_os = "linux")] pub mod ethernet; use secp256k1::XOnlyPublicKey; use udp::UdpTransport; +use tcp::TcpTransport; #[cfg(target_os = "linux")] use ethernet::EthernetTransport; use std::fmt; @@ -778,6 +780,15 @@ pub trait Transport { fn accept_connections(&self) -> bool { true } + + /// Close a specific connection (connection-oriented transports only). + /// + /// For connectionless transports (UDP, Ethernet), this is a no-op. + /// Connection-oriented transports (TCP, Tor) remove the connection + /// from their pool and drop the underlying stream. + fn close_connection(&self, _addr: &TransportAddr) { + // Default no-op for connectionless transports + } } // ============================================================================ @@ -794,6 +805,8 @@ pub enum TransportHandle { /// Raw Ethernet transport. #[cfg(target_os = "linux")] Ethernet(EthernetTransport), + /// TCP/IP transport. + Tcp(TcpTransport), } impl TransportHandle { @@ -803,6 +816,7 @@ impl TransportHandle { TransportHandle::Udp(t) => t.start_async().await, #[cfg(target_os = "linux")] TransportHandle::Ethernet(t) => t.start_async().await, + TransportHandle::Tcp(t) => t.start_async().await, } } @@ -812,6 +826,7 @@ impl TransportHandle { TransportHandle::Udp(t) => t.stop_async().await, #[cfg(target_os = "linux")] TransportHandle::Ethernet(t) => t.stop_async().await, + TransportHandle::Tcp(t) => t.stop_async().await, } } @@ -821,6 +836,7 @@ impl TransportHandle { TransportHandle::Udp(t) => t.send_async(addr, data).await, #[cfg(target_os = "linux")] TransportHandle::Ethernet(t) => t.send_async(addr, data).await, + TransportHandle::Tcp(t) => t.send_async(addr, data).await, } } @@ -830,6 +846,7 @@ impl TransportHandle { TransportHandle::Udp(t) => t.transport_id(), #[cfg(target_os = "linux")] TransportHandle::Ethernet(t) => t.transport_id(), + TransportHandle::Tcp(t) => t.transport_id(), } } @@ -839,6 +856,7 @@ impl TransportHandle { TransportHandle::Udp(t) => t.name(), #[cfg(target_os = "linux")] TransportHandle::Ethernet(t) => t.name(), + TransportHandle::Tcp(t) => t.name(), } } @@ -848,6 +866,7 @@ impl TransportHandle { TransportHandle::Udp(t) => t.transport_type(), #[cfg(target_os = "linux")] TransportHandle::Ethernet(t) => t.transport_type(), + TransportHandle::Tcp(t) => t.transport_type(), } } @@ -857,6 +876,7 @@ impl TransportHandle { TransportHandle::Udp(t) => t.state(), #[cfg(target_os = "linux")] TransportHandle::Ethernet(t) => t.state(), + TransportHandle::Tcp(t) => t.state(), } } @@ -866,6 +886,7 @@ impl TransportHandle { TransportHandle::Udp(t) => t.mtu(), #[cfg(target_os = "linux")] TransportHandle::Ethernet(t) => t.mtu(), + TransportHandle::Tcp(t) => t.mtu(), } } @@ -878,6 +899,7 @@ impl TransportHandle { TransportHandle::Udp(t) => t.link_mtu(addr), #[cfg(target_os = "linux")] TransportHandle::Ethernet(t) => t.link_mtu(addr), + TransportHandle::Tcp(t) => t.link_mtu(addr), } } @@ -887,6 +909,7 @@ impl TransportHandle { TransportHandle::Udp(t) => t.local_addr(), #[cfg(target_os = "linux")] TransportHandle::Ethernet(_) => None, + TransportHandle::Tcp(t) => t.local_addr(), } } @@ -896,6 +919,7 @@ impl TransportHandle { TransportHandle::Udp(_) => None, #[cfg(target_os = "linux")] TransportHandle::Ethernet(t) => Some(t.interface_name()), + TransportHandle::Tcp(_) => None, } } @@ -905,6 +929,7 @@ impl TransportHandle { TransportHandle::Udp(t) => t.discover(), #[cfg(target_os = "linux")] TransportHandle::Ethernet(t) => t.discover(), + TransportHandle::Tcp(t) => t.discover(), } } @@ -914,6 +939,7 @@ impl TransportHandle { TransportHandle::Udp(t) => t.auto_connect(), #[cfg(target_os = "linux")] TransportHandle::Ethernet(t) => t.auto_connect(), + TransportHandle::Tcp(t) => t.auto_connect(), } } @@ -923,6 +949,20 @@ impl TransportHandle { TransportHandle::Udp(t) => t.accept_connections(), #[cfg(target_os = "linux")] TransportHandle::Ethernet(t) => t.accept_connections(), + TransportHandle::Tcp(t) => t.accept_connections(), + } + } + + /// Close a specific connection on this transport. + /// + /// No-op for connectionless transports. For TCP, removes the + /// connection from the pool and drops the stream. + pub async fn close_connection(&self, addr: &TransportAddr) { + match self { + TransportHandle::Udp(t) => t.close_connection(addr), + #[cfg(target_os = "linux")] + TransportHandle::Ethernet(t) => t.close_connection(addr), + TransportHandle::Tcp(t) => t.close_connection_async(addr).await, } } diff --git a/src/transport/tcp/mod.rs b/src/transport/tcp/mod.rs new file mode 100644 index 0000000..7901038 --- /dev/null +++ b/src/transport/tcp/mod.rs @@ -0,0 +1,1087 @@ +//! TCP Transport Implementation +//! +//! Provides TCP-based transport for FIPS peer communication. TCP enables +//! firewall traversal (many networks allow TCP on port 443 but block UDP) +//! and serves as the foundation for the future Tor transport. +//! +//! FIPS protocols (FMP, FSP, MMP) are all unreliable datagrams. This +//! transport carries those datagrams over TCP — the main pathology is +//! head-of-line blocking, which adds latency jitter that MMP correctly +//! measures and cost-based parent selection correctly penalizes. +//! +//! ## Architecture +//! +//! Unlike UDP (one socket serves all peers), TCP requires one `TcpStream` +//! per peer. The transport maintains a connection pool mapping +//! `TransportAddr` to per-connection state, plus an optional `TcpListener` +//! for inbound connections. +//! +//! ## Framing +//! +//! Uses the existing 4-byte FMP common prefix to recover packet boundaries. +//! No additional framing overhead — packets are written directly to the +//! TCP stream and the receiver uses phase-dependent size computation. + +pub mod stream; + +use super::{ + DiscoveredPeer, PacketTx, ReceivedPacket, Transport, TransportAddr, TransportError, + TransportId, TransportState, TransportType, +}; +use crate::config::TcpConfig; +use stream::read_fmp_packet; + +use socket2::TcpKeepalive; +use std::collections::HashMap; +use std::net::SocketAddr; +use std::sync::Arc; +use std::time::Duration; +use tokio::io::AsyncWriteExt; +use tokio::net::{TcpListener, TcpStream}; +use tokio::net::tcp::OwnedWriteHalf; +use tokio::sync::Mutex; +use tokio::task::JoinHandle; +use tokio::time::Instant; +use tracing::{debug, info, trace, warn}; + +// ============================================================================ +// Connection Pool +// ============================================================================ + +/// State for a single TCP connection to a peer. +struct TcpConnection { + /// Write half of the split stream. + writer: Arc>, + /// Receive task for this connection. + recv_task: JoinHandle<()>, + /// MSS-derived MTU for this connection (used for dynamic MTU re-reading). + #[allow(dead_code)] + mtu: u16, + /// When the connection was established. + #[allow(dead_code)] + established_at: Instant, +} + +/// Shared connection pool. +type ConnectionPool = Arc>>; + +// ============================================================================ +// TCP Transport +// ============================================================================ + +/// TCP transport for FIPS. +/// +/// Provides connection-oriented, reliable byte stream delivery over TCP/IP. +/// Each peer has its own TCP connection; links are managed per-connection +/// with a connection pool keyed by `TransportAddr`. +pub struct TcpTransport { + /// Unique transport identifier. + transport_id: TransportId, + /// Optional instance name (for named instances in config). + name: Option, + /// Configuration. + config: TcpConfig, + /// Current state. + state: TransportState, + /// Connection pool: addr -> per-connection state. + pool: ConnectionPool, + /// Channel for delivering received packets to Node. + packet_tx: PacketTx, + /// Accept loop task handle (if listener bound). + accept_task: Option>, + /// Local listener address (after start, if bind_addr configured). + local_addr: Option, +} + +impl TcpTransport { + /// Create a new TCP transport. + pub fn new( + transport_id: TransportId, + name: Option, + config: TcpConfig, + packet_tx: PacketTx, + ) -> Self { + Self { + transport_id, + name, + config, + state: TransportState::Configured, + pool: Arc::new(Mutex::new(HashMap::new())), + packet_tx, + accept_task: None, + local_addr: None, + } + } + + /// Get the instance name (if configured as a named instance). + pub fn name(&self) -> Option<&str> { + self.name.as_deref() + } + + /// Get the local listener address (only valid after start with bind_addr). + pub fn local_addr(&self) -> Option { + self.local_addr + } + + /// Start the transport asynchronously. + /// + /// If `bind_addr` is configured, binds a TCP listener and spawns + /// the accept loop. Otherwise, operates in outbound-only mode. + pub async fn start_async(&mut self) -> Result<(), TransportError> { + if !self.state.can_start() { + return Err(TransportError::AlreadyStarted); + } + + self.state = TransportState::Starting; + + // Bind listener if configured + if let Some(ref bind_addr) = self.config.bind_addr { + let addr: SocketAddr = bind_addr + .parse() + .map_err(|e| TransportError::StartFailed(format!("invalid bind address: {}", e)))?; + + let listener = TcpListener::bind(addr) + .await + .map_err(|e| TransportError::StartFailed(format!("bind failed: {}", e)))?; + + self.local_addr = Some( + listener + .local_addr() + .map_err(|e| TransportError::StartFailed(format!("get local addr: {}", e)))?, + ); + + // Spawn accept loop + let transport_id = self.transport_id; + let packet_tx = self.packet_tx.clone(); + let pool = self.pool.clone(); + let cfg = AcceptConfig { + mtu: self.config.mtu(), + max_inbound: self.config.max_inbound_connections(), + nodelay: self.config.nodelay(), + keepalive_secs: self.config.keepalive_secs(), + recv_buf: self.config.recv_buf_size(), + send_buf: self.config.send_buf_size(), + }; + + let accept_task = tokio::spawn(async move { + accept_loop(listener, transport_id, packet_tx, pool, cfg).await; + }); + self.accept_task = Some(accept_task); + } + + self.state = TransportState::Up; + + if let Some(ref name) = self.name { + info!( + name = %name, + local_addr = ?self.local_addr, + mtu = self.config.mtu(), + "TCP transport started" + ); + } else { + info!( + local_addr = ?self.local_addr, + mtu = self.config.mtu(), + "TCP transport started" + ); + } + + Ok(()) + } + + /// Stop the transport asynchronously. + pub async fn stop_async(&mut self) -> Result<(), TransportError> { + if !self.state.is_operational() { + return Err(TransportError::NotStarted); + } + + // Abort accept loop + if let Some(task) = self.accept_task.take() { + task.abort(); + let _ = task.await; + } + + // Close all connections + let mut pool = self.pool.lock().await; + for (addr, conn) in pool.drain() { + conn.recv_task.abort(); + let _ = conn.recv_task.await; + debug!( + transport_id = %self.transport_id, + remote_addr = %addr, + "TCP connection closed (transport stopping)" + ); + } + drop(pool); + + self.local_addr = None; + self.state = TransportState::Down; + + info!( + transport_id = %self.transport_id, + "TCP transport stopped" + ); + + Ok(()) + } + + /// Send a packet asynchronously. + /// + /// If no connection exists to the given address, performs connect-on-send: + /// establishes a new TCP connection, configures socket options, splits the + /// stream, spawns a receive task, and stores the connection in the pool. + pub async fn send_async( + &self, + addr: &TransportAddr, + data: &[u8], + ) -> Result { + if !self.state.is_operational() { + return Err(TransportError::NotStarted); + } + + // Pre-send MTU check: reject oversize packets before writing them + // to the TCP stream. Without this, the receiver's FMP stream reader + // would see payload_len > max and close the connection, causing a + // disruptive reset-reconnect cycle. + let mtu = self.config.mtu() as usize; + if data.len() > mtu { + return Err(TransportError::MtuExceeded { + packet_size: data.len(), + mtu: self.config.mtu(), + }); + } + + // Get or create connection + let writer = { + let pool = self.pool.lock().await; + pool.get(addr).map(|c| c.writer.clone()) + }; + + let writer = match writer { + Some(w) => w, + None => { + // Connect-on-send + self.connect(addr).await? + } + }; + + // Write packet directly (no framing transformation needed) + let mut w = writer.lock().await; + match w.write_all(data).await { + Ok(()) => { + trace!( + transport_id = %self.transport_id, + remote_addr = %addr, + bytes = data.len(), + "TCP packet sent" + ); + Ok(data.len()) + } + Err(e) => { + drop(w); + // Remove failed connection from pool + let mut pool = self.pool.lock().await; + if let Some(conn) = pool.remove(addr) { + conn.recv_task.abort(); + } + Err(TransportError::SendFailed(format!("{}", e))) + } + } + } + + /// Establish a new TCP connection to the given address. + /// + /// Configures socket options, reads TCP_MAXSEG for MTU, splits the + /// stream, spawns a receive task, and stores in the pool. + async fn connect( + &self, + addr: &TransportAddr, + ) -> Result>, TransportError> { + let socket_addr = parse_socket_addr(addr)?; + let timeout_ms = self.config.connect_timeout_ms(); + + // Connect with timeout + let stream = tokio::time::timeout( + Duration::from_millis(timeout_ms), + TcpStream::connect(socket_addr), + ) + .await + .map_err(|_| TransportError::Timeout)? + .map_err(|_| TransportError::ConnectionRefused)?; + + // Configure socket options via socket2 + let std_stream = stream.into_std() + .map_err(|e| TransportError::StartFailed(format!("into_std: {}", e)))?; + configure_socket(&std_stream, &self.config)?; + + // Read TCP_MAXSEG for per-connection MTU + let mss_mtu = read_mss_mtu(&std_stream, self.config.mtu()); + + // Convert back to tokio + let stream = TcpStream::from_std(std_stream) + .map_err(|e| TransportError::StartFailed(format!("from_std: {}", e)))?; + + // Split and spawn receive task + let (read_half, write_half) = stream.into_split(); + let writer = Arc::new(Mutex::new(write_half)); + + let transport_id = self.transport_id; + let packet_tx = self.packet_tx.clone(); + let pool = self.pool.clone(); + let remote_addr = addr.clone(); + let mtu = mss_mtu; + + let recv_task = tokio::spawn(async move { + tcp_receive_loop(read_half, transport_id, remote_addr.clone(), packet_tx, pool, mtu).await; + }); + + let conn = TcpConnection { + writer: writer.clone(), + recv_task, + mtu: mss_mtu, + established_at: Instant::now(), + }; + + let mut pool = self.pool.lock().await; + pool.insert(addr.clone(), conn); + + debug!( + transport_id = %self.transport_id, + remote_addr = %addr, + mtu = mss_mtu, + "TCP connection established (connect-on-send)" + ); + + Ok(writer) + } + + /// Close a specific connection asynchronously. + /// + /// Removes the connection from the pool, aborts its receive task, + /// and drops the write half (sends FIN to remote). + pub async fn close_connection_async(&self, addr: &TransportAddr) { + let mut pool = self.pool.lock().await; + if let Some(conn) = pool.remove(addr) { + conn.recv_task.abort(); + debug!( + transport_id = %self.transport_id, + remote_addr = %addr, + "TCP connection closed (close_connection)" + ); + } + } +} + +impl Transport for TcpTransport { + fn transport_id(&self) -> TransportId { + self.transport_id + } + + fn transport_type(&self) -> &TransportType { + &TransportType::TCP + } + + fn state(&self) -> TransportState { + self.state + } + + fn mtu(&self) -> u16 { + self.config.mtu() + } + + fn link_mtu(&self, _addr: &TransportAddr) -> u16 { + // Per-link MTU would require synchronous pool access. + // For now, return the configured default. The async send path + // uses the per-connection MSS-derived MTU for validation. + self.config.mtu() + } + + fn start(&mut self) -> Result<(), TransportError> { + Err(TransportError::NotSupported( + "use start_async() for TCP transport".into(), + )) + } + + fn stop(&mut self) -> Result<(), TransportError> { + Err(TransportError::NotSupported( + "use stop_async() for TCP transport".into(), + )) + } + + fn send(&self, _addr: &TransportAddr, _data: &[u8]) -> Result<(), TransportError> { + Err(TransportError::NotSupported( + "use send_async() for TCP transport".into(), + )) + } + + fn discover(&self) -> Result, TransportError> { + // TCP has no discovery mechanism + Ok(Vec::new()) + } + + fn accept_connections(&self) -> bool { + // If bind_addr is configured, we accept inbound connections + self.config.bind_addr.is_some() + } +} + +// ============================================================================ +// Accept Loop +// ============================================================================ + +/// Socket configuration parameters passed to the accept loop. +struct AcceptConfig { + mtu: u16, + max_inbound: usize, + nodelay: bool, + keepalive_secs: u64, + recv_buf: usize, + send_buf: usize, +} + +/// TCP accept loop — runs as a spawned task when bind_addr is configured. +#[allow(clippy::too_many_arguments)] +async fn accept_loop( + listener: TcpListener, + transport_id: TransportId, + packet_tx: PacketTx, + pool: ConnectionPool, + cfg: AcceptConfig, +) { + let AcceptConfig { mtu, max_inbound, nodelay, keepalive_secs, recv_buf, send_buf } = cfg; + debug!(transport_id = %transport_id, "TCP accept loop starting"); + + loop { + match listener.accept().await { + Ok((stream, peer_addr)) => { + // Check connection limit + { + let pool_guard = pool.lock().await; + if pool_guard.len() >= max_inbound { + warn!( + transport_id = %transport_id, + peer_addr = %peer_addr, + max = max_inbound, + "Rejecting inbound TCP connection (max_inbound_connections reached)" + ); + continue; + } + } + + // Configure socket options + let std_stream = match stream.into_std() { + Ok(s) => s, + Err(e) => { + warn!( + transport_id = %transport_id, + error = %e, + "Failed to convert accepted stream to std" + ); + continue; + } + }; + + if let Err(e) = configure_accepted_socket(&std_stream, nodelay, keepalive_secs, recv_buf, send_buf) { + warn!( + transport_id = %transport_id, + peer_addr = %peer_addr, + error = %e, + "Failed to configure accepted socket" + ); + continue; + } + + // Read MSS for per-connection MTU + let conn_mtu = read_mss_mtu(&std_stream, mtu); + + let stream = match TcpStream::from_std(std_stream) { + Ok(s) => s, + Err(e) => { + warn!( + transport_id = %transport_id, + error = %e, + "Failed to convert accepted stream back to tokio" + ); + continue; + } + }; + + let remote_addr = TransportAddr::from_string(&peer_addr.to_string()); + + // Split and spawn receive task + let (read_half, write_half) = stream.into_split(); + let writer = Arc::new(Mutex::new(write_half)); + + let recv_pool = pool.clone(); + let recv_packet_tx = packet_tx.clone(); + let recv_addr = remote_addr.clone(); + + let recv_task = tokio::spawn(async move { + tcp_receive_loop( + read_half, + transport_id, + recv_addr, + recv_packet_tx, + recv_pool, + conn_mtu, + ) + .await; + }); + + let conn = TcpConnection { + writer, + recv_task, + mtu: conn_mtu, + established_at: Instant::now(), + }; + + let mut pool_guard = pool.lock().await; + pool_guard.insert(remote_addr.clone(), conn); + + debug!( + transport_id = %transport_id, + remote_addr = %remote_addr, + mtu = conn_mtu, + "Accepted inbound TCP connection" + ); + } + Err(e) => { + warn!( + transport_id = %transport_id, + error = %e, + "TCP accept error" + ); + } + } + } +} + +// ============================================================================ +// Receive Loop (per-connection) +// ============================================================================ + +/// Per-connection TCP receive loop. +/// +/// Reads complete FMP packets using the stream reader, delivers them to +/// the node via the packet channel. On error or EOF, removes the +/// connection from the pool and exits. +async fn tcp_receive_loop( + mut reader: tokio::net::tcp::OwnedReadHalf, + transport_id: TransportId, + remote_addr: TransportAddr, + packet_tx: PacketTx, + pool: ConnectionPool, + mtu: u16, +) { + debug!( + transport_id = %transport_id, + remote_addr = %remote_addr, + "TCP receive loop starting" + ); + + loop { + match read_fmp_packet(&mut reader, mtu).await { + Ok(data) => { + trace!( + transport_id = %transport_id, + remote_addr = %remote_addr, + bytes = data.len(), + "TCP packet received" + ); + + let packet = ReceivedPacket::new( + transport_id, + remote_addr.clone(), + data, + ); + + if packet_tx.send(packet).await.is_err() { + info!( + transport_id = %transport_id, + "Packet channel closed, stopping TCP receive loop" + ); + break; + } + } + Err(e) => { + // EOF or protocol error — remove connection from pool + debug!( + transport_id = %transport_id, + remote_addr = %remote_addr, + error = %e, + "TCP receive error, removing connection" + ); + break; + } + } + } + + // Clean up: remove ourselves from the pool + let mut pool_guard = pool.lock().await; + pool_guard.remove(&remote_addr); + + debug!( + transport_id = %transport_id, + remote_addr = %remote_addr, + "TCP receive loop stopped" + ); +} + +// ============================================================================ +// Socket Configuration Helpers +// ============================================================================ + +/// Parse a TransportAddr as SocketAddr. +fn parse_socket_addr(addr: &TransportAddr) -> Result { + addr.as_str() + .ok_or_else(|| TransportError::InvalidAddress("not valid UTF-8".into()))? + .parse() + .map_err(|e| TransportError::InvalidAddress(format!("{}", e))) +} + +/// Configure a TCP socket with the transport's settings. +fn configure_socket( + stream: &std::net::TcpStream, + config: &TcpConfig, +) -> Result<(), TransportError> { + let socket = socket2::SockRef::from(stream).try_clone() + .map_err(|e| TransportError::StartFailed(format!("clone socket: {}", e)))?; + + // TCP_NODELAY + socket.set_tcp_nodelay(config.nodelay()) + .map_err(|e| TransportError::StartFailed(format!("set nodelay: {}", e)))?; + + // Keepalive + let keepalive_secs = config.keepalive_secs(); + if keepalive_secs > 0 { + let keepalive = TcpKeepalive::new() + .with_time(Duration::from_secs(keepalive_secs)); + socket.set_tcp_keepalive(&keepalive) + .map_err(|e| TransportError::StartFailed(format!("set keepalive: {}", e)))?; + } + + // Buffer sizes + socket.set_recv_buffer_size(config.recv_buf_size()) + .map_err(|e| TransportError::StartFailed(format!("set recv buffer: {}", e)))?; + socket.set_send_buffer_size(config.send_buf_size()) + .map_err(|e| TransportError::StartFailed(format!("set send buffer: {}", e)))?; + + Ok(()) +} + +/// Configure an accepted TCP socket (without TcpConfig reference). +fn configure_accepted_socket( + stream: &std::net::TcpStream, + nodelay: bool, + keepalive_secs: u64, + recv_buf: usize, + send_buf: usize, +) -> Result<(), TransportError> { + let socket = socket2::SockRef::from(stream).try_clone() + .map_err(|e| TransportError::StartFailed(format!("clone socket: {}", e)))?; + + socket.set_tcp_nodelay(nodelay) + .map_err(|e| TransportError::StartFailed(format!("set nodelay: {}", e)))?; + + if keepalive_secs > 0 { + let keepalive = TcpKeepalive::new() + .with_time(Duration::from_secs(keepalive_secs)); + socket.set_tcp_keepalive(&keepalive) + .map_err(|e| TransportError::StartFailed(format!("set keepalive: {}", e)))?; + } + + socket.set_recv_buffer_size(recv_buf) + .map_err(|e| TransportError::StartFailed(format!("set recv buffer: {}", e)))?; + socket.set_send_buffer_size(send_buf) + .map_err(|e| TransportError::StartFailed(format!("set send buffer: {}", e)))?; + + Ok(()) +} + +/// Read TCP_MAXSEG and derive per-connection MTU, falling back to default. +fn read_mss_mtu(stream: &std::net::TcpStream, default_mtu: u16) -> u16 { + // Try to read TCP_MAXSEG. Not all platforms support this. + #[cfg(target_os = "linux")] + { + use std::os::unix::io::AsRawFd; + unsafe { + let mut mss: libc::c_int = 0; + let mut len: libc::socklen_t = std::mem::size_of::() as libc::socklen_t; + let fd = stream.as_raw_fd(); + let ret = libc::getsockopt( + fd, + libc::IPPROTO_TCP, + libc::TCP_MAXSEG, + &mut mss as *mut libc::c_int as *mut libc::c_void, + &mut len, + ); + if ret == 0 && mss > 0 { + let mss_mtu = (mss as u32).min(u16::MAX as u32) as u16; + // Use the smaller of MSS and configured default + return mss_mtu.min(default_mtu); + } + } + } + + #[cfg(not(target_os = "linux"))] + let _ = stream; + + // Fallback: use configured default MTU + default_mtu +} + +// ============================================================================ +// Tests +// ============================================================================ + +#[cfg(test)] +mod tests { + use super::*; + use crate::transport::packet_channel; + use tokio::time::{timeout, Duration}; + + fn make_config() -> TcpConfig { + TcpConfig { + bind_addr: Some("127.0.0.1:0".to_string()), + mtu: Some(1400), + ..Default::default() + } + } + + fn make_outbound_config() -> TcpConfig { + TcpConfig { + bind_addr: None, + mtu: Some(1400), + ..Default::default() + } + } + + #[tokio::test] + async fn test_start_stop() { + let (tx, _rx) = packet_channel(100); + let mut transport = TcpTransport::new(TransportId::new(1), None, make_config(), tx); + + assert_eq!(transport.state(), TransportState::Configured); + + transport.start_async().await.unwrap(); + assert_eq!(transport.state(), TransportState::Up); + assert!(transport.local_addr().is_some()); + + transport.stop_async().await.unwrap(); + assert_eq!(transport.state(), TransportState::Down); + } + + #[tokio::test] + async fn test_start_outbound_only() { + let (tx, _rx) = packet_channel(100); + let mut transport = TcpTransport::new(TransportId::new(1), None, make_outbound_config(), tx); + + transport.start_async().await.unwrap(); + assert_eq!(transport.state(), TransportState::Up); + // No listener, so no local_addr + assert!(transport.local_addr().is_none()); + + transport.stop_async().await.unwrap(); + } + + #[tokio::test] + async fn test_double_start_fails() { + let (tx, _rx) = packet_channel(100); + let mut transport = TcpTransport::new(TransportId::new(1), None, make_config(), tx); + + transport.start_async().await.unwrap(); + + let result = transport.start_async().await; + assert!(matches!(result, Err(TransportError::AlreadyStarted))); + + transport.stop_async().await.unwrap(); + } + + #[tokio::test] + async fn test_stop_not_started_fails() { + let (tx, _rx) = packet_channel(100); + let mut transport = TcpTransport::new(TransportId::new(1), None, make_config(), tx); + + let result = transport.stop_async().await; + assert!(matches!(result, Err(TransportError::NotStarted))); + } + + #[tokio::test] + async fn test_send_not_started() { + let (tx, _rx) = packet_channel(100); + let transport = TcpTransport::new(TransportId::new(1), None, make_config(), tx); + + let result = transport + .send_async(&TransportAddr::from_string("127.0.0.1:9999"), b"test") + .await; + + assert!(matches!(result, Err(TransportError::NotStarted))); + } + + #[tokio::test] + async fn test_send_recv() { + let (tx1, _rx1) = packet_channel(100); + let (tx2, mut rx2) = packet_channel(100); + + let mut t1 = TcpTransport::new(TransportId::new(1), None, make_outbound_config(), tx1); + let mut t2 = TcpTransport::new(TransportId::new(2), None, make_config(), tx2); + + t1.start_async().await.unwrap(); + t2.start_async().await.unwrap(); + + let addr2 = t2.local_addr().unwrap(); + + // Build a valid FMP established frame to send + // [ver+phase:1][flags:1][payload_len:2 LE][12 bytes header][payload bytes][16 bytes tag] + let payload_len = 4u16; + let total = 4 + 12 + payload_len as usize + 16; + let mut frame = vec![0u8; total]; + frame[0] = 0x00; // ver=0, phase=0 (established) + frame[1] = 0x00; // flags + frame[2..4].copy_from_slice(&payload_len.to_le_bytes()); + // Fill the rest with a recognizable pattern + for i in 4..total { + frame[i] = (i & 0xFF) as u8; + } + + let bytes_sent = t1 + .send_async( + &TransportAddr::from_string(&addr2.to_string()), + &frame, + ) + .await + .unwrap(); + assert_eq!(bytes_sent, frame.len()); + + // Receive on t2 + let packet = timeout(Duration::from_secs(2), rx2.recv()) + .await + .expect("timeout") + .expect("channel closed"); + + assert_eq!(packet.data, frame); + + t1.stop_async().await.unwrap(); + t2.stop_async().await.unwrap(); + } + + #[tokio::test] + async fn test_bidirectional() { + let (tx1, mut rx1) = packet_channel(100); + let (tx2, mut rx2) = packet_channel(100); + + let mut t1 = TcpTransport::new(TransportId::new(1), None, make_config(), tx1); + let mut t2 = TcpTransport::new(TransportId::new(2), None, make_config(), tx2); + + t1.start_async().await.unwrap(); + t2.start_async().await.unwrap(); + + let addr1 = t1.local_addr().unwrap(); + let addr2 = t2.local_addr().unwrap(); + + // Build valid FMP msg1 frame (114 bytes) + let mut msg1_frame = vec![0xAA; 114]; + msg1_frame[0] = 0x01; // phase=msg1 + msg1_frame[1] = 0x00; + msg1_frame[2..4].copy_from_slice(&110u16.to_le_bytes()); // payload_len = 110 + + // Send from t1 to t2 + t1.send_async( + &TransportAddr::from_string(&addr2.to_string()), + &msg1_frame, + ) + .await + .unwrap(); + + let packet = timeout(Duration::from_secs(2), rx2.recv()) + .await + .expect("timeout") + .expect("channel closed"); + assert_eq!(packet.data, msg1_frame); + + // Build valid FMP msg2 frame (69 bytes) + let mut msg2_frame = vec![0xBB; 69]; + msg2_frame[0] = 0x02; // phase=msg2 + msg2_frame[1] = 0x00; + msg2_frame[2..4].copy_from_slice(&65u16.to_le_bytes()); // payload_len = 65 + + // Send from t2 to t1 + t2.send_async( + &TransportAddr::from_string(&addr1.to_string()), + &msg2_frame, + ) + .await + .unwrap(); + + let packet = timeout(Duration::from_secs(2), rx1.recv()) + .await + .expect("timeout") + .expect("channel closed"); + assert_eq!(packet.data, msg2_frame); + + t1.stop_async().await.unwrap(); + t2.stop_async().await.unwrap(); + } + + #[tokio::test] + async fn test_connect_timeout() { + let (tx, _rx) = packet_channel(100); + let config = TcpConfig { + bind_addr: None, + connect_timeout_ms: Some(100), // Very short timeout + ..Default::default() + }; + let mut transport = TcpTransport::new(TransportId::new(1), None, config, tx); + transport.start_async().await.unwrap(); + + // Try to connect to a non-routable address (should timeout) + let result = transport + .send_async(&TransportAddr::from_string("192.0.2.1:2121"), b"\x00\x00\x04\x00test1234567890123456789012345678") + .await; + + assert!(result.is_err()); + + transport.stop_async().await.unwrap(); + } + + #[tokio::test] + async fn test_close_connection() { + let (tx1, _rx1) = packet_channel(100); + let (tx2, _rx2) = packet_channel(100); + + let mut t1 = TcpTransport::new(TransportId::new(1), None, make_outbound_config(), tx1); + let mut t2 = TcpTransport::new(TransportId::new(2), None, make_config(), tx2); + + t1.start_async().await.unwrap(); + t2.start_async().await.unwrap(); + + let addr2 = t2.local_addr().unwrap(); + let remote = TransportAddr::from_string(&addr2.to_string()); + + // Build valid msg1 frame to establish connection + let mut msg1 = vec![0xAA; 114]; + msg1[0] = 0x01; + msg1[1] = 0x00; + msg1[2..4].copy_from_slice(&110u16.to_le_bytes()); + + t1.send_async(&remote, &msg1).await.unwrap(); + + // Connection should exist + { + let pool = t1.pool.lock().await; + assert!(pool.contains_key(&remote)); + } + + // Close it + t1.close_connection_async(&remote).await; + + // Connection should be gone + { + let pool = t1.pool.lock().await; + assert!(!pool.contains_key(&remote)); + } + + t1.stop_async().await.unwrap(); + t2.stop_async().await.unwrap(); + } + + #[tokio::test] + async fn test_discover_returns_empty() { + let (tx, _rx) = packet_channel(100); + let transport = TcpTransport::new(TransportId::new(1), None, make_config(), tx); + + let peers = transport.discover().unwrap(); + assert!(peers.is_empty()); + } + + #[test] + fn test_transport_type() { + let (tx, _rx) = packet_channel(100); + let transport = TcpTransport::new(TransportId::new(1), None, make_config(), tx); + + assert_eq!(transport.transport_type().name, "tcp"); + assert!(transport.transport_type().connection_oriented); + assert!(transport.transport_type().reliable); + } + + #[test] + fn test_sync_methods_return_not_supported() { + let (tx, _rx) = packet_channel(100); + let mut transport = TcpTransport::new(TransportId::new(1), None, make_config(), tx); + + assert!(matches!( + transport.start(), + Err(TransportError::NotSupported(_)) + )); + assert!(matches!( + transport.stop(), + Err(TransportError::NotSupported(_)) + )); + assert!(matches!( + transport.send(&TransportAddr::from_string("test"), b"data"), + Err(TransportError::NotSupported(_)) + )); + } + + #[test] + fn test_accept_connections_with_bind() { + let (tx, _rx) = packet_channel(100); + let config = TcpConfig { + bind_addr: Some("0.0.0.0:0".to_string()), + ..Default::default() + }; + let transport = TcpTransport::new(TransportId::new(1), None, config, tx); + assert!(transport.accept_connections()); + } + + #[test] + fn test_accept_connections_without_bind() { + let (tx, _rx) = packet_channel(100); + let config = TcpConfig { + bind_addr: None, + ..Default::default() + }; + let transport = TcpTransport::new(TransportId::new(1), None, config, tx); + assert!(!transport.accept_connections()); + } + + #[tokio::test] + async fn test_connection_drop_and_reconnect() { + let (tx1, _rx1) = packet_channel(100); + let (tx2, mut rx2) = packet_channel(100); + + let mut t1 = TcpTransport::new(TransportId::new(1), None, make_outbound_config(), tx1); + let mut t2 = TcpTransport::new(TransportId::new(2), None, make_config(), tx2); + + t1.start_async().await.unwrap(); + t2.start_async().await.unwrap(); + + let addr2 = t2.local_addr().unwrap(); + let remote = TransportAddr::from_string(&addr2.to_string()); + + // Build valid msg1 frame + let mut msg1 = vec![0xAA; 114]; + msg1[0] = 0x01; + msg1[1] = 0x00; + msg1[2..4].copy_from_slice(&110u16.to_le_bytes()); + + // First send establishes connection + t1.send_async(&remote, &msg1).await.unwrap(); + let _ = timeout(Duration::from_secs(1), rx2.recv()).await; + + // Force-close the connection + t1.close_connection_async(&remote).await; + + // Second send should reconnect (connect-on-send) + t1.send_async(&remote, &msg1).await.unwrap(); + + let packet = timeout(Duration::from_secs(2), rx2.recv()) + .await + .expect("timeout") + .expect("channel closed"); + assert_eq!(packet.data, msg1); + + t1.stop_async().await.unwrap(); + t2.stop_async().await.unwrap(); + } +} diff --git a/src/transport/tcp/stream.rs b/src/transport/tcp/stream.rs new file mode 100644 index 0000000..5a15960 --- /dev/null +++ b/src/transport/tcp/stream.rs @@ -0,0 +1,361 @@ +//! FMP-Aware Stream Reader +//! +//! Recovers FIPS packet boundaries from a TCP byte stream using the +//! existing 4-byte FMP common prefix `[ver+phase:1][flags:1][payload_len:2 LE]`. +//! +//! This module is deliberately separate from the TCP transport so it can +//! be reused by the future Tor transport. + +use tokio::io::{AsyncRead, AsyncReadExt}; + +/// FMP phase values (low nibble of byte 0). +const PHASE_ESTABLISHED: u8 = 0x0; +const PHASE_MSG1: u8 = 0x1; +const PHASE_MSG2: u8 = 0x2; + +/// Size of the FMP common prefix. +const PREFIX_SIZE: usize = 4; + +/// Overhead for established frames: 12 bytes remaining header + 16 bytes AEAD tag. +/// The full established header is 16 bytes (PREFIX_SIZE + 12), so after reading +/// the 4-byte prefix, 12 more header bytes remain. Then payload_len bytes of +/// ciphertext, then 16 bytes of AEAD tag. +const ESTABLISHED_REMAINING_HEADER: usize = 12; +const AEAD_TAG_SIZE: usize = 16; + +/// Errors from the FMP stream reader. +#[derive(Debug)] +pub enum StreamError { + /// Unknown FMP phase byte — protocol error, close connection. + UnknownPhase(u8), + /// Payload length exceeds the connection's MTU — corrupted or malicious. + PayloadTooLarge { payload_len: u16, max_payload_len: u16 }, + /// Handshake packet has unexpected payload_len for its phase. + HandshakeSizeMismatch { phase: u8, expected: u16, got: u16 }, + /// I/O error (including EOF). + Io(std::io::Error), +} + +impl std::fmt::Display for StreamError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + StreamError::UnknownPhase(p) => write!(f, "unknown FMP phase: 0x{:02x}", p), + StreamError::PayloadTooLarge { payload_len, max_payload_len } => { + write!(f, "payload_len {} exceeds max {}", payload_len, max_payload_len) + } + StreamError::HandshakeSizeMismatch { phase, expected, got } => { + write!(f, "handshake phase 0x{:x}: expected payload_len {}, got {}", phase, expected, got) + } + StreamError::Io(e) => write!(f, "io: {}", e), + } + } +} + +impl std::error::Error for StreamError {} + +impl From for StreamError { + fn from(e: std::io::Error) -> Self { + StreamError::Io(e) + } +} + +/// Known wire sizes for handshake messages. +/// msg1: 4 (prefix) + 4 (sender_idx) + 106 (noise_msg1) = 114 bytes +/// msg2: 4 (prefix) + 4 (sender_idx) + 4 (receiver_idx) + 57 (noise_msg2) = 69 bytes +const MSG1_WIRE_SIZE: usize = 114; +const MSG2_WIRE_SIZE: usize = 69; + +/// Expected payload_len for msg1: sender_idx(4) + noise_msg1(106) = 110. +const MSG1_PAYLOAD_LEN: u16 = (MSG1_WIRE_SIZE - PREFIX_SIZE) as u16; + +/// Expected payload_len for msg2: sender_idx(4) + receiver_idx(4) + noise_msg2(57) = 65. +const MSG2_PAYLOAD_LEN: u16 = (MSG2_WIRE_SIZE - PREFIX_SIZE) as u16; + +/// Read one complete FMP packet from an async reader. +/// +/// Uses the 4-byte FMP common prefix to determine the total packet size, +/// then reads the remaining bytes. Returns the complete packet as a `Vec`. +/// +/// # Arguments +/// +/// * `reader` - Any async reader (typically an `OwnedReadHalf`) +/// * `mtu` - The connection's MTU for validation of established frame sizes +/// +/// # Errors +/// +/// * `UnknownPhase` — unrecognized phase nibble (protocol error) +/// * `PayloadTooLarge` — established frame exceeds MTU +/// * `HandshakeSizeMismatch` — handshake packet has wrong payload_len +/// * `Io` — underlying read error (including EOF) +pub async fn read_fmp_packet( + reader: &mut R, + mtu: u16, +) -> Result, StreamError> { + // Read the 4-byte FMP common prefix + let mut prefix = [0u8; PREFIX_SIZE]; + reader.read_exact(&mut prefix).await?; + + let phase = prefix[0] & 0x0F; + let payload_len = u16::from_le_bytes([prefix[2], prefix[3]]); + + // Compute remaining bytes based on phase + let remaining = match phase { + PHASE_ESTABLISHED => { + // Validate payload_len against MTU: + // total packet = 16 (header) + payload_len + 16 (tag) = payload_len + 32 + // max_payload_len = mtu - 32 + let max_payload_len = mtu.saturating_sub( + (ESTABLISHED_REMAINING_HEADER + PREFIX_SIZE + AEAD_TAG_SIZE) as u16, + ); + if payload_len > max_payload_len { + return Err(StreamError::PayloadTooLarge { + payload_len, + max_payload_len, + }); + } + // remaining = 12 (rest of header) + payload_len + 16 (AEAD tag) + ESTABLISHED_REMAINING_HEADER + payload_len as usize + AEAD_TAG_SIZE + } + PHASE_MSG1 => { + if payload_len != MSG1_PAYLOAD_LEN { + return Err(StreamError::HandshakeSizeMismatch { + phase, + expected: MSG1_PAYLOAD_LEN, + got: payload_len, + }); + } + payload_len as usize + } + PHASE_MSG2 => { + if payload_len != MSG2_PAYLOAD_LEN { + return Err(StreamError::HandshakeSizeMismatch { + phase, + expected: MSG2_PAYLOAD_LEN, + got: payload_len, + }); + } + payload_len as usize + } + _ => { + return Err(StreamError::UnknownPhase(phase)); + } + }; + + // Allocate buffer for the complete packet (prefix + remaining) + let total = PREFIX_SIZE + remaining; + let mut packet = vec![0u8; total]; + packet[..PREFIX_SIZE].copy_from_slice(&prefix); + reader.read_exact(&mut packet[PREFIX_SIZE..]).await?; + + Ok(packet) +} + +// ============================================================================ +// Tests +// ============================================================================ + +#[cfg(test)] +mod tests { + use super::*; + use std::io::Cursor; + + /// Build a minimal established frame with the given payload_len. + /// Layout: [ver+phase:1][flags:1][payload_len:2 LE][12 bytes header][payload_len bytes][16 bytes tag] + fn build_established_frame(payload_len: u16) -> Vec { + let total = PREFIX_SIZE + ESTABLISHED_REMAINING_HEADER + payload_len as usize + AEAD_TAG_SIZE; + let mut frame = vec![0u8; total]; + frame[0] = 0x00; // ver=0, phase=0 (established) + frame[1] = 0x00; // flags + frame[2..4].copy_from_slice(&payload_len.to_le_bytes()); + // Fill remaining with pattern for verification + for i in PREFIX_SIZE..total { + frame[i] = (i & 0xFF) as u8; + } + frame + } + + /// Build a msg1 frame (114 bytes total). + fn build_msg1_frame() -> Vec { + let mut frame = vec![0xAA; MSG1_WIRE_SIZE]; + frame[0] = 0x01; // ver=0, phase=1 + frame[1] = 0x00; // flags + frame[2..4].copy_from_slice(&MSG1_PAYLOAD_LEN.to_le_bytes()); + frame + } + + /// Build a msg2 frame (69 bytes total). + fn build_msg2_frame() -> Vec { + let mut frame = vec![0xBB; MSG2_WIRE_SIZE]; + frame[0] = 0x02; // ver=0, phase=2 + frame[1] = 0x00; // flags + frame[2..4].copy_from_slice(&MSG2_PAYLOAD_LEN.to_le_bytes()); + frame + } + + #[tokio::test] + async fn test_read_established_frame() { + let payload_len = 64u16; + let frame = build_established_frame(payload_len); + let expected = frame.clone(); + + let mut cursor = Cursor::new(frame); + let packet = read_fmp_packet(&mut cursor, 1400).await.unwrap(); + assert_eq!(packet, expected); + } + + #[tokio::test] + async fn test_read_msg1_frame() { + let frame = build_msg1_frame(); + let expected = frame.clone(); + + let mut cursor = Cursor::new(frame); + let packet = read_fmp_packet(&mut cursor, 1400).await.unwrap(); + assert_eq!(packet.len(), MSG1_WIRE_SIZE); + assert_eq!(packet, expected); + } + + #[tokio::test] + async fn test_read_msg2_frame() { + let frame = build_msg2_frame(); + let expected = frame.clone(); + + let mut cursor = Cursor::new(frame); + let packet = read_fmp_packet(&mut cursor, 1400).await.unwrap(); + assert_eq!(packet.len(), MSG2_WIRE_SIZE); + assert_eq!(packet, expected); + } + + #[tokio::test] + async fn test_read_multiple_packets() { + let mut data = Vec::new(); + let msg1 = build_msg1_frame(); + let est = build_established_frame(32); + let msg2 = build_msg2_frame(); + data.extend_from_slice(&msg1); + data.extend_from_slice(&est); + data.extend_from_slice(&msg2); + + let mut cursor = Cursor::new(data); + let p1 = read_fmp_packet(&mut cursor, 1400).await.unwrap(); + assert_eq!(p1.len(), MSG1_WIRE_SIZE); + + let p2 = read_fmp_packet(&mut cursor, 1400).await.unwrap(); + assert_eq!(p2, est); + + let p3 = read_fmp_packet(&mut cursor, 1400).await.unwrap(); + assert_eq!(p3.len(), MSG2_WIRE_SIZE); + } + + #[tokio::test] + async fn test_unknown_phase_error() { + let mut frame = vec![0u8; 100]; + frame[0] = 0x05; // unknown phase + frame[2..4].copy_from_slice(&10u16.to_le_bytes()); + + let mut cursor = Cursor::new(frame); + let err = read_fmp_packet(&mut cursor, 1400).await.unwrap_err(); + assert!(matches!(err, StreamError::UnknownPhase(0x5))); + } + + #[tokio::test] + async fn test_payload_too_large() { + // mtu=100, max_payload_len = 100 - 32 = 68 + let payload_len = 100u16; // exceeds max of 68 + let mut prefix = [0u8; 4]; + prefix[0] = 0x00; // established + prefix[2..4].copy_from_slice(&payload_len.to_le_bytes()); + + // Provide enough bytes for the reader to read prefix + let mut data = prefix.to_vec(); + data.extend_from_slice(&vec![0u8; 200]); // extra bytes + + let mut cursor = Cursor::new(data); + let err = read_fmp_packet(&mut cursor, 100).await.unwrap_err(); + assert!(matches!(err, StreamError::PayloadTooLarge { .. })); + } + + #[tokio::test] + async fn test_handshake_size_mismatch_msg1() { + let mut frame = vec![0u8; 200]; + frame[0] = 0x01; // msg1 + // Wrong payload_len (should be 110) + frame[2..4].copy_from_slice(&50u16.to_le_bytes()); + + let mut cursor = Cursor::new(frame); + let err = read_fmp_packet(&mut cursor, 1400).await.unwrap_err(); + assert!(matches!(err, StreamError::HandshakeSizeMismatch { phase: 0x1, .. })); + } + + #[tokio::test] + async fn test_handshake_size_mismatch_msg2() { + let mut frame = vec![0u8; 200]; + frame[0] = 0x02; // msg2 + // Wrong payload_len (should be 65) + frame[2..4].copy_from_slice(&50u16.to_le_bytes()); + + let mut cursor = Cursor::new(frame); + let err = read_fmp_packet(&mut cursor, 1400).await.unwrap_err(); + assert!(matches!(err, StreamError::HandshakeSizeMismatch { phase: 0x2, .. })); + } + + #[tokio::test] + async fn test_eof_on_prefix() { + // Only 2 bytes available (need 4 for prefix) + let data = vec![0u8; 2]; + let mut cursor = Cursor::new(data); + let err = read_fmp_packet(&mut cursor, 1400).await.unwrap_err(); + assert!(matches!(err, StreamError::Io(_))); + } + + #[tokio::test] + async fn test_eof_on_body() { + // Valid msg1 prefix but truncated body + let mut data = vec![0u8; 10]; // need 114 total + data[0] = 0x01; // msg1 + data[2..4].copy_from_slice(&MSG1_PAYLOAD_LEN.to_le_bytes()); + + let mut cursor = Cursor::new(data); + let err = read_fmp_packet(&mut cursor, 1400).await.unwrap_err(); + assert!(matches!(err, StreamError::Io(_))); + } + + #[tokio::test] + async fn test_zero_payload_established() { + // payload_len = 0 is valid (header-only encrypted frame with tag) + let frame = build_established_frame(0); + let expected_len = PREFIX_SIZE + ESTABLISHED_REMAINING_HEADER + AEAD_TAG_SIZE; + assert_eq!(frame.len(), expected_len); + + let mut cursor = Cursor::new(frame.clone()); + let packet = read_fmp_packet(&mut cursor, 1400).await.unwrap(); + assert_eq!(packet.len(), expected_len); + assert_eq!(packet, frame); + } + + #[tokio::test] + async fn test_max_payload_at_mtu_boundary() { + // mtu=1400, max_payload_len = 1400 - 32 = 1368 + let max_payload = 1400u16 - 32; + let frame = build_established_frame(max_payload); + + let mut cursor = Cursor::new(frame.clone()); + let packet = read_fmp_packet(&mut cursor, 1400).await.unwrap(); + assert_eq!(packet, frame); + } + + #[tokio::test] + async fn test_payload_one_over_mtu() { + // mtu=1400, max_payload_len = 1368, try 1369 + let over = 1400u16 - 32 + 1; + let mut prefix = [0u8; 4]; + prefix[0] = 0x00; // established + prefix[2..4].copy_from_slice(&over.to_le_bytes()); + + let mut data = prefix.to_vec(); + data.extend_from_slice(&vec![0u8; 2000]); + + let mut cursor = Cursor::new(data); + let err = read_fmp_packet(&mut cursor, 1400).await.unwrap_err(); + assert!(matches!(err, StreamError::PayloadTooLarge { .. })); + } +} diff --git a/testing/chaos/scenarios/tcp-chain.yaml b/testing/chaos/scenarios/tcp-chain.yaml new file mode 100644 index 0000000..6630a3c --- /dev/null +++ b/testing/chaos/scenarios/tcp-chain.yaml @@ -0,0 +1,40 @@ +# TCP chain: 4-node linear topology, all TCP transport +# +# Tests basic TCP transport connectivity, spanning tree convergence, +# and multi-hop routing over TCP links. +# +# Topology: +# +# n01 ---tcp--- n02 ---tcp--- n03 ---tcp--- n04 + +scenario: + name: "tcp-chain" + seed: 42 + duration_secs: 90 + +topology: + algorithm: explicit + num_nodes: 4 + default_transport: tcp + params: + adjacency: + - [n01, n02] + - [n02, n03] + - [n03, n04] + +netem: + enabled: true + default_policy: + delay_ms: [1, 5] + jitter_ms: [0, 1] + loss_pct: [0, 0.5] + +link_flaps: + enabled: false + +traffic: + enabled: false + +logging: + rust_log: "info" + output_dir: "./sim-results" diff --git a/testing/chaos/scenarios/tcp-mesh.yaml b/testing/chaos/scenarios/tcp-mesh.yaml new file mode 100644 index 0000000..0010a29 --- /dev/null +++ b/testing/chaos/scenarios/tcp-mesh.yaml @@ -0,0 +1,66 @@ +# Mixed transport mesh: 6 nodes, UDP + TCP edges +# +# Exercises both transports in a single mesh. UDP and TCP edges both +# use static peer config. Tests that the spanning tree converges +# across heterogeneous transports with netem and link flaps active. +# +# Topology: +# +# n01 ---udp--- n02 ---udp--- n03 +# | | | +# tcp tcp udp +# | | | +# n04 ---tcp--- n05 ---udp--- n06 +# +# UDP edges: n01-n02, n02-n03, n03-n06, n05-n06 +# TCP edges: n01-n04, n02-n05, n04-n05 + +scenario: + name: "tcp-mesh" + seed: 42 + duration_secs: 120 + +topology: + algorithm: explicit + num_nodes: 6 + params: + adjacency: + - [n01, n02, udp] + - [n02, n03, udp] + - [n03, n06, udp] + - [n05, n06, udp] + - [n01, n04, tcp] + - [n02, n05, tcp] + - [n04, n05, tcp] + +netem: + enabled: true + default_policy: + delay_ms: [1, 10] + jitter_ms: [0, 2] + loss_pct: [0, 1] + mutation: + interval_secs: {min: 20, max: 40} + fraction: 0.3 + policies: + normal: + delay_ms: [1, 10] + loss_pct: [0, 1] + degraded: + delay_ms: [30, 80] + jitter_ms: [5, 20] + loss_pct: [3, 8] + +link_flaps: + enabled: true + interval_secs: {min: 20, max: 40} + max_down_links: 1 + down_duration_secs: {min: 10, max: 20} + protect_connectivity: true + +traffic: + enabled: false + +logging: + rust_log: "info" + output_dir: "./sim-results" diff --git a/testing/chaos/scenarios/tcp-only.yaml b/testing/chaos/scenarios/tcp-only.yaml new file mode 100644 index 0000000..530a145 --- /dev/null +++ b/testing/chaos/scenarios/tcp-only.yaml @@ -0,0 +1,45 @@ +# TCP-only: 4-node ring, all TCP transport +# +# Tests pure-TCP mesh with netem impairment. Nodes have no UDP +# transport — only TCP with static peer config. +# +# Topology: +# +# n01 ---tcp--- n02 +# | | +# tcp tcp +# | | +# n04 ---tcp--- n03 + +scenario: + name: "tcp-only" + seed: 42 + duration_secs: 90 + +topology: + algorithm: explicit + num_nodes: 4 + default_transport: tcp + params: + adjacency: + - [n01, n02] + - [n02, n03] + - [n03, n04] + - [n04, n01] + +netem: + enabled: true + default_policy: + delay_ms: [1, 10] + jitter_ms: [0, 2] + loss_pct: [0, 1] + +link_flaps: + enabled: false + +traffic: + enabled: false + +logging: + rust_log: "info" + output_dir: "./sim-results" diff --git a/testing/chaos/sim/config_gen.py b/testing/chaos/sim/config_gen.py index 97297da..e2ade0e 100644 --- a/testing/chaos/sim/config_gen.py +++ b/testing/chaos/sim/config_gen.py @@ -31,6 +31,12 @@ def _load_template() -> str: return f.read() +_TRANSPORT_PORTS = { + "udp": 2121, + "tcp": 443, +} + + def generate_peers_block( topology: SimTopology, node_id: str, outbound_peers: list[str] ) -> str: @@ -38,6 +44,7 @@ def generate_peers_block( Only includes peers that this node is responsible for connecting to (outbound direction). The link is still bidirectional once established. + Transport type and port are determined per-edge from the topology. """ if not outbound_peers: return " []" @@ -45,11 +52,13 @@ def generate_peers_block( lines = [] for peer_id in sorted(outbound_peers): peer = topology.nodes[peer_id] + transport = topology.transport_for_edge(node_id, peer_id) + port = _TRANSPORT_PORTS.get(transport, 2121) lines.append(f' - npub: "{peer.npub}"') lines.append(f' alias: "{peer_id}"') lines.append(f" addresses:") - lines.append(f" - transport: udp") - lines.append(f' addr: "{peer.docker_ip}:2121"') + lines.append(f" - transport: {transport}") + lines.append(f' addr: "{peer.docker_ip}:{port}"') lines.append(f" connect_policy: auto_connect") return "\n".join(lines) @@ -85,6 +94,14 @@ def _inject_ethernet_transports(parsed: dict, eth_ifaces: list[str]): } +def _inject_tcp_transport(parsed: dict): + """Inject TCP transport config into a parsed FIPS config.""" + transports = parsed.setdefault("transports", {}) + transports["tcp"] = { + "bind_addr": "0.0.0.0:443", + } + + def generate_node_config( topology: SimTopology, node_id: str, @@ -103,34 +120,36 @@ def generate_node_config( config = config.replace("{{NSEC}}", node.nsec) config = config.replace("{{PEERS}}", peers_yaml) - # Inject Ethernet transport config if this node has Ethernet edges + # Determine which transports this node participates in eth_ifaces = topology.ethernet_interfaces(node_id) - has_udp_peers = bool(outbound_peers) or _has_inbound_udp_peers(topology, node_id) + has_tcp = bool(topology.tcp_peers(node_id)) + has_udp = _has_transport_peers(topology, node_id, "udp") - if eth_ifaces or not has_udp_peers: + # Inject non-UDP transport configs and handle pure-transport nodes + needs_yaml_rewrite = eth_ifaces or has_tcp or not has_udp or fips_overrides + + if needs_yaml_rewrite: parsed = yaml.safe_load(config) if fips_overrides: parsed = _deep_merge(parsed, fips_overrides) if eth_ifaces: _inject_ethernet_transports(parsed, eth_ifaces) - if not has_udp_peers and eth_ifaces: - # Pure-Ethernet node: remove UDP transport + if has_tcp: + _inject_tcp_transport(parsed) + if not has_udp: + # No UDP edges: remove UDP transport transports = parsed.get("transports", {}) transports.pop("udp", None) config = yaml.dump(parsed, default_flow_style=False, sort_keys=False) - elif fips_overrides: - parsed = yaml.safe_load(config) - merged = _deep_merge(parsed, fips_overrides) - config = yaml.dump(merged, default_flow_style=False, sort_keys=False) return config -def _has_inbound_udp_peers(topology: SimTopology, node_id: str) -> bool: - """Check if any other node has a UDP outbound edge to this node.""" +def _has_transport_peers(topology: SimTopology, node_id: str, transport: str) -> bool: + """Check if a node has any edges (inbound or outbound) using the given transport.""" for peer_id in topology.nodes[node_id].peers: edge = (min(node_id, peer_id), max(node_id, peer_id)) - if topology.edge_transport.get(edge, "udp") == "udp": + if topology.edge_transport.get(edge, "udp") == transport: return True return False diff --git a/testing/chaos/sim/links.py b/testing/chaos/sim/links.py index 283864d..577ded5 100644 --- a/testing/chaos/sim/links.py +++ b/testing/chaos/sim/links.py @@ -178,7 +178,7 @@ class LinkManager: docker_exec_quiet(container, cmd) return prev_args else: - # UDP: HTB class on eth0 + # IP-based (UDP/TCP): HTB class on eth0 dest_ip = self.topology.nodes[dst_node].docker_ip states = self.netem_mgr.states.get(container, {}) diff --git a/testing/chaos/sim/netem.py b/testing/chaos/sim/netem.py index a3ec00c..d2de330 100644 --- a/testing/chaos/sim/netem.py +++ b/testing/chaos/sim/netem.py @@ -151,18 +151,18 @@ class NetemManager: node = self.topology.nodes[node_id] container = self.topology.container_name(node_id) - # Split peers by transport type - udp_peers = {} + # Split peers by transport type: IP-based (UDP/TCP) vs Ethernet (veth) + ip_peers = {} eth_peers = [] for peer_id in sorted(node.peers): transport = self.topology.transport_for_edge(node_id, peer_id) if transport == "ethernet": eth_peers.append(peer_id) else: - udp_peers[peer_id] = self.topology.nodes[peer_id].docker_ip + ip_peers[peer_id] = self.topology.nodes[peer_id].docker_ip - # --- UDP peers: HTB + u32 on eth0 --- - if udp_peers: + # --- IP-based peers (UDP/TCP): HTB + u32 on eth0 --- + if ip_peers: cmds = [f"tc qdisc del dev {IFACE} root 2>/dev/null || true"] cmds.append( f"tc qdisc add dev {IFACE} root handle 1: htb default 99" @@ -173,7 +173,7 @@ class NetemManager: container_states = {} - for idx, (peer_id, dest_ip) in enumerate(udp_peers.items(), start=1): + for idx, (peer_id, dest_ip) in enumerate(ip_peers.items(), start=1): class_id = f"1:{idx}" netem_handle = f"{idx + 10}:" @@ -208,9 +208,9 @@ class NetemManager: result = docker_exec_quiet(container, full_cmd, timeout=30) if result is not None: log.info( - "Configured per-link netem on %s (%d UDP peers)", + "Configured per-link netem on %s (%d IP peers)", container, - len(udp_peers), + len(ip_peers), ) else: log.warning("Failed to configure netem on %s", container) @@ -256,7 +256,7 @@ class NetemManager: """ container = self.topology.container_name(node_id) - # Re-apply UDP netem (eth0 HTB + u32) + # Re-apply IP-based netem (eth0 HTB + u32) for UDP/TCP peers container_states = self.states.get(container) if container_states: cmds = [f"tc qdisc del dev {IFACE} root 2>/dev/null || true"] @@ -286,12 +286,12 @@ class NetemManager: result = docker_exec_quiet(container, full_cmd, timeout=30) if result is not None: log.info( - "Re-applied UDP netem on %s (%d peers)", + "Re-applied IP netem on %s (%d peers)", container, len(container_states), ) else: - log.warning("Failed to re-apply UDP netem on %s", container) + log.warning("Failed to re-apply IP netem on %s", container) # Re-apply Ethernet veth netem veth_states = self.veth_states.get(container) @@ -376,7 +376,7 @@ class NetemManager: state.params = params log.debug("Updated veth netem %s:%s -> %s", src, iface, params.to_tc_args()) else: - # UDP: HTB class-based netem on eth0 + # IP-based (UDP/TCP): HTB class-based netem on eth0 dest_ip = self.topology.nodes[dst].docker_ip states = self.states.get(container, {}) state = states.get(dest_ip) diff --git a/testing/chaos/sim/scenario.py b/testing/chaos/sim/scenario.py index b730dc1..8d24bf8 100644 --- a/testing/chaos/sim/scenario.py +++ b/testing/chaos/sim/scenario.py @@ -22,7 +22,7 @@ class Range: raise ValueError(f"{name}: min ({self.min}) must be >= 0") -VALID_TRANSPORTS = ("udp", "ethernet") +VALID_TRANSPORTS = ("udp", "ethernet", "tcp") @dataclass diff --git a/testing/chaos/sim/topology.py b/testing/chaos/sim/topology.py index 37f8943..b9a26d0 100644 --- a/testing/chaos/sim/topology.py +++ b/testing/chaos/sim/topology.py @@ -42,6 +42,26 @@ class SimTopology: """Check if any edges use Ethernet transport.""" return any(t == "ethernet" for t in self.edge_transport.values()) + def tcp_edges(self) -> list[tuple[str, str]]: + """Return all edges using TCP transport.""" + return [e for e, t in self.edge_transport.items() if t == "tcp"] + + def has_tcp(self) -> bool: + """Check if any edges use TCP transport.""" + return any(t == "tcp" for t in self.edge_transport.values()) + + def tcp_peers(self, node_id: str) -> list[str]: + """Return peer IDs connected to this node via TCP.""" + peers = [] + for (a, b), transport in self.edge_transport.items(): + if transport != "tcp": + continue + if a == node_id: + peers.append(b) + elif b == node_id: + peers.append(a) + return sorted(peers) + def ethernet_interfaces(self, node_id: str) -> list[str]: """Return the veth interface names for a node's Ethernet edges.""" ifaces = [] @@ -90,7 +110,7 @@ class SimTopology: return f"fips-node-{node_id}" def directed_outbound(self) -> dict[str, list[str]]: - """Assign each UDP edge to exactly one node for outbound connection. + """Assign each static-config edge to exactly one node for outbound connection. Returns a mapping from node_id to the list of peers that node should connect to (outbound only). Every edge appears in exactly @@ -98,27 +118,27 @@ class SimTopology: down, only A (the outbound owner) will attempt to reconnect. Ethernet edges are excluded — they use beacon discovery instead - of static peer configuration. + of static peer configuration. UDP and TCP edges use static config. Strategy: BFS spanning tree edges go parent→child. Non-tree edges go from the lower node ID to the higher. This guarantees every node is reachable via at least one inbound connection. """ - # Only consider UDP edges for static peer config - udp_edges = { + # Consider all edges that use static peer config (not Ethernet/discovery) + static_edges = { e for e in self.edges - if self.edge_transport.get(e, "udp") == "udp" + if self.edge_transport.get(e, "udp") != "ethernet" } outbound: dict[str, list[str]] = {nid: [] for nid in self.nodes} - # Build UDP-only adjacency for BFS - udp_adj: dict[str, list[str]] = {nid: [] for nid in self.nodes} - for a, b in udp_edges: - udp_adj[a].append(b) - udp_adj[b].append(a) + # Build static-config adjacency for BFS + static_adj: dict[str, list[str]] = {nid: [] for nid in self.nodes} + for a, b in static_edges: + static_adj[a].append(b) + static_adj[b].append(a) - # BFS spanning tree from first node (over UDP edges only) + # BFS spanning tree from first node (over static-config edges only) root = min(self.nodes) visited: set[str] = set() tree_edges: set[tuple[str, str]] = set() @@ -126,15 +146,15 @@ class SimTopology: visited.add(root) while queue: node = queue.popleft() - for peer in udp_adj[node]: + for peer in static_adj[node]: if peer not in visited: visited.add(peer) queue.append(peer) tree_edges.add((node, peer)) # parent → child outbound[node].append(peer) - # Non-tree UDP edges: lower ID → higher ID - for a, b in udp_edges: + # Non-tree static-config edges: lower ID → higher ID + for a, b in static_edges: if (a, b) not in tree_edges and (b, a) not in tree_edges: outbound[a].append(b) # a < b by _make_edge convention diff --git a/testing/static/configs/topologies/tcp-chain.yaml b/testing/static/configs/topologies/tcp-chain.yaml new file mode 100644 index 0000000..a36fa6d --- /dev/null +++ b/testing/static/configs/topologies/tcp-chain.yaml @@ -0,0 +1,29 @@ +# TCP Chain Topology Definition +# +# Three nodes with TCP links forming a linear chain. Tests basic TCP +# transport connectivity, spanning tree convergence, and multi-hop +# routing over TCP. All peer connections use TCP on port 443. +# +# default_transport: tcp tells the config generator to use TCP +# transport and port 443 instead of the default UDP/2121. + +default_transport: tcp + +nodes: + a: + nsec: "0102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f20" + npub: "npub1sjlh2c3x9w7kjsqg2ay080n2lff2uvt325vpan33ke34rn8l5jcqawh57m" + docker_ip: "172.20.0.10" + peers: [b] + + b: + nsec: "b102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1fb0" + npub: "npub1tdwa4vjrjl33pcjdpf2t4p027nl86xrx24g4d3avg4vwvayr3g8qhd84le" + docker_ip: "172.20.0.11" + peers: [a, c] + + c: + nsec: "c102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1fc0" + npub: "npub1cld9yay0u24davpu6c35l4vldrhzvaq66pcqtg9a0j2cnjrn9rtsxx2pe6" + docker_ip: "172.20.0.12" + peers: [b] diff --git a/testing/static/docker-compose.yml b/testing/static/docker-compose.yml index 5b4424a..fbc2604 100644 --- a/testing/static/docker-compose.yml +++ b/testing/static/docker-compose.yml @@ -203,3 +203,40 @@ services: networks: fips-net: ipv4_address: 172.20.0.14 + + # ── TCP chain topology (A-B-C) ─────────────────────────────── + tcp-a: + <<: *fips-common + profiles: ["tcp-chain"] + container_name: fips-node-a + hostname: node-a + volumes: + - ./resolv.conf:/etc/resolv.conf:ro + - ./generated-configs/tcp-chain/node-a.yaml:/etc/fips/fips.yaml:ro + networks: + fips-net: + ipv4_address: 172.20.0.10 + + tcp-b: + <<: *fips-common + profiles: ["tcp-chain"] + container_name: fips-node-b + hostname: node-b + volumes: + - ./resolv.conf:/etc/resolv.conf:ro + - ./generated-configs/tcp-chain/node-b.yaml:/etc/fips/fips.yaml:ro + networks: + fips-net: + ipv4_address: 172.20.0.11 + + tcp-c: + <<: *fips-common + profiles: ["tcp-chain"] + container_name: fips-node-c + hostname: node-c + volumes: + - ./resolv.conf:/etc/resolv.conf:ro + - ./generated-configs/tcp-chain/node-c.yaml:/etc/fips/fips.yaml:ro + networks: + fips-net: + ipv4_address: 172.20.0.12 diff --git a/testing/static/scripts/generate-configs.sh b/testing/static/scripts/generate-configs.sh index c0daf65..038c12c 100755 --- a/testing/static/scripts/generate-configs.sh +++ b/testing/static/scripts/generate-configs.sh @@ -77,19 +77,37 @@ resolve_keys() { fi } +# Get the default transport from topology file (defaults to "udp") +get_default_transport() { + local topology_file="$1" + local transport=$(grep "^default_transport:" "$topology_file" | head -1 | sed 's/.*: *\([a-z]*\).*/\1/') + echo "${transport:-udp}" +} + +# Get the port for a given transport type +transport_port() { + local transport="$1" + case "$transport" in + tcp) echo "443" ;; + *) echo "2121" ;; + esac +} + generate_peer_block() { local topology_file="$1" local peer_id="$2" local peer_npub="$(get_key RESOLVED_NPUB "$peer_id")" local peer_ip=$(get_node_attr "$topology_file" "$peer_id" "address") + local transport=$(get_default_transport "$topology_file") + local port=$(transport_port "$transport") cat < "$output_file" + + # Post-process: inject TCP transport config for TCP topologies + local transport + transport=$(get_default_transport "$topology_file") + if [ "$transport" = "tcp" ]; then + # Add TCP transport section and remove UDP transport + python3 -c " +import yaml, sys +with open('$output_file') as f: + cfg = yaml.safe_load(f) +cfg.setdefault('transports', {})['tcp'] = {'bind_addr': '0.0.0.0:443'} +cfg.get('transports', {}).pop('udp', None) +with open('$output_file', 'w') as f: + yaml.dump(cfg, f, default_flow_style=False, sort_keys=False) +" + fi } # Key storage for bash 3.2 compatibility (using prefixed variables instead of associative arrays)