mirror of
https://github.com/jmcorgan/fips.git
synced 2026-07-30 19:46:15 +00:00
Add TCP transport implementation and test harness support
Implement TCP transport for FIPS enabling firewall traversal and serving as the foundation for future Tor transport. This is the first connection-oriented transport in the system. Key design decisions: - FMP header-based framing: reuses existing 4-byte FMP common prefix for packet boundary recovery with zero framing overhead - Session survives TCP reconnection: Noise/MMP/FSP state bound to npub, not TCP connection; MMP liveness is sole authority for peer death - Connect-on-send: fresh connection on first send, transparent reconnect - close_connection() trait method for cross-connection deduplication cleanup New transport files: - src/transport/tcp/mod.rs: TcpTransport, connection pool, accept loop - src/transport/tcp/stream.rs: FMP-aware stream reader (shared with Tor) Modified: transport trait (close_connection), TcpConfig, TransportHandle match arms, create_transports(), initiate_connection() for connection- oriented links, cross-connection tie-breaker cleanup, design docs. Tree announce loop and TCP stability fixes: - Preserve tree announce rate-limit state across reconnection: carry forward last_tree_announce_sent_ms when a peer reconnects so the rate-limit window isn't reset to zero - Drop oversize TCP packets at sender: pre-send MTU check returns MtuExceeded instead of writing to the stream, preventing receiver-side connection teardown and reset-reconnect cycles Chaos harness: - TCP transport support: tcp_edges/has_tcp/tcp_peers in SimTopology, transport-aware config_gen with per-edge transport type, TCP port 443, pure-TCP node support - Include all non-Ethernet edges in directed_outbound() - Fix netem/links log messages to say "IP-based" instead of "UDP" - Add tcp-chain, tcp-only, and tcp-mesh scenario files Static harness: - Transport-aware config generation (get_default_transport, transport_port) - TCP transport injection via Python post-processing - Add tcp-chain topology and docker-compose profile
This commit is contained in:
Johnathan Corgan
@@ -285,6 +285,37 @@ Each named instance operates independently with its own socket and
|
||||
discovery state. The instance name is used in log messages and the
|
||||
`name()` method on the Transport trait.
|
||||
|
||||
### TCP (`transports.tcp.*`)
|
||||
|
||||
TCP transport enables firewall traversal on networks that block UDP but
|
||||
allow TCP (e.g., port 443). Uses FMP header-based framing with zero
|
||||
overhead.
|
||||
|
||||
| Parameter | Type | Default | Description |
|
||||
|-----------|------|---------|-------------|
|
||||
| `transports.tcp.bind_addr` | string | *(none)* | Listen address (e.g., `"0.0.0.0:443"`). If omitted, outbound-only mode. |
|
||||
| `transports.tcp.mtu` | u16 | `1400` | Default MTU. Per-connection MTU derived from `TCP_MAXSEG` when available. |
|
||||
| `transports.tcp.connect_timeout_ms` | u64 | `5000` | Outbound connect timeout in milliseconds |
|
||||
| `transports.tcp.nodelay` | bool | `true` | `TCP_NODELAY` (disable Nagle for low latency) |
|
||||
| `transports.tcp.keepalive_secs` | u64 | `30` | TCP keepalive interval in seconds (0 = disabled) |
|
||||
| `transports.tcp.recv_buf_size` | usize | `2097152` | Socket receive buffer size in bytes (2 MB) |
|
||||
| `transports.tcp.send_buf_size` | usize | `2097152` | Socket send buffer size in bytes (2 MB) |
|
||||
| `transports.tcp.max_inbound_connections` | usize | `256` | Maximum simultaneous inbound connections |
|
||||
| `transports.tcp.socks5_proxy` | string | *(none)* | SOCKS5 proxy for outbound connections (implementation deferred) |
|
||||
|
||||
**Named instances.** Like other transports, multiple TCP instances can
|
||||
be configured with named sub-keys:
|
||||
|
||||
```yaml
|
||||
transports:
|
||||
tcp:
|
||||
public:
|
||||
bind_addr: "0.0.0.0:443"
|
||||
tor:
|
||||
socks5_proxy: "127.0.0.1:9050"
|
||||
connect_timeout_ms: 30000
|
||||
```
|
||||
|
||||
## Peers (`peers[]`)
|
||||
|
||||
Static peer list. Each entry defines a peer to connect to.
|
||||
@@ -293,8 +324,8 @@ Static peer list. Each entry defines a peer to connect to.
|
||||
|-----------|------|---------|-------------|
|
||||
| `peers[].npub` | string | *(required)* | Peer's Nostr public key (npub-encoded) |
|
||||
| `peers[].alias` | string | *(none)* | Human-readable name for logging |
|
||||
| `peers[].addresses[].transport` | string | *(required)* | Transport type: `udp` or `ethernet` |
|
||||
| `peers[].addresses[].addr` | string | *(required)* | Transport address. UDP: `"ip:port"`. Ethernet: `"interface/mac"` (e.g., `"eth0/aa:bb:cc:dd:ee:ff"`) |
|
||||
| `peers[].addresses[].transport` | string | *(required)* | Transport type: `udp`, `tcp`, or `ethernet` |
|
||||
| `peers[].addresses[].addr` | string | *(required)* | Transport address. UDP/TCP: `"ip:port"`. Ethernet: `"interface/mac"` (e.g., `"eth0/aa:bb:cc:dd:ee:ff"`) |
|
||||
| `peers[].addresses[].priority` | u8 | `100` | Address priority (lower = preferred) |
|
||||
| `peers[].connect_policy` | string | `"auto_connect"` | Connection policy: `auto_connect`, `on_demand`, or `manual` |
|
||||
| `peers[].auto_reconnect` | bool | `true` | Automatically reconnect after MMP link-dead removal (exponential backoff, unlimited retries) |
|
||||
@@ -483,6 +514,16 @@ transports:
|
||||
# auto_connect: false # connect to discovered peers
|
||||
# accept_connections: false # accept inbound handshakes
|
||||
# beacon_interval_secs: 30 # beacon interval (min 10)
|
||||
# tcp: # uncomment to enable TCP transport
|
||||
# bind_addr: "0.0.0.0:443" # listen address (omit for outbound-only)
|
||||
# mtu: 1400 # default MTU
|
||||
# connect_timeout_ms: 5000 # outbound connect timeout
|
||||
# nodelay: true # TCP_NODELAY
|
||||
# keepalive_secs: 30 # keepalive interval (0 = disabled)
|
||||
# recv_buf_size: 2097152 # 2 MB
|
||||
# send_buf_size: 2097152 # 2 MB
|
||||
# max_inbound_connections: 256 # resource protection limit
|
||||
# socks5_proxy: null # SOCKS5 for outbound (deferred)
|
||||
|
||||
peers: # static peer list
|
||||
# - npub: "npub1..."
|
||||
|
||||
@@ -312,6 +312,98 @@ Startup logging:
|
||||
Ethernet transport started name=eth0 interface=eth0 mac=aa:bb:cc:dd:ee:ff mtu=1499 if_mtu=1500
|
||||
```
|
||||
|
||||
## TCP/IP: Firewall Traversal Transport
|
||||
|
||||
For networks where UDP is blocked but TCP port 443 is open, the TCP
|
||||
transport provides an alternative path. It also serves as the foundation
|
||||
for the future Tor transport.
|
||||
|
||||
FIPS protocols (FMP, FSP, MMP) are all unreliable datagrams. Running them
|
||||
over TCP introduces head-of-line blocking, which adds latency jitter. MMP
|
||||
correctly measures this jitter, and cost-based parent selection naturally
|
||||
penalizes TCP links (higher SRTT leads to higher link cost). ETX will be
|
||||
1.0 over TCP since TCP handles retransmission.
|
||||
|
||||
### Architecture
|
||||
|
||||
Unlike UDP (one socket serves all peers), TCP requires one `TcpStream` per
|
||||
peer. The transport maintains a connection pool (`HashMap<TransportAddr,
|
||||
TcpConnection>`) plus an optional `TcpListener` for inbound connections.
|
||||
|
||||
| Property | Value |
|
||||
| -------- | ----- |
|
||||
| Addressing | IP:port (same as UDP) |
|
||||
| Default MTU | 1400 bytes |
|
||||
| Per-link MTU | Derived from `TCP_MAXSEG` socket option |
|
||||
| Framing | FMP header-based (zero overhead) |
|
||||
| Connection model | Connect-on-send, optional listener |
|
||||
| Platform | Cross-platform (no `#[cfg]` gates) |
|
||||
|
||||
### FMP Header-Based Framing
|
||||
|
||||
TCP is a byte stream; FIPS packets need delineation. Rather than adding a
|
||||
separate length-prefix layer, the TCP transport uses the existing 4-byte
|
||||
FMP common prefix `[ver+phase:1][flags:1][payload_len:2 LE]` to determine
|
||||
packet boundaries:
|
||||
|
||||
- **Phase 0x0 (established)**: remaining = 12 + payload_len + 16 (header + AEAD tag)
|
||||
- **Phase 0x1 (msg1)**: remaining = payload_len (fixed at 110, total 114 bytes)
|
||||
- **Phase 0x2 (msg2)**: remaining = payload_len (fixed at 65, total 69 bytes)
|
||||
- **Unknown phase**: close connection (protocol error)
|
||||
|
||||
This provides zero framing overhead and built-in phase validation. The
|
||||
stream reader is implemented in a separate module (`stream.rs`) for reuse
|
||||
by the future Tor transport.
|
||||
|
||||
### Connect-on-Send
|
||||
|
||||
When `send(addr, data)` is called with no existing connection:
|
||||
|
||||
1. Connect with configurable timeout (default 5s)
|
||||
2. Configure socket: `TCP_NODELAY`, keepalive, buffer sizes
|
||||
3. Read `TCP_MAXSEG` for per-connection MTU
|
||||
4. Split stream into read/write halves
|
||||
5. Spawn per-connection receive task
|
||||
6. Store connection in pool
|
||||
7. Write packet directly to stream
|
||||
|
||||
If connect fails, return error. The node's handshake retry mechanism
|
||||
handles re-attempts.
|
||||
|
||||
### Session Independence
|
||||
|
||||
TCP connection loss does **not** tear down the FIPS peer. Noise keys, MMP
|
||||
state, and FSP sessions are bound to the peer's npub, not the TCP
|
||||
connection. The transport reconnects transparently on the next send via
|
||||
connect-on-send. MMP liveness timeout is the sole authority for peer death.
|
||||
|
||||
### Connection Deduplication
|
||||
|
||||
Simultaneous outbound connections from both sides are resolved by the
|
||||
existing cross-connection tie-breaker in `promote_connection`. The losing
|
||||
TCP connection is closed via `Transport::close_connection(addr)`, which
|
||||
removes it from the pool and aborts its receive task.
|
||||
|
||||
### Configuration
|
||||
|
||||
```yaml
|
||||
transports:
|
||||
tcp:
|
||||
bind_addr: "0.0.0.0:443" # Listen address (omit for outbound-only)
|
||||
mtu: 1400 # Default MTU
|
||||
connect_timeout_ms: 5000 # Outbound connect timeout
|
||||
nodelay: true # TCP_NODELAY (disable Nagle)
|
||||
keepalive_secs: 30 # TCP keepalive interval (0 = disabled)
|
||||
recv_buf_size: 2097152 # SO_RCVBUF (2 MB)
|
||||
send_buf_size: 2097152 # SO_SNDBUF (2 MB)
|
||||
max_inbound_connections: 256 # Resource protection limit
|
||||
socks5_proxy: "127.0.0.1:9050" # SOCKS5 for outbound (deferred)
|
||||
```
|
||||
|
||||
If `bind_addr` is configured, the transport accepts inbound connections.
|
||||
Without it, the transport operates in outbound-only mode (no listener
|
||||
socket is created).
|
||||
|
||||
## Discovery
|
||||
|
||||
Discovery determines that a FIPS-capable endpoint is reachable at a given
|
||||
@@ -369,12 +461,13 @@ Key properties:
|
||||
|
||||
### Current State
|
||||
|
||||
> **Implemented**: UDP peers are configured via YAML. Ethernet peers are
|
||||
> discovered via beacon broadcast — the `discover()` trait method returns
|
||||
> newly seen endpoints, and per-transport `auto_connect()` /
|
||||
> `accept_connections()` policies control whether discovered peers are
|
||||
> connected automatically or require explicit configuration. Nostr relay
|
||||
> discovery is not yet implemented.
|
||||
> **Implemented**: UDP and TCP peers are configured via YAML. Ethernet
|
||||
> peers are discovered via beacon broadcast — the `discover()` trait
|
||||
> method returns newly seen endpoints, and per-transport `auto_connect()`
|
||||
> / `accept_connections()` policies control whether discovered peers are
|
||||
> connected automatically or require explicit configuration. TCP has no
|
||||
> discovery mechanism (peers are configured). Nostr relay discovery is
|
||||
> not yet implemented.
|
||||
|
||||
## Transport Interface
|
||||
|
||||
@@ -392,6 +485,7 @@ link_mtu(addr) → u16 Per-link MTU (defaults to mtu())
|
||||
start() → lifecycle Bring transport up (bind socket, open device)
|
||||
stop() → lifecycle Bring transport down
|
||||
send(addr, data) → delivery Send datagram to transport address
|
||||
close_connection(addr)→ () Close a specific connection (no-op for connectionless)
|
||||
discover() → Vec<DiscoveredPeer> Report discovered FIPS endpoints (optional)
|
||||
auto_connect() → bool Auto-connect discovered peers (default: false)
|
||||
accept_connections() → bool Accept inbound handshakes (default: true)
|
||||
@@ -449,7 +543,7 @@ transitions through `Starting` to `Up` (operational). `stop()` moves to
|
||||
| Transport | Status | Notes |
|
||||
| --------- | ------ | ----- |
|
||||
| UDP/IP | **Implemented** | Primary transport, async send/receive, configurable MTU |
|
||||
| TCP/IP | Future direction | Requires stream framing, TCP-over-TCP concern |
|
||||
| TCP/IP | **Implemented** | FMP header-based framing, connect-on-send, per-connection MSS MTU |
|
||||
| Ethernet | **Implemented** | AF_PACKET SOCK_DGRAM, EtherType 0x2121, beacon discovery, Linux only |
|
||||
| WiFi | Future direction | Infrastructure mode = Ethernet driver |
|
||||
| Tor | Future direction | High latency, .onion addressing |
|
||||
|
||||
+1
-1
@@ -33,7 +33,7 @@ pub use node::{
|
||||
NodeConfig, RateLimitConfig, RetryConfig, SessionConfig, SessionMmpConfig, TreeConfig,
|
||||
};
|
||||
pub use peer::{ConnectPolicy, PeerAddress, PeerConfig};
|
||||
pub use transport::{EthernetConfig, TransportInstances, TransportsConfig, UdpConfig};
|
||||
pub use transport::{EthernetConfig, TcpConfig, TransportInstances, TransportsConfig, UdpConfig};
|
||||
|
||||
/// Default config filename.
|
||||
const CONFIG_FILENAME: &str = "fips.yaml";
|
||||
|
||||
+113
-1
@@ -238,6 +238,111 @@ impl EthernetConfig {
|
||||
}
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// TCP Transport Configuration
|
||||
// ============================================================================
|
||||
|
||||
/// Default TCP MTU (conservative, matches typical Ethernet MSS minus overhead).
|
||||
const DEFAULT_TCP_MTU: u16 = 1400;
|
||||
|
||||
/// Default TCP connect timeout in milliseconds.
|
||||
const DEFAULT_TCP_CONNECT_TIMEOUT_MS: u64 = 5000;
|
||||
|
||||
/// Default TCP keepalive interval in seconds.
|
||||
const DEFAULT_TCP_KEEPALIVE_SECS: u64 = 30;
|
||||
|
||||
/// Default TCP receive buffer size (2 MB).
|
||||
const DEFAULT_TCP_RECV_BUF: usize = 2 * 1024 * 1024;
|
||||
|
||||
/// Default TCP send buffer size (2 MB).
|
||||
const DEFAULT_TCP_SEND_BUF: usize = 2 * 1024 * 1024;
|
||||
|
||||
/// Default maximum inbound TCP connections.
|
||||
const DEFAULT_TCP_MAX_INBOUND: usize = 256;
|
||||
|
||||
/// TCP transport instance configuration.
|
||||
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
|
||||
#[serde(deny_unknown_fields)]
|
||||
pub struct TcpConfig {
|
||||
/// Listen address (e.g., "0.0.0.0:443"). If not set, outbound-only.
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub bind_addr: Option<String>,
|
||||
|
||||
/// Default MTU for TCP connections. Defaults to 1400.
|
||||
/// Per-connection MTU is derived from TCP_MAXSEG when available.
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub mtu: Option<u16>,
|
||||
|
||||
/// Outbound connect timeout in milliseconds. Defaults to 5000.
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub connect_timeout_ms: Option<u64>,
|
||||
|
||||
/// Enable TCP_NODELAY (disable Nagle). Defaults to true.
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub nodelay: Option<bool>,
|
||||
|
||||
/// TCP keepalive interval in seconds. 0 = disabled. Defaults to 30.
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub keepalive_secs: Option<u64>,
|
||||
|
||||
/// TCP receive buffer size in bytes. Defaults to 2 MB.
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub recv_buf_size: Option<usize>,
|
||||
|
||||
/// TCP send buffer size in bytes. Defaults to 2 MB.
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub send_buf_size: Option<usize>,
|
||||
|
||||
/// SOCKS5 proxy for outbound connections (placeholder; not yet implemented).
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub socks5_proxy: Option<String>,
|
||||
|
||||
/// Maximum simultaneous inbound connections. Defaults to 256.
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub max_inbound_connections: Option<usize>,
|
||||
}
|
||||
|
||||
impl TcpConfig {
|
||||
/// Get the default MTU.
|
||||
pub fn mtu(&self) -> u16 {
|
||||
self.mtu.unwrap_or(DEFAULT_TCP_MTU)
|
||||
}
|
||||
|
||||
/// Get the connect timeout in milliseconds.
|
||||
pub fn connect_timeout_ms(&self) -> u64 {
|
||||
self.connect_timeout_ms.unwrap_or(DEFAULT_TCP_CONNECT_TIMEOUT_MS)
|
||||
}
|
||||
|
||||
/// Whether TCP_NODELAY is enabled. Default: true.
|
||||
pub fn nodelay(&self) -> bool {
|
||||
self.nodelay.unwrap_or(true)
|
||||
}
|
||||
|
||||
/// Get the keepalive interval in seconds. 0 = disabled. Default: 30.
|
||||
pub fn keepalive_secs(&self) -> u64 {
|
||||
self.keepalive_secs.unwrap_or(DEFAULT_TCP_KEEPALIVE_SECS)
|
||||
}
|
||||
|
||||
/// Get the receive buffer size. Default: 2 MB.
|
||||
pub fn recv_buf_size(&self) -> usize {
|
||||
self.recv_buf_size.unwrap_or(DEFAULT_TCP_RECV_BUF)
|
||||
}
|
||||
|
||||
/// Get the send buffer size. Default: 2 MB.
|
||||
pub fn send_buf_size(&self) -> usize {
|
||||
self.send_buf_size.unwrap_or(DEFAULT_TCP_SEND_BUF)
|
||||
}
|
||||
|
||||
/// Get the maximum number of inbound connections. Default: 256.
|
||||
pub fn max_inbound_connections(&self) -> usize {
|
||||
self.max_inbound_connections.unwrap_or(DEFAULT_TCP_MAX_INBOUND)
|
||||
}
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// TransportsConfig
|
||||
// ============================================================================
|
||||
|
||||
/// Transports configuration section.
|
||||
///
|
||||
/// Each transport type can have either a single instance (config directly
|
||||
@@ -251,6 +356,10 @@ pub struct TransportsConfig {
|
||||
/// Ethernet transport instances.
|
||||
#[serde(default, skip_serializing_if = "is_transport_empty")]
|
||||
pub ethernet: TransportInstances<EthernetConfig>,
|
||||
|
||||
/// TCP transport instances.
|
||||
#[serde(default, skip_serializing_if = "is_transport_empty")]
|
||||
pub tcp: TransportInstances<TcpConfig>,
|
||||
}
|
||||
|
||||
/// Helper for skip_serializing_if on TransportInstances.
|
||||
@@ -261,7 +370,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.ethernet.is_empty()
|
||||
self.udp.is_empty() && self.ethernet.is_empty() && self.tcp.is_empty()
|
||||
}
|
||||
|
||||
/// Merge another TransportsConfig into this one.
|
||||
@@ -274,5 +383,8 @@ impl TransportsConfig {
|
||||
if !other.ethernet.is_empty() {
|
||||
self.ethernet = other.ethernet;
|
||||
}
|
||||
if !other.tcp.is_empty() {
|
||||
self.tcp = other.tcp;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -280,6 +280,14 @@ impl Node {
|
||||
if let Some(peer) = self.peers.get_mut(&node_addr) {
|
||||
peer.set_handshake_msg2(wire_msg2.clone());
|
||||
}
|
||||
// Close the losing TCP connection (no-op for connectionless)
|
||||
if let Some(loser_link) = self.links.get(&loser_link_id) {
|
||||
let loser_tid = loser_link.transport_id();
|
||||
let loser_addr = loser_link.remote_addr().clone();
|
||||
if let Some(transport) = self.transports.get(&loser_tid) {
|
||||
transport.close_connection(&loser_addr).await;
|
||||
}
|
||||
}
|
||||
// Clean up the losing connection's link
|
||||
self.remove_link(&loser_link_id);
|
||||
info!(
|
||||
@@ -295,6 +303,10 @@ impl Node {
|
||||
self.bloom_state.mark_update_needed(node_addr);
|
||||
}
|
||||
PromotionResult::CrossConnectionLost { winner_link_id } => {
|
||||
// Close the losing TCP connection (no-op for connectionless)
|
||||
if let Some(transport) = self.transports.get(&packet.transport_id) {
|
||||
transport.close_connection(&packet.remote_addr).await;
|
||||
}
|
||||
// This connection lost — clean up its link
|
||||
self.remove_link(&link_id);
|
||||
// Restore addr_to_link for the winner's link
|
||||
@@ -519,6 +531,14 @@ impl Node {
|
||||
|
||||
// Clean up outbound connection state
|
||||
self.pending_outbound.remove(&key);
|
||||
// Close the losing TCP connection (no-op for connectionless)
|
||||
if let Some(link) = self.links.get(&link_id) {
|
||||
let tid = link.transport_id();
|
||||
let addr = link.remote_addr().clone();
|
||||
if let Some(transport) = self.transports.get(&tid) {
|
||||
transport.close_connection(&addr).await;
|
||||
}
|
||||
}
|
||||
self.remove_link(&link_id);
|
||||
|
||||
// Send TreeAnnounce now that sessions are aligned
|
||||
@@ -550,6 +570,14 @@ impl Node {
|
||||
self.bloom_state.mark_update_needed(node_addr);
|
||||
}
|
||||
PromotionResult::CrossConnectionWon { loser_link_id, node_addr } => {
|
||||
// Close the losing TCP connection (no-op for connectionless)
|
||||
if let Some(loser_link) = self.links.get(&loser_link_id) {
|
||||
let loser_tid = loser_link.transport_id();
|
||||
let loser_addr = loser_link.remote_addr().clone();
|
||||
if let Some(transport) = self.transports.get(&loser_tid) {
|
||||
transport.close_connection(&loser_addr).await;
|
||||
}
|
||||
}
|
||||
// Clean up the losing connection's link
|
||||
self.remove_link(&loser_link_id);
|
||||
// Ensure addr_to_link points to the winning link
|
||||
@@ -570,6 +598,10 @@ impl Node {
|
||||
self.bloom_state.mark_update_needed(node_addr);
|
||||
}
|
||||
PromotionResult::CrossConnectionLost { winner_link_id } => {
|
||||
// Close the losing TCP connection (no-op for connectionless)
|
||||
if let Some(transport) = self.transports.get(&packet.transport_id) {
|
||||
transport.close_connection(&packet.remote_addr).await;
|
||||
}
|
||||
// This connection lost — clean up its link
|
||||
self.remove_link(&link_id);
|
||||
// Ensure addr_to_link points to the winner's link
|
||||
@@ -756,6 +788,12 @@ impl Node {
|
||||
return Err(NodeError::MaxPeersExceeded { max: self.max_peers });
|
||||
}
|
||||
|
||||
// Preserve tree announce rate-limit state from old peer (if reconnecting).
|
||||
// Without this, reconnection resets the rate limit window to zero,
|
||||
// allowing an immediate announce that can feed an announce loop.
|
||||
let old_announce_ts = self.peers.get(&peer_node_addr)
|
||||
.map(|p| p.last_tree_announce_sent_ms());
|
||||
|
||||
let mut new_peer = ActivePeer::with_session(
|
||||
verified_identity,
|
||||
link_id,
|
||||
@@ -771,6 +809,9 @@ impl Node {
|
||||
remote_epoch,
|
||||
);
|
||||
new_peer.set_tree_announce_min_interval_ms(self.config.node.tree.announce_min_interval_ms);
|
||||
if let Some(ts) = old_announce_ts {
|
||||
new_peer.set_last_tree_announce_sent_ms(ts);
|
||||
}
|
||||
|
||||
self.peers.insert(peer_node_addr, new_peer);
|
||||
self.peers_by_index
|
||||
|
||||
+17
-2
@@ -157,13 +157,28 @@ impl Node {
|
||||
// Allocate link ID and create link
|
||||
let link_id = self.allocate_link_id();
|
||||
|
||||
let link = Link::connectionless(
|
||||
// Use Link::new() for connection-oriented transports (Connecting state)
|
||||
// and Link::connectionless() for connectionless transports (Connected state)
|
||||
let link = if self.transports.get(&transport_id)
|
||||
.map(|t| t.transport_type().connection_oriented)
|
||||
.unwrap_or(false)
|
||||
{
|
||||
Link::new(
|
||||
link_id,
|
||||
transport_id,
|
||||
remote_addr.clone(),
|
||||
LinkDirection::Outbound,
|
||||
Duration::from_millis(self.config.node.base_rtt_ms),
|
||||
);
|
||||
)
|
||||
} else {
|
||||
Link::connectionless(
|
||||
link_id,
|
||||
transport_id,
|
||||
remote_addr.clone(),
|
||||
LinkDirection::Outbound,
|
||||
Duration::from_millis(self.config.node.base_rtt_ms),
|
||||
)
|
||||
};
|
||||
|
||||
self.links.insert(link_id, link);
|
||||
|
||||
|
||||
@@ -28,6 +28,7 @@ use crate::transport::{
|
||||
Link, LinkId, PacketRx, PacketTx, TransportAddr, TransportError, TransportHandle, TransportId,
|
||||
};
|
||||
use crate::transport::udp::UdpTransport;
|
||||
use crate::transport::tcp::TcpTransport;
|
||||
#[cfg(target_os = "linux")]
|
||||
use crate::transport::ethernet::EthernetTransport;
|
||||
use crate::tree::TreeState;
|
||||
@@ -598,6 +599,21 @@ impl Node {
|
||||
}
|
||||
}
|
||||
|
||||
// Create TCP transport instances
|
||||
let tcp_instances: Vec<_> = self
|
||||
.config
|
||||
.transports
|
||||
.tcp
|
||||
.iter()
|
||||
.map(|(name, config)| (name.map(|s| s.to_string()), config.clone()))
|
||||
.collect();
|
||||
|
||||
for (name, tcp_config) in tcp_instances {
|
||||
let transport_id = self.allocate_transport_id();
|
||||
let tcp = TcpTransport::new(transport_id, name, tcp_config, packet_tx.clone());
|
||||
transports.push(TransportHandle::Tcp(tcp));
|
||||
}
|
||||
|
||||
transports
|
||||
}
|
||||
|
||||
|
||||
@@ -622,6 +622,16 @@ impl ActivePeer {
|
||||
self.tree_announce_min_interval_ms = ms;
|
||||
}
|
||||
|
||||
/// Get the last tree announce send timestamp (for carrying across reconnection).
|
||||
pub fn last_tree_announce_sent_ms(&self) -> u64 {
|
||||
self.last_tree_announce_sent_ms
|
||||
}
|
||||
|
||||
/// Set the last tree announce send timestamp (to preserve rate limit across reconnection).
|
||||
pub fn set_last_tree_announce_sent_ms(&mut self, ms: u64) {
|
||||
self.last_tree_announce_sent_ms = ms;
|
||||
}
|
||||
|
||||
/// Check if we can send a TreeAnnounce now (rate limiting).
|
||||
pub fn can_send_tree_announce(&self, now_ms: u64) -> bool {
|
||||
now_ms.saturating_sub(self.last_tree_announce_sent_ms) >= self.tree_announce_min_interval_ms
|
||||
|
||||
@@ -5,12 +5,14 @@
|
||||
//! which FIPS links are established.
|
||||
|
||||
pub mod udp;
|
||||
pub mod tcp;
|
||||
|
||||
#[cfg(target_os = "linux")]
|
||||
pub mod ethernet;
|
||||
|
||||
use secp256k1::XOnlyPublicKey;
|
||||
use udp::UdpTransport;
|
||||
use tcp::TcpTransport;
|
||||
#[cfg(target_os = "linux")]
|
||||
use ethernet::EthernetTransport;
|
||||
use std::fmt;
|
||||
@@ -778,6 +780,15 @@ pub trait Transport {
|
||||
fn accept_connections(&self) -> bool {
|
||||
true
|
||||
}
|
||||
|
||||
/// Close a specific connection (connection-oriented transports only).
|
||||
///
|
||||
/// For connectionless transports (UDP, Ethernet), this is a no-op.
|
||||
/// Connection-oriented transports (TCP, Tor) remove the connection
|
||||
/// from their pool and drop the underlying stream.
|
||||
fn close_connection(&self, _addr: &TransportAddr) {
|
||||
// Default no-op for connectionless transports
|
||||
}
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
@@ -794,6 +805,8 @@ pub enum TransportHandle {
|
||||
/// Raw Ethernet transport.
|
||||
#[cfg(target_os = "linux")]
|
||||
Ethernet(EthernetTransport),
|
||||
/// TCP/IP transport.
|
||||
Tcp(TcpTransport),
|
||||
}
|
||||
|
||||
impl TransportHandle {
|
||||
@@ -803,6 +816,7 @@ impl TransportHandle {
|
||||
TransportHandle::Udp(t) => t.start_async().await,
|
||||
#[cfg(target_os = "linux")]
|
||||
TransportHandle::Ethernet(t) => t.start_async().await,
|
||||
TransportHandle::Tcp(t) => t.start_async().await,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -812,6 +826,7 @@ impl TransportHandle {
|
||||
TransportHandle::Udp(t) => t.stop_async().await,
|
||||
#[cfg(target_os = "linux")]
|
||||
TransportHandle::Ethernet(t) => t.stop_async().await,
|
||||
TransportHandle::Tcp(t) => t.stop_async().await,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -821,6 +836,7 @@ impl TransportHandle {
|
||||
TransportHandle::Udp(t) => t.send_async(addr, data).await,
|
||||
#[cfg(target_os = "linux")]
|
||||
TransportHandle::Ethernet(t) => t.send_async(addr, data).await,
|
||||
TransportHandle::Tcp(t) => t.send_async(addr, data).await,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -830,6 +846,7 @@ impl TransportHandle {
|
||||
TransportHandle::Udp(t) => t.transport_id(),
|
||||
#[cfg(target_os = "linux")]
|
||||
TransportHandle::Ethernet(t) => t.transport_id(),
|
||||
TransportHandle::Tcp(t) => t.transport_id(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -839,6 +856,7 @@ impl TransportHandle {
|
||||
TransportHandle::Udp(t) => t.name(),
|
||||
#[cfg(target_os = "linux")]
|
||||
TransportHandle::Ethernet(t) => t.name(),
|
||||
TransportHandle::Tcp(t) => t.name(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -848,6 +866,7 @@ impl TransportHandle {
|
||||
TransportHandle::Udp(t) => t.transport_type(),
|
||||
#[cfg(target_os = "linux")]
|
||||
TransportHandle::Ethernet(t) => t.transport_type(),
|
||||
TransportHandle::Tcp(t) => t.transport_type(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -857,6 +876,7 @@ impl TransportHandle {
|
||||
TransportHandle::Udp(t) => t.state(),
|
||||
#[cfg(target_os = "linux")]
|
||||
TransportHandle::Ethernet(t) => t.state(),
|
||||
TransportHandle::Tcp(t) => t.state(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -866,6 +886,7 @@ impl TransportHandle {
|
||||
TransportHandle::Udp(t) => t.mtu(),
|
||||
#[cfg(target_os = "linux")]
|
||||
TransportHandle::Ethernet(t) => t.mtu(),
|
||||
TransportHandle::Tcp(t) => t.mtu(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -878,6 +899,7 @@ impl TransportHandle {
|
||||
TransportHandle::Udp(t) => t.link_mtu(addr),
|
||||
#[cfg(target_os = "linux")]
|
||||
TransportHandle::Ethernet(t) => t.link_mtu(addr),
|
||||
TransportHandle::Tcp(t) => t.link_mtu(addr),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -887,6 +909,7 @@ impl TransportHandle {
|
||||
TransportHandle::Udp(t) => t.local_addr(),
|
||||
#[cfg(target_os = "linux")]
|
||||
TransportHandle::Ethernet(_) => None,
|
||||
TransportHandle::Tcp(t) => t.local_addr(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -896,6 +919,7 @@ impl TransportHandle {
|
||||
TransportHandle::Udp(_) => None,
|
||||
#[cfg(target_os = "linux")]
|
||||
TransportHandle::Ethernet(t) => Some(t.interface_name()),
|
||||
TransportHandle::Tcp(_) => None,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -905,6 +929,7 @@ impl TransportHandle {
|
||||
TransportHandle::Udp(t) => t.discover(),
|
||||
#[cfg(target_os = "linux")]
|
||||
TransportHandle::Ethernet(t) => t.discover(),
|
||||
TransportHandle::Tcp(t) => t.discover(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -914,6 +939,7 @@ impl TransportHandle {
|
||||
TransportHandle::Udp(t) => t.auto_connect(),
|
||||
#[cfg(target_os = "linux")]
|
||||
TransportHandle::Ethernet(t) => t.auto_connect(),
|
||||
TransportHandle::Tcp(t) => t.auto_connect(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -923,6 +949,20 @@ impl TransportHandle {
|
||||
TransportHandle::Udp(t) => t.accept_connections(),
|
||||
#[cfg(target_os = "linux")]
|
||||
TransportHandle::Ethernet(t) => t.accept_connections(),
|
||||
TransportHandle::Tcp(t) => t.accept_connections(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Close a specific connection on this transport.
|
||||
///
|
||||
/// No-op for connectionless transports. For TCP, removes the
|
||||
/// connection from the pool and drops the stream.
|
||||
pub async fn close_connection(&self, addr: &TransportAddr) {
|
||||
match self {
|
||||
TransportHandle::Udp(t) => t.close_connection(addr),
|
||||
#[cfg(target_os = "linux")]
|
||||
TransportHandle::Ethernet(t) => t.close_connection(addr),
|
||||
TransportHandle::Tcp(t) => t.close_connection_async(addr).await,
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,361 @@
|
||||
//! FMP-Aware Stream Reader
|
||||
//!
|
||||
//! Recovers FIPS packet boundaries from a TCP byte stream using the
|
||||
//! existing 4-byte FMP common prefix `[ver+phase:1][flags:1][payload_len:2 LE]`.
|
||||
//!
|
||||
//! This module is deliberately separate from the TCP transport so it can
|
||||
//! be reused by the future Tor transport.
|
||||
|
||||
use tokio::io::{AsyncRead, AsyncReadExt};
|
||||
|
||||
/// FMP phase values (low nibble of byte 0).
|
||||
const PHASE_ESTABLISHED: u8 = 0x0;
|
||||
const PHASE_MSG1: u8 = 0x1;
|
||||
const PHASE_MSG2: u8 = 0x2;
|
||||
|
||||
/// Size of the FMP common prefix.
|
||||
const PREFIX_SIZE: usize = 4;
|
||||
|
||||
/// Overhead for established frames: 12 bytes remaining header + 16 bytes AEAD tag.
|
||||
/// The full established header is 16 bytes (PREFIX_SIZE + 12), so after reading
|
||||
/// the 4-byte prefix, 12 more header bytes remain. Then payload_len bytes of
|
||||
/// ciphertext, then 16 bytes of AEAD tag.
|
||||
const ESTABLISHED_REMAINING_HEADER: usize = 12;
|
||||
const AEAD_TAG_SIZE: usize = 16;
|
||||
|
||||
/// Errors from the FMP stream reader.
|
||||
#[derive(Debug)]
|
||||
pub enum StreamError {
|
||||
/// Unknown FMP phase byte — protocol error, close connection.
|
||||
UnknownPhase(u8),
|
||||
/// Payload length exceeds the connection's MTU — corrupted or malicious.
|
||||
PayloadTooLarge { payload_len: u16, max_payload_len: u16 },
|
||||
/// Handshake packet has unexpected payload_len for its phase.
|
||||
HandshakeSizeMismatch { phase: u8, expected: u16, got: u16 },
|
||||
/// I/O error (including EOF).
|
||||
Io(std::io::Error),
|
||||
}
|
||||
|
||||
impl std::fmt::Display for StreamError {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
match self {
|
||||
StreamError::UnknownPhase(p) => write!(f, "unknown FMP phase: 0x{:02x}", p),
|
||||
StreamError::PayloadTooLarge { payload_len, max_payload_len } => {
|
||||
write!(f, "payload_len {} exceeds max {}", payload_len, max_payload_len)
|
||||
}
|
||||
StreamError::HandshakeSizeMismatch { phase, expected, got } => {
|
||||
write!(f, "handshake phase 0x{:x}: expected payload_len {}, got {}", phase, expected, got)
|
||||
}
|
||||
StreamError::Io(e) => write!(f, "io: {}", e),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl std::error::Error for StreamError {}
|
||||
|
||||
impl From<std::io::Error> for StreamError {
|
||||
fn from(e: std::io::Error) -> Self {
|
||||
StreamError::Io(e)
|
||||
}
|
||||
}
|
||||
|
||||
/// Known wire sizes for handshake messages.
|
||||
/// msg1: 4 (prefix) + 4 (sender_idx) + 106 (noise_msg1) = 114 bytes
|
||||
/// msg2: 4 (prefix) + 4 (sender_idx) + 4 (receiver_idx) + 57 (noise_msg2) = 69 bytes
|
||||
const MSG1_WIRE_SIZE: usize = 114;
|
||||
const MSG2_WIRE_SIZE: usize = 69;
|
||||
|
||||
/// Expected payload_len for msg1: sender_idx(4) + noise_msg1(106) = 110.
|
||||
const MSG1_PAYLOAD_LEN: u16 = (MSG1_WIRE_SIZE - PREFIX_SIZE) as u16;
|
||||
|
||||
/// Expected payload_len for msg2: sender_idx(4) + receiver_idx(4) + noise_msg2(57) = 65.
|
||||
const MSG2_PAYLOAD_LEN: u16 = (MSG2_WIRE_SIZE - PREFIX_SIZE) as u16;
|
||||
|
||||
/// Read one complete FMP packet from an async reader.
|
||||
///
|
||||
/// Uses the 4-byte FMP common prefix to determine the total packet size,
|
||||
/// then reads the remaining bytes. Returns the complete packet as a `Vec<u8>`.
|
||||
///
|
||||
/// # Arguments
|
||||
///
|
||||
/// * `reader` - Any async reader (typically an `OwnedReadHalf`)
|
||||
/// * `mtu` - The connection's MTU for validation of established frame sizes
|
||||
///
|
||||
/// # Errors
|
||||
///
|
||||
/// * `UnknownPhase` — unrecognized phase nibble (protocol error)
|
||||
/// * `PayloadTooLarge` — established frame exceeds MTU
|
||||
/// * `HandshakeSizeMismatch` — handshake packet has wrong payload_len
|
||||
/// * `Io` — underlying read error (including EOF)
|
||||
pub async fn read_fmp_packet<R: AsyncRead + Unpin>(
|
||||
reader: &mut R,
|
||||
mtu: u16,
|
||||
) -> Result<Vec<u8>, StreamError> {
|
||||
// Read the 4-byte FMP common prefix
|
||||
let mut prefix = [0u8; PREFIX_SIZE];
|
||||
reader.read_exact(&mut prefix).await?;
|
||||
|
||||
let phase = prefix[0] & 0x0F;
|
||||
let payload_len = u16::from_le_bytes([prefix[2], prefix[3]]);
|
||||
|
||||
// Compute remaining bytes based on phase
|
||||
let remaining = match phase {
|
||||
PHASE_ESTABLISHED => {
|
||||
// Validate payload_len against MTU:
|
||||
// total packet = 16 (header) + payload_len + 16 (tag) = payload_len + 32
|
||||
// max_payload_len = mtu - 32
|
||||
let max_payload_len = mtu.saturating_sub(
|
||||
(ESTABLISHED_REMAINING_HEADER + PREFIX_SIZE + AEAD_TAG_SIZE) as u16,
|
||||
);
|
||||
if payload_len > max_payload_len {
|
||||
return Err(StreamError::PayloadTooLarge {
|
||||
payload_len,
|
||||
max_payload_len,
|
||||
});
|
||||
}
|
||||
// remaining = 12 (rest of header) + payload_len + 16 (AEAD tag)
|
||||
ESTABLISHED_REMAINING_HEADER + payload_len as usize + AEAD_TAG_SIZE
|
||||
}
|
||||
PHASE_MSG1 => {
|
||||
if payload_len != MSG1_PAYLOAD_LEN {
|
||||
return Err(StreamError::HandshakeSizeMismatch {
|
||||
phase,
|
||||
expected: MSG1_PAYLOAD_LEN,
|
||||
got: payload_len,
|
||||
});
|
||||
}
|
||||
payload_len as usize
|
||||
}
|
||||
PHASE_MSG2 => {
|
||||
if payload_len != MSG2_PAYLOAD_LEN {
|
||||
return Err(StreamError::HandshakeSizeMismatch {
|
||||
phase,
|
||||
expected: MSG2_PAYLOAD_LEN,
|
||||
got: payload_len,
|
||||
});
|
||||
}
|
||||
payload_len as usize
|
||||
}
|
||||
_ => {
|
||||
return Err(StreamError::UnknownPhase(phase));
|
||||
}
|
||||
};
|
||||
|
||||
// Allocate buffer for the complete packet (prefix + remaining)
|
||||
let total = PREFIX_SIZE + remaining;
|
||||
let mut packet = vec![0u8; total];
|
||||
packet[..PREFIX_SIZE].copy_from_slice(&prefix);
|
||||
reader.read_exact(&mut packet[PREFIX_SIZE..]).await?;
|
||||
|
||||
Ok(packet)
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Tests
|
||||
// ============================================================================
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use std::io::Cursor;
|
||||
|
||||
/// Build a minimal established frame with the given payload_len.
|
||||
/// Layout: [ver+phase:1][flags:1][payload_len:2 LE][12 bytes header][payload_len bytes][16 bytes tag]
|
||||
fn build_established_frame(payload_len: u16) -> Vec<u8> {
|
||||
let total = PREFIX_SIZE + ESTABLISHED_REMAINING_HEADER + payload_len as usize + AEAD_TAG_SIZE;
|
||||
let mut frame = vec![0u8; total];
|
||||
frame[0] = 0x00; // ver=0, phase=0 (established)
|
||||
frame[1] = 0x00; // flags
|
||||
frame[2..4].copy_from_slice(&payload_len.to_le_bytes());
|
||||
// Fill remaining with pattern for verification
|
||||
for i in PREFIX_SIZE..total {
|
||||
frame[i] = (i & 0xFF) as u8;
|
||||
}
|
||||
frame
|
||||
}
|
||||
|
||||
/// Build a msg1 frame (114 bytes total).
|
||||
fn build_msg1_frame() -> Vec<u8> {
|
||||
let mut frame = vec![0xAA; MSG1_WIRE_SIZE];
|
||||
frame[0] = 0x01; // ver=0, phase=1
|
||||
frame[1] = 0x00; // flags
|
||||
frame[2..4].copy_from_slice(&MSG1_PAYLOAD_LEN.to_le_bytes());
|
||||
frame
|
||||
}
|
||||
|
||||
/// Build a msg2 frame (69 bytes total).
|
||||
fn build_msg2_frame() -> Vec<u8> {
|
||||
let mut frame = vec![0xBB; MSG2_WIRE_SIZE];
|
||||
frame[0] = 0x02; // ver=0, phase=2
|
||||
frame[1] = 0x00; // flags
|
||||
frame[2..4].copy_from_slice(&MSG2_PAYLOAD_LEN.to_le_bytes());
|
||||
frame
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_read_established_frame() {
|
||||
let payload_len = 64u16;
|
||||
let frame = build_established_frame(payload_len);
|
||||
let expected = frame.clone();
|
||||
|
||||
let mut cursor = Cursor::new(frame);
|
||||
let packet = read_fmp_packet(&mut cursor, 1400).await.unwrap();
|
||||
assert_eq!(packet, expected);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_read_msg1_frame() {
|
||||
let frame = build_msg1_frame();
|
||||
let expected = frame.clone();
|
||||
|
||||
let mut cursor = Cursor::new(frame);
|
||||
let packet = read_fmp_packet(&mut cursor, 1400).await.unwrap();
|
||||
assert_eq!(packet.len(), MSG1_WIRE_SIZE);
|
||||
assert_eq!(packet, expected);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_read_msg2_frame() {
|
||||
let frame = build_msg2_frame();
|
||||
let expected = frame.clone();
|
||||
|
||||
let mut cursor = Cursor::new(frame);
|
||||
let packet = read_fmp_packet(&mut cursor, 1400).await.unwrap();
|
||||
assert_eq!(packet.len(), MSG2_WIRE_SIZE);
|
||||
assert_eq!(packet, expected);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_read_multiple_packets() {
|
||||
let mut data = Vec::new();
|
||||
let msg1 = build_msg1_frame();
|
||||
let est = build_established_frame(32);
|
||||
let msg2 = build_msg2_frame();
|
||||
data.extend_from_slice(&msg1);
|
||||
data.extend_from_slice(&est);
|
||||
data.extend_from_slice(&msg2);
|
||||
|
||||
let mut cursor = Cursor::new(data);
|
||||
let p1 = read_fmp_packet(&mut cursor, 1400).await.unwrap();
|
||||
assert_eq!(p1.len(), MSG1_WIRE_SIZE);
|
||||
|
||||
let p2 = read_fmp_packet(&mut cursor, 1400).await.unwrap();
|
||||
assert_eq!(p2, est);
|
||||
|
||||
let p3 = read_fmp_packet(&mut cursor, 1400).await.unwrap();
|
||||
assert_eq!(p3.len(), MSG2_WIRE_SIZE);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_unknown_phase_error() {
|
||||
let mut frame = vec![0u8; 100];
|
||||
frame[0] = 0x05; // unknown phase
|
||||
frame[2..4].copy_from_slice(&10u16.to_le_bytes());
|
||||
|
||||
let mut cursor = Cursor::new(frame);
|
||||
let err = read_fmp_packet(&mut cursor, 1400).await.unwrap_err();
|
||||
assert!(matches!(err, StreamError::UnknownPhase(0x5)));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_payload_too_large() {
|
||||
// mtu=100, max_payload_len = 100 - 32 = 68
|
||||
let payload_len = 100u16; // exceeds max of 68
|
||||
let mut prefix = [0u8; 4];
|
||||
prefix[0] = 0x00; // established
|
||||
prefix[2..4].copy_from_slice(&payload_len.to_le_bytes());
|
||||
|
||||
// Provide enough bytes for the reader to read prefix
|
||||
let mut data = prefix.to_vec();
|
||||
data.extend_from_slice(&vec![0u8; 200]); // extra bytes
|
||||
|
||||
let mut cursor = Cursor::new(data);
|
||||
let err = read_fmp_packet(&mut cursor, 100).await.unwrap_err();
|
||||
assert!(matches!(err, StreamError::PayloadTooLarge { .. }));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_handshake_size_mismatch_msg1() {
|
||||
let mut frame = vec![0u8; 200];
|
||||
frame[0] = 0x01; // msg1
|
||||
// Wrong payload_len (should be 110)
|
||||
frame[2..4].copy_from_slice(&50u16.to_le_bytes());
|
||||
|
||||
let mut cursor = Cursor::new(frame);
|
||||
let err = read_fmp_packet(&mut cursor, 1400).await.unwrap_err();
|
||||
assert!(matches!(err, StreamError::HandshakeSizeMismatch { phase: 0x1, .. }));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_handshake_size_mismatch_msg2() {
|
||||
let mut frame = vec![0u8; 200];
|
||||
frame[0] = 0x02; // msg2
|
||||
// Wrong payload_len (should be 65)
|
||||
frame[2..4].copy_from_slice(&50u16.to_le_bytes());
|
||||
|
||||
let mut cursor = Cursor::new(frame);
|
||||
let err = read_fmp_packet(&mut cursor, 1400).await.unwrap_err();
|
||||
assert!(matches!(err, StreamError::HandshakeSizeMismatch { phase: 0x2, .. }));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_eof_on_prefix() {
|
||||
// Only 2 bytes available (need 4 for prefix)
|
||||
let data = vec![0u8; 2];
|
||||
let mut cursor = Cursor::new(data);
|
||||
let err = read_fmp_packet(&mut cursor, 1400).await.unwrap_err();
|
||||
assert!(matches!(err, StreamError::Io(_)));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_eof_on_body() {
|
||||
// Valid msg1 prefix but truncated body
|
||||
let mut data = vec![0u8; 10]; // need 114 total
|
||||
data[0] = 0x01; // msg1
|
||||
data[2..4].copy_from_slice(&MSG1_PAYLOAD_LEN.to_le_bytes());
|
||||
|
||||
let mut cursor = Cursor::new(data);
|
||||
let err = read_fmp_packet(&mut cursor, 1400).await.unwrap_err();
|
||||
assert!(matches!(err, StreamError::Io(_)));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_zero_payload_established() {
|
||||
// payload_len = 0 is valid (header-only encrypted frame with tag)
|
||||
let frame = build_established_frame(0);
|
||||
let expected_len = PREFIX_SIZE + ESTABLISHED_REMAINING_HEADER + AEAD_TAG_SIZE;
|
||||
assert_eq!(frame.len(), expected_len);
|
||||
|
||||
let mut cursor = Cursor::new(frame.clone());
|
||||
let packet = read_fmp_packet(&mut cursor, 1400).await.unwrap();
|
||||
assert_eq!(packet.len(), expected_len);
|
||||
assert_eq!(packet, frame);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_max_payload_at_mtu_boundary() {
|
||||
// mtu=1400, max_payload_len = 1400 - 32 = 1368
|
||||
let max_payload = 1400u16 - 32;
|
||||
let frame = build_established_frame(max_payload);
|
||||
|
||||
let mut cursor = Cursor::new(frame.clone());
|
||||
let packet = read_fmp_packet(&mut cursor, 1400).await.unwrap();
|
||||
assert_eq!(packet, frame);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_payload_one_over_mtu() {
|
||||
// mtu=1400, max_payload_len = 1368, try 1369
|
||||
let over = 1400u16 - 32 + 1;
|
||||
let mut prefix = [0u8; 4];
|
||||
prefix[0] = 0x00; // established
|
||||
prefix[2..4].copy_from_slice(&over.to_le_bytes());
|
||||
|
||||
let mut data = prefix.to_vec();
|
||||
data.extend_from_slice(&vec![0u8; 2000]);
|
||||
|
||||
let mut cursor = Cursor::new(data);
|
||||
let err = read_fmp_packet(&mut cursor, 1400).await.unwrap_err();
|
||||
assert!(matches!(err, StreamError::PayloadTooLarge { .. }));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,40 @@
|
||||
# TCP chain: 4-node linear topology, all TCP transport
|
||||
#
|
||||
# Tests basic TCP transport connectivity, spanning tree convergence,
|
||||
# and multi-hop routing over TCP links.
|
||||
#
|
||||
# Topology:
|
||||
#
|
||||
# n01 ---tcp--- n02 ---tcp--- n03 ---tcp--- n04
|
||||
|
||||
scenario:
|
||||
name: "tcp-chain"
|
||||
seed: 42
|
||||
duration_secs: 90
|
||||
|
||||
topology:
|
||||
algorithm: explicit
|
||||
num_nodes: 4
|
||||
default_transport: tcp
|
||||
params:
|
||||
adjacency:
|
||||
- [n01, n02]
|
||||
- [n02, n03]
|
||||
- [n03, n04]
|
||||
|
||||
netem:
|
||||
enabled: true
|
||||
default_policy:
|
||||
delay_ms: [1, 5]
|
||||
jitter_ms: [0, 1]
|
||||
loss_pct: [0, 0.5]
|
||||
|
||||
link_flaps:
|
||||
enabled: false
|
||||
|
||||
traffic:
|
||||
enabled: false
|
||||
|
||||
logging:
|
||||
rust_log: "info"
|
||||
output_dir: "./sim-results"
|
||||
@@ -0,0 +1,66 @@
|
||||
# Mixed transport mesh: 6 nodes, UDP + TCP edges
|
||||
#
|
||||
# Exercises both transports in a single mesh. UDP and TCP edges both
|
||||
# use static peer config. Tests that the spanning tree converges
|
||||
# across heterogeneous transports with netem and link flaps active.
|
||||
#
|
||||
# Topology:
|
||||
#
|
||||
# n01 ---udp--- n02 ---udp--- n03
|
||||
# | | |
|
||||
# tcp tcp udp
|
||||
# | | |
|
||||
# n04 ---tcp--- n05 ---udp--- n06
|
||||
#
|
||||
# UDP edges: n01-n02, n02-n03, n03-n06, n05-n06
|
||||
# TCP edges: n01-n04, n02-n05, n04-n05
|
||||
|
||||
scenario:
|
||||
name: "tcp-mesh"
|
||||
seed: 42
|
||||
duration_secs: 120
|
||||
|
||||
topology:
|
||||
algorithm: explicit
|
||||
num_nodes: 6
|
||||
params:
|
||||
adjacency:
|
||||
- [n01, n02, udp]
|
||||
- [n02, n03, udp]
|
||||
- [n03, n06, udp]
|
||||
- [n05, n06, udp]
|
||||
- [n01, n04, tcp]
|
||||
- [n02, n05, tcp]
|
||||
- [n04, n05, tcp]
|
||||
|
||||
netem:
|
||||
enabled: true
|
||||
default_policy:
|
||||
delay_ms: [1, 10]
|
||||
jitter_ms: [0, 2]
|
||||
loss_pct: [0, 1]
|
||||
mutation:
|
||||
interval_secs: {min: 20, max: 40}
|
||||
fraction: 0.3
|
||||
policies:
|
||||
normal:
|
||||
delay_ms: [1, 10]
|
||||
loss_pct: [0, 1]
|
||||
degraded:
|
||||
delay_ms: [30, 80]
|
||||
jitter_ms: [5, 20]
|
||||
loss_pct: [3, 8]
|
||||
|
||||
link_flaps:
|
||||
enabled: true
|
||||
interval_secs: {min: 20, max: 40}
|
||||
max_down_links: 1
|
||||
down_duration_secs: {min: 10, max: 20}
|
||||
protect_connectivity: true
|
||||
|
||||
traffic:
|
||||
enabled: false
|
||||
|
||||
logging:
|
||||
rust_log: "info"
|
||||
output_dir: "./sim-results"
|
||||
@@ -0,0 +1,45 @@
|
||||
# TCP-only: 4-node ring, all TCP transport
|
||||
#
|
||||
# Tests pure-TCP mesh with netem impairment. Nodes have no UDP
|
||||
# transport — only TCP with static peer config.
|
||||
#
|
||||
# Topology:
|
||||
#
|
||||
# n01 ---tcp--- n02
|
||||
# | |
|
||||
# tcp tcp
|
||||
# | |
|
||||
# n04 ---tcp--- n03
|
||||
|
||||
scenario:
|
||||
name: "tcp-only"
|
||||
seed: 42
|
||||
duration_secs: 90
|
||||
|
||||
topology:
|
||||
algorithm: explicit
|
||||
num_nodes: 4
|
||||
default_transport: tcp
|
||||
params:
|
||||
adjacency:
|
||||
- [n01, n02]
|
||||
- [n02, n03]
|
||||
- [n03, n04]
|
||||
- [n04, n01]
|
||||
|
||||
netem:
|
||||
enabled: true
|
||||
default_policy:
|
||||
delay_ms: [1, 10]
|
||||
jitter_ms: [0, 2]
|
||||
loss_pct: [0, 1]
|
||||
|
||||
link_flaps:
|
||||
enabled: false
|
||||
|
||||
traffic:
|
||||
enabled: false
|
||||
|
||||
logging:
|
||||
rust_log: "info"
|
||||
output_dir: "./sim-results"
|
||||
@@ -31,6 +31,12 @@ def _load_template() -> str:
|
||||
return f.read()
|
||||
|
||||
|
||||
_TRANSPORT_PORTS = {
|
||||
"udp": 2121,
|
||||
"tcp": 443,
|
||||
}
|
||||
|
||||
|
||||
def generate_peers_block(
|
||||
topology: SimTopology, node_id: str, outbound_peers: list[str]
|
||||
) -> str:
|
||||
@@ -38,6 +44,7 @@ def generate_peers_block(
|
||||
|
||||
Only includes peers that this node is responsible for connecting to
|
||||
(outbound direction). The link is still bidirectional once established.
|
||||
Transport type and port are determined per-edge from the topology.
|
||||
"""
|
||||
if not outbound_peers:
|
||||
return " []"
|
||||
@@ -45,11 +52,13 @@ def generate_peers_block(
|
||||
lines = []
|
||||
for peer_id in sorted(outbound_peers):
|
||||
peer = topology.nodes[peer_id]
|
||||
transport = topology.transport_for_edge(node_id, peer_id)
|
||||
port = _TRANSPORT_PORTS.get(transport, 2121)
|
||||
lines.append(f' - npub: "{peer.npub}"')
|
||||
lines.append(f' alias: "{peer_id}"')
|
||||
lines.append(f" addresses:")
|
||||
lines.append(f" - transport: udp")
|
||||
lines.append(f' addr: "{peer.docker_ip}:2121"')
|
||||
lines.append(f" - transport: {transport}")
|
||||
lines.append(f' addr: "{peer.docker_ip}:{port}"')
|
||||
lines.append(f" connect_policy: auto_connect")
|
||||
return "\n".join(lines)
|
||||
|
||||
@@ -85,6 +94,14 @@ def _inject_ethernet_transports(parsed: dict, eth_ifaces: list[str]):
|
||||
}
|
||||
|
||||
|
||||
def _inject_tcp_transport(parsed: dict):
|
||||
"""Inject TCP transport config into a parsed FIPS config."""
|
||||
transports = parsed.setdefault("transports", {})
|
||||
transports["tcp"] = {
|
||||
"bind_addr": "0.0.0.0:443",
|
||||
}
|
||||
|
||||
|
||||
def generate_node_config(
|
||||
topology: SimTopology,
|
||||
node_id: str,
|
||||
@@ -103,34 +120,36 @@ def generate_node_config(
|
||||
config = config.replace("{{NSEC}}", node.nsec)
|
||||
config = config.replace("{{PEERS}}", peers_yaml)
|
||||
|
||||
# Inject Ethernet transport config if this node has Ethernet edges
|
||||
# Determine which transports this node participates in
|
||||
eth_ifaces = topology.ethernet_interfaces(node_id)
|
||||
has_udp_peers = bool(outbound_peers) or _has_inbound_udp_peers(topology, node_id)
|
||||
has_tcp = bool(topology.tcp_peers(node_id))
|
||||
has_udp = _has_transport_peers(topology, node_id, "udp")
|
||||
|
||||
if eth_ifaces or not has_udp_peers:
|
||||
# Inject non-UDP transport configs and handle pure-transport nodes
|
||||
needs_yaml_rewrite = eth_ifaces or has_tcp or not has_udp or fips_overrides
|
||||
|
||||
if needs_yaml_rewrite:
|
||||
parsed = yaml.safe_load(config)
|
||||
if fips_overrides:
|
||||
parsed = _deep_merge(parsed, fips_overrides)
|
||||
if eth_ifaces:
|
||||
_inject_ethernet_transports(parsed, eth_ifaces)
|
||||
if not has_udp_peers and eth_ifaces:
|
||||
# Pure-Ethernet node: remove UDP transport
|
||||
if has_tcp:
|
||||
_inject_tcp_transport(parsed)
|
||||
if not has_udp:
|
||||
# No UDP edges: remove UDP transport
|
||||
transports = parsed.get("transports", {})
|
||||
transports.pop("udp", None)
|
||||
config = yaml.dump(parsed, default_flow_style=False, sort_keys=False)
|
||||
elif fips_overrides:
|
||||
parsed = yaml.safe_load(config)
|
||||
merged = _deep_merge(parsed, fips_overrides)
|
||||
config = yaml.dump(merged, default_flow_style=False, sort_keys=False)
|
||||
|
||||
return config
|
||||
|
||||
|
||||
def _has_inbound_udp_peers(topology: SimTopology, node_id: str) -> bool:
|
||||
"""Check if any other node has a UDP outbound edge to this node."""
|
||||
def _has_transport_peers(topology: SimTopology, node_id: str, transport: str) -> bool:
|
||||
"""Check if a node has any edges (inbound or outbound) using the given transport."""
|
||||
for peer_id in topology.nodes[node_id].peers:
|
||||
edge = (min(node_id, peer_id), max(node_id, peer_id))
|
||||
if topology.edge_transport.get(edge, "udp") == "udp":
|
||||
if topology.edge_transport.get(edge, "udp") == transport:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
@@ -178,7 +178,7 @@ class LinkManager:
|
||||
docker_exec_quiet(container, cmd)
|
||||
return prev_args
|
||||
else:
|
||||
# UDP: HTB class on eth0
|
||||
# IP-based (UDP/TCP): HTB class on eth0
|
||||
dest_ip = self.topology.nodes[dst_node].docker_ip
|
||||
|
||||
states = self.netem_mgr.states.get(container, {})
|
||||
|
||||
+12
-12
@@ -151,18 +151,18 @@ class NetemManager:
|
||||
node = self.topology.nodes[node_id]
|
||||
container = self.topology.container_name(node_id)
|
||||
|
||||
# Split peers by transport type
|
||||
udp_peers = {}
|
||||
# Split peers by transport type: IP-based (UDP/TCP) vs Ethernet (veth)
|
||||
ip_peers = {}
|
||||
eth_peers = []
|
||||
for peer_id in sorted(node.peers):
|
||||
transport = self.topology.transport_for_edge(node_id, peer_id)
|
||||
if transport == "ethernet":
|
||||
eth_peers.append(peer_id)
|
||||
else:
|
||||
udp_peers[peer_id] = self.topology.nodes[peer_id].docker_ip
|
||||
ip_peers[peer_id] = self.topology.nodes[peer_id].docker_ip
|
||||
|
||||
# --- UDP peers: HTB + u32 on eth0 ---
|
||||
if udp_peers:
|
||||
# --- IP-based peers (UDP/TCP): HTB + u32 on eth0 ---
|
||||
if ip_peers:
|
||||
cmds = [f"tc qdisc del dev {IFACE} root 2>/dev/null || true"]
|
||||
cmds.append(
|
||||
f"tc qdisc add dev {IFACE} root handle 1: htb default 99"
|
||||
@@ -173,7 +173,7 @@ class NetemManager:
|
||||
|
||||
container_states = {}
|
||||
|
||||
for idx, (peer_id, dest_ip) in enumerate(udp_peers.items(), start=1):
|
||||
for idx, (peer_id, dest_ip) in enumerate(ip_peers.items(), start=1):
|
||||
class_id = f"1:{idx}"
|
||||
netem_handle = f"{idx + 10}:"
|
||||
|
||||
@@ -208,9 +208,9 @@ class NetemManager:
|
||||
result = docker_exec_quiet(container, full_cmd, timeout=30)
|
||||
if result is not None:
|
||||
log.info(
|
||||
"Configured per-link netem on %s (%d UDP peers)",
|
||||
"Configured per-link netem on %s (%d IP peers)",
|
||||
container,
|
||||
len(udp_peers),
|
||||
len(ip_peers),
|
||||
)
|
||||
else:
|
||||
log.warning("Failed to configure netem on %s", container)
|
||||
@@ -256,7 +256,7 @@ class NetemManager:
|
||||
"""
|
||||
container = self.topology.container_name(node_id)
|
||||
|
||||
# Re-apply UDP netem (eth0 HTB + u32)
|
||||
# Re-apply IP-based netem (eth0 HTB + u32) for UDP/TCP peers
|
||||
container_states = self.states.get(container)
|
||||
if container_states:
|
||||
cmds = [f"tc qdisc del dev {IFACE} root 2>/dev/null || true"]
|
||||
@@ -286,12 +286,12 @@ class NetemManager:
|
||||
result = docker_exec_quiet(container, full_cmd, timeout=30)
|
||||
if result is not None:
|
||||
log.info(
|
||||
"Re-applied UDP netem on %s (%d peers)",
|
||||
"Re-applied IP netem on %s (%d peers)",
|
||||
container,
|
||||
len(container_states),
|
||||
)
|
||||
else:
|
||||
log.warning("Failed to re-apply UDP netem on %s", container)
|
||||
log.warning("Failed to re-apply IP netem on %s", container)
|
||||
|
||||
# Re-apply Ethernet veth netem
|
||||
veth_states = self.veth_states.get(container)
|
||||
@@ -376,7 +376,7 @@ class NetemManager:
|
||||
state.params = params
|
||||
log.debug("Updated veth netem %s:%s -> %s", src, iface, params.to_tc_args())
|
||||
else:
|
||||
# UDP: HTB class-based netem on eth0
|
||||
# IP-based (UDP/TCP): HTB class-based netem on eth0
|
||||
dest_ip = self.topology.nodes[dst].docker_ip
|
||||
states = self.states.get(container, {})
|
||||
state = states.get(dest_ip)
|
||||
|
||||
@@ -22,7 +22,7 @@ class Range:
|
||||
raise ValueError(f"{name}: min ({self.min}) must be >= 0")
|
||||
|
||||
|
||||
VALID_TRANSPORTS = ("udp", "ethernet")
|
||||
VALID_TRANSPORTS = ("udp", "ethernet", "tcp")
|
||||
|
||||
|
||||
@dataclass
|
||||
|
||||
@@ -42,6 +42,26 @@ class SimTopology:
|
||||
"""Check if any edges use Ethernet transport."""
|
||||
return any(t == "ethernet" for t in self.edge_transport.values())
|
||||
|
||||
def tcp_edges(self) -> list[tuple[str, str]]:
|
||||
"""Return all edges using TCP transport."""
|
||||
return [e for e, t in self.edge_transport.items() if t == "tcp"]
|
||||
|
||||
def has_tcp(self) -> bool:
|
||||
"""Check if any edges use TCP transport."""
|
||||
return any(t == "tcp" for t in self.edge_transport.values())
|
||||
|
||||
def tcp_peers(self, node_id: str) -> list[str]:
|
||||
"""Return peer IDs connected to this node via TCP."""
|
||||
peers = []
|
||||
for (a, b), transport in self.edge_transport.items():
|
||||
if transport != "tcp":
|
||||
continue
|
||||
if a == node_id:
|
||||
peers.append(b)
|
||||
elif b == node_id:
|
||||
peers.append(a)
|
||||
return sorted(peers)
|
||||
|
||||
def ethernet_interfaces(self, node_id: str) -> list[str]:
|
||||
"""Return the veth interface names for a node's Ethernet edges."""
|
||||
ifaces = []
|
||||
@@ -90,7 +110,7 @@ class SimTopology:
|
||||
return f"fips-node-{node_id}"
|
||||
|
||||
def directed_outbound(self) -> dict[str, list[str]]:
|
||||
"""Assign each UDP edge to exactly one node for outbound connection.
|
||||
"""Assign each static-config edge to exactly one node for outbound connection.
|
||||
|
||||
Returns a mapping from node_id to the list of peers that node
|
||||
should connect to (outbound only). Every edge appears in exactly
|
||||
@@ -98,27 +118,27 @@ class SimTopology:
|
||||
down, only A (the outbound owner) will attempt to reconnect.
|
||||
|
||||
Ethernet edges are excluded — they use beacon discovery instead
|
||||
of static peer configuration.
|
||||
of static peer configuration. UDP and TCP edges use static config.
|
||||
|
||||
Strategy: BFS spanning tree edges go parent→child. Non-tree
|
||||
edges go from the lower node ID to the higher. This guarantees
|
||||
every node is reachable via at least one inbound connection.
|
||||
"""
|
||||
# Only consider UDP edges for static peer config
|
||||
udp_edges = {
|
||||
# Consider all edges that use static peer config (not Ethernet/discovery)
|
||||
static_edges = {
|
||||
e for e in self.edges
|
||||
if self.edge_transport.get(e, "udp") == "udp"
|
||||
if self.edge_transport.get(e, "udp") != "ethernet"
|
||||
}
|
||||
|
||||
outbound: dict[str, list[str]] = {nid: [] for nid in self.nodes}
|
||||
|
||||
# Build UDP-only adjacency for BFS
|
||||
udp_adj: dict[str, list[str]] = {nid: [] for nid in self.nodes}
|
||||
for a, b in udp_edges:
|
||||
udp_adj[a].append(b)
|
||||
udp_adj[b].append(a)
|
||||
# Build static-config adjacency for BFS
|
||||
static_adj: dict[str, list[str]] = {nid: [] for nid in self.nodes}
|
||||
for a, b in static_edges:
|
||||
static_adj[a].append(b)
|
||||
static_adj[b].append(a)
|
||||
|
||||
# BFS spanning tree from first node (over UDP edges only)
|
||||
# BFS spanning tree from first node (over static-config edges only)
|
||||
root = min(self.nodes)
|
||||
visited: set[str] = set()
|
||||
tree_edges: set[tuple[str, str]] = set()
|
||||
@@ -126,15 +146,15 @@ class SimTopology:
|
||||
visited.add(root)
|
||||
while queue:
|
||||
node = queue.popleft()
|
||||
for peer in udp_adj[node]:
|
||||
for peer in static_adj[node]:
|
||||
if peer not in visited:
|
||||
visited.add(peer)
|
||||
queue.append(peer)
|
||||
tree_edges.add((node, peer)) # parent → child
|
||||
outbound[node].append(peer)
|
||||
|
||||
# Non-tree UDP edges: lower ID → higher ID
|
||||
for a, b in udp_edges:
|
||||
# Non-tree static-config edges: lower ID → higher ID
|
||||
for a, b in static_edges:
|
||||
if (a, b) not in tree_edges and (b, a) not in tree_edges:
|
||||
outbound[a].append(b) # a < b by _make_edge convention
|
||||
|
||||
|
||||
@@ -0,0 +1,29 @@
|
||||
# TCP Chain Topology Definition
|
||||
#
|
||||
# Three nodes with TCP links forming a linear chain. Tests basic TCP
|
||||
# transport connectivity, spanning tree convergence, and multi-hop
|
||||
# routing over TCP. All peer connections use TCP on port 443.
|
||||
#
|
||||
# default_transport: tcp tells the config generator to use TCP
|
||||
# transport and port 443 instead of the default UDP/2121.
|
||||
|
||||
default_transport: tcp
|
||||
|
||||
nodes:
|
||||
a:
|
||||
nsec: "0102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f20"
|
||||
npub: "npub1sjlh2c3x9w7kjsqg2ay080n2lff2uvt325vpan33ke34rn8l5jcqawh57m"
|
||||
docker_ip: "172.20.0.10"
|
||||
peers: [b]
|
||||
|
||||
b:
|
||||
nsec: "b102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1fb0"
|
||||
npub: "npub1tdwa4vjrjl33pcjdpf2t4p027nl86xrx24g4d3avg4vwvayr3g8qhd84le"
|
||||
docker_ip: "172.20.0.11"
|
||||
peers: [a, c]
|
||||
|
||||
c:
|
||||
nsec: "c102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1fc0"
|
||||
npub: "npub1cld9yay0u24davpu6c35l4vldrhzvaq66pcqtg9a0j2cnjrn9rtsxx2pe6"
|
||||
docker_ip: "172.20.0.12"
|
||||
peers: [b]
|
||||
@@ -203,3 +203,40 @@ services:
|
||||
networks:
|
||||
fips-net:
|
||||
ipv4_address: 172.20.0.14
|
||||
|
||||
# ── TCP chain topology (A-B-C) ───────────────────────────────
|
||||
tcp-a:
|
||||
<<: *fips-common
|
||||
profiles: ["tcp-chain"]
|
||||
container_name: fips-node-a
|
||||
hostname: node-a
|
||||
volumes:
|
||||
- ./resolv.conf:/etc/resolv.conf:ro
|
||||
- ./generated-configs/tcp-chain/node-a.yaml:/etc/fips/fips.yaml:ro
|
||||
networks:
|
||||
fips-net:
|
||||
ipv4_address: 172.20.0.10
|
||||
|
||||
tcp-b:
|
||||
<<: *fips-common
|
||||
profiles: ["tcp-chain"]
|
||||
container_name: fips-node-b
|
||||
hostname: node-b
|
||||
volumes:
|
||||
- ./resolv.conf:/etc/resolv.conf:ro
|
||||
- ./generated-configs/tcp-chain/node-b.yaml:/etc/fips/fips.yaml:ro
|
||||
networks:
|
||||
fips-net:
|
||||
ipv4_address: 172.20.0.11
|
||||
|
||||
tcp-c:
|
||||
<<: *fips-common
|
||||
profiles: ["tcp-chain"]
|
||||
container_name: fips-node-c
|
||||
hostname: node-c
|
||||
volumes:
|
||||
- ./resolv.conf:/etc/resolv.conf:ro
|
||||
- ./generated-configs/tcp-chain/node-c.yaml:/etc/fips/fips.yaml:ro
|
||||
networks:
|
||||
fips-net:
|
||||
ipv4_address: 172.20.0.12
|
||||
|
||||
@@ -77,19 +77,37 @@ resolve_keys() {
|
||||
fi
|
||||
}
|
||||
|
||||
# Get the default transport from topology file (defaults to "udp")
|
||||
get_default_transport() {
|
||||
local topology_file="$1"
|
||||
local transport=$(grep "^default_transport:" "$topology_file" | head -1 | sed 's/.*: *\([a-z]*\).*/\1/')
|
||||
echo "${transport:-udp}"
|
||||
}
|
||||
|
||||
# Get the port for a given transport type
|
||||
transport_port() {
|
||||
local transport="$1"
|
||||
case "$transport" in
|
||||
tcp) echo "443" ;;
|
||||
*) echo "2121" ;;
|
||||
esac
|
||||
}
|
||||
|
||||
generate_peer_block() {
|
||||
local topology_file="$1"
|
||||
local peer_id="$2"
|
||||
|
||||
local peer_npub="$(get_key RESOLVED_NPUB "$peer_id")"
|
||||
local peer_ip=$(get_node_attr "$topology_file" "$peer_id" "address")
|
||||
local transport=$(get_default_transport "$topology_file")
|
||||
local port=$(transport_port "$transport")
|
||||
|
||||
cat <<EOF
|
||||
- npub: "$peer_npub"
|
||||
alias: "node-$peer_id"
|
||||
addresses:
|
||||
- transport: udp
|
||||
addr: "$peer_ip:2121"
|
||||
- transport: $transport
|
||||
addr: "$peer_ip:$port"
|
||||
connect_policy: auto_connect
|
||||
EOF
|
||||
}
|
||||
@@ -103,7 +121,8 @@ generate_config() {
|
||||
node_npub="$(get_key RESOLVED_NPUB "$node_id")"
|
||||
local node_nsec
|
||||
node_nsec="$(get_key RESOLVED_NSEC "$node_id")"
|
||||
local peers=$(get_peers "$topology_file" "$node_id")
|
||||
local peers
|
||||
peers=$(get_peers "$topology_file" "$node_id")
|
||||
|
||||
# Generate peers section
|
||||
local peers_config=""
|
||||
@@ -129,6 +148,22 @@ generate_config() {
|
||||
config="${config//\{\{PEERS\}\}/$peers_config}"
|
||||
|
||||
echo "$config" > "$output_file"
|
||||
|
||||
# Post-process: inject TCP transport config for TCP topologies
|
||||
local transport
|
||||
transport=$(get_default_transport "$topology_file")
|
||||
if [ "$transport" = "tcp" ]; then
|
||||
# Add TCP transport section and remove UDP transport
|
||||
python3 -c "
|
||||
import yaml, sys
|
||||
with open('$output_file') as f:
|
||||
cfg = yaml.safe_load(f)
|
||||
cfg.setdefault('transports', {})['tcp'] = {'bind_addr': '0.0.0.0:443'}
|
||||
cfg.get('transports', {}).pop('udp', None)
|
||||
with open('$output_file', 'w') as f:
|
||||
yaml.dump(cfg, f, default_flow_style=False, sort_keys=False)
|
||||
"
|
||||
fi
|
||||
}
|
||||
|
||||
# Key storage for bash 3.2 compatibility (using prefixed variables instead of associative arrays)
|
||||
|
||||
Reference in New Issue
Block a user