Add Ethernet transport with beacon discovery

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
This commit is contained in:
Johnathan Corgan
2026-02-26 00:03:14 +00:00
parent 7260ad2878
commit d29da442ac
30 changed files with 2967 additions and 305 deletions
+88 -3
View File
@@ -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..."
+83 -6
View File
@@ -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<DiscoveredPeer> 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 |
+6
View File
@@ -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
+1 -1
View File
@@ -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";
+114 -10
View File
@@ -131,6 +131,113 @@ impl<T> Default for TransportInstances<T> {
}
}
/// 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<u16>,
/// 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<u16>,
/// Receive buffer size in bytes. Default: 2 MB.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub recv_buf_size: Option<usize>,
/// Send buffer size in bytes. Default: 2 MB.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub send_buf_size: Option<usize>,
/// Listen for discovery beacons from other nodes. Default: true.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub discovery: Option<bool>,
/// Broadcast announcement beacons on the LAN. Default: false.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub announce: Option<bool>,
/// Auto-connect to discovered peers. Default: false.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub auto_connect: Option<bool>,
/// Accept incoming connection attempts. Default: false.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub accept_connections: Option<bool>,
/// Announcement beacon interval in seconds. Default: 30.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub beacon_interval_secs: Option<u64>,
}
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<UdpConfig>,
// Future transport types:
// #[serde(default, skip_serializing_if = "is_transport_empty")]
// pub tcp: TransportInstances<TcpConfig>,
//
// #[serde(default, skip_serializing_if = "is_transport_empty")]
// pub tor: TransportInstances<TorConfig>,
/// Ethernet transport instances.
#[serde(default, skip_serializing_if = "is_transport_empty")]
pub ethernet: TransportInstances<EthernetConfig>,
}
/// Helper for skip_serializing_if on TransportInstances.
@@ -157,9 +261,7 @@ fn is_transport_empty<T>(instances: &TransportInstances<T>) -> 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;
}
}
}
+8
View File
@@ -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,
+1
View File
@@ -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;
}
}
}
+218 -122
View File
@@ -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.
+92 -12
View File
@@ -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 ===
+210
View File
@@ -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;
}
+2
View File
@@ -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;
+4 -4
View File
@@ -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);
+166
View File
@@ -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<XOnlyPublicKey> {
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<Vec<DiscoveredPeer>>,
}
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<DiscoveredPeer> {
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);
}
}
+629
View File
@@ -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<String>,
/// Configuration.
config: EthernetConfig,
/// Current state.
state: TransportState,
/// Async socket (None until started).
socket: Option<Arc<AsyncPacketSocket>>,
/// Channel for delivering received packets to Node.
packet_tx: PacketTx,
/// Receive loop task handle.
recv_task: Option<JoinHandle<()>>,
/// Beacon sender task handle.
beacon_task: Option<JoinHandle<()>>,
/// 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<DiscoveryBuffer>,
/// Transport-level statistics.
stats: Arc<EthernetStats>,
/// Node's public key for beacon construction.
local_pubkey: Option<XOnlyPublicKey>,
}
impl EthernetTransport {
/// Create a new Ethernet transport.
pub fn new(
transport_id: TransportId,
name: Option<String>,
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<EthernetStats> {
&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<usize, TransportError> {
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<Vec<DiscoveredPeer>, 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<AsyncPacketSocket>,
transport_id: TransportId,
packet_tx: PacketTx,
mtu: u16,
discovery_enabled: bool,
discovery_buffer: Arc<DiscoveryBuffer>,
stats: Arc<EthernetStats>,
) {
// 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<AsyncPacketSocket>,
pubkey: XOnlyPublicKey,
interval_secs: u64,
stats: Arc<EthernetStats>,
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, &ETHERNET_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, &ETHERNET_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);
}
}
+388
View File
@@ -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<Self, TransportError> {
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 <binary>)"
.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::<libc::sockaddr_ll>() 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<u16, TransportError> {
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::<libc::c_int>() 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::<libc::c_int>() 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<usize> {
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::<libc::sockaddr_ll>() 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::<libc::sockaddr_ll>() 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<AsyncPacketSocket, TransportError> {
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<PacketSocket>,
}
impl AsyncPacketSocket {
/// Send a payload to a destination MAC address.
pub async fn send_to(&self, data: &[u8], dest_mac: &[u8; 6]) -> Result<usize, TransportError> {
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<i32, TransportError> {
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<u16, TransportError> {
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)
}
+107
View File
@@ -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,
}
+77 -2
View File
@@ -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<Vec<DiscoveredPeer>, 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<usize, TransportError> {
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<std::net::SocketAddr> {
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<Vec<DiscoveredPeer>, 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(),
}
}
+4 -2
View File
@@ -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"]
+51
View File
@@ -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: <name>" 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"
@@ -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"
@@ -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"
+1
View File
@@ -25,6 +25,7 @@ x-fips-common: &fips-common
context: ../..
cap_add:
- NET_ADMIN
- NET_RAW
devices:
- /dev/net/tun:/dev/net/tun
sysctls:
+56 -1
View File
@@ -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_<ID>=<npub> for all nodes."""
lines = []
+34 -17
View File
@@ -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."""
+184 -104
View File
@@ -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."""
+9 -2
View File
@@ -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:
+19 -1
View File
@@ -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(
+22 -4
View File
@@ -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 "
+87 -14
View File
@@ -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]:
+193
View File
@@ -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