From d29da442ac8722fc8dd8831003abc4e5cd342d42 Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Thu, 26 Feb 2026 00:03:14 +0000 Subject: [PATCH] Add Ethernet transport with beacon discovery MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implement raw Ethernet transport using AF_PACKET SOCK_DGRAM on Linux with EtherType 0x88B5 (IEEE experimental range) and 1-byte frame type prefix (0x00=data, 0x01=beacon). Transport implementation: - EthernetConfig with interface, ethertype, MTU, buffer sizes, and four independent discovery knobs (discovery, announce, auto_connect, accept_connections) - PacketSocket/AsyncPacketSocket wrappers with ioctl helpers for interface index, MAC address, and MTU queries - EthernetTransport with Transport trait impl, async start/stop/send, receive loop dispatching data frames and discovery beacons - Discovery beacons (34 bytes: type + version + x-only pubkey) with DiscoveryBuffer for peer accumulation and dedup - Atomic statistics counters (frames, bytes, errors, beacons) - Platform-gated with #[cfg(target_os = "linux")] Transport-layer discovery integration: - Promote auto_connect() and accept_connections() to Transport trait with default implementations and TransportHandle dispatch - Extract initiate_connection() so both static peer config and discovery auto-connect share the same handshake initiation path - Add poll_transport_discovery() to the tick handler to drain discovery buffers and auto-connect to discovered peers - Enforce accept_connections() in handle_msg1() — transports with accept_connections=false silently drop inbound handshakes Node integration: - create_transports() handles Ethernet named instances - resolve_ethernet_addr() parses "interface/mac" address format - transport_mtu() generalized for multi-transport operation Test harness: - VethPair RAII struct for veth pair lifecycle management - Three #[ignore] integration tests requiring root/CAP_NET_RAW: two-node handshake, data exchange, mixed transport coexistence - Chaos harness: transport-aware topology model, VethManager for veth pairs between Docker containers, Ethernet-aware config gen, netem split (HTB+u32 for UDP, root netem for veth), transport-aware link flaps and node churn with veth re-setup - Container entrypoint waits for configured Ethernet interfaces before starting FIPS (handles veth creation timing) - New scenarios: ethernet-only (4-node ring), ethernet-mesh (6-node mixed UDP+Ethernet with netem and link flaps) Documentation: - fips-transport-layer.md: Ethernet section, beacon discovery, WiFi compatibility, updated discovery state, trait surface additions, implementation status table - fips-configuration.md: Ethernet parameter table, named instances, peer address format, mixed UDP+Ethernet example, complete reference - fips-wire-formats.md: Ethernet frame type prefix note --- docs/design/fips-configuration.md | 91 ++- docs/design/fips-transport-layer.md | 89 ++- docs/design/fips-wire-formats.md | 6 + src/config/mod.rs | 2 +- src/config/transport.rs | 124 +++- src/node/handlers/handshake.rs | 8 + src/node/handlers/rx_loop.rs | 1 + src/node/lifecycle.rs | 340 +++++++---- src/node/mod.rs | 104 +++- src/node/tests/ethernet.rs | 210 +++++++ src/node/tests/mod.rs | 2 + src/node/tree.rs | 8 +- src/transport/ethernet/discovery.rs | 166 ++++++ src/transport/ethernet/mod.rs | 629 +++++++++++++++++++++ src/transport/ethernet/socket.rs | 388 +++++++++++++ src/transport/ethernet/stats.rs | 107 ++++ src/transport/mod.rs | 79 ++- testing/chaos/Dockerfile | 6 +- testing/chaos/entrypoint.sh | 51 ++ testing/chaos/scenarios/ethernet-mesh.yaml | 67 +++ testing/chaos/scenarios/ethernet-only.yaml | 46 ++ testing/chaos/sim/compose.py | 1 + testing/chaos/sim/config_gen.py | 57 +- testing/chaos/sim/links.py | 51 +- testing/chaos/sim/netem.py | 288 ++++++---- testing/chaos/sim/nodes.py | 11 +- testing/chaos/sim/runner.py | 20 +- testing/chaos/sim/scenario.py | 26 +- testing/chaos/sim/topology.py | 101 +++- testing/chaos/sim/veth.py | 193 +++++++ 30 files changed, 2967 insertions(+), 305 deletions(-) create mode 100644 src/node/tests/ethernet.rs create mode 100644 src/transport/ethernet/discovery.rs create mode 100644 src/transport/ethernet/mod.rs create mode 100644 src/transport/ethernet/socket.rs create mode 100644 src/transport/ethernet/stats.rs create mode 100755 testing/chaos/entrypoint.sh create mode 100644 testing/chaos/scenarios/ethernet-mesh.yaml create mode 100644 testing/chaos/scenarios/ethernet-only.yaml create mode 100644 testing/chaos/sim/veth.py diff --git a/docs/design/fips-configuration.md b/docs/design/fips-configuration.md index 4d3f00a..7e01b45 100644 --- a/docs/design/fips-configuration.md +++ b/docs/design/fips-configuration.md @@ -45,7 +45,7 @@ The configuration is organized into five top-level sections: node: # Node behavior, protocol parameters, and tuning tun: # TUN virtual interface dns: # DNS responder for .fips domain -transports: # Network transports (UDP, future: TCP, Tor) +transports: # Network transports (UDP, Ethernet, Bluetooth, Tor, ...) peers: # Static peer list ``` @@ -248,6 +248,43 @@ stale address mappings. | `transports.udp.recv_buf_size` | usize | `2097152` | UDP socket receive buffer size in bytes (2 MB). Linux kernel doubles the requested value internally. Host `net.core.rmem_max` must be >= this value. | | `transports.udp.send_buf_size` | usize | `2097152` | UDP socket send buffer size in bytes (2 MB). Host `net.core.wmem_max` must be >= this value. | +### Ethernet (`transports.ethernet.*`) + +Ethernet transport sends raw frames via AF_PACKET SOCK_DGRAM sockets. +Requires `CAP_NET_RAW` or running as root. Linux only. + +| Parameter | Type | Default | Description | +|-----------|------|---------|-------------| +| `interface` | string | *(required)* | Network interface name (e.g., `"eth0"`, `"enp3s0"`) | +| `ethertype` | u16 | `0x88B5` | IEEE EtherType (802 experimental range) | +| `mtu` | u16 | *(auto)* | Override MTU. Default: interface MTU minus 1 (for frame type prefix) | +| `recv_buf_size` | usize | `2097152` | Socket receive buffer size in bytes (2 MB) | +| `send_buf_size` | usize | `2097152` | Socket send buffer size in bytes (2 MB) | +| `discovery` | bool | `true` | Listen for discovery beacons from other nodes | +| `announce` | bool | `false` | Broadcast announcement beacons on the LAN | +| `auto_connect` | bool | `false` | Auto-connect to discovered peers | +| `accept_connections` | bool | `false` | Accept incoming connection attempts from discovered peers | +| `beacon_interval_secs` | u64 | `30` | Announcement beacon interval in seconds (minimum 10) | + +**Named instances.** Multiple Ethernet interfaces can be configured by +using named sub-keys instead of flat parameters: + +```yaml +transports: + ethernet: + lan: + interface: "eth0" + discovery: true + announce: true + backbone: + interface: "eth1" + announce: false +``` + +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. + ## Peers (`peers[]`) Static peer list. Each entry defines a peer to connect to. @@ -256,8 +293,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`) | -| `peers[].addresses[].addr` | string | *(required)* | Transport address (e.g., `"10.0.0.2:4000"`) | +| `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[].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) | @@ -295,6 +332,43 @@ peers: connect_policy: auto_connect ``` +### Mixed UDP + Ethernet Example + +A node bridging internet peers (UDP) and a local Ethernet segment with +beacon discovery: + +```yaml +node: + identity: + nsec: "0102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f20" + +tun: + enabled: true + +transports: + udp: + bind_addr: "0.0.0.0:4000" + mtu: 1472 + ethernet: + interface: "eth0" + discovery: true + announce: true + auto_connect: true + accept_connections: true + +peers: + - npub: "npub1tdwa4vjrjl33pcjdpf2t4p027nl86xrx24g4d3avg4vwvayr3g8qhd84le" + alias: "internet-peer" + addresses: + - transport: udp + addr: "203.0.113.5:4000" + connect_policy: auto_connect +``` + +Ethernet peers on the local segment are discovered automatically via +beacons — no static peer entries needed. Internet peers still require +explicit configuration. + All `node.*` parameters use their defaults. To override specific values, add only the relevant sections: @@ -398,6 +472,17 @@ transports: mtu: 1280 recv_buf_size: 2097152 # 2 MB (kernel doubles to 4 MB actual) send_buf_size: 2097152 # 2 MB + # ethernet: # uncomment to enable (requires CAP_NET_RAW) + # interface: "eth0" # required: network interface name + # ethertype: 0x88B5 # IEEE 802 experimental EtherType + # mtu: null # null = interface MTU - 1 (typically 1499) + # recv_buf_size: 2097152 # 2 MB + # send_buf_size: 2097152 # 2 MB + # discovery: true # listen for beacons + # announce: false # broadcast beacons + # auto_connect: false # connect to discovered peers + # accept_connections: false # accept inbound handshakes + # beacon_interval_secs: 30 # beacon interval (min 10) peers: # static peer list # - npub: "npub1..." diff --git a/docs/design/fips-transport-layer.md b/docs/design/fips-transport-layer.md index 4343917..250c568 100644 --- a/docs/design/fips-transport-layer.md +++ b/docs/design/fips-transport-layer.md @@ -242,6 +242,76 @@ Actual buffer sizes are logged at startup: UDP transport started local_addr=0.0.0.0:4000 recv_buf=4194304 send_buf=4194304 ``` +## Ethernet: The Local Network Transport + +For nodes on the same LAN segment, raw Ethernet provides a direct transport +without IP/UDP overhead — 28 bytes more FIPS payload per frame compared to +UDP (1500 vs 1472 MTU). + +- **No IP dependency**: Operates below the IP layer. Nodes on the same + Ethernet segment can communicate without IP addresses or routing + infrastructure +- **Broadcast discovery**: Nodes discover each other via periodic beacon + broadcasts on the shared medium, with no static peer configuration required +- **Higher MTU**: Standard Ethernet frames carry 1500 bytes of payload, + yielding an effective FIPS MTU of 1499 after the frame type prefix +- **Matches FIPS model**: Like UDP, Ethernet is connectionless and + unreliable — datagrams flow immediately to any MAC address on the segment + +### Implementation + +The Ethernet transport uses Linux AF_PACKET sockets in SOCK_DGRAM mode with +EtherType 0x88B5 (IEEE 802 experimental/local use range). SOCK_DGRAM mode +lets the kernel handle Ethernet header construction and parsing — the +transport deals only with payloads and MAC addresses. + +A 1-byte frame type prefix disambiguates data frames (0x00) from discovery +beacons (0x01) on the receive path. This costs one byte of MTU but allows +beacons and data to share the same EtherType and socket. + +| Property | Value | +| -------- | ----- | +| EtherType | 0x88B5 (IEEE 802 experimental) | +| Socket type | AF_PACKET SOCK_DGRAM | +| Frame type prefix | 0x00 = data, 0x01 = beacon | +| Effective MTU | Interface MTU - 1 (typically 1499) | +| Addressing | 6-byte MAC address | +| Platform | Linux only (`CAP_NET_RAW` required) | + +### Beacon Discovery + +Ethernet nodes discover peers via broadcast beacons sent to +ff:ff:ff:ff:ff:ff. Each beacon is a 34-byte frame containing the sender's +x-only public key. Receiving nodes extract the MAC source address from the +frame and the public key from the payload, then report the discovered peer +to FMP. + +Four configuration flags control discovery behavior: + +| Flag | Default | Description | +| ---- | ------- | ----------- | +| `discovery` | true | Listen for beacons from other nodes | +| `announce` | false | Broadcast beacons periodically | +| `auto_connect` | false | Initiate handshakes to discovered peers | +| `accept_connections` | false | Accept inbound handshake attempts | + +A typical discoverable node sets `announce: true`, `auto_connect: true`, and +`accept_connections: true`. A passive listener uses just `discovery: true` to +observe the network without announcing itself. + +### WiFi Compatibility + +WiFi interfaces in infrastructure (managed) mode work transparently for +unicast — the mac80211 subsystem handles frame translation between 802.11 +and 802.3. Broadcast beacon discovery is unreliable in managed mode because +access points commonly isolate clients from each other's broadcast traffic. + +Startup logging: + +```text +Ethernet transport started name=eth0 interface=eth0 mac=aa:bb:cc:dd:ee:ff mtu=1499 if_mtu=1500 +``` + ## Discovery Discovery determines that a FIPS-capable endpoint is reachable at a given @@ -255,7 +325,7 @@ FMP handles both cases uniformly: with discovery, it waits for events then initiates link setup; without discovery, it initiates link setup directly to configured addresses. -### Local/Medium Discovery *(future direction)* +### Local/Medium Discovery For transports where endpoints share a physical or link-layer medium — LAN broadcast, radio, BLE — discovery uses beacon and query mechanisms: @@ -289,6 +359,7 @@ is reachable at UDP 1.2.3.4:9735, then establishes the link over the UDP transport. Key properties: + - Identity is built in — Nostr events are signed, so discovery information is authenticated - Relay selection acts as scoping — which relays a node publishes to and @@ -298,10 +369,12 @@ Key properties: ### Current State -> **Implemented**: Peer addresses come from YAML configuration. The -> transport trait's `discover()` method exists but returns an empty list for -> UDP. Transport-level discovery (beacon/query, Nostr relay) is not yet -> implemented. +> **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. ## Transport Interface @@ -312,6 +385,7 @@ The transport interface defines what every transport driver must provide. ```text transport_id() → TransportId Unique identifier for this transport instance transport_type() → &TransportType Static metadata (name, connection-oriented, reliable) +name() → Option<&str> Instance name (for multi-instance transports) state() → TransportState Current lifecycle state mtu() → u16 Transport-wide default MTU link_mtu(addr) → u16 Per-link MTU (defaults to mtu()) @@ -319,6 +393,8 @@ start() → lifecycle Bring transport up (bind socket, o stop() → lifecycle Bring transport down send(addr, data) → delivery Send datagram to transport address discover() → Vec Report discovered FIPS endpoints (optional) +auto_connect() → bool Auto-connect discovered peers (default: false) +accept_connections() → bool Accept inbound handshakes (default: true) ``` ### Receive Path @@ -330,6 +406,7 @@ node's main event loop reads from the corresponding receiver, which aggregates datagrams from all active transports into a single stream. Each inbound datagram carries: + - **transport_id** — which transport it arrived on - **remote_addr** — the transport address of the sender - **data** — the raw datagram bytes @@ -373,7 +450,7 @@ transitions through `Starting` to `Up` (operational). `stop()` moves to | --------- | ------ | ----- | | UDP/IP | **Implemented** | Primary transport, async send/receive, configurable MTU | | TCP/IP | Future direction | Requires stream framing, TCP-over-TCP concern | -| Ethernet | Future direction | AF_PACKET raw frames, EtherType TBD | +| Ethernet | **Implemented** | AF_PACKET SOCK_DGRAM, EtherType 0x88B5, beacon discovery, Linux only | | WiFi | Future direction | Infrastructure mode = Ethernet driver | | Tor | Future direction | High latency, .onion addressing | | BLE | Future direction | ATT_MTU negotiation, per-link MTU | diff --git a/docs/design/fips-wire-formats.md b/docs/design/fips-wire-formats.md index db22c2b..c8a7c30 100644 --- a/docs/design/fips-wire-formats.md +++ b/docs/design/fips-wire-formats.md @@ -23,6 +23,12 @@ transports (TCP, WebSocket, Tor) must delineate FIPS packets within the byte stream; the common prefix `payload_len` field provides this framing directly. +**Ethernet frame type prefix.** The Ethernet transport prepends a 1-byte +frame type before the FMP payload: `0x00` for data frames and `0x01` for +beacon (discovery) frames. This byte is consumed by the transport layer +and is not visible to FMP. The effective MTU for FMP is the interface +MTU minus one byte (typically 1499). + ## Link-Layer Formats All FMP packets begin with a **4-byte common prefix** that identifies the diff --git a/src/config/mod.rs b/src/config/mod.rs index 36ffd9e..d8daa63 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::{TransportInstances, TransportsConfig, UdpConfig}; +pub use transport::{EthernetConfig, 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 fbca658..c3cd508 100644 --- a/src/config/transport.rs +++ b/src/config/transport.rs @@ -131,6 +131,113 @@ impl Default for TransportInstances { } } +/// Default Ethernet EtherType (IEEE 802 experimental). +const DEFAULT_ETHERNET_ETHERTYPE: u16 = 0x88B5; + +/// Default Ethernet receive buffer size (2 MB). +const DEFAULT_ETHERNET_RECV_BUF: usize = 2 * 1024 * 1024; + +/// Default Ethernet send buffer size (2 MB). +const DEFAULT_ETHERNET_SEND_BUF: usize = 2 * 1024 * 1024; + +/// Default beacon announcement interval in seconds. +const DEFAULT_BEACON_INTERVAL_SECS: u64 = 30; + +/// Minimum beacon announcement interval in seconds. +const MIN_BEACON_INTERVAL_SECS: u64 = 10; + +/// Ethernet transport instance configuration. +/// +/// EthernetConfig is always compiled (for config parsing on any platform), +/// but the transport runtime requires Linux (`#[cfg(target_os = "linux")]`). +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct EthernetConfig { + /// Network interface name (e.g., "eth0", "enp3s0"). Required. + pub interface: String, + + /// Custom EtherType (default: 0x88B5). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub ethertype: Option, + + /// MTU override. Defaults to the interface's MTU minus 1 (for frame type prefix). + /// Cannot exceed the interface's actual MTU. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub mtu: Option, + + /// Receive buffer size in bytes. Default: 2 MB. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub recv_buf_size: Option, + + /// Send buffer size in bytes. Default: 2 MB. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub send_buf_size: Option, + + /// Listen for discovery beacons from other nodes. Default: true. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub discovery: Option, + + /// Broadcast announcement beacons on the LAN. Default: false. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub announce: Option, + + /// Auto-connect to discovered peers. Default: false. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub auto_connect: Option, + + /// Accept incoming connection attempts. Default: false. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub accept_connections: Option, + + /// Announcement beacon interval in seconds. Default: 30. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub beacon_interval_secs: Option, +} + +impl EthernetConfig { + /// Get the EtherType, using default if not configured. + pub fn ethertype(&self) -> u16 { + self.ethertype.unwrap_or(DEFAULT_ETHERNET_ETHERTYPE) + } + + /// Get the receive buffer size, using default if not configured. + pub fn recv_buf_size(&self) -> usize { + self.recv_buf_size.unwrap_or(DEFAULT_ETHERNET_RECV_BUF) + } + + /// Get the send buffer size, using default if not configured. + pub fn send_buf_size(&self) -> usize { + self.send_buf_size.unwrap_or(DEFAULT_ETHERNET_SEND_BUF) + } + + /// Whether to listen for discovery beacons. Default: true. + pub fn discovery(&self) -> bool { + self.discovery.unwrap_or(true) + } + + /// Whether to broadcast announcement beacons. Default: false. + pub fn announce(&self) -> bool { + self.announce.unwrap_or(false) + } + + /// Whether to auto-connect to discovered peers. Default: false. + pub fn auto_connect(&self) -> bool { + self.auto_connect.unwrap_or(false) + } + + /// Whether to accept incoming connections. Default: false. + pub fn accept_connections(&self) -> bool { + self.accept_connections.unwrap_or(false) + } + + /// Get the beacon interval, clamped to minimum. Default: 30s. + pub fn beacon_interval_secs(&self) -> u64 { + self.beacon_interval_secs + .unwrap_or(DEFAULT_BEACON_INTERVAL_SECS) + .max(MIN_BEACON_INTERVAL_SECS) + } +} + /// Transports configuration section. /// /// Each transport type can have either a single instance (config directly @@ -141,12 +248,9 @@ pub struct TransportsConfig { #[serde(default, skip_serializing_if = "is_transport_empty")] pub udp: TransportInstances, - // Future transport types: - // #[serde(default, skip_serializing_if = "is_transport_empty")] - // pub tcp: TransportInstances, - // - // #[serde(default, skip_serializing_if = "is_transport_empty")] - // pub tor: TransportInstances, + /// Ethernet transport instances. + #[serde(default, skip_serializing_if = "is_transport_empty")] + pub ethernet: TransportInstances, } /// Helper for skip_serializing_if on TransportInstances. @@ -157,9 +261,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.tcp.is_empty() - // && self.tor.is_empty() + self.udp.is_empty() && self.ethernet.is_empty() } /// Merge another TransportsConfig into this one. @@ -169,6 +271,8 @@ impl TransportsConfig { if !other.udp.is_empty() { self.udp = other.udp; } - // Future: same for tcp, tor, etc. + if !other.ethernet.is_empty() { + self.ethernet = other.ethernet; + } } } diff --git a/src/node/handlers/handshake.rs b/src/node/handlers/handshake.rs index a06cabb..51dbe09 100644 --- a/src/node/handlers/handshake.rs +++ b/src/node/handlers/handshake.rs @@ -26,6 +26,14 @@ impl Node { return; } + // Check if this transport accepts inbound connections + if let Some(transport) = self.transports.get(&packet.transport_id) + && !transport.accept_connections() + { + self.msg1_rate_limiter.complete_handshake(); + return; + } + // Parse header let header = match Msg1Header::parse(&packet.data) { Some(h) => h, diff --git a/src/node/handlers/rx_loop.rs b/src/node/handlers/rx_loop.rs index a52ddaa..225bf15 100644 --- a/src/node/handlers/rx_loop.rs +++ b/src/node/handlers/rx_loop.rs @@ -117,6 +117,7 @@ impl Node { self.check_session_mmp_reports().await; self.check_link_heartbeats().await; self.purge_stale_lookups(now_ms); + self.poll_transport_discovery().await; } } } diff --git a/src/node/lifecycle.rs b/src/node/lifecycle.rs index 0a6c9dc..5323fa1 100644 --- a/src/node/lifecycle.rs +++ b/src/node/lifecycle.rs @@ -3,7 +3,7 @@ use super::{Node, NodeError, NodeState}; use crate::peer::PeerConnection; use crate::protocol::{Disconnect, DisconnectReason}; -use crate::transport::{packet_channel, Link, LinkDirection, TransportAddr}; +use crate::transport::{packet_channel, Link, LinkDirection, TransportAddr, TransportId}; use crate::upper::tun::{run_tun_reader, shutdown_tun_interface, TunDevice, TunState}; use crate::node::wire::build_msg1; use crate::{NodeAddr, PeerIdentity}; @@ -89,133 +89,49 @@ impl Node { // Try addresses in priority order until one works for addr in peer_config.addresses_by_priority() { - // Find a transport matching this address type - let transport_id = match self.find_transport_for_type(&addr.transport) { - Some(id) => id, - None => { - debug!( - transport = %addr.transport, - addr = %addr.addr, - "No operational transport for address type" - ); - continue; - } - }; - - // Allocate link ID and create link - let link_id = self.allocate_link_id(); - let remote_addr = TransportAddr::from_string(&addr.addr); - - // For UDP, links are immediately "connected" (connectionless) - // TODO: For connection-oriented transports, state would be Connecting - let link = 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); - - // Add reverse lookup for packet dispatch - self.addr_to_link - .insert((transport_id, remote_addr.clone()), link_id); - - // Create connection in handshake phase (outbound knows expected identity) - let current_time_ms = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .map(|d| d.as_millis() as u64) - .unwrap_or(0); - let mut connection = PeerConnection::outbound(link_id, peer_identity, current_time_ms); - - // Allocate a session index for this handshake - let our_index = match self.index_allocator.allocate() { - Ok(idx) => idx, - Err(e) => { - warn!( - npub = %peer_config.npub, - error = %e, - "Failed to allocate session index" - ); - // Clean up the link we just created - self.links.remove(&link_id); - self.addr_to_link.remove(&(transport_id, remote_addr)); - continue; - } - }; - - // Start the Noise handshake and get message 1 - let our_keypair = self.identity.keypair(); - let noise_msg1 = match connection.start_handshake(our_keypair, self.startup_epoch, current_time_ms) { - Ok(msg) => msg, - Err(e) => { - warn!( - npub = %peer_config.npub, - error = %e, - "Failed to start handshake" - ); - // Clean up the index and link - let _ = self.index_allocator.free(our_index); - self.links.remove(&link_id); - self.addr_to_link.remove(&(transport_id, remote_addr)); - continue; - } - }; - - // Set index and transport info on the connection - connection.set_our_index(our_index); - connection.set_transport_id(transport_id); - connection.set_source_addr(remote_addr.clone()); - - // Build wire format msg1: [0x01][sender_idx:4 LE][noise_msg1:82] - let wire_msg1 = build_msg1(our_index, &noise_msg1); - - debug!( - peer = %self.peer_display_name(&peer_node_addr), - transport = %addr.transport, - addr = %addr.addr, - link_id = %link_id, - our_index = %our_index, - "Peer connection initiated" - ); - - // Store msg1 for resend and schedule first resend - let resend_interval = self.config.node.rate_limit.handshake_resend_interval_ms; - connection.set_handshake_msg1(wire_msg1.clone(), current_time_ms + resend_interval); - - // Track in pending_outbound for msg2 dispatch - self.pending_outbound.insert((transport_id, our_index.as_u32()), link_id); - self.connections.insert(link_id, connection); - - // Send the wire format handshake message - if let Some(transport) = self.transports.get(&transport_id) { - match transport.send(&remote_addr, &wire_msg1).await { - Ok(bytes) => { - debug!( - link_id = %link_id, - our_index = %our_index, - bytes, - "Sent Noise handshake message 1 (wire format)" - ); - } + // For Ethernet addresses ("interface/mac"), find the transport + // instance matching the interface name and parse the MAC. + let (transport_id, remote_addr) = if addr.transport == "ethernet" { + match self.resolve_ethernet_addr(&addr.addr) { + Ok(result) => result, Err(e) => { - warn!( - link_id = %link_id, + debug!( + transport = %addr.transport, + addr = %addr.addr, error = %e, - "Failed to send handshake message" + "Failed to resolve Ethernet address" ); - // Mark connection as failed but don't remove it yet - // The event loop can handle retry logic - if let Some(conn) = self.connections.get_mut(&link_id) { - conn.mark_failed(); - } + continue; } } + } else { + // Find a transport matching this address type + let tid = match self.find_transport_for_type(&addr.transport) { + Some(id) => id, + None => { + debug!( + transport = %addr.transport, + addr = %addr.addr, + "No operational transport for address type" + ); + continue; + } + }; + (tid, TransportAddr::from_string(&addr.addr)) + }; + + match self.initiate_connection(transport_id, remote_addr, peer_identity).await { + Ok(()) => return Ok(()), + Err(e) => { + debug!( + npub = %peer_config.npub, + transport_id = %transport_id, + error = %e, + "Connection attempt failed, trying next address" + ); + continue; + } } - - // Successfully initiated connection via this address - return Ok(()); } // No address worked @@ -225,6 +141,186 @@ impl Node { ))) } + /// Initiate a connection to a peer on a specific transport and address. + /// + /// Allocates a link, starts the Noise IK handshake, sends msg1, and + /// registers the connection for msg2 dispatch. Used by both static peer + /// config and transport discovery auto-connect paths. + pub(super) async fn initiate_connection( + &mut self, + transport_id: TransportId, + remote_addr: TransportAddr, + peer_identity: PeerIdentity, + ) -> Result<(), NodeError> { + let peer_node_addr = *peer_identity.node_addr(); + + // 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), + ); + + self.links.insert(link_id, link); + + // Add reverse lookup for packet dispatch + self.addr_to_link + .insert((transport_id, remote_addr.clone()), link_id); + + // Create connection in handshake phase (outbound knows expected identity) + let current_time_ms = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0); + let mut connection = PeerConnection::outbound(link_id, peer_identity, current_time_ms); + + // Allocate a session index for this handshake + let our_index = match self.index_allocator.allocate() { + Ok(idx) => idx, + Err(e) => { + // Clean up the link we just created + self.links.remove(&link_id); + self.addr_to_link.remove(&(transport_id, remote_addr)); + return Err(NodeError::IndexAllocationFailed(e.to_string())); + } + }; + + // Start the Noise handshake and get message 1 + let our_keypair = self.identity.keypair(); + let noise_msg1 = match connection.start_handshake(our_keypair, self.startup_epoch, current_time_ms) { + Ok(msg) => msg, + Err(e) => { + // Clean up the index and link + let _ = self.index_allocator.free(our_index); + self.links.remove(&link_id); + self.addr_to_link.remove(&(transport_id, remote_addr)); + return Err(NodeError::HandshakeFailed(e.to_string())); + } + }; + + // Set index and transport info on the connection + connection.set_our_index(our_index); + connection.set_transport_id(transport_id); + connection.set_source_addr(remote_addr.clone()); + + // Build wire format msg1: [0x01][sender_idx:4 LE][noise_msg1:82] + let wire_msg1 = build_msg1(our_index, &noise_msg1); + + debug!( + peer = %self.peer_display_name(&peer_node_addr), + transport_id = %transport_id, + remote_addr = %remote_addr, + link_id = %link_id, + our_index = %our_index, + "Connection initiated" + ); + + // Store msg1 for resend and schedule first resend + let resend_interval = self.config.node.rate_limit.handshake_resend_interval_ms; + connection.set_handshake_msg1(wire_msg1.clone(), current_time_ms + resend_interval); + + // Track in pending_outbound for msg2 dispatch + self.pending_outbound.insert((transport_id, our_index.as_u32()), link_id); + self.connections.insert(link_id, connection); + + // Send the wire format handshake message + if let Some(transport) = self.transports.get(&transport_id) { + match transport.send(&remote_addr, &wire_msg1).await { + Ok(bytes) => { + debug!( + link_id = %link_id, + our_index = %our_index, + bytes, + "Sent Noise handshake message 1 (wire format)" + ); + } + Err(e) => { + warn!( + link_id = %link_id, + error = %e, + "Failed to send handshake message" + ); + // Mark connection as failed but don't remove it yet + // The event loop can handle retry logic + if let Some(conn) = self.connections.get_mut(&link_id) { + conn.mark_failed(); + } + } + } + } + + Ok(()) + } + + /// Poll all transports for discovered peers and auto-connect. + /// + /// Called from the tick handler. Iterates operational transports, + /// drains their discovery buffers, and initiates connections to + /// newly discovered peers (if auto_connect is enabled). + pub(super) async fn poll_transport_discovery(&mut self) { + // Collect discoveries first to avoid borrow conflict with self + let mut to_connect = Vec::new(); + + for (transport_id, transport) in &self.transports { + if !transport.is_operational() { + continue; + } + if !transport.auto_connect() { + // Still drain the buffer so it doesn't grow unbounded + let _ = transport.discover(); + continue; + } + let discovered = match transport.discover() { + Ok(peers) => peers, + Err(_) => continue, + }; + for peer in discovered { + let pubkey = match peer.pubkey_hint { + Some(pk) => pk, + None => continue, + }; + let identity = PeerIdentity::from_pubkey(pubkey); + let node_addr = *identity.node_addr(); + + // Skip self + if node_addr == *self.identity.node_addr() { + continue; + } + // Skip if already connected + if self.peers.contains_key(&node_addr) { + continue; + } + // Skip if connection already in progress + let connecting = self.connections.values().any(|c| { + c.expected_identity() + .map(|id| id.node_addr() == &node_addr) + .unwrap_or(false) + }); + if connecting { + continue; + } + + to_connect.push((*transport_id, peer.addr, identity)); + } + } + + for (transport_id, remote_addr, identity) in to_connect { + info!( + peer = %self.peer_display_name(identity.node_addr()), + transport_id = %transport_id, + remote_addr = %remote_addr, + "Auto-connecting to discovered peer" + ); + if let Err(e) = self.initiate_connection(transport_id, remote_addr, identity).await { + warn!(error = %e, "Failed to auto-connect to discovered peer"); + } + } + } + // === State Transitions === /// Start the node. diff --git a/src/node/mod.rs b/src/node/mod.rs index 5b40fdf..2eceb73 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -28,6 +28,8 @@ use crate::transport::{ Link, LinkId, PacketRx, PacketTx, TransportAddr, TransportError, TransportHandle, TransportId, }; use crate::transport::udp::UdpTransport; +#[cfg(target_os = "linux")] +use crate::transport::ethernet::EthernetTransport; use crate::tree::TreeState; use crate::upper::icmp_rate_limit::IcmpRateLimiter; use crate::upper::tun::{TunError, TunOutboundRx, TunState, TunTx}; @@ -107,6 +109,12 @@ pub enum NodeError { #[error("TUN error: {0}")] Tun(#[from] TunError), + + #[error("index allocation failed: {0}")] + IndexAllocationFailed(String), + + #[error("handshake failed: {0}")] + HandshakeFailed(String), } /// Node operational state. @@ -570,8 +578,25 @@ impl Node { transports.push(TransportHandle::Udp(udp)); } - // Future transports follow same pattern: - // for (name, tcp_config) in self.config.transports.tcp.iter() { ... } + // Create Ethernet transport instances + #[cfg(target_os = "linux")] + { + let eth_instances: Vec<_> = self + .config + .transports + .ethernet + .iter() + .map(|(name, config)| (name.map(|s| s.to_string()), config.clone())) + .collect(); + + let xonly = self.identity.pubkey(); + for (name, eth_config) in eth_instances { + let transport_id = self.allocate_transport_id(); + let mut eth = EthernetTransport::new(transport_id, name, eth_config, packet_tx.clone()); + eth.set_local_pubkey(xonly); + transports.push(TransportHandle::Ethernet(eth)); + } + } transports } @@ -586,6 +611,55 @@ impl Node { .map(|(id, _)| *id) } + /// Resolve an Ethernet peer address ("interface/mac") to a transport ID + /// and binary TransportAddr. + /// + /// Finds the Ethernet transport instance bound to the named interface + /// and parses the MAC portion into a 6-byte TransportAddr. + fn resolve_ethernet_addr( + &self, + addr_str: &str, + ) -> Result<(TransportId, TransportAddr), NodeError> { + let (iface, mac_str) = addr_str.split_once('/').ok_or_else(|| { + NodeError::NoTransportForType(format!( + "invalid Ethernet address format '{}': expected 'interface/mac'", + addr_str + )) + })?; + + // Find the Ethernet transport bound to this interface + let transport_id = self + .transports + .iter() + .find(|(_, handle)| { + handle.transport_type().name == "ethernet" + && handle.is_operational() + && handle.interface_name() == Some(iface) + }) + .map(|(id, _)| *id) + .ok_or_else(|| { + NodeError::NoTransportForType(format!( + "no operational Ethernet transport for interface '{}'", + iface + )) + })?; + + // Parse the MAC address + #[cfg(target_os = "linux")] + let mac = crate::transport::ethernet::parse_mac_string(mac_str).map_err(|e| { + NodeError::NoTransportForType(format!("invalid MAC in '{}': {}", addr_str, e)) + })?; + #[cfg(not(target_os = "linux"))] + let mac: [u8; 6] = { + let _ = mac_str; + return Err(NodeError::NoTransportForType( + "Ethernet transport not available on this platform".into(), + )); + }; + + Ok((transport_id, TransportAddr::from_bytes(&mac))) + } + // === Identity Accessors === /// Get this node's identity. @@ -640,18 +714,24 @@ impl Node { crate::upper::icmp::effective_ipv6_mtu(self.transport_mtu()) } - /// Get the transport MTU from configuration. + /// Get the transport MTU for a specific transport. /// - /// Returns the MTU of the first configured UDP transport, or 1280 - /// (IPv6 minimum) as fallback. + /// When called without a specific transport context, returns the MTU + /// of the first operational transport, or 1280 (IPv6 minimum) as + /// fallback. This is used for initial TUN configuration where a + /// specific transport isn't yet known. pub fn transport_mtu(&self) -> u16 { - self.config - .transports - .udp - .iter() - .next() - .map(|(_, cfg)| cfg.mtu()) - .unwrap_or(1280) + // Prefer the MTU from the first operational transport + for handle in self.transports.values() { + if handle.is_operational() { + return handle.mtu(); + } + } + // Fallback to config: try UDP first, then Ethernet + if let Some((_, cfg)) = self.config.transports.udp.iter().next() { + return cfg.mtu(); + } + 1280 } // === State === diff --git a/src/node/tests/ethernet.rs b/src/node/tests/ethernet.rs new file mode 100644 index 0000000..5e2edf8 --- /dev/null +++ b/src/node/tests/ethernet.rs @@ -0,0 +1,210 @@ +//! Ethernet transport integration tests. +//! +//! Tests that the Ethernet transport works end-to-end using veth pairs. +//! All tests require root or CAP_NET_RAW and are marked `#[ignore]`. + +use super::*; +use crate::config::EthernetConfig; +use crate::transport::ethernet::EthernetTransport; +use crate::transport::{packet_channel, TransportAddr, TransportHandle, TransportId}; +use spanning_tree::{cleanup_nodes, drain_all_packets, initiate_handshake, TestNode}; + +use std::process::Command; +use std::sync::atomic::{AtomicU32, Ordering}; + +/// Atomic counter for unique veth names across tests. +static VETH_COUNTER: AtomicU32 = AtomicU32::new(0); + +/// RAII wrapper for a veth pair. +/// +/// Creates a pair of connected virtual Ethernet interfaces. Destroying +/// one end automatically destroys the other. +struct VethPair { + name_a: String, + name_b: String, +} + +impl VethPair { + /// Create a new veth pair with unique interface names. + /// + /// Names are kept under 15 chars (IFNAMSIZ limit). Format: `ftXXa`/`ftXXb` + /// where XX is an atomic counter combined with PID for cross-process uniqueness. + fn create() -> Self { + let id = VETH_COUNTER.fetch_add(1, Ordering::Relaxed); + let pid = std::process::id() % 10000; + let name_a = format!("ft{}{}a", pid, id); + let name_b = format!("ft{}{}b", pid, id); + + assert!(name_a.len() <= 15, "veth name too long: {}", name_a); + assert!(name_b.len() <= 15, "veth name too long: {}", name_b); + + // Create veth pair + let status = Command::new("ip") + .args(["link", "add", &name_a, "type", "veth", "peer", "name", &name_b]) + .status() + .expect("failed to run 'ip link add'"); + assert!(status.success(), "failed to create veth pair"); + + // Bring both ends up + let status = Command::new("ip") + .args(["link", "set", &name_a, "up"]) + .status() + .expect("failed to run 'ip link set up'"); + assert!(status.success(), "failed to bring up {}", name_a); + + let status = Command::new("ip") + .args(["link", "set", &name_b, "up"]) + .status() + .expect("failed to run 'ip link set up'"); + assert!(status.success(), "failed to bring up {}", name_b); + + VethPair { name_a, name_b } + } +} + +impl Drop for VethPair { + fn drop(&mut self) { + // Deleting one end destroys both + let _ = Command::new("ip") + .args(["link", "delete", &self.name_a]) + .status(); + } +} + +/// Create a test node with a live Ethernet transport on the given interface. +/// +/// Parallel to `make_test_node()` in spanning_tree.rs but uses +/// EthernetTransport instead of UDP. +async fn make_test_node_ethernet(interface: &str) -> TestNode { + let mut node = make_node(); + let transport_id = TransportId::new(1); + + let config = EthernetConfig { + interface: interface.to_string(), + discovery: Some(false), + announce: Some(false), + accept_connections: Some(true), + ..Default::default() + }; + + let (packet_tx, packet_rx) = packet_channel(256); + let mut transport = EthernetTransport::new(transport_id, None, config, packet_tx); + transport.start_async().await.unwrap(); + + let mac = transport.local_mac().expect("transport should have MAC after start"); + let addr = TransportAddr::from_bytes(&mac); + + node.transports + .insert(transport_id, TransportHandle::Ethernet(transport)); + + TestNode { + node, + transport_id, + packet_rx, + addr, + } +} + +/// Two nodes on a veth pair complete a Noise handshake and establish peering. +#[tokio::test] +#[ignore] // Requires root or CAP_NET_RAW +async fn test_ethernet_two_node_handshake() { + let veth = VethPair::create(); + + let mut nodes = vec![ + make_test_node_ethernet(&veth.name_a).await, + make_test_node_ethernet(&veth.name_b).await, + ]; + + // Initiate handshake from node 0 to node 1 + initiate_handshake(&mut nodes, 0, 1).await; + + // Drain all packets (handshake + tree announce) + let total = drain_all_packets(&mut nodes, false).await; + assert!(total > 0, "should have processed packets"); + + // Verify bidirectional peering + let addr_0 = *nodes[0].node.node_addr(); + let addr_1 = *nodes[1].node.node_addr(); + assert!( + nodes[0].node.get_peer(&addr_1).is_some(), + "node 0 should have node 1 as peer" + ); + assert!( + nodes[1].node.get_peer(&addr_0).is_some(), + "node 1 should have node 0 as peer" + ); + + cleanup_nodes(&mut nodes).await; +} + +/// Two Ethernet nodes converge to a correct spanning tree (2-node tree). +#[tokio::test] +#[ignore] // Requires root or CAP_NET_RAW +async fn test_ethernet_data_exchange() { + use spanning_tree::verify_tree_convergence; + + let veth = VethPair::create(); + + let mut nodes = vec![ + make_test_node_ethernet(&veth.name_a).await, + make_test_node_ethernet(&veth.name_b).await, + ]; + + initiate_handshake(&mut nodes, 0, 1).await; + let total = drain_all_packets(&mut nodes, false).await; + assert!(total > 0); + + // Verify spanning tree convergence + verify_tree_convergence(&nodes); + + // The root should be the node with the smallest NodeAddr + let expected_root = std::cmp::min(*nodes[0].node.node_addr(), *nodes[1].node.node_addr()); + assert_eq!(*nodes[0].node.tree_state().root(), expected_root); + assert_eq!(*nodes[1].node.tree_state().root(), expected_root); + + cleanup_nodes(&mut nodes).await; +} + +/// Mixed transport: 2 Ethernet nodes + 2 UDP nodes coexist. +/// +/// Each transport forms its own connected component. Validates that +/// `process_available_packets()` handles heterogeneous transport types. +#[tokio::test] +#[ignore] // Requires root or CAP_NET_RAW +async fn test_mixed_transport_coexistence() { + use spanning_tree::{make_test_node, verify_tree_convergence_components}; + + let veth = VethPair::create(); + + // Create 2 Ethernet nodes and 2 UDP nodes + let eth_0 = make_test_node_ethernet(&veth.name_a).await; + let eth_1 = make_test_node_ethernet(&veth.name_b).await; + let udp_0 = make_test_node().await; + let udp_1 = make_test_node().await; + + let mut nodes = vec![eth_0, eth_1, udp_0, udp_1]; + + // Handshake within each component + initiate_handshake(&mut nodes, 0, 1).await; // Ethernet pair + initiate_handshake(&mut nodes, 2, 3).await; // UDP pair + + // Drain all packets across both transports + let total = drain_all_packets(&mut nodes, false).await; + assert!(total > 0); + + // Verify each component converges independently + verify_tree_convergence_components(&nodes, &[vec![0, 1], vec![2, 3]]); + + // Ethernet component has its own root + let eth_root = std::cmp::min(*nodes[0].node.node_addr(), *nodes[1].node.node_addr()); + assert_eq!(*nodes[0].node.tree_state().root(), eth_root); + assert_eq!(*nodes[1].node.tree_state().root(), eth_root); + + // UDP component has its own root + let udp_root = std::cmp::min(*nodes[2].node.node_addr(), *nodes[3].node.node_addr()); + assert_eq!(*nodes[2].node.tree_state().root(), udp_root); + assert_eq!(*nodes[3].node.tree_state().root(), udp_root); + + cleanup_nodes(&mut nodes).await; +} diff --git a/src/node/tests/mod.rs b/src/node/tests/mod.rs index 65d72df..51443f1 100644 --- a/src/node/tests/mod.rs +++ b/src/node/tests/mod.rs @@ -7,6 +7,8 @@ use std::time::Duration; mod bloom; mod disconnect; mod discovery; +#[cfg(target_os = "linux")] +mod ethernet; mod forwarding; mod handshake; mod routing; diff --git a/src/node/tree.rs b/src/node/tree.rs index 8a5ac63..7e4abb7 100644 --- a/src/node/tree.rs +++ b/src/node/tree.rs @@ -302,10 +302,10 @@ impl Node { let now = std::time::Instant::now(); let interval = std::time::Duration::from_secs(interval_secs); - if let Some(last) = self.last_parent_reeval { - if now.duration_since(last) < interval { - return; - } + if let Some(last) = self.last_parent_reeval + && now.duration_since(last) < interval + { + return; } self.last_parent_reeval = Some(now); diff --git a/src/transport/ethernet/discovery.rs b/src/transport/ethernet/discovery.rs new file mode 100644 index 0000000..7130b53 --- /dev/null +++ b/src/transport/ethernet/discovery.rs @@ -0,0 +1,166 @@ +//! Ethernet LAN discovery via broadcast beacons. +//! +//! Beacon format (34 bytes total): +//! - `0x01` (1 byte): frame type = discovery announcement +//! - `0x01` (1 byte): discovery protocol version +//! - x-only public key (32 bytes): node's Nostr identity + +use crate::transport::{DiscoveredPeer, TransportAddr, TransportId}; +use secp256k1::XOnlyPublicKey; +use std::sync::Mutex; + +/// Discovery protocol version. +pub const DISCOVERY_VERSION: u8 = 0x01; + +/// Frame type prefix for discovery announcement beacons. +pub const FRAME_TYPE_BEACON: u8 = 0x01; + +/// Frame type prefix for FIPS data frames. +pub const FRAME_TYPE_DATA: u8 = 0x00; + +/// Total beacon payload size: type(1) + version(1) + pubkey(32). +pub const BEACON_SIZE: usize = 34; + +/// Build a discovery announcement beacon payload. +pub fn build_beacon(pubkey: &XOnlyPublicKey) -> [u8; BEACON_SIZE] { + let mut buf = [0u8; BEACON_SIZE]; + buf[0] = FRAME_TYPE_BEACON; + buf[1] = DISCOVERY_VERSION; + buf[2..BEACON_SIZE].copy_from_slice(&pubkey.serialize()); + buf +} + +/// Parse a discovery announcement beacon payload. +/// +/// Returns the sender's public key, or None if the payload is invalid. +pub fn parse_beacon(data: &[u8]) -> Option { + if data.len() < BEACON_SIZE { + return None; + } + if data[0] != FRAME_TYPE_BEACON { + return None; + } + if data[1] != DISCOVERY_VERSION { + return None; + } + XOnlyPublicKey::from_slice(&data[2..34]).ok() +} + +/// Buffer for discovered peers, drained by `discover()`. +pub struct DiscoveryBuffer { + transport_id: TransportId, + peers: Mutex>, +} + +impl DiscoveryBuffer { + /// Create a new empty discovery buffer. + pub fn new(transport_id: TransportId) -> Self { + Self { + transport_id, + peers: Mutex::new(Vec::new()), + } + } + + /// Add a discovered peer from a received beacon. + pub fn add_peer(&self, src_mac: [u8; 6], pubkey: XOnlyPublicKey) { + let addr = TransportAddr::from_bytes(&src_mac); + let peer = DiscoveredPeer::with_hint(self.transport_id, addr, pubkey); + let mut peers = self.peers.lock().unwrap(); + // Deduplicate by MAC address — keep the latest + peers.retain(|p| p.addr.as_bytes() != src_mac); + peers.push(peer); + } + + /// Drain all discovered peers since the last call. + pub fn take(&self) -> Vec { + let mut peers = self.peers.lock().unwrap(); + std::mem::take(&mut *peers) + } +} + +// ============================================================================ +// Tests +// ============================================================================ + +#[cfg(test)] +mod tests { + use super::*; + use secp256k1::{Secp256k1, SecretKey}; + + fn test_pubkey() -> XOnlyPublicKey { + let secp = Secp256k1::new(); + let sk = SecretKey::from_slice(&[0x42; 32]).unwrap(); + let (xonly, _) = sk.public_key(&secp).x_only_public_key(); + xonly + } + + #[test] + fn test_build_parse_beacon() { + let pubkey = test_pubkey(); + let beacon = build_beacon(&pubkey); + + assert_eq!(beacon.len(), BEACON_SIZE); + assert_eq!(beacon[0], FRAME_TYPE_BEACON); + assert_eq!(beacon[1], DISCOVERY_VERSION); + + let parsed = parse_beacon(&beacon).unwrap(); + assert_eq!(parsed, pubkey); + } + + #[test] + fn test_parse_beacon_too_short() { + assert!(parse_beacon(&[0x01, 0x01]).is_none()); + assert!(parse_beacon(&[]).is_none()); + } + + #[test] + fn test_parse_beacon_wrong_type() { + let mut beacon = build_beacon(&test_pubkey()); + beacon[0] = 0x00; // data frame, not beacon + assert!(parse_beacon(&beacon).is_none()); + } + + #[test] + fn test_parse_beacon_wrong_version() { + let mut beacon = build_beacon(&test_pubkey()); + beacon[1] = 0xFF; + assert!(parse_beacon(&beacon).is_none()); + } + + #[test] + fn test_frame_type_prefix() { + assert_eq!(FRAME_TYPE_DATA, 0x00); + assert_eq!(FRAME_TYPE_BEACON, 0x01); + } + + #[test] + fn test_discovery_buffer() { + let buffer = DiscoveryBuffer::new(TransportId::new(1)); + let pubkey = test_pubkey(); + let mac = [0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xff]; + + buffer.add_peer(mac, pubkey); + + let peers = buffer.take(); + assert_eq!(peers.len(), 1); + assert_eq!(peers[0].addr.as_bytes(), &mac); + assert_eq!(peers[0].pubkey_hint, Some(pubkey)); + + // Second take should be empty + let peers = buffer.take(); + assert!(peers.is_empty()); + } + + #[test] + fn test_discovery_buffer_dedup() { + let buffer = DiscoveryBuffer::new(TransportId::new(1)); + let pubkey = test_pubkey(); + let mac = [0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xff]; + + buffer.add_peer(mac, pubkey); + buffer.add_peer(mac, pubkey); // same MAC again + + let peers = buffer.take(); + assert_eq!(peers.len(), 1); + } +} diff --git a/src/transport/ethernet/mod.rs b/src/transport/ethernet/mod.rs new file mode 100644 index 0000000..5b0d180 --- /dev/null +++ b/src/transport/ethernet/mod.rs @@ -0,0 +1,629 @@ +//! Ethernet Transport Implementation +//! +//! Provides raw Ethernet transport for FIPS peer communication using +//! AF_PACKET sockets with SOCK_DGRAM. Works on wired Ethernet and WiFi +//! interfaces (kernel mac80211 abstracts 802.11 transparently). + +pub mod discovery; +pub mod socket; +pub mod stats; + +use super::{ + DiscoveredPeer, PacketTx, ReceivedPacket, Transport, TransportAddr, TransportError, + TransportId, TransportState, TransportType, +}; +use crate::config::EthernetConfig; +use discovery::{ + build_beacon, parse_beacon, DiscoveryBuffer, FRAME_TYPE_BEACON, FRAME_TYPE_DATA, +}; +use socket::{AsyncPacketSocket, PacketSocket, ETHERNET_BROADCAST}; +use stats::EthernetStats; + +use secp256k1::XOnlyPublicKey; +use std::sync::Arc; +use tokio::task::JoinHandle; +use tracing::{debug, info, trace, warn}; + +/// Ethernet transport for FIPS. +/// +/// Uses AF_PACKET with SOCK_DGRAM for raw Ethernet frame I/O. A single +/// socket per interface serves all peers; links are virtual tuples of +/// (transport_id, remote_mac). +pub struct EthernetTransport { + /// Unique transport identifier. + transport_id: TransportId, + /// Optional instance name (for named instances in config). + name: Option, + /// Configuration. + config: EthernetConfig, + /// Current state. + state: TransportState, + /// Async socket (None until started). + socket: Option>, + /// Channel for delivering received packets to Node. + packet_tx: PacketTx, + /// Receive loop task handle. + recv_task: Option>, + /// Beacon sender task handle. + beacon_task: Option>, + /// Local MAC address (after start). + local_mac: Option<[u8; 6]>, + /// Interface name (from config). + interface: String, + /// Effective MTU (interface MTU - 1 for frame type prefix). + effective_mtu: u16, + /// Discovery buffer for discovered peers. + discovery_buffer: Arc, + /// Transport-level statistics. + stats: Arc, + /// Node's public key for beacon construction. + local_pubkey: Option, +} + +impl EthernetTransport { + /// Create a new Ethernet transport. + pub fn new( + transport_id: TransportId, + name: Option, + config: EthernetConfig, + packet_tx: PacketTx, + ) -> Self { + let interface = config.interface.clone(); + let discovery_buffer = Arc::new(DiscoveryBuffer::new(transport_id)); + let stats = Arc::new(EthernetStats::new()); + + Self { + transport_id, + name, + config, + state: TransportState::Configured, + socket: None, + packet_tx, + recv_task: None, + beacon_task: None, + local_mac: None, + interface, + effective_mtu: 1499, // default, updated on start + discovery_buffer, + stats, + local_pubkey: None, + } + } + + /// Get the instance name (if configured as a named instance). + pub fn name(&self) -> Option<&str> { + self.name.as_deref() + } + + /// Get the interface name. + pub fn interface_name(&self) -> &str { + &self.interface + } + + /// Get the local MAC address (only valid after start). + pub fn local_mac(&self) -> Option<[u8; 6]> { + self.local_mac + } + + /// Set the node's public key for beacon construction. + /// + /// Must be called before start if announce is enabled. + pub fn set_local_pubkey(&mut self, pubkey: XOnlyPublicKey) { + self.local_pubkey = Some(pubkey); + } + + /// Get a reference to the statistics. + pub fn stats(&self) -> &Arc { + &self.stats + } + + /// Start the transport asynchronously. + /// + /// Creates the AF_PACKET socket, spawns the receive loop, and + /// optionally spawns the beacon sender task. + pub async fn start_async(&mut self) -> Result<(), TransportError> { + if !self.state.can_start() { + return Err(TransportError::AlreadyStarted); + } + + self.state = TransportState::Starting; + + // Create and bind AF_PACKET socket + let raw_socket = PacketSocket::open(&self.config.interface, self.config.ethertype())?; + + // Get local MAC and MTU + let local_mac = raw_socket.local_mac()?; + let if_mtu = raw_socket.interface_mtu()?; + + // Effective MTU: interface MTU minus 1 byte for frame type prefix + let effective_mtu = if let Some(configured_mtu) = self.config.mtu { + // Config MTU cannot exceed interface MTU - 1 + configured_mtu.min(if_mtu.saturating_sub(1)) + } else { + if_mtu.saturating_sub(1) + }; + self.effective_mtu = effective_mtu; + self.local_mac = Some(local_mac); + + // Set buffer sizes + raw_socket.set_recv_buffer_size(self.config.recv_buf_size())?; + raw_socket.set_send_buffer_size(self.config.send_buf_size())?; + + // Wrap in async + let async_socket = raw_socket.into_async()?; + let socket = Arc::new(async_socket); + self.socket = Some(socket.clone()); + + // Spawn receive loop + let transport_id = self.transport_id; + let packet_tx = self.packet_tx.clone(); + let mtu = self.effective_mtu; + let discovery_enabled = self.config.discovery(); + let discovery_buffer = self.discovery_buffer.clone(); + let stats = self.stats.clone(); + let recv_socket = socket.clone(); + + let recv_task = tokio::spawn(async move { + ethernet_receive_loop( + recv_socket, + transport_id, + packet_tx, + mtu, + discovery_enabled, + discovery_buffer, + stats, + ) + .await; + }); + self.recv_task = Some(recv_task); + + // Spawn beacon sender if announce is enabled + if self.config.announce() { + if let Some(pubkey) = self.local_pubkey { + let beacon_socket = socket.clone(); + let interval_secs = self.config.beacon_interval_secs(); + let beacon_stats = self.stats.clone(); + let beacon_transport_id = self.transport_id; + + let beacon_task = tokio::spawn(async move { + beacon_sender_loop( + beacon_socket, + pubkey, + interval_secs, + beacon_stats, + beacon_transport_id, + ) + .await; + }); + self.beacon_task = Some(beacon_task); + } else { + warn!( + transport_id = %self.transport_id, + "Announce enabled but no local pubkey set; beacons disabled" + ); + } + } + + self.state = TransportState::Up; + + if let Some(ref name) = self.name { + info!( + name = %name, + interface = %self.interface, + mac = %format_mac(&local_mac), + mtu = effective_mtu, + if_mtu = if_mtu, + "Ethernet transport started" + ); + } else { + info!( + interface = %self.interface, + mac = %format_mac(&local_mac), + mtu = effective_mtu, + if_mtu = if_mtu, + "Ethernet 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 beacon task + if let Some(task) = self.beacon_task.take() { + task.abort(); + let _ = task.await; + } + + // Abort receive task + if let Some(task) = self.recv_task.take() { + task.abort(); + let _ = task.await; + } + + // Drop socket + self.socket.take(); + self.local_mac = None; + + self.state = TransportState::Down; + + info!( + transport_id = %self.transport_id, + interface = %self.interface, + "Ethernet transport stopped" + ); + + Ok(()) + } + + /// Send a packet asynchronously. + /// + /// The data is prepended with a FRAME_TYPE_DATA prefix byte before + /// transmission. + pub async fn send_async( + &self, + addr: &TransportAddr, + data: &[u8], + ) -> Result { + if !self.state.is_operational() { + return Err(TransportError::NotStarted); + } + + if data.len() > self.effective_mtu as usize { + return Err(TransportError::MtuExceeded { + packet_size: data.len(), + mtu: self.effective_mtu, + }); + } + + let dest_mac = parse_mac_addr(addr)?; + let socket = self.socket.as_ref().ok_or(TransportError::NotStarted)?; + + // Prepend frame type prefix + let mut frame = Vec::with_capacity(1 + data.len()); + frame.push(FRAME_TYPE_DATA); + frame.extend_from_slice(data); + + let bytes_sent = socket.send_to(&frame, &dest_mac).await?; + self.stats.record_send(bytes_sent); + + trace!( + transport_id = %self.transport_id, + remote_mac = %format_mac(&dest_mac), + bytes = bytes_sent, + "Ethernet frame sent" + ); + + // Return the data bytes sent (excluding frame type prefix) + Ok(bytes_sent.saturating_sub(1)) + } +} + +impl Transport for EthernetTransport { + fn transport_id(&self) -> TransportId { + self.transport_id + } + + fn transport_type(&self) -> &TransportType { + &TransportType::ETHERNET + } + + fn state(&self) -> TransportState { + self.state + } + + fn mtu(&self) -> u16 { + self.effective_mtu + } + + fn start(&mut self) -> Result<(), TransportError> { + Err(TransportError::NotSupported( + "use start_async() for Ethernet transport".into(), + )) + } + + fn stop(&mut self) -> Result<(), TransportError> { + Err(TransportError::NotSupported( + "use stop_async() for Ethernet transport".into(), + )) + } + + fn send(&self, _addr: &TransportAddr, _data: &[u8]) -> Result<(), TransportError> { + Err(TransportError::NotSupported( + "use send_async() for Ethernet transport".into(), + )) + } + + fn discover(&self) -> Result, TransportError> { + Ok(self.discovery_buffer.take()) + } + + fn auto_connect(&self) -> bool { + self.config.auto_connect() + } + + fn accept_connections(&self) -> bool { + self.config.accept_connections() + } +} + +// ============================================================================ +// Receive Loop +// ============================================================================ + +/// Ethernet receive loop — runs as a spawned task. +async fn ethernet_receive_loop( + socket: Arc, + transport_id: TransportId, + packet_tx: PacketTx, + mtu: u16, + discovery_enabled: bool, + discovery_buffer: Arc, + stats: Arc, +) { + // Buffer with headroom: frame type prefix + MTU + some extra + let mut buf = vec![0u8; mtu as usize + 100]; + + debug!(transport_id = %transport_id, "Ethernet receive loop starting"); + + loop { + match socket.recv_from(&mut buf).await { + Ok((len, src_mac)) => { + if len == 0 { + continue; + } + + stats.record_recv(len); + + let frame_type = buf[0]; + match frame_type { + FRAME_TYPE_DATA => { + // Strip the frame type prefix, deliver payload + let data = buf[1..len].to_vec(); + let addr = TransportAddr::from_bytes(&src_mac); + let packet = ReceivedPacket::new(transport_id, addr, data); + + trace!( + transport_id = %transport_id, + remote_mac = %format_mac(&src_mac), + bytes = len - 1, + "Ethernet data frame received" + ); + + if packet_tx.send(packet).await.is_err() { + info!( + transport_id = %transport_id, + "Packet channel closed, stopping receive loop" + ); + break; + } + } + FRAME_TYPE_BEACON => { + stats.record_beacon_recv(); + + if discovery_enabled + && let Some(pubkey) = parse_beacon(&buf[..len]) + { + discovery_buffer.add_peer(src_mac, pubkey); + trace!( + transport_id = %transport_id, + remote_mac = %format_mac(&src_mac), + "Discovery beacon received" + ); + } + } + _ => { + // Unknown frame type, ignore + trace!( + transport_id = %transport_id, + frame_type = frame_type, + "Unknown frame type, dropping" + ); + } + } + } + Err(e) => { + stats.record_recv_error(); + warn!( + transport_id = %transport_id, + error = %e, + "Ethernet receive error" + ); + } + } + } + + debug!(transport_id = %transport_id, "Ethernet receive loop stopped"); +} + +// ============================================================================ +// Beacon Sender +// ============================================================================ + +/// Periodic beacon sender loop. +async fn beacon_sender_loop( + socket: Arc, + pubkey: XOnlyPublicKey, + interval_secs: u64, + stats: Arc, + transport_id: TransportId, +) { + let beacon = build_beacon(&pubkey); + let interval = tokio::time::Duration::from_secs(interval_secs); + + debug!( + transport_id = %transport_id, + interval_secs, + "Beacon sender starting" + ); + + // Send an initial beacon immediately at startup + if let Err(e) = socket.send_to(&beacon, ÐERNET_BROADCAST).await { + warn!( + transport_id = %transport_id, + error = %e, + "Failed to send initial beacon" + ); + } else { + stats.record_beacon_sent(); + } + + let mut interval_timer = tokio::time::interval(interval); + interval_timer.tick().await; // consume the immediate first tick + + loop { + interval_timer.tick().await; + + match socket.send_to(&beacon, ÐERNET_BROADCAST).await { + Ok(_) => { + stats.record_beacon_sent(); + trace!( + transport_id = %transport_id, + "Beacon sent" + ); + } + Err(e) => { + stats.record_send_error(); + warn!( + transport_id = %transport_id, + error = %e, + "Failed to send beacon" + ); + } + } + } +} + +// ============================================================================ +// MAC Address Helpers +// ============================================================================ + +/// Parse a TransportAddr as a 6-byte MAC address. +fn parse_mac_addr(addr: &TransportAddr) -> Result<[u8; 6], TransportError> { + let bytes = addr.as_bytes(); + if bytes.len() != 6 { + return Err(TransportError::InvalidAddress(format!( + "expected 6-byte MAC, got {} bytes", + bytes.len() + ))); + } + if bytes == [0, 0, 0, 0, 0, 0] { + return Err(TransportError::InvalidAddress( + "destination MAC is all zeros".into(), + )); + } + let mut mac = [0u8; 6]; + mac.copy_from_slice(bytes); + Ok(mac) +} + +/// Format a MAC address as colon-separated hex for display. +pub fn format_mac(mac: &[u8; 6]) -> String { + format!( + "{:02x}:{:02x}:{:02x}:{:02x}:{:02x}:{:02x}", + mac[0], mac[1], mac[2], mac[3], mac[4], mac[5] + ) +} + +/// Parse a colon-separated MAC string (e.g., "aa:bb:cc:dd:ee:ff") into bytes. +pub fn parse_mac_string(s: &str) -> Result<[u8; 6], TransportError> { + let parts: Vec<&str> = s.split(':').collect(); + if parts.len() != 6 { + return Err(TransportError::InvalidAddress(format!( + "invalid MAC format: expected 6 colon-separated hex bytes, got '{}'", + s + ))); + } + let mut mac = [0u8; 6]; + for (i, part) in parts.iter().enumerate() { + mac[i] = u8::from_str_radix(part, 16).map_err(|_| { + TransportError::InvalidAddress(format!("invalid hex byte '{}' in MAC address", part)) + })?; + } + Ok(mac) +} + +// ============================================================================ +// Tests +// ============================================================================ + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_parse_mac_addr_valid() { + let addr = TransportAddr::from_bytes(&[0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xff]); + let mac = parse_mac_addr(&addr).unwrap(); + assert_eq!(mac, [0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xff]); + } + + #[test] + fn test_parse_mac_addr_wrong_length() { + let addr = TransportAddr::from_bytes(&[0xaa, 0xbb, 0xcc]); + assert!(parse_mac_addr(&addr).is_err()); + + let addr = TransportAddr::from_string("192.168.1.1:4000"); + assert!(parse_mac_addr(&addr).is_err()); + } + + #[test] + fn test_parse_mac_addr_all_zeros() { + let addr = TransportAddr::from_bytes(&[0, 0, 0, 0, 0, 0]); + assert!(parse_mac_addr(&addr).is_err()); + } + + #[test] + fn test_format_mac() { + let mac = [0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xff]; + assert_eq!(format_mac(&mac), "aa:bb:cc:dd:ee:ff"); + } + + #[test] + fn test_format_mac_leading_zeros() { + let mac = [0x01, 0x02, 0x03, 0x04, 0x05, 0x06]; + assert_eq!(format_mac(&mac), "01:02:03:04:05:06"); + } + + #[test] + fn test_parse_mac_string_valid() { + let mac = parse_mac_string("aa:bb:cc:dd:ee:ff").unwrap(); + assert_eq!(mac, [0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xff]); + } + + #[test] + fn test_parse_mac_string_uppercase() { + let mac = parse_mac_string("AA:BB:CC:DD:EE:FF").unwrap(); + assert_eq!(mac, [0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xff]); + } + + #[test] + fn test_parse_mac_string_invalid() { + assert!(parse_mac_string("aa:bb:cc").is_err()); + assert!(parse_mac_string("not:a:mac:at:all:x").is_err()); + assert!(parse_mac_string("").is_err()); + assert!(parse_mac_string("aa-bb-cc-dd-ee-ff").is_err()); + } + + #[test] + fn test_frame_type_data_prefix() { + // Verify data frames are prefixed with 0x00 + let data = vec![1, 2, 3, 4]; + let mut frame = Vec::with_capacity(1 + data.len()); + frame.push(FRAME_TYPE_DATA); + frame.extend_from_slice(&data); + + assert_eq!(frame[0], 0x00); + assert_eq!(&frame[1..], &[1, 2, 3, 4]); + } + + #[test] + fn test_beacon_size() { + assert_eq!(discovery::BEACON_SIZE, 34); + } +} diff --git a/src/transport/ethernet/socket.rs b/src/transport/ethernet/socket.rs new file mode 100644 index 0000000..c346382 --- /dev/null +++ b/src/transport/ethernet/socket.rs @@ -0,0 +1,388 @@ +//! AF_PACKET socket creation, binding, and ioctl helpers. + +use crate::transport::TransportError; +use std::os::unix::io::{AsRawFd, RawFd}; +use tokio::io::unix::AsyncFd; + +/// Broadcast MAC address. +pub const ETHERNET_BROADCAST: [u8; 6] = [0xff; 6]; + +/// Wrapper around an AF_PACKET SOCK_DGRAM file descriptor. +/// +/// Owns the fd and closes it on drop. Provides synchronous send/recv +/// methods used by the async wrappers via `AsyncFd`. +pub struct PacketSocket { + fd: RawFd, + if_index: i32, + ethertype: u16, +} + +impl PacketSocket { + /// Create and bind an AF_PACKET SOCK_DGRAM socket. + /// + /// Returns an error with a clear message if CAP_NET_RAW is missing. + pub fn open(interface: &str, ethertype: u16) -> Result { + let fd = unsafe { + libc::socket( + libc::AF_PACKET, + libc::SOCK_DGRAM, + (ethertype).to_be() as i32, + ) + }; + if fd < 0 { + let err = std::io::Error::last_os_error(); + if err.raw_os_error() == Some(libc::EPERM) { + return Err(TransportError::StartFailed( + "AF_PACKET requires CAP_NET_RAW capability \ + (run as root or use: setcap cap_net_raw=ep )" + .into(), + )); + } + return Err(TransportError::StartFailed(format!( + "socket(AF_PACKET) failed: {}", + err + ))); + } + + // Look up interface index + let if_index = get_if_index(fd, interface)?; + + // Bind to the interface + let mut sll: libc::sockaddr_ll = unsafe { std::mem::zeroed() }; + sll.sll_family = libc::AF_PACKET as u16; + sll.sll_protocol = ethertype.to_be(); + sll.sll_ifindex = if_index; + + let ret = unsafe { + libc::bind( + fd, + &sll as *const libc::sockaddr_ll as *const libc::sockaddr, + std::mem::size_of::() as libc::socklen_t, + ) + }; + if ret < 0 { + let err = std::io::Error::last_os_error(); + unsafe { libc::close(fd) }; + return Err(TransportError::StartFailed(format!( + "bind(AF_PACKET, {}) failed: {}", + interface, err + ))); + } + + // Set non-blocking for async integration + let flags = unsafe { libc::fcntl(fd, libc::F_GETFL) }; + if flags < 0 { + let err = std::io::Error::last_os_error(); + unsafe { libc::close(fd) }; + return Err(TransportError::StartFailed(format!( + "fcntl(F_GETFL) failed: {}", + err + ))); + } + let ret = unsafe { libc::fcntl(fd, libc::F_SETFL, flags | libc::O_NONBLOCK) }; + if ret < 0 { + let err = std::io::Error::last_os_error(); + unsafe { libc::close(fd) }; + return Err(TransportError::StartFailed(format!( + "fcntl(F_SETFL, O_NONBLOCK) failed: {}", + err + ))); + } + + Ok(Self { + fd, + if_index, + ethertype, + }) + } + + /// Get the interface index. + pub fn if_index(&self) -> i32 { + self.if_index + } + + /// Get the local MAC address of the bound interface. + pub fn local_mac(&self) -> Result<[u8; 6], TransportError> { + get_mac_addr(self.fd, self.if_index) + } + + /// Get the interface MTU. + pub fn interface_mtu(&self) -> Result { + get_if_mtu(self.fd, self.if_index) + } + + /// Set the socket receive buffer size. + pub fn set_recv_buffer_size(&self, size: usize) -> Result<(), TransportError> { + let size = size as libc::c_int; + let ret = unsafe { + libc::setsockopt( + self.fd, + libc::SOL_SOCKET, + libc::SO_RCVBUF, + &size as *const libc::c_int as *const libc::c_void, + std::mem::size_of::() as libc::socklen_t, + ) + }; + if ret < 0 { + return Err(TransportError::StartFailed(format!( + "setsockopt(SO_RCVBUF) failed: {}", + std::io::Error::last_os_error() + ))); + } + Ok(()) + } + + /// Set the socket send buffer size. + pub fn set_send_buffer_size(&self, size: usize) -> Result<(), TransportError> { + let size = size as libc::c_int; + let ret = unsafe { + libc::setsockopt( + self.fd, + libc::SOL_SOCKET, + libc::SO_SNDBUF, + &size as *const libc::c_int as *const libc::c_void, + std::mem::size_of::() as libc::socklen_t, + ) + }; + if ret < 0 { + return Err(TransportError::StartFailed(format!( + "setsockopt(SO_SNDBUF) failed: {}", + std::io::Error::last_os_error() + ))); + } + Ok(()) + } + + /// Send a payload to a destination MAC address. + /// + /// Returns the number of bytes sent, or an io::Error. + pub fn send_to(&self, data: &[u8], dest_mac: &[u8; 6]) -> std::io::Result { + let mut sll: libc::sockaddr_ll = unsafe { std::mem::zeroed() }; + sll.sll_family = libc::AF_PACKET as u16; + sll.sll_protocol = self.ethertype.to_be(); + sll.sll_ifindex = self.if_index; + sll.sll_halen = 6; + sll.sll_addr[..6].copy_from_slice(dest_mac); + + let ret = unsafe { + libc::sendto( + self.fd, + data.as_ptr() as *const libc::c_void, + data.len(), + 0, + &sll as *const libc::sockaddr_ll as *const libc::sockaddr, + std::mem::size_of::() as libc::socklen_t, + ) + }; + if ret < 0 { + Err(std::io::Error::last_os_error()) + } else { + Ok(ret as usize) + } + } + + /// Receive a payload and source MAC address. + /// + /// Returns (bytes_read, source_mac), or an io::Error. + pub fn recv_from(&self, buf: &mut [u8]) -> std::io::Result<(usize, [u8; 6])> { + let mut sll: libc::sockaddr_ll = unsafe { std::mem::zeroed() }; + let mut sll_len = std::mem::size_of::() as libc::socklen_t; + + let ret = unsafe { + libc::recvfrom( + self.fd, + buf.as_mut_ptr() as *mut libc::c_void, + buf.len(), + 0, + &mut sll as *mut libc::sockaddr_ll as *mut libc::sockaddr, + &mut sll_len, + ) + }; + if ret < 0 { + return Err(std::io::Error::last_os_error()); + } + + let mut src_mac = [0u8; 6]; + src_mac.copy_from_slice(&sll.sll_addr[..6]); + + Ok((ret as usize, src_mac)) + } + + /// Wrap this socket in a tokio AsyncFd for async I/O. + pub fn into_async(self) -> Result { + let async_fd = AsyncFd::new(self) + .map_err(|e| TransportError::StartFailed(format!("AsyncFd::new failed: {}", e)))?; + Ok(AsyncPacketSocket { inner: async_fd }) + } +} + +impl AsRawFd for PacketSocket { + fn as_raw_fd(&self) -> RawFd { + self.fd + } +} + +impl Drop for PacketSocket { + fn drop(&mut self) { + unsafe { + libc::close(self.fd); + } + } +} + +/// Async wrapper around PacketSocket using tokio's AsyncFd. +pub struct AsyncPacketSocket { + inner: AsyncFd, +} + +impl AsyncPacketSocket { + /// Send a payload to a destination MAC address. + pub async fn send_to(&self, data: &[u8], dest_mac: &[u8; 6]) -> Result { + loop { + let mut guard = self + .inner + .writable() + .await + .map_err(|e| TransportError::SendFailed(format!("writable wait: {}", e)))?; + + match guard.try_io(|inner| inner.get_ref().send_to(data, dest_mac)) { + Ok(Ok(n)) => return Ok(n), + Ok(Err(e)) => return Err(TransportError::SendFailed(format!("{}", e))), + Err(_would_block) => continue, + } + } + } + + /// Receive a payload and source MAC address. + pub async fn recv_from( + &self, + buf: &mut [u8], + ) -> Result<(usize, [u8; 6]), TransportError> { + loop { + let mut guard = self + .inner + .readable() + .await + .map_err(|e| TransportError::RecvFailed(format!("readable wait: {}", e)))?; + + match guard.try_io(|inner| inner.get_ref().recv_from(buf)) { + Ok(Ok(result)) => return Ok(result), + Ok(Err(e)) => return Err(TransportError::RecvFailed(format!("{}", e))), + Err(_would_block) => continue, + } + } + } + + /// Get a reference to the inner PacketSocket. + pub fn get_ref(&self) -> &PacketSocket { + self.inner.get_ref() + } +} + +// ============================================================================ +// ioctl helpers +// ============================================================================ + +/// Get the interface index by name. +fn get_if_index(_fd: RawFd, interface: &str) -> Result { + let c_name = std::ffi::CString::new(interface).map_err(|_| { + TransportError::StartFailed(format!("invalid interface name: {}", interface)) + })?; + + let idx = unsafe { libc::if_nametoindex(c_name.as_ptr()) }; + if idx == 0 { + return Err(TransportError::StartFailed(format!( + "interface not found: {} ({})", + interface, + std::io::Error::last_os_error() + ))); + } + Ok(idx as i32) +} + +/// Get the MAC address of an interface by its index. +fn get_mac_addr(fd: RawFd, if_index: i32) -> Result<[u8; 6], TransportError> { + // First get the interface name from the index + let mut ifr: libc::ifreq = unsafe { std::mem::zeroed() }; + + // Use if_indextoname to get the name + let mut name_buf = [0u8; libc::IFNAMSIZ]; + let ret = unsafe { + libc::if_indextoname(if_index as libc::c_uint, name_buf.as_mut_ptr() as *mut libc::c_char) + }; + if ret.is_null() { + return Err(TransportError::StartFailed(format!( + "if_indextoname({}) failed: {}", + if_index, + std::io::Error::last_os_error() + ))); + } + + // Copy name into ifreq + let name_len = name_buf.iter().position(|&b| b == 0).unwrap_or(name_buf.len()); + let copy_len = name_len.min(libc::IFNAMSIZ - 1); + unsafe { + std::ptr::copy_nonoverlapping( + name_buf.as_ptr(), + ifr.ifr_name.as_mut_ptr() as *mut u8, + copy_len, + ); + } + + let ret = unsafe { libc::ioctl(fd, libc::SIOCGIFHWADDR as libc::c_ulong, &ifr) }; + if ret < 0 { + return Err(TransportError::StartFailed(format!( + "ioctl(SIOCGIFHWADDR) failed: {}", + std::io::Error::last_os_error() + ))); + } + + let mut mac = [0u8; 6]; + unsafe { + let sa_data = ifr.ifr_ifru.ifru_hwaddr.sa_data; + for (i, byte) in mac.iter_mut().enumerate() { + *byte = sa_data[i] as u8; + } + } + + Ok(mac) +} + +/// Get the MTU of an interface by its index. +fn get_if_mtu(fd: RawFd, if_index: i32) -> Result { + let mut ifr: libc::ifreq = unsafe { std::mem::zeroed() }; + + // Get the interface name from index + let mut name_buf = [0u8; libc::IFNAMSIZ]; + let ret = unsafe { + libc::if_indextoname(if_index as libc::c_uint, name_buf.as_mut_ptr() as *mut libc::c_char) + }; + if ret.is_null() { + return Err(TransportError::StartFailed(format!( + "if_indextoname({}) failed: {}", + if_index, + std::io::Error::last_os_error() + ))); + } + + let name_len = name_buf.iter().position(|&b| b == 0).unwrap_or(name_buf.len()); + let copy_len = name_len.min(libc::IFNAMSIZ - 1); + unsafe { + std::ptr::copy_nonoverlapping( + name_buf.as_ptr(), + ifr.ifr_name.as_mut_ptr() as *mut u8, + copy_len, + ); + } + + let ret = unsafe { libc::ioctl(fd, libc::SIOCGIFMTU as libc::c_ulong, &ifr) }; + if ret < 0 { + return Err(TransportError::StartFailed(format!( + "ioctl(SIOCGIFMTU) failed: {}", + std::io::Error::last_os_error() + ))); + } + + let mtu = unsafe { ifr.ifr_ifru.ifru_mtu } as u16; + Ok(mtu) +} diff --git a/src/transport/ethernet/stats.rs b/src/transport/ethernet/stats.rs new file mode 100644 index 0000000..67e542d --- /dev/null +++ b/src/transport/ethernet/stats.rs @@ -0,0 +1,107 @@ +//! Ethernet transport statistics. + +use std::sync::atomic::{AtomicU64, Ordering}; + +/// Statistics for an Ethernet transport instance. +/// +/// Uses atomic counters for lock-free updates from the receive loop +/// and send path concurrently. +pub struct EthernetStats { + pub frames_sent: AtomicU64, + pub frames_recv: AtomicU64, + pub bytes_sent: AtomicU64, + pub bytes_recv: AtomicU64, + pub send_errors: AtomicU64, + pub recv_errors: AtomicU64, + pub beacons_sent: AtomicU64, + pub beacons_recv: AtomicU64, + pub frames_too_short: AtomicU64, + pub frames_too_long: AtomicU64, +} + +impl EthernetStats { + /// Create a new stats instance with all counters at zero. + pub fn new() -> Self { + Self { + frames_sent: AtomicU64::new(0), + frames_recv: AtomicU64::new(0), + bytes_sent: AtomicU64::new(0), + bytes_recv: AtomicU64::new(0), + send_errors: AtomicU64::new(0), + recv_errors: AtomicU64::new(0), + beacons_sent: AtomicU64::new(0), + beacons_recv: AtomicU64::new(0), + frames_too_short: AtomicU64::new(0), + frames_too_long: AtomicU64::new(0), + } + } + + /// Record a successful send. + pub fn record_send(&self, bytes: usize) { + self.frames_sent.fetch_add(1, Ordering::Relaxed); + self.bytes_sent.fetch_add(bytes as u64, Ordering::Relaxed); + } + + /// Record a successful receive. + pub fn record_recv(&self, bytes: usize) { + self.frames_recv.fetch_add(1, Ordering::Relaxed); + self.bytes_recv.fetch_add(bytes as u64, Ordering::Relaxed); + } + + /// Record a send error. + pub fn record_send_error(&self) { + self.send_errors.fetch_add(1, Ordering::Relaxed); + } + + /// Record a receive error. + pub fn record_recv_error(&self) { + self.recv_errors.fetch_add(1, Ordering::Relaxed); + } + + /// Record a sent beacon. + pub fn record_beacon_sent(&self) { + self.beacons_sent.fetch_add(1, Ordering::Relaxed); + } + + /// Record a received beacon. + pub fn record_beacon_recv(&self) { + self.beacons_recv.fetch_add(1, Ordering::Relaxed); + } + + /// Take a snapshot of all counters. + pub fn snapshot(&self) -> EthernetStatsSnapshot { + EthernetStatsSnapshot { + frames_sent: self.frames_sent.load(Ordering::Relaxed), + frames_recv: self.frames_recv.load(Ordering::Relaxed), + bytes_sent: self.bytes_sent.load(Ordering::Relaxed), + bytes_recv: self.bytes_recv.load(Ordering::Relaxed), + send_errors: self.send_errors.load(Ordering::Relaxed), + recv_errors: self.recv_errors.load(Ordering::Relaxed), + beacons_sent: self.beacons_sent.load(Ordering::Relaxed), + beacons_recv: self.beacons_recv.load(Ordering::Relaxed), + frames_too_short: self.frames_too_short.load(Ordering::Relaxed), + frames_too_long: self.frames_too_long.load(Ordering::Relaxed), + } + } +} + +impl Default for EthernetStats { + fn default() -> Self { + Self::new() + } +} + +/// Point-in-time snapshot of Ethernet stats (non-atomic, copyable). +#[derive(Clone, Debug, Default)] +pub struct EthernetStatsSnapshot { + pub frames_sent: u64, + pub frames_recv: u64, + pub bytes_sent: u64, + pub bytes_recv: u64, + pub send_errors: u64, + pub recv_errors: u64, + pub beacons_sent: u64, + pub beacons_recv: u64, + pub frames_too_short: u64, + pub frames_too_long: u64, +} diff --git a/src/transport/mod.rs b/src/transport/mod.rs index 3b667d3..e050699 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -6,8 +6,13 @@ pub mod udp; +#[cfg(target_os = "linux")] +pub mod ethernet; + use secp256k1::XOnlyPublicKey; use udp::UdpTransport; +#[cfg(target_os = "linux")] +use ethernet::EthernetTransport; use std::fmt; use std::time::{Duration, SystemTime, UNIX_EPOCH}; use thiserror::Error; @@ -761,6 +766,18 @@ pub trait Transport { /// Discover potential peers (if supported). fn discover(&self) -> Result, TransportError>; + + /// Whether to auto-connect to peers returned by discover(). + /// Default: false. Concrete transports read from their own config. + fn auto_connect(&self) -> bool { + false + } + + /// Whether to accept inbound handshake initiations on this transport. + /// Default: true (preserves UDP's current implicit behavior). + fn accept_connections(&self) -> bool { + true + } } // ============================================================================ @@ -774,7 +791,9 @@ pub trait Transport { pub enum TransportHandle { /// UDP/IP transport. Udp(UdpTransport), - // Future: Tcp(TcpTransport), Tor(TorTransport), etc. + /// Raw Ethernet transport. + #[cfg(target_os = "linux")] + Ethernet(EthernetTransport), } impl TransportHandle { @@ -782,6 +801,8 @@ impl TransportHandle { pub async fn start(&mut self) -> Result<(), TransportError> { match self { TransportHandle::Udp(t) => t.start_async().await, + #[cfg(target_os = "linux")] + TransportHandle::Ethernet(t) => t.start_async().await, } } @@ -789,6 +810,8 @@ impl TransportHandle { pub async fn stop(&mut self) -> Result<(), TransportError> { match self { TransportHandle::Udp(t) => t.stop_async().await, + #[cfg(target_os = "linux")] + TransportHandle::Ethernet(t) => t.stop_async().await, } } @@ -796,6 +819,8 @@ impl TransportHandle { pub async fn send(&self, addr: &TransportAddr, data: &[u8]) -> Result { match self { TransportHandle::Udp(t) => t.send_async(addr, data).await, + #[cfg(target_os = "linux")] + TransportHandle::Ethernet(t) => t.send_async(addr, data).await, } } @@ -803,6 +828,8 @@ impl TransportHandle { pub fn transport_id(&self) -> TransportId { match self { TransportHandle::Udp(t) => t.transport_id(), + #[cfg(target_os = "linux")] + TransportHandle::Ethernet(t) => t.transport_id(), } } @@ -810,6 +837,8 @@ impl TransportHandle { pub fn name(&self) -> Option<&str> { match self { TransportHandle::Udp(t) => t.name(), + #[cfg(target_os = "linux")] + TransportHandle::Ethernet(t) => t.name(), } } @@ -817,6 +846,8 @@ impl TransportHandle { pub fn transport_type(&self) -> &TransportType { match self { TransportHandle::Udp(t) => t.transport_type(), + #[cfg(target_os = "linux")] + TransportHandle::Ethernet(t) => t.transport_type(), } } @@ -824,6 +855,8 @@ impl TransportHandle { pub fn state(&self) -> TransportState { match self { TransportHandle::Udp(t) => t.state(), + #[cfg(target_os = "linux")] + TransportHandle::Ethernet(t) => t.state(), } } @@ -831,6 +864,8 @@ impl TransportHandle { pub fn mtu(&self) -> u16 { match self { TransportHandle::Udp(t) => t.mtu(), + #[cfg(target_os = "linux")] + TransportHandle::Ethernet(t) => t.mtu(), } } @@ -841,13 +876,53 @@ impl TransportHandle { pub fn link_mtu(&self, addr: &TransportAddr) -> u16 { match self { TransportHandle::Udp(t) => t.link_mtu(addr), + #[cfg(target_os = "linux")] + TransportHandle::Ethernet(t) => t.link_mtu(addr), } } - /// Get the local bound address (only valid after start). + /// Get the local bound address (UDP only, returns None for other transports). pub fn local_addr(&self) -> Option { match self { TransportHandle::Udp(t) => t.local_addr(), + #[cfg(target_os = "linux")] + TransportHandle::Ethernet(_) => None, + } + } + + /// Get the interface name (Ethernet only, returns None for other transports). + pub fn interface_name(&self) -> Option<&str> { + match self { + TransportHandle::Udp(_) => None, + #[cfg(target_os = "linux")] + TransportHandle::Ethernet(t) => Some(t.interface_name()), + } + } + + /// Drain discovered peers from this transport. + pub fn discover(&self) -> Result, TransportError> { + match self { + TransportHandle::Udp(t) => t.discover(), + #[cfg(target_os = "linux")] + TransportHandle::Ethernet(t) => t.discover(), + } + } + + /// Whether this transport auto-connects to discovered peers. + pub fn auto_connect(&self) -> bool { + match self { + TransportHandle::Udp(t) => t.auto_connect(), + #[cfg(target_os = "linux")] + TransportHandle::Ethernet(t) => t.auto_connect(), + } + } + + /// Whether this transport accepts inbound connections. + pub fn accept_connections(&self) -> bool { + match self { + TransportHandle::Udp(t) => t.accept_connections(), + #[cfg(target_os = "linux")] + TransportHandle::Ethernet(t) => t.accept_connections(), } } diff --git a/testing/chaos/Dockerfile b/testing/chaos/Dockerfile index 27ca15a..8eb90b4 100644 --- a/testing/chaos/Dockerfile +++ b/testing/chaos/Dockerfile @@ -27,8 +27,10 @@ RUN printf '%s\n' \ COPY fips fipsctl /usr/local/bin/ RUN chmod +x /usr/local/bin/fips /usr/local/bin/fipsctl +COPY entrypoint.sh /usr/local/bin/entrypoint.sh +RUN chmod +x /usr/local/bin/entrypoint.sh + # Static web page served via Python HTTP server RUN printf 'Fuck IPs!\n' > /root/index.html -# Start dnsmasq, SSH server, iperf3, and HTTP server in background, then run FIPS -ENTRYPOINT ["/bin/bash", "-c", "dnsmasq && /usr/sbin/sshd && iperf3 -s -D && python3 -m http.server 8000 -d /root -b :: &>/dev/null & exec fips --config /etc/fips/fips.yaml"] +ENTRYPOINT ["/usr/local/bin/entrypoint.sh"] diff --git a/testing/chaos/entrypoint.sh b/testing/chaos/entrypoint.sh new file mode 100755 index 0000000..98c776c --- /dev/null +++ b/testing/chaos/entrypoint.sh @@ -0,0 +1,51 @@ +#!/bin/bash +# Container entrypoint: start services and wait for Ethernet interfaces. +# +# If the FIPS config references Ethernet transports, wait for the +# interfaces to appear before starting the FIPS daemon. This handles +# the case where veth pairs are created from the host after the +# container starts. + +set -e + +# Start background services +dnsmasq +/usr/sbin/sshd +iperf3 -s -D +python3 -m http.server 8000 -d /root -b :: &>/dev/null & + +CONFIG="/etc/fips/fips.yaml" + +# Extract Ethernet interface names from the config file. +# Matches "interface: " lines that appear under transports.ethernet. +# The sed strips the key prefix and any whitespace. +ETH_IFACES="" +if grep -q 'ethernet:' "$CONFIG" 2>/dev/null; then + ETH_IFACES=$(grep '^\s*interface:' "$CONFIG" \ + | sed 's/.*interface:\s*//' \ + | tr -d ' ' || true) +fi + +if [ -n "$ETH_IFACES" ]; then + echo "Waiting for Ethernet interfaces: $ETH_IFACES" + DEADLINE=$((SECONDS + 30)) + while [ $SECONDS -lt $DEADLINE ]; do + ALL_FOUND=true + for iface in $ETH_IFACES; do + if [ ! -e "/sys/class/net/$iface" ]; then + ALL_FOUND=false + break + fi + done + if $ALL_FOUND; then + echo "All Ethernet interfaces ready" + break + fi + sleep 0.2 + done + if ! $ALL_FOUND; then + echo "WARNING: Timed out waiting for Ethernet interfaces" + fi +fi + +exec fips --config "$CONFIG" diff --git a/testing/chaos/scenarios/ethernet-mesh.yaml b/testing/chaos/scenarios/ethernet-mesh.yaml new file mode 100644 index 0000000..00cf009 --- /dev/null +++ b/testing/chaos/scenarios/ethernet-mesh.yaml @@ -0,0 +1,67 @@ +# Mixed transport mesh: 6 nodes, UDP + Ethernet edges +# +# Exercises both transports in a single mesh. UDP edges use static +# peer config; Ethernet edges use beacon discovery. Tests that the +# spanning tree converges across heterogeneous transports with netem +# and link flaps active. +# +# Topology: +# +# n01 ---udp--- n02 ---udp--- n03 +# | | | +# eth eth udp +# | | | +# n04 ---eth--- n05 ---udp--- n06 +# +# UDP edges: n01-n02, n02-n03, n03-n06, n05-n06 +# Ethernet edges: n01-n04, n02-n05, n04-n05 + +scenario: + name: "ethernet-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, ethernet] + - [n02, n05, ethernet] + - [n04, n05, ethernet] + +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/ethernet-only.yaml b/testing/chaos/scenarios/ethernet-only.yaml new file mode 100644 index 0000000..e3b0dc7 --- /dev/null +++ b/testing/chaos/scenarios/ethernet-only.yaml @@ -0,0 +1,46 @@ +# Ethernet-only: 4-node ring, all Ethernet transport +# +# All links use raw Ethernet (AF_PACKET) over veth pairs. Nodes +# discover each other via beacons (no static peer config). Tests +# that Ethernet transport and beacon discovery work end-to-end. +# +# Topology: +# +# n01 ---eth--- n02 +# | | +# eth eth +# | | +# n04 ---eth--- n03 + +scenario: + name: "ethernet-only" + seed: 42 + duration_secs: 90 + +topology: + algorithm: explicit + num_nodes: 4 + default_transport: ethernet + params: + adjacency: + - [n01, n02] + - [n02, n03] + - [n03, n04] + - [n04, n01] + +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/sim/compose.py b/testing/chaos/sim/compose.py index 1f21b6b..38a2e56 100644 --- a/testing/chaos/sim/compose.py +++ b/testing/chaos/sim/compose.py @@ -25,6 +25,7 @@ x-fips-common: &fips-common context: ../.. cap_add: - NET_ADMIN + - NET_RAW devices: - /dev/net/tun:/dev/net/tun sysctls: diff --git a/testing/chaos/sim/config_gen.py b/testing/chaos/sim/config_gen.py index b42ae08..f1c02c2 100644 --- a/testing/chaos/sim/config_gen.py +++ b/testing/chaos/sim/config_gen.py @@ -54,6 +54,37 @@ def generate_peers_block( return "\n".join(lines) +def _build_ethernet_config(iface: str) -> dict: + """Build an Ethernet transport config dict for a single interface.""" + return { + "interface": iface, + "discovery": True, + "announce": True, + "auto_connect": True, + "accept_connections": True, + "beacon_interval_secs": 10, + } + + +def _inject_ethernet_transports(parsed: dict, eth_ifaces: list[str]): + """Inject Ethernet transport config into a parsed FIPS config. + + For a single interface, uses the single-instance format. + For multiple interfaces, uses the named-instances format. + Pure-Ethernet nodes (no UDP peers) have their UDP transport removed. + """ + if not eth_ifaces: + return + + transports = parsed.setdefault("transports", {}) + if len(eth_ifaces) == 1: + transports["ethernet"] = _build_ethernet_config(eth_ifaces[0]) + else: + transports["ethernet"] = { + iface: _build_ethernet_config(iface) for iface in eth_ifaces + } + + def generate_node_config( topology: SimTopology, node_id: str, @@ -72,7 +103,22 @@ def generate_node_config( config = config.replace("{{NSEC}}", node.nsec) config = config.replace("{{PEERS}}", peers_yaml) - if fips_overrides: + # Inject Ethernet transport config if this node has Ethernet edges + eth_ifaces = topology.ethernet_interfaces(node_id) + has_udp_peers = bool(outbound_peers) or _has_inbound_udp_peers(topology, node_id) + + if eth_ifaces or not has_udp_peers: + 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 + 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) @@ -80,6 +126,15 @@ def generate_node_config( 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.""" + 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": + return True + return False + + def generate_npubs_env(topology: SimTopology) -> str: """Generate npubs.env content mapping NPUB_= for all nodes.""" lines = [] diff --git a/testing/chaos/sim/links.py b/testing/chaos/sim/links.py index 7417533..283864d 100644 --- a/testing/chaos/sim/links.py +++ b/testing/chaos/sim/links.py @@ -15,7 +15,7 @@ from dataclasses import dataclass, field from .docker_exec import docker_exec_quiet, is_container_running from .scenario import LinkFlapsConfig -from .topology import SimTopology +from .topology import SimTopology, veth_interface_name log = logging.getLogger(__name__) @@ -138,7 +138,11 @@ class LinkManager: log.info("Link UP: %s -- %s (was down %.0fs)", a, b, down_for) def _set_loss(self, src_node: str, dst_node: str, netem_args: str) -> str | None: - """Set netem args on the tc class for src->dst. Returns the previous netem args.""" + """Set netem args on the link for src->dst. Returns the previous netem args. + + Transport-aware: UDP links use tc class on eth0, Ethernet links + use tc qdisc replace on the dedicated veth interface. + """ if not self.netem_mgr: return None @@ -158,25 +162,38 @@ class LinkManager: self.netem_mgr.down_nodes.add(src_node) return None - dest_ip = self.topology.nodes[dst_node].docker_ip + transport = self.topology.transport_for_edge(src_node, dst_node) - states = self.netem_mgr.states.get(container, {}) - link_state = states.get(dest_ip) - if link_state is None: - log.warning("No netem state for %s -> %s", src_node, dst_node) - return None + if transport == "ethernet": + # Ethernet: simple netem on veth + iface = veth_interface_name(src_node, dst_node) + veth_states = self.netem_mgr.veth_states.get(container, {}) + veth_state = veth_states.get(iface) + if veth_state is None: + log.warning("No veth netem state for %s -> %s (%s)", src_node, dst_node, iface) + return None - # Save current params - prev_args = link_state.params.to_tc_args() + prev_args = veth_state.params.to_tc_args() + cmd = f"tc qdisc replace dev {iface} root netem {netem_args}" + docker_exec_quiet(container, cmd) + return prev_args + else: + # UDP: HTB class on eth0 + dest_ip = self.topology.nodes[dst_node].docker_ip - # Apply new netem - cmd = ( - f"tc qdisc replace dev {IFACE} parent {link_state.class_id} " - f"handle {link_state.netem_handle} netem {netem_args}" - ) - docker_exec_quiet(container, cmd) + states = self.netem_mgr.states.get(container, {}) + link_state = states.get(dest_ip) + if link_state is None: + log.warning("No netem state for %s -> %s", src_node, dst_node) + return None - return prev_args + prev_args = link_state.params.to_tc_args() + cmd = ( + f"tc qdisc replace dev {IFACE} parent {link_state.class_id} " + f"handle {link_state.netem_handle} netem {netem_args}" + ) + docker_exec_quiet(container, cmd) + return prev_args def _would_disconnect(self, edge: tuple[str, str]) -> bool: """Check if removing this edge (plus currently-down edges) disconnects the graph.""" diff --git a/testing/chaos/sim/netem.py b/testing/chaos/sim/netem.py index 59e9f1a..a3ec00c 100644 --- a/testing/chaos/sim/netem.py +++ b/testing/chaos/sim/netem.py @@ -18,7 +18,7 @@ from dataclasses import dataclass, field from .docker_exec import docker_exec_quiet, is_container_running from .scenario import BandwidthConfig, LinkPolicyOverride, NetemConfig, NetemPolicy -from .topology import SimTopology +from .topology import SimTopology, veth_interface_name log = logging.getLogger(__name__) @@ -67,6 +67,15 @@ class LinkNetemState: rate_mbit: int = 0 # 0 = unlimited (1gbit default) +@dataclass +class VethNetemState: + """Tracks the netem state for a veth interface (Ethernet link direction).""" + + container: str + iface: str # e.g., "ve-n01-n02" + params: NetemParams = field(default_factory=NetemParams) + + class NetemManager: """Manages per-link netem impairment across all containers.""" @@ -80,8 +89,10 @@ class NetemManager: self.topology = topology self.config = config self.rng = rng - # Per-container, per-dest-ip netem state + # Per-container, per-dest-ip netem state (UDP links on eth0) self.states: dict[str, dict[str, LinkNetemState]] = {} + # Per-container, per-veth netem state (Ethernet links) + self.veth_states: dict[str, dict[str, VethNetemState]] = {} # Nodes currently down (updated by NodeManager) — skip tc ops on these self.down_nodes: set[str] = set() # Per-edge bandwidth: (node_a, node_b) -> rate in mbit @@ -127,7 +138,11 @@ class NetemManager: return self.config.default_policy def setup_initial(self): - """Set up HTB qdiscs and initial netem on all containers.""" + """Set up HTB qdiscs and initial netem on all containers. + + UDP peers use HTB + u32 filters on eth0. Ethernet peers use a + simple root netem qdisc on their dedicated veth interface. + """ if self._edge_rates: log.info("Bandwidth pacing enabled (%d edges with rate limits)", len(self._edge_rates) // 2) @@ -135,69 +150,103 @@ class NetemManager: for node_id in sorted(self.topology.nodes): node = self.topology.nodes[node_id] container = self.topology.container_name(node_id) - peer_ips = {} + + # Split peers by transport type + udp_peers = {} + eth_peers = [] for peer_id in sorted(node.peers): - peer_ips[peer_id] = self.topology.nodes[peer_id].docker_ip + 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 - if not peer_ips: - continue - - # Build all tc commands for this container - 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" - ) - cmds.append( - f"tc class add dev {IFACE} parent 1: classid 1:99 htb rate 1gbit" - ) - - container_states = {} - - for idx, (peer_id, dest_ip) in enumerate(peer_ips.items(), start=1): - class_id = f"1:{idx}" - netem_handle = f"{idx + 10}:" - - # Sample initial params (per-edge override or default policy) - policy = self._policy_for_edge(node_id, peer_id) - params = self._sample_policy(policy) - - rate = self._htb_rate(node_id, peer_id) - rate_mbit = self._edge_rates.get((node_id, peer_id), 0) + # --- UDP peers: HTB + u32 on eth0 --- + if udp_peers: + cmds = [f"tc qdisc del dev {IFACE} root 2>/dev/null || true"] cmds.append( - f"tc class add dev {IFACE} parent 1: classid {class_id} htb rate {rate}" + f"tc qdisc add dev {IFACE} root handle 1: htb default 99" ) cmds.append( - f"tc qdisc add dev {IFACE} parent {class_id} " - f"handle {netem_handle} netem {params.to_tc_args()}" - ) - cmds.append( - f"tc filter add dev {IFACE} parent 1: protocol ip " - f"prio {idx} u32 match ip dst {dest_ip}/32 flowid {class_id}" + f"tc class add dev {IFACE} parent 1: classid 1:99 htb rate 1gbit" ) - state = LinkNetemState( - container=container, - dest_ip=dest_ip, - class_id=class_id, - netem_handle=netem_handle, - params=params, - rate_mbit=rate_mbit, - ) - container_states[dest_ip] = state + container_states = {} - # Execute all commands in one docker exec - full_cmd = " && ".join(cmds) - result = docker_exec_quiet(container, full_cmd, timeout=30) - if result is not None: + for idx, (peer_id, dest_ip) in enumerate(udp_peers.items(), start=1): + class_id = f"1:{idx}" + netem_handle = f"{idx + 10}:" + + policy = self._policy_for_edge(node_id, peer_id) + params = self._sample_policy(policy) + + rate = self._htb_rate(node_id, peer_id) + rate_mbit = self._edge_rates.get((node_id, peer_id), 0) + cmds.append( + f"tc class add dev {IFACE} parent 1: classid {class_id} htb rate {rate}" + ) + cmds.append( + f"tc qdisc add dev {IFACE} parent {class_id} " + f"handle {netem_handle} netem {params.to_tc_args()}" + ) + cmds.append( + f"tc filter add dev {IFACE} parent 1: protocol ip " + f"prio {idx} u32 match ip dst {dest_ip}/32 flowid {class_id}" + ) + + state = LinkNetemState( + container=container, + dest_ip=dest_ip, + class_id=class_id, + netem_handle=netem_handle, + params=params, + rate_mbit=rate_mbit, + ) + container_states[dest_ip] = state + + full_cmd = " && ".join(cmds) + 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)", + container, + len(udp_peers), + ) + else: + log.warning("Failed to configure netem on %s", container) + + self.states[container] = container_states + + # --- Ethernet peers: simple netem on veth --- + if eth_peers: + container_veth_states = {} + for peer_id in eth_peers: + iface = veth_interface_name(node_id, peer_id) + policy = self._policy_for_edge(node_id, peer_id) + params = self._sample_policy(policy) + + cmd = ( + f"tc qdisc del dev {iface} root 2>/dev/null || true && " + f"tc qdisc add dev {iface} root netem {params.to_tc_args()}" + ) + result = docker_exec_quiet(container, cmd, timeout=10) + if result is not None: + log.debug("Veth netem on %s:%s -> %s", container, iface, params.to_tc_args()) + else: + log.warning("Failed to configure veth netem on %s:%s", container, iface) + + container_veth_states[iface] = VethNetemState( + container=container, + iface=iface, + params=params, + ) + + self.veth_states[container] = container_veth_states log.info( - "Configured per-link netem on %s (%d peers)", + "Configured veth netem on %s (%d Ethernet peers)", container, - len(peer_ips), + len(eth_peers), ) - else: - log.warning("Failed to configure netem on %s", container) - - self.states[container] = container_states def setup_node(self, node_id: str): """Re-apply HTB/netem/filters for a single node (after container restart). @@ -206,45 +255,62 @@ class NetemManager: class IDs and current netem params it had before going down. """ container = self.topology.container_name(node_id) + + # Re-apply UDP netem (eth0 HTB + u32) container_states = self.states.get(container) - if not container_states: - log.warning("No netem state for %s, skipping setup", container) - return - - 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" - ) - cmds.append( - f"tc class add dev {IFACE} parent 1: classid 1:99 htb rate 1gbit" - ) - - for dest_ip, state in container_states.items(): - rate = f"{state.rate_mbit}mbit" if state.rate_mbit > 0 else "1gbit" + if container_states: + cmds = [f"tc qdisc del dev {IFACE} root 2>/dev/null || true"] cmds.append( - f"tc class add dev {IFACE} parent 1: classid {state.class_id} htb rate {rate}" + f"tc qdisc add dev {IFACE} root handle 1: htb default 99" ) cmds.append( - f"tc qdisc add dev {IFACE} parent {state.class_id} " - f"handle {state.netem_handle} netem {state.params.to_tc_args()}" - ) - # Extract the priority from class_id (e.g., "1:3" -> 3) - prio = state.class_id.split(":")[1] - cmds.append( - f"tc filter add dev {IFACE} parent 1: protocol ip " - f"prio {prio} u32 match ip dst {dest_ip}/32 flowid {state.class_id}" + f"tc class add dev {IFACE} parent 1: classid 1:99 htb rate 1gbit" ) - full_cmd = " && ".join(cmds) - result = docker_exec_quiet(container, full_cmd, timeout=30) - if result is not None: + for dest_ip, state in container_states.items(): + rate = f"{state.rate_mbit}mbit" if state.rate_mbit > 0 else "1gbit" + cmds.append( + f"tc class add dev {IFACE} parent 1: classid {state.class_id} htb rate {rate}" + ) + cmds.append( + f"tc qdisc add dev {IFACE} parent {state.class_id} " + f"handle {state.netem_handle} netem {state.params.to_tc_args()}" + ) + prio = state.class_id.split(":")[1] + cmds.append( + f"tc filter add dev {IFACE} parent 1: protocol ip " + f"prio {prio} u32 match ip dst {dest_ip}/32 flowid {state.class_id}" + ) + + full_cmd = " && ".join(cmds) + result = docker_exec_quiet(container, full_cmd, timeout=30) + if result is not None: + log.info( + "Re-applied UDP netem on %s (%d peers)", + container, + len(container_states), + ) + else: + log.warning("Failed to re-apply UDP netem on %s", container) + + # Re-apply Ethernet veth netem + veth_states = self.veth_states.get(container) + if veth_states: + for iface, state in veth_states.items(): + cmd = ( + f"tc qdisc del dev {iface} root 2>/dev/null || true && " + f"tc qdisc add dev {iface} root netem {state.params.to_tc_args()}" + ) + result = docker_exec_quiet(container, cmd, timeout=10) + if result is not None: + log.debug("Re-applied veth netem on %s:%s", container, iface) + else: + log.warning("Failed to re-apply veth netem on %s:%s", container, iface) log.info( - "Re-applied netem on %s (%d peers)", + "Re-applied veth netem on %s (%d Ethernet peers)", container, - len(container_states), + len(veth_states), ) - else: - log.warning("Failed to re-apply netem on %s", container) def mutate(self): """Randomly mutate netem params on a fraction of links.""" @@ -280,6 +346,8 @@ class NetemManager: def _update_link(self, node_a: str, node_b: str, params: NetemParams): """Update netem on both directions of a link.""" + transport = self.topology.transport_for_edge(node_a, node_b) + for src, dst in [(node_a, node_b), (node_b, node_a)]: if src in self.down_nodes: continue @@ -295,26 +363,38 @@ class NetemManager: self.down_nodes.add(src) continue - dest_ip = self.topology.nodes[dst].docker_ip - - states = self.states.get(container, {}) - state = states.get(dest_ip) - if state is None: - continue - - cmd = ( - f"tc qdisc replace dev {IFACE} parent {state.class_id} " - f"handle {state.netem_handle} netem {params.to_tc_args()}" - ) - result = docker_exec_quiet(container, cmd) - if result is not None: - state.params = params - log.debug( - "Updated netem %s -> %s: %s", - src, - dst, - params.to_tc_args(), + if transport == "ethernet": + # Ethernet: simple netem replace on veth + iface = veth_interface_name(src, dst) + veth_states = self.veth_states.get(container, {}) + state = veth_states.get(iface) + if state is None: + continue + cmd = f"tc qdisc replace dev {iface} root netem {params.to_tc_args()}" + result = docker_exec_quiet(container, cmd) + if result is not None: + 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 + dest_ip = self.topology.nodes[dst].docker_ip + states = self.states.get(container, {}) + state = states.get(dest_ip) + if state is None: + continue + cmd = ( + f"tc qdisc replace dev {IFACE} parent {state.class_id} " + f"handle {state.netem_handle} netem {params.to_tc_args()}" ) + result = docker_exec_quiet(container, cmd) + if result is not None: + state.params = params + log.debug( + "Updated netem %s -> %s: %s", + src, + dst, + params.to_tc_args(), + ) def _sample_policy(self, policy: NetemPolicy) -> NetemParams: """Sample concrete params from a policy's ranges.""" diff --git a/testing/chaos/sim/nodes.py b/testing/chaos/sim/nodes.py index 4ca6972..8648107 100644 --- a/testing/chaos/sim/nodes.py +++ b/testing/chaos/sim/nodes.py @@ -42,11 +42,13 @@ class NodeManager: rng: random.Random, netem_mgr=None, down_nodes: set[str] | None = None, + veth_mgr=None, ): self.topology = topology self.config = config self.rng = rng self.netem_mgr = netem_mgr + self.veth_mgr = veth_mgr self.down_nodes = down_nodes or set() self.node_states: dict[str, NodeState] = { nid: NodeState(node_id=nid) for nid in topology.nodes @@ -149,10 +151,15 @@ class NodeManager: log.info("Node STARTED: %s (was down %.0fs)", node_id, down_for) + # Re-create veth pairs (container restart destroys netns) + if self.veth_mgr: + time.sleep(1) + self.veth_mgr.setup_node(node_id) + # Re-apply netem after a brief delay for the container to initialize if self.netem_mgr: - # Small delay for container networking to be ready - time.sleep(1) + if not self.veth_mgr: + time.sleep(1) self.netem_mgr.setup_node(node_id) def _would_disconnect(self, node_id: str) -> bool: diff --git a/testing/chaos/sim/runner.py b/testing/chaos/sim/runner.py index 9ce648a..6c94dec 100644 --- a/testing/chaos/sim/runner.py +++ b/testing/chaos/sim/runner.py @@ -21,6 +21,7 @@ from .nodes import NodeManager from .scenario import Scenario from .topology import SimTopology, generate_topology from .traffic import TrafficManager +from .veth import VethManager log = logging.getLogger(__name__) @@ -39,6 +40,7 @@ class SimRunner: self._down_nodes: set[str] = set() # Managers (initialized during setup) + self.veth_mgr: VethManager | None = None self.netem_mgr: NetemManager | None = None self.link_mgr: LinkManager | None = None self.traffic_mgr: TrafficManager | None = None @@ -125,7 +127,17 @@ class SimRunner: log.info("Starting %d containers...", len(self.topology.nodes)) docker_compose(self.compose_file, ["up", "-d"]) - # 6. Initialize managers + # 6. Set up veth pairs for Ethernet edges (before netem) + # + # The entrypoint script waits for configured Ethernet interfaces + # to appear before starting FIPS, so we just need to create the + # veth pairs promptly after containers are running. + if self.topology.has_ethernet(): + self.veth_mgr = VethManager(self.topology) + log.info("Setting up Ethernet veth pairs...") + self.veth_mgr.setup_all() + + # 7. Initialize managers if s.netem.enabled: bw = s.bandwidth if s.bandwidth.enabled else None self.netem_mgr = NetemManager(self.topology, s.netem, self.rng, bandwidth=bw) @@ -147,6 +159,7 @@ class SimRunner: self.node_mgr = NodeManager( self.topology, s.node_churn, self.rng, netem_mgr=self.netem_mgr, down_nodes=self._down_nodes, + veth_mgr=self.veth_mgr, ) def _warmup(self): @@ -269,6 +282,11 @@ class SimRunner: topology=self.topology, ) + # Clean up veth pairs + if self.veth_mgr: + log.info("Cleaning up veth pairs...") + self.veth_mgr.teardown_all() + # Stop containers log.info("Stopping containers...") docker_compose( diff --git a/testing/chaos/sim/scenario.py b/testing/chaos/sim/scenario.py index 409ba82..b730dc1 100644 --- a/testing/chaos/sim/scenario.py +++ b/testing/chaos/sim/scenario.py @@ -22,6 +22,9 @@ class Range: raise ValueError(f"{name}: min ({self.min}) must be >= 0") +VALID_TRANSPORTS = ("udp", "ethernet") + + @dataclass class TopologyConfig: num_nodes: int = 10 @@ -30,6 +33,7 @@ class TopologyConfig: ensure_connected: bool = True subnet: str = "172.20.0.0/24" ip_start: int = 10 + default_transport: str = "udp" @dataclass @@ -186,6 +190,7 @@ def load_scenario(path: str) -> Scenario: s.topology.ensure_connected = tc.get("ensure_connected", True) s.topology.subnet = tc.get("subnet", "172.20.0.0/24") s.topology.ip_start = int(tc.get("ip_start", 10)) + s.topology.default_transport = tc.get("default_transport", "udp") # Netem section nc = raw.get("netem", {}) @@ -279,17 +284,30 @@ def _validate(s: Scenario): raise ValueError("topology.num_nodes must be <= 250 (subnet limit)") if s.topology.algorithm not in ("random_geometric", "erdos_renyi", "chain", "explicit"): raise ValueError(f"Unknown topology algorithm: {s.topology.algorithm}") + if s.topology.default_transport not in VALID_TRANSPORTS: + raise ValueError( + f"topology.default_transport: '{s.topology.default_transport}' " + f"not in {VALID_TRANSPORTS}" + ) if s.topology.algorithm == "explicit": adj = s.topology.params.get("adjacency") if not adj or not isinstance(adj, list): raise ValueError("explicit topology requires params.adjacency list") node_ids = set() - for i, pair in enumerate(adj): - if not isinstance(pair, (list, tuple)) or len(pair) != 2: + for i, entry in enumerate(adj): + if not isinstance(entry, (list, tuple)) or len(entry) not in (2, 3): raise ValueError( - f"explicit adjacency[{i}]: expected [nodeA, nodeB], got {pair}" + f"explicit adjacency[{i}]: expected [nodeA, nodeB] or " + f"[nodeA, nodeB, transport], got {entry}" ) - node_ids.update(str(p) for p in pair) + node_ids.update(str(p) for p in entry[:2]) + if len(entry) == 3: + transport = str(entry[2]) + if transport not in VALID_TRANSPORTS: + raise ValueError( + f"explicit adjacency[{i}]: transport '{transport}' " + f"not in {VALID_TRANSPORTS}" + ) if len(node_ids) != s.topology.num_nodes: raise ValueError( f"explicit adjacency references {len(node_ids)} nodes " diff --git a/testing/chaos/sim/topology.py b/testing/chaos/sim/topology.py index 29db6ff..37f8943 100644 --- a/testing/chaos/sim/topology.py +++ b/testing/chaos/sim/topology.py @@ -18,12 +18,41 @@ class SimNode: nsec: str # 64-char hex npub: str # bech32 npub1... peers: list[str] = field(default_factory=list) + # MAC addresses for Ethernet veth interfaces, keyed by peer_id + ethernet_macs: dict[str, str] = field(default_factory=dict) @dataclass class SimTopology: nodes: dict[str, SimNode] = field(default_factory=dict) edges: set[tuple[str, str]] = field(default_factory=set) + # Per-edge transport type; edges not in this dict default to "udp" + edge_transport: dict[tuple[str, str], str] = field(default_factory=dict) + + def transport_for_edge(self, a: str, b: str) -> str: + """Get the transport type for an edge (defaults to 'udp').""" + edge = _make_edge(a, b) + return self.edge_transport.get(edge, "udp") + + def ethernet_edges(self) -> list[tuple[str, str]]: + """Return all edges using Ethernet transport.""" + return [e for e, t in self.edge_transport.items() if t == "ethernet"] + + def has_ethernet(self) -> bool: + """Check if any edges use Ethernet transport.""" + return any(t == "ethernet" for t in self.edge_transport.values()) + + def ethernet_interfaces(self, node_id: str) -> list[str]: + """Return the veth interface names for a node's Ethernet edges.""" + ifaces = [] + for (a, b), transport in self.edge_transport.items(): + if transport != "ethernet": + continue + if a == node_id: + ifaces.append(veth_interface_name(a, b)) + elif b == node_id: + ifaces.append(veth_interface_name(b, a)) + return sorted(ifaces) def is_connected(self) -> bool: """BFS connectivity check.""" @@ -61,20 +90,35 @@ class SimTopology: return f"fips-node-{node_id}" def directed_outbound(self) -> dict[str, list[str]]: - """Assign each edge to exactly one node for outbound connection. + """Assign each UDP 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 one direction, ensuring auto-reconnect is testable — if B goes down, only A (the outbound owner) will attempt to reconnect. + Ethernet edges are excluded — they use beacon discovery instead + of static peer configuration. + 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 = { + e for e in self.edges + if self.edge_transport.get(e, "udp") == "udp" + } + outbound: dict[str, list[str]] = {nid: [] for nid in self.nodes} - # BFS spanning tree from first node + # 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) + + # BFS spanning tree from first node (over UDP edges only) root = min(self.nodes) visited: set[str] = set() tree_edges: set[tuple[str, str]] = set() @@ -82,15 +126,15 @@ class SimTopology: visited.add(root) while queue: node = queue.popleft() - for peer in self.nodes[node].peers: + for peer in udp_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 edges: lower ID → higher ID - for a, b in self.edges: + # Non-tree UDP edges: lower ID → higher ID + for a, b in udp_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 @@ -134,7 +178,9 @@ def generate_topology( adjacency = config.params.get("adjacency") if not adjacency: raise ValueError("explicit topology requires params.adjacency") - edges = _generate_explicit(adjacency) + edges, edge_transport = _generate_explicit( + adjacency, config.default_transport + ) # Validate all referenced nodes exist for a, b in edges: if a not in nodes: @@ -144,12 +190,16 @@ def generate_topology( else: raise ValueError(f"Unknown algorithm: {config.algorithm}") + # For non-explicit topologies, all edges use the default transport + if config.algorithm != "explicit": + edge_transport = {e: config.default_transport for e in edges} + # Build peer lists from edges for a, b in edges: nodes[a].peers.append(b) nodes[b].peers.append(a) - topo = SimTopology(nodes=nodes, edges=edges) + topo = SimTopology(nodes=nodes, edges=edges, edge_transport=edge_transport) # Connectivity check with retry if config.ensure_connected: @@ -223,19 +273,42 @@ def _generate_erdos_renyi( return edges -def _generate_explicit(adjacency: list) -> set[tuple[str, str]]: +def _generate_explicit( + adjacency: list, default_transport: str = "udp" +) -> tuple[set[tuple[str, str]], dict[tuple[str, str], str]]: """Build edges from an explicit adjacency list. - Each entry should be a 2-element list like ["n01", "n02"]. + Each entry is a 2-element list ``[nodeA, nodeB]`` (uses default + transport) or a 3-element list ``[nodeA, nodeB, transport]``. + + Returns ``(edges, edge_transport)`` where ``edge_transport`` maps + each edge to its transport type. """ edges = set() - for i, pair in enumerate(adjacency): - if not isinstance(pair, (list, tuple)) or len(pair) != 2: + edge_transport: dict[tuple[str, str], str] = {} + for i, entry in enumerate(adjacency): + if not isinstance(entry, (list, tuple)) or len(entry) not in (2, 3): raise ValueError( - f"explicit adjacency[{i}]: expected [nodeA, nodeB], got {pair}" + f"explicit adjacency[{i}]: expected [nodeA, nodeB] or " + f"[nodeA, nodeB, transport], got {entry}" ) - edges.add(_make_edge(str(pair[0]), str(pair[1]))) - return edges + edge = _make_edge(str(entry[0]), str(entry[1])) + edges.add(edge) + transport = str(entry[2]) if len(entry) == 3 else default_transport + edge_transport[edge] = transport + return edges, edge_transport + + +def veth_interface_name(local: str, peer: str) -> str: + """Generate the veth interface name inside a container. + + Format: ``ve-{local}-{peer}`` (max 15 chars for IFNAMSIZ). + For typical node IDs like "n01", this yields "ve-n01-n02" (10 chars). + """ + name = f"ve-{local}-{peer}" + if len(name) > 15: + raise ValueError(f"veth interface name too long: {name!r} ({len(name)} > 15)") + return name def _make_edge(a: str, b: str) -> tuple[str, str]: diff --git a/testing/chaos/sim/veth.py b/testing/chaos/sim/veth.py new file mode 100644 index 0000000..ed34ee4 --- /dev/null +++ b/testing/chaos/sim/veth.py @@ -0,0 +1,193 @@ +"""Veth pair management for Ethernet transport edges. + +Creates veth pairs between Docker containers for Ethernet-transport +edges. Each Ethernet edge gets a veth pair with one end moved into +each container's network namespace. Naming: + + Host (temporary): vh{NN}{MM}a / vh{NN}{MM}b + Container: ve-{local}-{peer} (via veth_interface_name()) + +After creation, the container-side MAC addresses are queried and +stored in SimNode.ethernet_macs for use in config generation. +""" + +from __future__ import annotations + +import logging +import subprocess + +from .docker_exec import docker_exec_quiet +from .topology import SimTopology, veth_interface_name + +log = logging.getLogger(__name__) + + +class VethManager: + """Manages veth pairs for Ethernet-transport edges.""" + + def __init__(self, topology: SimTopology): + self.topology = topology + # Track created host-side temp names for cleanup + self._host_pairs: list[tuple[str, str, str, str]] = [] + # (node_a, node_b, host_name_a, host_name_b) + + def setup_all(self): + """Create veth pairs for all Ethernet edges. + + For each Ethernet edge: + 1. Get container PIDs + 2. Create veth pair on host with temp names + 3. Move ends into container network namespaces + 4. Rename to final names and bring up + 5. Query MACs and store in SimNode.ethernet_macs + """ + eth_edges = self.topology.ethernet_edges() + if not eth_edges: + return + + log.info("Setting up %d Ethernet veth pairs...", len(eth_edges)) + + for a, b in eth_edges: + self._create_veth_pair(a, b) + + log.info( + "Veth setup complete: %d pairs", + len(self._host_pairs), + ) + + def setup_node(self, node_id: str): + """Re-create veth endpoints for a single node after container restart. + + When a container restarts (node churn), its network namespace is + destroyed. We re-create the veth pairs for all Ethernet edges + involving this node. + """ + for a, b in self.topology.ethernet_edges(): + if a != node_id and b != node_id: + continue + # Remove existing pair if any (host-side might still exist) + nn_a = a.replace("n", "") + nn_b = b.replace("n", "") + host_a = f"vh{nn_a}{nn_b}a" + _run_host(["ip", "link", "delete", host_a], check=False) + # Re-create + self._create_veth_pair(a, b) + + def teardown_all(self): + """Clean up all veth pairs.""" + for _, _, host_a, _ in self._host_pairs: + _run_host(["ip", "link", "delete", host_a], check=False) + self._host_pairs.clear() + + def _create_veth_pair(self, node_a: str, node_b: str): + """Create a single veth pair between two containers.""" + container_a = self.topology.container_name(node_a) + container_b = self.topology.container_name(node_b) + + # Get container PIDs + pid_a = _get_container_pid(container_a) + pid_b = _get_container_pid(container_b) + if pid_a is None or pid_b is None: + log.warning( + "Cannot create veth %s--%s: container PID not found", node_a, node_b + ) + return + + # Generate names + nn_a = node_a.replace("n", "") + nn_b = node_b.replace("n", "") + host_a = f"vh{nn_a}{nn_b}a" + host_b = f"vh{nn_a}{nn_b}b" + final_a = veth_interface_name(node_a, node_b) + final_b = veth_interface_name(node_b, node_a) + + # Clean up any stale pair + _run_host(["ip", "link", "delete", host_a], check=False) + + # Create veth pair on host + ok = _run_host([ + "ip", "link", "add", host_a, "type", "veth", "peer", "name", host_b, + ]) + if not ok: + log.warning("Failed to create veth pair %s/%s", host_a, host_b) + return + + # Move into container namespaces + _run_host(["ip", "link", "set", host_a, "netns", str(pid_a)]) + _run_host(["ip", "link", "set", host_b, "netns", str(pid_b)]) + + # Rename and bring up inside containers + docker_exec_quiet( + container_a, + f"ip link set {host_a} name {final_a} && ip link set {final_a} up", + timeout=10, + ) + docker_exec_quiet( + container_b, + f"ip link set {host_b} name {final_b} && ip link set {final_b} up", + timeout=10, + ) + + # Query MAC addresses + mac_a = _get_mac_in_container(container_a, final_a) + mac_b = _get_mac_in_container(container_b, final_b) + + if mac_a: + self.topology.nodes[node_a].ethernet_macs[node_b] = mac_a + if mac_b: + self.topology.nodes[node_b].ethernet_macs[node_a] = mac_b + + self._host_pairs.append((node_a, node_b, host_a, host_b)) + + log.info( + "Veth %s(%s) -- %s(%s) MAC: %s / %s", + node_a, final_a, node_b, final_b, + mac_a or "?", mac_b or "?", + ) + + +def _get_container_pid(container: str) -> int | None: + """Get the PID of a running Docker container.""" + try: + result = subprocess.run( + ["docker", "inspect", "-f", "{{.State.Pid}}", container], + capture_output=True, + text=True, + timeout=10, + ) + if result.returncode == 0: + pid = int(result.stdout.strip()) + return pid if pid > 0 else None + except (subprocess.TimeoutExpired, ValueError): + pass + return None + + +def _get_mac_in_container(container: str, iface: str) -> str | None: + """Query the MAC address of an interface inside a container.""" + result = docker_exec_quiet( + container, + f"cat /sys/class/net/{iface}/address", + timeout=5, + ) + if result is not None: + return result.strip() + return None + + +def _run_host(cmd: list[str], check: bool = True) -> bool: + """Run a command on the host. Returns True on success.""" + try: + result = subprocess.run( + cmd, + capture_output=True, + text=True, + timeout=10, + ) + if check and result.returncode != 0: + log.debug("Host cmd failed: %s -> %s", " ".join(cmd), result.stderr.strip()) + return False + return result.returncode == 0 + except subprocess.TimeoutExpired: + log.warning("Host cmd timed out: %s", " ".join(cmd)) + return False