mirror of
https://github.com/jmcorgan/fips.git
synced 2026-08-09 08:14:42 +00:00
Module reorganization and clippy cleanup
Move single-consumer modules into node/:
- rate_limit.rs, wire.rs, dns.rs — exclusively used by node subsystem
- Reduces top-level lib.rs from 16 to 13 modules
Split large files into focused subdirectories:
- noise.rs (1475 lines) → noise/{mod, handshake, session, replay, tests}.rs
- tree.rs (1479 lines) → tree/{mod, coordinate, declaration, state, tests}.rs
- bloom.rs (849 lines) → bloom/{mod, filter, state, tests}.rs
- All public APIs re-exported from mod.rs, no external import changes
Remove unused rate_limit defaults:
- HANDSHAKE_TIMEOUT_SECS, MAX_PENDING_INBOUND constants
- Default constructor eliminated in favor of with_params() taking config values
Fix all clippy warnings across codebase:
- Remove .clone() on Copy types, collapse nested ifs, replace match-return-None
with ?, remove/gate unused code, fix loop indexing, remove unnecessary casts
- Box large PeerSlot enum variants to reduce size disparity
- cargo clippy --all-targets now reports zero warnings
This commit is contained in:
Johnathan Corgan
+327
@@ -0,0 +1,327 @@
|
||||
//! FIPS DNS Responder
|
||||
//!
|
||||
//! Resolves `<npub>.fips` queries to FipsAddress IPv6 addresses.
|
||||
//! The resolution is pure computation: npub → PublicKey → NodeAddr → FipsAddress.
|
||||
//! As a side effect, resolved identities are sent to the Node for identity
|
||||
//! cache population, enabling subsequent TUN packet routing.
|
||||
|
||||
use crate::{NodeAddr, PeerIdentity};
|
||||
use simple_dns::rdata::{RData, AAAA};
|
||||
use simple_dns::{Packet, Name, ResourceRecord, CLASS, RCODE, PacketFlag, QTYPE, TYPE};
|
||||
use std::net::Ipv6Addr;
|
||||
use tracing::{debug, warn};
|
||||
|
||||
/// Identity resolved by the DNS responder, sent to Node for cache population.
|
||||
pub struct DnsResolvedIdentity {
|
||||
pub node_addr: NodeAddr,
|
||||
pub pubkey: secp256k1::PublicKey,
|
||||
}
|
||||
|
||||
/// Channel sender for DNS → Node identity registration.
|
||||
pub type DnsIdentityTx = tokio::sync::mpsc::Sender<DnsResolvedIdentity>;
|
||||
|
||||
/// Channel receiver consumed by the Node RX event loop.
|
||||
pub type DnsIdentityRx = tokio::sync::mpsc::Receiver<DnsResolvedIdentity>;
|
||||
|
||||
/// Resolve a `.fips` domain name to an IPv6 address and identity.
|
||||
///
|
||||
/// The name should be `<npub>.fips` (with optional trailing dot).
|
||||
/// Returns the FipsAddress IPv6, NodeAddr, and full PublicKey on success.
|
||||
pub fn resolve_fips_query(name: &str) -> Option<(Ipv6Addr, NodeAddr, secp256k1::PublicKey)> {
|
||||
let name = name.strip_suffix('.').unwrap_or(name);
|
||||
let npub = name.strip_suffix(".fips")
|
||||
.or_else(|| name.strip_suffix(".FIPS"))
|
||||
.or_else(|| {
|
||||
// Case-insensitive check for .fips suffix
|
||||
let lower = name.to_ascii_lowercase();
|
||||
if lower.ends_with(".fips") {
|
||||
Some(&name[..name.len() - 5])
|
||||
} else {
|
||||
None
|
||||
}
|
||||
})?;
|
||||
|
||||
let peer = PeerIdentity::from_npub(npub).ok()?;
|
||||
let ipv6 = peer.address().to_ipv6();
|
||||
let node_addr = *peer.node_addr();
|
||||
let pubkey = peer.pubkey_full();
|
||||
|
||||
Some((ipv6, node_addr, pubkey))
|
||||
}
|
||||
|
||||
/// Handle a raw DNS query packet and produce a response.
|
||||
///
|
||||
/// Returns the response bytes and an optional resolved identity (for AAAA queries
|
||||
/// that successfully resolved a `.fips` name).
|
||||
pub fn handle_dns_packet(query_bytes: &[u8], ttl: u32) -> Option<(Vec<u8>, Option<DnsResolvedIdentity>)> {
|
||||
let query = Packet::parse(query_bytes).ok()?;
|
||||
let question = query.questions.first()?;
|
||||
|
||||
let qname = question.qname.to_string();
|
||||
let is_aaaa = matches!(question.qtype, QTYPE::TYPE(TYPE::AAAA));
|
||||
|
||||
let mut response = query.into_reply();
|
||||
response.set_flags(PacketFlag::AUTHORITATIVE_ANSWER);
|
||||
|
||||
if is_aaaa
|
||||
&& let Some((ipv6, node_addr, pubkey)) = resolve_fips_query(&qname)
|
||||
{
|
||||
let name = Name::new_unchecked(&qname).into_owned();
|
||||
let record = ResourceRecord::new(
|
||||
name,
|
||||
CLASS::IN,
|
||||
ttl,
|
||||
RData::AAAA(AAAA::from(ipv6)),
|
||||
);
|
||||
response.answers.push(record);
|
||||
|
||||
let identity = DnsResolvedIdentity { node_addr, pubkey };
|
||||
let bytes = response.build_bytes_vec_compressed().ok()?;
|
||||
return Some((bytes, Some(identity)));
|
||||
}
|
||||
|
||||
// Query we can't answer: NXDOMAIN
|
||||
*response.rcode_mut() = RCODE::NameError;
|
||||
let bytes = response.build_bytes_vec_compressed().ok()?;
|
||||
Some((bytes, None))
|
||||
}
|
||||
|
||||
/// Run the DNS responder UDP server loop.
|
||||
///
|
||||
/// Listens for DNS queries, resolves `.fips` names, and sends resolved
|
||||
/// identities to the Node via the identity channel.
|
||||
pub async fn run_dns_responder(
|
||||
socket: tokio::net::UdpSocket,
|
||||
identity_tx: DnsIdentityTx,
|
||||
ttl: u32,
|
||||
) {
|
||||
let mut buf = [0u8; 512]; // Standard DNS UDP max
|
||||
|
||||
loop {
|
||||
let (len, src) = match socket.recv_from(&mut buf).await {
|
||||
Ok(result) => result,
|
||||
Err(e) => {
|
||||
warn!(error = %e, "DNS socket recv error");
|
||||
continue;
|
||||
}
|
||||
};
|
||||
|
||||
let query_bytes = &buf[..len];
|
||||
|
||||
match handle_dns_packet(query_bytes, ttl) {
|
||||
Some((response_bytes, identity)) => {
|
||||
if let Some(id) = identity {
|
||||
debug!(
|
||||
node_addr = %id.node_addr,
|
||||
"DNS resolved .fips name, registering identity"
|
||||
);
|
||||
let _ = identity_tx.send(id).await;
|
||||
}
|
||||
|
||||
if let Err(e) = socket.send_to(&response_bytes, src).await {
|
||||
debug!(error = %e, "DNS send error");
|
||||
}
|
||||
}
|
||||
None => {
|
||||
debug!(len, "Failed to parse DNS query, dropping");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::Identity;
|
||||
|
||||
#[test]
|
||||
fn test_resolve_valid_npub() {
|
||||
let identity = Identity::generate();
|
||||
let npub = identity.npub();
|
||||
let expected_ipv6 = identity.address().to_ipv6();
|
||||
|
||||
let query = format!("{}.fips", npub);
|
||||
let result = resolve_fips_query(&query);
|
||||
|
||||
assert!(result.is_some(), "should resolve valid npub.fips");
|
||||
let (ipv6, node_addr, _pubkey) = result.unwrap();
|
||||
assert_eq!(ipv6, expected_ipv6);
|
||||
assert_eq!(node_addr, *identity.node_addr());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_resolve_trailing_dot() {
|
||||
let identity = Identity::generate();
|
||||
let npub = identity.npub();
|
||||
let expected_ipv6 = identity.address().to_ipv6();
|
||||
|
||||
let query = format!("{}.fips.", npub);
|
||||
let result = resolve_fips_query(&query);
|
||||
|
||||
assert!(result.is_some(), "should handle trailing dot");
|
||||
let (ipv6, _, _) = result.unwrap();
|
||||
assert_eq!(ipv6, expected_ipv6);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_resolve_case_insensitive() {
|
||||
let identity = Identity::generate();
|
||||
let npub = identity.npub();
|
||||
|
||||
// .FIPS
|
||||
let result = resolve_fips_query(&format!("{}.FIPS", npub));
|
||||
assert!(result.is_some(), "should handle .FIPS");
|
||||
|
||||
// .Fips
|
||||
let result = resolve_fips_query(&format!("{}.Fips", npub));
|
||||
assert!(result.is_some(), "should handle .Fips");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_resolve_invalid_npub() {
|
||||
let result = resolve_fips_query("not-a-valid-npub.fips");
|
||||
assert!(result.is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_resolve_wrong_suffix() {
|
||||
let identity = Identity::generate();
|
||||
let npub = identity.npub();
|
||||
|
||||
let result = resolve_fips_query(&format!("{}.com", npub));
|
||||
assert!(result.is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_resolve_empty_name() {
|
||||
assert!(resolve_fips_query("").is_none());
|
||||
assert!(resolve_fips_query(".fips").is_none());
|
||||
assert!(resolve_fips_query("fips").is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_handle_aaaa_query() {
|
||||
let identity = Identity::generate();
|
||||
let npub = identity.npub();
|
||||
let expected_ipv6 = identity.address().to_ipv6();
|
||||
|
||||
// Build a DNS AAAA query packet
|
||||
let query_name = format!("{}.fips", npub);
|
||||
let query_packet = build_test_query(&query_name, TYPE::AAAA);
|
||||
|
||||
let result = handle_dns_packet(&query_packet, 300);
|
||||
assert!(result.is_some(), "should handle AAAA query");
|
||||
|
||||
let (response_bytes, identity_opt) = result.unwrap();
|
||||
assert!(identity_opt.is_some(), "should produce identity");
|
||||
|
||||
// Parse the response
|
||||
let response = Packet::parse(&response_bytes).unwrap();
|
||||
assert_eq!(response.answers.len(), 1);
|
||||
|
||||
// Verify the AAAA record
|
||||
if let RData::AAAA(aaaa) = &response.answers[0].rdata {
|
||||
let addr = Ipv6Addr::from(aaaa.address);
|
||||
assert_eq!(addr, expected_ipv6);
|
||||
} else {
|
||||
panic!("expected AAAA record");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_handle_nxdomain_for_unknown() {
|
||||
let query_packet = build_test_query("unknown.fips", TYPE::AAAA);
|
||||
|
||||
let result = handle_dns_packet(&query_packet, 300);
|
||||
assert!(result.is_some());
|
||||
|
||||
let (response_bytes, identity_opt) = result.unwrap();
|
||||
assert!(identity_opt.is_none(), "should not produce identity for unknown");
|
||||
|
||||
let response = Packet::parse(&response_bytes).unwrap();
|
||||
assert_eq!(response.rcode(), RCODE::NameError);
|
||||
assert!(response.answers.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_handle_non_aaaa_query() {
|
||||
let identity = Identity::generate();
|
||||
let query_name = format!("{}.fips", identity.npub());
|
||||
let query_packet = build_test_query(&query_name, TYPE::A);
|
||||
|
||||
let result = handle_dns_packet(&query_packet, 300);
|
||||
assert!(result.is_some());
|
||||
|
||||
let (response_bytes, identity_opt) = result.unwrap();
|
||||
assert!(identity_opt.is_none(), "A query should not resolve .fips");
|
||||
|
||||
let response = Packet::parse(&response_bytes).unwrap();
|
||||
assert_eq!(response.rcode(), RCODE::NameError);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_dns_responder_udp() {
|
||||
let identity = Identity::generate();
|
||||
let npub = identity.npub();
|
||||
let expected_ipv6 = identity.address().to_ipv6();
|
||||
|
||||
// Bind responder on ephemeral port
|
||||
let server_socket = tokio::net::UdpSocket::bind("127.0.0.1:0").await.unwrap();
|
||||
let server_addr = server_socket.local_addr().unwrap();
|
||||
|
||||
let (identity_tx, mut identity_rx) = tokio::sync::mpsc::channel(16);
|
||||
|
||||
// Spawn the responder
|
||||
let responder_handle = tokio::spawn(run_dns_responder(server_socket, identity_tx, 300));
|
||||
|
||||
// Send a query
|
||||
let client_socket = tokio::net::UdpSocket::bind("127.0.0.1:0").await.unwrap();
|
||||
let query = build_test_query(&format!("{}.fips", npub), TYPE::AAAA);
|
||||
client_socket.send_to(&query, server_addr).await.unwrap();
|
||||
|
||||
// Receive response
|
||||
let mut buf = [0u8; 512];
|
||||
let (len, _) = tokio::time::timeout(
|
||||
std::time::Duration::from_secs(2),
|
||||
client_socket.recv_from(&mut buf),
|
||||
)
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
|
||||
let response = Packet::parse(&buf[..len]).unwrap();
|
||||
assert_eq!(response.answers.len(), 1);
|
||||
if let RData::AAAA(aaaa) = &response.answers[0].rdata {
|
||||
assert_eq!(Ipv6Addr::from(aaaa.address), expected_ipv6);
|
||||
} else {
|
||||
panic!("expected AAAA record");
|
||||
}
|
||||
|
||||
// Verify identity was sent through channel
|
||||
let resolved = tokio::time::timeout(
|
||||
std::time::Duration::from_secs(1),
|
||||
identity_rx.recv(),
|
||||
)
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
assert_eq!(resolved.node_addr, *identity.node_addr());
|
||||
|
||||
responder_handle.abort();
|
||||
}
|
||||
|
||||
/// Build a test DNS query packet for a given name and record type.
|
||||
fn build_test_query(name: &str, rtype: TYPE) -> Vec<u8> {
|
||||
use simple_dns::Question;
|
||||
|
||||
let mut packet = Packet::new_query(0x1234);
|
||||
let question = Question::new(
|
||||
Name::new_unchecked(name).into_owned(),
|
||||
QTYPE::TYPE(rtype),
|
||||
simple_dns::QCLASS::CLASS(CLASS::IN),
|
||||
false,
|
||||
);
|
||||
packet.questions.push(question);
|
||||
packet.build_bytes_vec().unwrap()
|
||||
}
|
||||
}
|
||||
@@ -300,10 +300,10 @@ impl Node {
|
||||
pub(in crate::node) async fn maybe_initiate_lookup(&mut self, dest: &NodeAddr) {
|
||||
let now_ms = Self::now_ms();
|
||||
let lookup_timeout_ms = self.config.node.discovery.timeout_secs * 1000;
|
||||
if let Some(&initiated_at) = self.pending_lookups.get(dest) {
|
||||
if now_ms.saturating_sub(initiated_at) < lookup_timeout_ms {
|
||||
return;
|
||||
}
|
||||
if let Some(&initiated_at) = self.pending_lookups.get(dest)
|
||||
&& now_ms.saturating_sub(initiated_at) < lookup_timeout_ms
|
||||
{
|
||||
return;
|
||||
}
|
||||
self.pending_lookups.insert(*dest, now_ms);
|
||||
let ttl = self.config.node.discovery.ttl;
|
||||
|
||||
@@ -2,7 +2,7 @@
|
||||
|
||||
use crate::node::Node;
|
||||
use crate::transport::ReceivedPacket;
|
||||
use crate::wire::EncryptedHeader;
|
||||
use crate::node::wire::EncryptedHeader;
|
||||
use tracing::{debug, warn};
|
||||
|
||||
impl Node {
|
||||
|
||||
@@ -5,7 +5,7 @@ use crate::peer::{
|
||||
cross_connection_winner, ActivePeer, PeerConnection, PromotionResult,
|
||||
};
|
||||
use crate::transport::{Link, LinkDirection, LinkId, ReceivedPacket};
|
||||
use crate::wire::{build_msg2, Msg1Header, Msg2Header};
|
||||
use crate::node::wire::{build_msg2, Msg1Header, Msg2Header};
|
||||
use crate::PeerIdentity;
|
||||
use std::time::Duration;
|
||||
use tracing::{debug, info, warn};
|
||||
@@ -43,25 +43,25 @@ impl Node {
|
||||
// (we initiated to them AND they initiated to us), this is a cross-connection.
|
||||
// Allow it to proceed — promote_connection() will resolve via tie-breaker.
|
||||
let addr_key = (packet.transport_id, packet.remote_addr.clone());
|
||||
if let Some(&existing_link_id) = self.addr_to_link.get(&addr_key) {
|
||||
if let Some(link) = self.links.get(&existing_link_id) {
|
||||
if link.direction() == LinkDirection::Inbound {
|
||||
self.msg1_rate_limiter.complete_handshake();
|
||||
debug!(
|
||||
transport_id = %packet.transport_id,
|
||||
remote_addr = %packet.remote_addr,
|
||||
"Already have inbound connection from this address"
|
||||
);
|
||||
return;
|
||||
}
|
||||
// Outbound link to this address — cross-connection, allow msg1
|
||||
if let Some(&existing_link_id) = self.addr_to_link.get(&addr_key)
|
||||
&& let Some(link) = self.links.get(&existing_link_id)
|
||||
{
|
||||
if link.direction() == LinkDirection::Inbound {
|
||||
self.msg1_rate_limiter.complete_handshake();
|
||||
debug!(
|
||||
transport_id = %packet.transport_id,
|
||||
remote_addr = %packet.remote_addr,
|
||||
existing_link_id = %existing_link_id,
|
||||
"Cross-connection detected: have outbound, received inbound msg1"
|
||||
"Already have inbound connection from this address"
|
||||
);
|
||||
return;
|
||||
}
|
||||
// Outbound link to this address — cross-connection, allow msg1
|
||||
debug!(
|
||||
transport_id = %packet.transport_id,
|
||||
remote_addr = %packet.remote_addr,
|
||||
existing_link_id = %existing_link_id,
|
||||
"Cross-connection detected: have outbound, received inbound msg1"
|
||||
);
|
||||
}
|
||||
|
||||
// === CRYPTO COST PAID HERE ===
|
||||
@@ -89,7 +89,7 @@ impl Node {
|
||||
|
||||
// Learn peer identity from msg1
|
||||
let peer_identity = match conn.expected_identity() {
|
||||
Some(id) => id.clone(),
|
||||
Some(id) => *id,
|
||||
None => {
|
||||
self.msg1_rate_limiter.complete_handshake();
|
||||
warn!("Identity not learned from msg1");
|
||||
@@ -275,7 +275,7 @@ impl Node {
|
||||
|
||||
// Get peer identity for promotion
|
||||
let peer_identity = match conn.expected_identity() {
|
||||
Some(id) => id.clone(),
|
||||
Some(id) => *id,
|
||||
None => {
|
||||
warn!(link_id = %link_id, "No identity after handshake");
|
||||
return;
|
||||
@@ -398,7 +398,7 @@ impl Node {
|
||||
}
|
||||
|
||||
// Normal path: promote to active peer
|
||||
match self.promote_connection(link_id, peer_identity.clone(), packet.timestamp_ms) {
|
||||
match self.promote_connection(link_id, peer_identity, packet.timestamp_ms) {
|
||||
Ok(result) => {
|
||||
// Clean up pending_outbound
|
||||
self.pending_outbound.remove(&key);
|
||||
|
||||
@@ -2,7 +2,7 @@
|
||||
|
||||
use crate::node::{Node, NodeError};
|
||||
use crate::transport::ReceivedPacket;
|
||||
use crate::wire::{DISCRIMINATOR_ENCRYPTED, DISCRIMINATOR_MSG1, DISCRIMINATOR_MSG2};
|
||||
use crate::node::wire::{DISCRIMINATOR_ENCRYPTED, DISCRIMINATOR_MSG1, DISCRIMINATOR_MSG2};
|
||||
use std::time::Duration;
|
||||
use tracing::{debug, info};
|
||||
|
||||
|
||||
@@ -379,10 +379,10 @@ impl Node {
|
||||
dest_pubkey: PublicKey,
|
||||
) -> Result<(), NodeError> {
|
||||
// Check for existing session
|
||||
if let Some(existing) = self.sessions.get(&dest_addr) {
|
||||
if existing.state().is_established() || existing.state().is_initiating() {
|
||||
return Ok(());
|
||||
}
|
||||
if let Some(existing) = self.sessions.get(&dest_addr)
|
||||
&& (existing.state().is_established() || existing.state().is_initiating())
|
||||
{
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
// Create Noise IK initiator handshake
|
||||
@@ -638,10 +638,10 @@ impl Node {
|
||||
};
|
||||
|
||||
// Skip if a session already exists
|
||||
if let Some(existing) = self.sessions.get(&dest_addr) {
|
||||
if existing.state().is_established() || existing.state().is_initiating() {
|
||||
return;
|
||||
}
|
||||
if let Some(existing) = self.sessions.get(&dest_addr)
|
||||
&& (existing.state().is_established() || existing.state().is_initiating())
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
match self.initiate_session(dest_addr, dest_pubkey).await {
|
||||
|
||||
@@ -46,10 +46,10 @@ impl Node {
|
||||
}
|
||||
|
||||
// Schedule retry for failed outbound auto-connect peers
|
||||
if conn.is_outbound() {
|
||||
if let Some(identity) = conn.expected_identity() {
|
||||
self.schedule_retry(*identity.node_addr(), now_ms);
|
||||
}
|
||||
if conn.is_outbound()
|
||||
&& let Some(identity) = conn.expected_identity()
|
||||
{
|
||||
self.schedule_retry(*identity.node_addr(), now_ms);
|
||||
}
|
||||
}
|
||||
self.cleanup_stale_connection(link_id, now_ms);
|
||||
|
||||
@@ -5,7 +5,7 @@ use crate::peer::PeerConnection;
|
||||
use crate::protocol::{Disconnect, DisconnectReason};
|
||||
use crate::transport::{packet_channel, Link, LinkDirection, TransportAddr};
|
||||
use crate::tun::{run_tun_reader, shutdown_tun_interface, TunDevice, TunState};
|
||||
use crate::wire::build_msg1;
|
||||
use crate::node::wire::build_msg1;
|
||||
use crate::{NodeAddr, PeerIdentity};
|
||||
use std::thread;
|
||||
use std::time::Duration;
|
||||
@@ -116,7 +116,7 @@ impl Node {
|
||||
.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.clone(), current_time_ms);
|
||||
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() {
|
||||
@@ -319,7 +319,7 @@ impl Node {
|
||||
let dns_channel_size = self.config.node.buffers.dns_channel;
|
||||
let (identity_tx, identity_rx) = tokio::sync::mpsc::channel(dns_channel_size);
|
||||
let dns_ttl = self.config.dns.ttl();
|
||||
let handle = tokio::spawn(crate::dns::run_dns_responder(socket, identity_tx, dns_ttl));
|
||||
let handle = tokio::spawn(crate::node::dns::run_dns_responder(socket, identity_tx, dns_ttl));
|
||||
self.dns_identity_rx = Some(identity_rx);
|
||||
self.dns_task = Some(handle);
|
||||
info!(bind = %bind, "DNS responder started for .fips domain");
|
||||
|
||||
+12
-11
@@ -5,10 +5,13 @@
|
||||
//! Bloom filters, coordinate caches, transports, links, and peers.
|
||||
|
||||
mod bloom;
|
||||
pub(crate) mod dns;
|
||||
mod handlers;
|
||||
mod lifecycle;
|
||||
mod retry;
|
||||
mod rate_limit;
|
||||
pub(crate) mod session;
|
||||
pub(crate) mod wire;
|
||||
mod tree;
|
||||
#[cfg(test)]
|
||||
mod tests;
|
||||
@@ -18,14 +21,14 @@ use crate::cache::{CoordCache, RouteCache};
|
||||
use crate::index::IndexAllocator;
|
||||
use crate::node::session::SessionEntry;
|
||||
use crate::peer::{ActivePeer, PeerConnection};
|
||||
use crate::rate_limit::HandshakeRateLimiter;
|
||||
use self::rate_limit::HandshakeRateLimiter;
|
||||
use crate::transport::{
|
||||
Link, LinkId, PacketRx, PacketTx, TransportAddr, TransportHandle, TransportId,
|
||||
};
|
||||
use crate::transport::udp::UdpTransport;
|
||||
use crate::tree::TreeState;
|
||||
use crate::tun::{TunError, TunOutboundRx, TunState, TunTx};
|
||||
use crate::wire::build_encrypted;
|
||||
use self::wire::build_encrypted;
|
||||
use crate::{Config, ConfigError, Identity, IdentityError, NodeAddr};
|
||||
use std::collections::{HashMap, VecDeque};
|
||||
use std::fmt;
|
||||
@@ -186,9 +189,7 @@ type AddrKey = (TransportId, TransportAddr);
|
||||
///
|
||||
/// The `addr_to_link` map enables dispatching incoming packets to the right
|
||||
/// connection before authentication completes.
|
||||
///
|
||||
// Discovery lookup constants moved to config: node.discovery.timeout_secs, node.discovery.ttl
|
||||
|
||||
pub struct Node {
|
||||
// === Identity ===
|
||||
/// This node's cryptographic identity.
|
||||
@@ -296,7 +297,7 @@ pub struct Node {
|
||||
|
||||
// === DNS Responder ===
|
||||
/// Receiver for resolved identities from the DNS responder.
|
||||
dns_identity_rx: Option<crate::dns::DnsIdentityRx>,
|
||||
dns_identity_rx: Option<dns::DnsIdentityRx>,
|
||||
/// DNS responder task handle.
|
||||
dns_task: Option<tokio::task::JoinHandle<()>>,
|
||||
|
||||
@@ -360,7 +361,7 @@ impl Node {
|
||||
let route_cache = RouteCache::new(config.node.cache.route_size);
|
||||
let rl = &config.node.rate_limit;
|
||||
let msg1_rate_limiter = HandshakeRateLimiter::with_params(
|
||||
crate::rate_limit::TokenBucket::with_params(rl.handshake_burst, rl.handshake_rate),
|
||||
rate_limit::TokenBucket::with_params(rl.handshake_burst, rl.handshake_rate),
|
||||
config.node.limits.max_pending_inbound,
|
||||
);
|
||||
|
||||
@@ -437,7 +438,7 @@ impl Node {
|
||||
let route_cache = RouteCache::new(config.node.cache.route_size);
|
||||
let rl = &config.node.rate_limit;
|
||||
let msg1_rate_limiter = HandshakeRateLimiter::with_params(
|
||||
crate::rate_limit::TokenBucket::with_params(rl.handshake_burst, rl.handshake_rate),
|
||||
rate_limit::TokenBucket::with_params(rl.handshake_burst, rl.handshake_rate),
|
||||
config.node.limits.max_pending_inbound,
|
||||
);
|
||||
|
||||
@@ -903,10 +904,10 @@ impl Node {
|
||||
}
|
||||
|
||||
// 2. Direct peer
|
||||
if let Some(peer) = self.peers.get(dest_node_addr) {
|
||||
if peer.can_send() {
|
||||
return Some(peer);
|
||||
}
|
||||
if let Some(peer) = self.peers.get(dest_node_addr)
|
||||
&& peer.can_send()
|
||||
{
|
||||
return Some(peer);
|
||||
}
|
||||
|
||||
// Look up destination coords (required by both bloom and tree paths).
|
||||
|
||||
@@ -0,0 +1,404 @@
|
||||
//! Rate Limiting for FIPS Protocol
|
||||
//!
|
||||
//! Provides token bucket rate limiting for protecting against DoS attacks,
|
||||
//! particularly on the Noise handshake path where msg1 processing involves
|
||||
//! expensive cryptographic operations.
|
||||
//!
|
||||
//! ## Design
|
||||
//!
|
||||
//! - Token bucket algorithm with configurable burst and refill rate
|
||||
//! - Global rate limit (not per-source, since UDP sources are spoofable)
|
||||
//! - Applied before expensive DH operations in handshake processing
|
||||
//!
|
||||
//! ## Default Parameters
|
||||
//!
|
||||
//! - Burst capacity: 100 tokens (max concurrent handshakes)
|
||||
//! - Refill rate: 10 tokens/second (sustained handshake rate)
|
||||
//! - This allows handling burst traffic while limiting sustained attack impact
|
||||
|
||||
use std::time::Instant;
|
||||
|
||||
/// Default burst capacity (max tokens).
|
||||
pub const DEFAULT_BURST_CAPACITY: u32 = 100;
|
||||
|
||||
/// Default refill rate (tokens per second).
|
||||
pub const DEFAULT_REFILL_RATE: f64 = 10.0;
|
||||
|
||||
/// Token bucket rate limiter.
|
||||
///
|
||||
/// Uses a classic token bucket algorithm where tokens are consumed for each
|
||||
/// operation and refilled at a constant rate. When tokens are exhausted,
|
||||
/// operations are rate-limited until tokens refill.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct TokenBucket {
|
||||
/// Maximum number of tokens (burst capacity).
|
||||
capacity: u32,
|
||||
/// Current number of available tokens (may be fractional during refill).
|
||||
tokens: f64,
|
||||
/// Tokens added per second.
|
||||
refill_rate: f64,
|
||||
/// Last time tokens were refilled.
|
||||
last_refill: Instant,
|
||||
}
|
||||
|
||||
impl TokenBucket {
|
||||
/// Create a new token bucket with default parameters.
|
||||
///
|
||||
/// - Burst capacity: 100 tokens
|
||||
/// - Refill rate: 10 tokens/second
|
||||
pub fn new() -> Self {
|
||||
Self::with_params(DEFAULT_BURST_CAPACITY, DEFAULT_REFILL_RATE)
|
||||
}
|
||||
|
||||
/// Create a token bucket with custom parameters.
|
||||
///
|
||||
/// # Arguments
|
||||
///
|
||||
/// * `capacity` - Maximum number of tokens (burst capacity)
|
||||
/// * `refill_rate` - Tokens added per second
|
||||
pub fn with_params(capacity: u32, refill_rate: f64) -> Self {
|
||||
Self {
|
||||
capacity,
|
||||
tokens: capacity as f64,
|
||||
refill_rate,
|
||||
last_refill: Instant::now(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Try to consume one token.
|
||||
///
|
||||
/// Returns `true` if a token was available and consumed, `false` if
|
||||
/// rate limited (no tokens available).
|
||||
pub fn try_acquire(&mut self) -> bool {
|
||||
self.try_acquire_n(1)
|
||||
}
|
||||
|
||||
/// Try to consume n tokens.
|
||||
///
|
||||
/// Returns `true` if n tokens were available and consumed, `false` if
|
||||
/// rate limited (insufficient tokens).
|
||||
pub fn try_acquire_n(&mut self, n: u32) -> bool {
|
||||
self.refill();
|
||||
|
||||
if self.tokens >= n as f64 {
|
||||
self.tokens -= n as f64;
|
||||
true
|
||||
} else {
|
||||
false
|
||||
}
|
||||
}
|
||||
|
||||
/// Check if tokens are available without consuming them.
|
||||
#[cfg(test)]
|
||||
pub fn available(&mut self) -> bool {
|
||||
self.refill();
|
||||
self.tokens >= 1.0
|
||||
}
|
||||
|
||||
/// Get the current number of available tokens.
|
||||
#[cfg(test)]
|
||||
pub fn tokens(&mut self) -> f64 {
|
||||
self.refill();
|
||||
self.tokens
|
||||
}
|
||||
|
||||
/// Get the capacity (max tokens).
|
||||
#[cfg(test)]
|
||||
pub fn capacity(&self) -> u32 {
|
||||
self.capacity
|
||||
}
|
||||
|
||||
/// Refill tokens based on elapsed time.
|
||||
fn refill(&mut self) {
|
||||
let now = Instant::now();
|
||||
let elapsed = now.duration_since(self.last_refill);
|
||||
let elapsed_secs = elapsed.as_secs_f64();
|
||||
|
||||
// Add tokens based on time elapsed
|
||||
self.tokens += elapsed_secs * self.refill_rate;
|
||||
|
||||
// Cap at capacity
|
||||
if self.tokens > self.capacity as f64 {
|
||||
self.tokens = self.capacity as f64;
|
||||
}
|
||||
|
||||
self.last_refill = now;
|
||||
}
|
||||
|
||||
/// Reset to full capacity.
|
||||
#[cfg(test)]
|
||||
pub fn reset(&mut self) {
|
||||
self.tokens = self.capacity as f64;
|
||||
self.last_refill = Instant::now();
|
||||
}
|
||||
|
||||
/// Time until the next token is available.
|
||||
///
|
||||
/// Returns `Duration::ZERO` if tokens are available, otherwise the
|
||||
/// estimated time until one token will be available.
|
||||
#[cfg(test)]
|
||||
pub fn time_until_available(&mut self) -> std::time::Duration {
|
||||
self.refill();
|
||||
|
||||
if self.tokens >= 1.0 {
|
||||
std::time::Duration::ZERO
|
||||
} else {
|
||||
let needed = 1.0 - self.tokens;
|
||||
let secs = needed / self.refill_rate;
|
||||
std::time::Duration::from_secs_f64(secs)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for TokenBucket {
|
||||
fn default() -> Self {
|
||||
Self::new()
|
||||
}
|
||||
}
|
||||
|
||||
/// Rate limiter for handshake message 1 processing.
|
||||
///
|
||||
/// Combines token bucket rate limiting with connection counting to
|
||||
/// protect against DoS attacks on the handshake path.
|
||||
#[derive(Debug)]
|
||||
pub struct HandshakeRateLimiter {
|
||||
/// Token bucket for rate limiting.
|
||||
bucket: TokenBucket,
|
||||
/// Current count of pending inbound connections.
|
||||
pending_count: usize,
|
||||
/// Maximum pending inbound connections.
|
||||
max_pending: usize,
|
||||
}
|
||||
|
||||
impl HandshakeRateLimiter {
|
||||
/// Create a handshake rate limiter with the given parameters.
|
||||
pub fn with_params(bucket: TokenBucket, max_pending: usize) -> Self {
|
||||
Self {
|
||||
bucket,
|
||||
pending_count: 0,
|
||||
max_pending,
|
||||
}
|
||||
}
|
||||
|
||||
/// Check if a new handshake can be started.
|
||||
///
|
||||
/// Returns `true` if:
|
||||
/// - Token bucket has available tokens (rate limit not exceeded)
|
||||
/// - Pending connection count is below maximum
|
||||
///
|
||||
/// Does NOT consume a token - call `start_handshake` for that.
|
||||
#[cfg(test)]
|
||||
pub fn can_start_handshake(&mut self) -> bool {
|
||||
self.bucket.available() && self.pending_count < self.max_pending
|
||||
}
|
||||
|
||||
/// Start a new handshake, consuming a token and incrementing pending count.
|
||||
///
|
||||
/// Returns `true` if the handshake was allowed, `false` if rate limited.
|
||||
pub fn start_handshake(&mut self) -> bool {
|
||||
if self.pending_count >= self.max_pending {
|
||||
return false;
|
||||
}
|
||||
|
||||
if self.bucket.try_acquire() {
|
||||
self.pending_count += 1;
|
||||
true
|
||||
} else {
|
||||
false
|
||||
}
|
||||
}
|
||||
|
||||
/// Mark a handshake as complete (successful or failed).
|
||||
///
|
||||
/// Decrements the pending connection count.
|
||||
pub fn complete_handshake(&mut self) {
|
||||
if self.pending_count > 0 {
|
||||
self.pending_count -= 1;
|
||||
}
|
||||
}
|
||||
|
||||
/// Get the current pending connection count.
|
||||
#[cfg(test)]
|
||||
pub fn pending_count(&self) -> usize {
|
||||
self.pending_count
|
||||
}
|
||||
|
||||
/// Get a reference to the token bucket.
|
||||
#[cfg(test)]
|
||||
pub fn bucket(&self) -> &TokenBucket {
|
||||
&self.bucket
|
||||
}
|
||||
|
||||
/// Reset the rate limiter.
|
||||
#[cfg(test)]
|
||||
pub fn reset(&mut self) {
|
||||
self.bucket.reset();
|
||||
self.pending_count = 0;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use std::thread;
|
||||
use std::time::Duration;
|
||||
|
||||
#[test]
|
||||
fn test_token_bucket_basic() {
|
||||
let mut bucket = TokenBucket::with_params(10, 1.0);
|
||||
|
||||
// Should have full capacity
|
||||
assert_eq!(bucket.capacity(), 10);
|
||||
assert!(bucket.tokens() >= 9.9); // Allow for timing
|
||||
|
||||
// Consume all tokens
|
||||
for _ in 0..10 {
|
||||
assert!(bucket.try_acquire());
|
||||
}
|
||||
|
||||
// Should be empty
|
||||
assert!(!bucket.try_acquire());
|
||||
assert!(!bucket.available());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_token_bucket_refill() {
|
||||
let mut bucket = TokenBucket::with_params(10, 100.0); // 100 tokens/sec
|
||||
|
||||
// Drain completely
|
||||
for _ in 0..10 {
|
||||
bucket.try_acquire();
|
||||
}
|
||||
assert!(!bucket.available());
|
||||
|
||||
// Wait for refill
|
||||
thread::sleep(Duration::from_millis(50)); // Should refill ~5 tokens
|
||||
|
||||
// Should have tokens now
|
||||
let tokens = bucket.tokens();
|
||||
assert!((4.0..=6.0).contains(&tokens), "tokens: {}", tokens);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_token_bucket_try_acquire_n() {
|
||||
let mut bucket = TokenBucket::with_params(10, 1.0);
|
||||
|
||||
// Acquire 5
|
||||
assert!(bucket.try_acquire_n(5));
|
||||
assert!(bucket.tokens() >= 4.9 && bucket.tokens() <= 5.1);
|
||||
|
||||
// Acquire 5 more
|
||||
assert!(bucket.try_acquire_n(5));
|
||||
|
||||
// Can't acquire more
|
||||
assert!(!bucket.try_acquire_n(1));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_token_bucket_reset() {
|
||||
let mut bucket = TokenBucket::with_params(10, 1.0);
|
||||
|
||||
// Drain
|
||||
for _ in 0..10 {
|
||||
bucket.try_acquire();
|
||||
}
|
||||
|
||||
// Reset
|
||||
bucket.reset();
|
||||
|
||||
// Should be full again
|
||||
assert!(bucket.tokens() >= 9.9);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_token_bucket_time_until_available() {
|
||||
let mut bucket = TokenBucket::with_params(10, 10.0); // 10 tokens/sec
|
||||
|
||||
// When full, should be zero
|
||||
assert_eq!(bucket.time_until_available(), Duration::ZERO);
|
||||
|
||||
// Drain completely
|
||||
for _ in 0..10 {
|
||||
bucket.try_acquire();
|
||||
}
|
||||
|
||||
// Should need ~100ms for one token at 10/sec
|
||||
let wait = bucket.time_until_available();
|
||||
assert!(wait.as_millis() >= 90 && wait.as_millis() <= 110);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_handshake_rate_limiter_basic() {
|
||||
let mut limiter = HandshakeRateLimiter::with_params(TokenBucket::new(), 100);
|
||||
|
||||
assert!(limiter.can_start_handshake());
|
||||
assert_eq!(limiter.pending_count(), 0);
|
||||
|
||||
// Start a handshake
|
||||
assert!(limiter.start_handshake());
|
||||
assert_eq!(limiter.pending_count(), 1);
|
||||
|
||||
// Complete it
|
||||
limiter.complete_handshake();
|
||||
assert_eq!(limiter.pending_count(), 0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_handshake_rate_limiter_max_pending() {
|
||||
let bucket = TokenBucket::with_params(1000, 100.0);
|
||||
let mut limiter = HandshakeRateLimiter::with_params(bucket, 3);
|
||||
|
||||
// Start 3 handshakes
|
||||
assert!(limiter.start_handshake());
|
||||
assert!(limiter.start_handshake());
|
||||
assert!(limiter.start_handshake());
|
||||
|
||||
// Fourth should fail (max pending)
|
||||
assert!(!limiter.can_start_handshake());
|
||||
assert!(!limiter.start_handshake());
|
||||
|
||||
// Complete one
|
||||
limiter.complete_handshake();
|
||||
|
||||
// Now should be able to start another
|
||||
assert!(limiter.can_start_handshake());
|
||||
assert!(limiter.start_handshake());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_handshake_rate_limiter_token_exhaustion() {
|
||||
let bucket = TokenBucket::with_params(5, 0.0); // No refill
|
||||
let mut limiter = HandshakeRateLimiter::with_params(bucket, 100);
|
||||
|
||||
// Start 5 handshakes (exhausts tokens)
|
||||
for _ in 0..5 {
|
||||
assert!(limiter.start_handshake());
|
||||
}
|
||||
|
||||
// Complete them all
|
||||
for _ in 0..5 {
|
||||
limiter.complete_handshake();
|
||||
}
|
||||
|
||||
// Tokens exhausted, even though pending is 0
|
||||
assert!(!limiter.can_start_handshake());
|
||||
assert!(!limiter.start_handshake());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_handshake_rate_limiter_reset() {
|
||||
let mut limiter = HandshakeRateLimiter::with_params(TokenBucket::new(), 100);
|
||||
|
||||
// Start some handshakes
|
||||
limiter.start_handshake();
|
||||
limiter.start_handshake();
|
||||
assert_eq!(limiter.pending_count(), 2);
|
||||
|
||||
// Reset
|
||||
limiter.reset();
|
||||
|
||||
assert_eq!(limiter.pending_count(), 0);
|
||||
assert!(limiter.bucket().tokens >= DEFAULT_BURST_CAPACITY as f64 - 0.1);
|
||||
}
|
||||
}
|
||||
+1
-1
@@ -103,7 +103,7 @@ impl SessionEntry {
|
||||
|
||||
/// Check if the session is established.
|
||||
pub(crate) fn is_established(&self) -> bool {
|
||||
self.state.as_ref().map_or(false, |s| s.is_established())
|
||||
self.state.as_ref().is_some_and(|s| s.is_established())
|
||||
}
|
||||
|
||||
/// Get creation time.
|
||||
|
||||
@@ -120,11 +120,12 @@ async fn test_bloom_filter_star() {
|
||||
let filter = peer.inbound_filter().unwrap();
|
||||
|
||||
// Filter from hub should contain all OTHER spokes
|
||||
for other in 1..5 {
|
||||
for (other, other_node) in nodes[1..5].iter().enumerate() {
|
||||
let other = other + 1; // adjust for slice offset
|
||||
if other == spoke {
|
||||
continue;
|
||||
}
|
||||
let other_addr = *nodes[other].node.node_addr();
|
||||
let other_addr = *other_node.node.node_addr();
|
||||
assert!(
|
||||
filter.contains(&other_addr),
|
||||
"Spoke {}'s filter from hub should contain spoke {} (addr={})",
|
||||
@@ -168,12 +169,12 @@ async fn test_bloom_filter_chain_propagation() {
|
||||
// Entries propagate through the full chain because each
|
||||
// intermediate node merges its peer's filter into its outgoing
|
||||
// filter. Verify all nodes are reachable from the endpoints.
|
||||
for i in 2..8 {
|
||||
for (i, addr) in addrs[2..8].iter().enumerate() {
|
||||
assert!(
|
||||
filter.contains(&addrs[i]),
|
||||
filter.contains(addr),
|
||||
"Node 0's filter from node 1 should contain node {} \
|
||||
(chain merge propagation)",
|
||||
i
|
||||
i + 2
|
||||
);
|
||||
}
|
||||
|
||||
|
||||
@@ -100,20 +100,21 @@ async fn test_disconnect_star_hub_departs() {
|
||||
process_available_packets(&mut nodes).await;
|
||||
|
||||
// All spokes should have removed the hub
|
||||
for spoke_idx in 1..4 {
|
||||
for (spoke_idx, spoke) in nodes[1..4].iter().enumerate() {
|
||||
let spoke_idx = spoke_idx + 1; // adjust for slice offset
|
||||
assert!(
|
||||
nodes[spoke_idx].node.get_peer(&hub_addr).is_none(),
|
||||
spoke.node.get_peer(&hub_addr).is_none(),
|
||||
"Spoke {} should have removed hub",
|
||||
spoke_idx
|
||||
);
|
||||
assert_eq!(
|
||||
nodes[spoke_idx].node.peer_count(),
|
||||
spoke.node.peer_count(),
|
||||
0,
|
||||
"Spoke {} should have no peers (no spoke-spoke links)",
|
||||
spoke_idx
|
||||
);
|
||||
assert!(
|
||||
nodes[spoke_idx].node.tree_state().is_root(),
|
||||
spoke.node.tree_state().is_root(),
|
||||
"Isolated spoke {} should become root",
|
||||
spoke_idx
|
||||
);
|
||||
|
||||
+10
-10
@@ -6,7 +6,7 @@ use super::*;
|
||||
async fn test_two_node_handshake_udp() {
|
||||
use crate::config::UdpConfig;
|
||||
use crate::transport::udp::UdpTransport;
|
||||
use crate::wire::{build_encrypted, build_msg1};
|
||||
use crate::node::wire::{build_encrypted, build_msg1};
|
||||
use tokio::time::{timeout, Duration};
|
||||
|
||||
// === Setup: Two nodes with UDP transports on localhost ===
|
||||
@@ -55,7 +55,7 @@ async fn test_two_node_handshake_udp() {
|
||||
let link_id_a = node_a.allocate_link_id();
|
||||
let mut conn_a = PeerConnection::outbound(
|
||||
link_id_a,
|
||||
peer_b_identity.clone(),
|
||||
peer_b_identity,
|
||||
1000,
|
||||
);
|
||||
|
||||
@@ -233,7 +233,7 @@ async fn test_two_node_handshake_udp() {
|
||||
async fn test_run_rx_loop_handshake() {
|
||||
use crate::config::UdpConfig;
|
||||
use crate::transport::udp::UdpTransport;
|
||||
use crate::wire::build_msg1;
|
||||
use crate::node::wire::build_msg1;
|
||||
use tokio::time::Duration;
|
||||
|
||||
// === Setup: Two nodes with UDP transports on localhost ===
|
||||
@@ -287,7 +287,7 @@ async fn test_run_rx_loop_handshake() {
|
||||
let link_id_a = node_a.allocate_link_id();
|
||||
let mut conn_a = PeerConnection::outbound(
|
||||
link_id_a,
|
||||
peer_b_identity.clone(),
|
||||
peer_b_identity,
|
||||
1000,
|
||||
);
|
||||
|
||||
@@ -423,7 +423,7 @@ async fn test_run_rx_loop_handshake() {
|
||||
async fn test_cross_connection_both_initiate() {
|
||||
use crate::config::UdpConfig;
|
||||
use crate::transport::udp::UdpTransport;
|
||||
use crate::wire::build_msg1;
|
||||
use crate::node::wire::build_msg1;
|
||||
use tokio::time::{timeout, Duration};
|
||||
|
||||
// === Setup: Two nodes with UDP transports on localhost ===
|
||||
@@ -474,7 +474,7 @@ async fn test_cross_connection_both_initiate() {
|
||||
|
||||
// Node A initiates to Node B
|
||||
let link_id_a_out = node_a.allocate_link_id();
|
||||
let mut conn_a = PeerConnection::outbound(link_id_a_out, peer_b_identity.clone(), 1000);
|
||||
let mut conn_a = PeerConnection::outbound(link_id_a_out, peer_b_identity, 1000);
|
||||
let our_index_a = node_a.index_allocator.allocate().unwrap();
|
||||
let our_keypair_a = node_a.identity.keypair();
|
||||
let noise_msg1_a = conn_a.start_handshake(our_keypair_a, 1000).unwrap();
|
||||
@@ -495,7 +495,7 @@ async fn test_cross_connection_both_initiate() {
|
||||
|
||||
// Node B initiates to Node A
|
||||
let link_id_b_out = node_b.allocate_link_id();
|
||||
let mut conn_b = PeerConnection::outbound(link_id_b_out, peer_a_identity.clone(), 1000);
|
||||
let mut conn_b = PeerConnection::outbound(link_id_b_out, peer_a_identity, 1000);
|
||||
let our_index_b = node_b.index_allocator.allocate().unwrap();
|
||||
let our_keypair_b = node_b.identity.keypair();
|
||||
let noise_msg1_b = conn_b.start_handshake(our_keypair_b, 1000).unwrap();
|
||||
@@ -595,7 +595,7 @@ async fn test_stale_connection_cleanup() {
|
||||
// Create outbound connection with a timestamp far in the past
|
||||
let past_time_ms = 1000; // A very early timestamp
|
||||
let link_id = node.allocate_link_id();
|
||||
let mut conn = PeerConnection::outbound(link_id, peer_identity.clone(), past_time_ms);
|
||||
let mut conn = PeerConnection::outbound(link_id, peer_identity, past_time_ms);
|
||||
|
||||
// Allocate session index and set transport info
|
||||
let our_index = node.index_allocator.allocate().unwrap();
|
||||
@@ -631,7 +631,7 @@ async fn test_stale_connection_cleanup() {
|
||||
assert!(!node.pending_outbound.contains_key(&(transport_id, our_index.as_u32())),
|
||||
"pending_outbound should be cleaned up");
|
||||
assert_eq!(node.index_allocator.count(), 0, "Session index should be freed");
|
||||
assert!(node.addr_to_link.get(&(transport_id, remote_addr)).is_none(),
|
||||
assert!(!node.addr_to_link.contains_key(&(transport_id, remote_addr)),
|
||||
"addr_to_link should be cleaned up");
|
||||
}
|
||||
|
||||
@@ -650,7 +650,7 @@ async fn test_failed_connection_cleanup() {
|
||||
.map(|d| d.as_millis() as u64)
|
||||
.unwrap_or(0);
|
||||
let link_id = node.allocate_link_id();
|
||||
let mut conn = PeerConnection::outbound(link_id, peer_identity.clone(), now_ms);
|
||||
let mut conn = PeerConnection::outbound(link_id, peer_identity, now_ms);
|
||||
|
||||
let our_index = node.index_allocator.allocate().unwrap();
|
||||
let our_keypair = node.identity.keypair();
|
||||
|
||||
@@ -46,7 +46,7 @@ pub(super) fn make_completed_connection(
|
||||
let peer_identity = PeerIdentity::from_pubkey_full(peer_identity_full.pubkey_full());
|
||||
|
||||
// Create outbound connection
|
||||
let mut conn = PeerConnection::outbound(link_id, peer_identity.clone(), current_time_ms);
|
||||
let mut conn = PeerConnection::outbound(link_id, peer_identity, current_time_ms);
|
||||
|
||||
// Run initiator side of handshake
|
||||
let our_keypair = node.identity.keypair();
|
||||
|
||||
@@ -562,7 +562,7 @@ async fn test_routing_reachability_100_nodes() {
|
||||
.collect();
|
||||
|
||||
for node in &mut nodes {
|
||||
for &(ref addr, ref coords) in &all_coords {
|
||||
for (addr, coords) in &all_coords {
|
||||
if addr != node.node.node_addr() {
|
||||
node.node.coord_cache_mut().insert(*addr, coords.clone(), now_ms);
|
||||
}
|
||||
@@ -696,7 +696,7 @@ async fn test_routing_stops_after_peer_removal() {
|
||||
.collect();
|
||||
|
||||
for node in &mut nodes {
|
||||
for &(ref addr, ref coords) in &all_coords {
|
||||
for (addr, coords) in &all_coords {
|
||||
if addr != node.node.node_addr() {
|
||||
node.node.coord_cache_mut().insert(*addr, coords.clone(), now_ms);
|
||||
}
|
||||
@@ -931,7 +931,7 @@ async fn test_routing_source_only_coords_100_nodes() {
|
||||
|
||||
// Now compare: inject coords at ALL nodes (full cache) and verify 100%
|
||||
for node in &mut nodes {
|
||||
for &(ref addr, ref coords) in &all_coords {
|
||||
for (addr, coords) in &all_coords {
|
||||
if addr != node.node.node_addr() {
|
||||
node.node.coord_cache_mut().insert(*addr, coords.clone(), now_ms);
|
||||
}
|
||||
|
||||
@@ -639,13 +639,13 @@ async fn test_session_100_nodes() {
|
||||
let fwd_payload = format!("fwd-{}", pair_idx).into_bytes();
|
||||
let rev_payload = format!("rev-{}", pair_idx).into_bytes();
|
||||
|
||||
if delivered_per_node[dst].iter().any(|p| *p == fwd_payload) {
|
||||
if delivered_per_node[dst].contains(&fwd_payload) {
|
||||
fwd_delivered += 1;
|
||||
} else if fwd_missing.len() < 20 {
|
||||
fwd_missing.push((src, dst));
|
||||
}
|
||||
|
||||
if delivered_per_node[src].iter().any(|p| *p == rev_payload) {
|
||||
if delivered_per_node[src].contains(&rev_payload) {
|
||||
rev_delivered += 1;
|
||||
} else if rev_missing.len() < 20 {
|
||||
rev_missing.push((src, dst));
|
||||
|
||||
@@ -48,7 +48,7 @@ pub(super) async fn make_test_node() -> TestNode {
|
||||
/// Sends msg1 over UDP. The drain loop will handle msg1 processing,
|
||||
/// msg2 response, and subsequent TreeAnnounce exchange.
|
||||
pub(super) async fn initiate_handshake(nodes: &mut [TestNode], i: usize, j: usize) {
|
||||
use crate::wire::build_msg1;
|
||||
use crate::node::wire::build_msg1;
|
||||
|
||||
// Extract responder info before mutably borrowing initiator
|
||||
let responder_addr = nodes[j].addr.clone();
|
||||
@@ -203,19 +203,19 @@ pub(super) fn print_tree_snapshot(label: &str, nodes: &[TestNode]) {
|
||||
///
|
||||
/// Returns the number of packets processed.
|
||||
pub(super) async fn process_available_packets(nodes: &mut [TestNode]) -> usize {
|
||||
use crate::wire::{DISCRIMINATOR_ENCRYPTED, DISCRIMINATOR_MSG1, DISCRIMINATOR_MSG2};
|
||||
use crate::node::wire::{DISCRIMINATOR_ENCRYPTED, DISCRIMINATOR_MSG1, DISCRIMINATOR_MSG2};
|
||||
|
||||
let mut count = 0;
|
||||
for i in 0..nodes.len() {
|
||||
while let Ok(packet) = nodes[i].packet_rx.try_recv() {
|
||||
for node in nodes.iter_mut() {
|
||||
while let Ok(packet) = node.packet_rx.try_recv() {
|
||||
if packet.data.is_empty() {
|
||||
continue;
|
||||
}
|
||||
match packet.data[0] {
|
||||
DISCRIMINATOR_MSG1 => nodes[i].node.handle_msg1(packet).await,
|
||||
DISCRIMINATOR_MSG2 => nodes[i].node.handle_msg2(packet).await,
|
||||
DISCRIMINATOR_MSG1 => node.node.handle_msg1(packet).await,
|
||||
DISCRIMINATOR_MSG2 => node.node.handle_msg2(packet).await,
|
||||
DISCRIMINATOR_ENCRYPTED => {
|
||||
nodes[i].node.handle_encrypted_frame(packet).await
|
||||
node.node.handle_encrypted_frame(packet).await
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
|
||||
@@ -152,7 +152,7 @@ fn test_node_connection_duplicate() {
|
||||
|
||||
let identity = make_peer_identity();
|
||||
let link_id = LinkId::new(1);
|
||||
let conn1 = PeerConnection::outbound(link_id, identity.clone(), 1000);
|
||||
let conn1 = PeerConnection::outbound(link_id, identity, 1000);
|
||||
let conn2 = PeerConnection::outbound(link_id, identity, 2000);
|
||||
|
||||
node.add_connection(conn1).unwrap();
|
||||
@@ -206,7 +206,7 @@ fn test_node_cross_connection_resolution() {
|
||||
let node_addr = *identity.node_addr();
|
||||
|
||||
node.add_connection(conn1).unwrap();
|
||||
node.promote_connection(link_id1, identity.clone(), 1500).unwrap();
|
||||
node.promote_connection(link_id1, identity, 1500).unwrap();
|
||||
|
||||
assert_eq!(node.peer_count(), 1);
|
||||
assert_eq!(node.get_peer(&node_addr).unwrap().link_id(), link_id1);
|
||||
@@ -447,7 +447,7 @@ fn test_promote_cleans_up_pending_outbound_to_same_peer() {
|
||||
let pending_link_id = LinkId::new(1);
|
||||
let pending_time_ms = 1000;
|
||||
let mut pending_conn =
|
||||
PeerConnection::outbound(pending_link_id, peer_b_identity.clone(), pending_time_ms);
|
||||
PeerConnection::outbound(pending_link_id, peer_b_identity, pending_time_ms);
|
||||
|
||||
let our_keypair = node.identity.keypair();
|
||||
let _msg1 = pending_conn.start_handshake(our_keypair, pending_time_ms).unwrap();
|
||||
@@ -485,7 +485,7 @@ fn test_promote_cleans_up_pending_outbound_to_same_peer() {
|
||||
|
||||
let mut completing_conn = PeerConnection::outbound(
|
||||
completing_link_id,
|
||||
peer_b_identity.clone(),
|
||||
peer_b_identity,
|
||||
completing_time_ms,
|
||||
);
|
||||
|
||||
@@ -519,7 +519,7 @@ fn test_promote_cleans_up_pending_outbound_to_same_peer() {
|
||||
|
||||
// --- Promote the completing connection ---
|
||||
let result = node
|
||||
.promote_connection(completing_link_id, peer_b_identity.clone(), completing_time_ms)
|
||||
.promote_connection(completing_link_id, peer_b_identity, completing_time_ms)
|
||||
.unwrap();
|
||||
|
||||
assert!(matches!(result, PromotionResult::Promoted(_)));
|
||||
|
||||
@@ -0,0 +1,360 @@
|
||||
//! Wire Format Parsing and Serialization
|
||||
//!
|
||||
//! Defines the FIPS link-layer wire format for packet dispatch.
|
||||
//! All packets begin with a discriminator byte followed by type-specific payload.
|
||||
//!
|
||||
//! ## Packet Types
|
||||
//!
|
||||
//! | Byte | Type | Size | Description |
|
||||
//! |------|-----------------|-----------|--------------------------------|
|
||||
//! | 0x00 | Encrypted frame | 29+ bytes | Post-handshake encrypted data |
|
||||
//! | 0x01 | Noise IK msg1 | 87 bytes | Handshake initiation |
|
||||
//! | 0x02 | Noise IK msg2 | 42 bytes | Handshake response |
|
||||
|
||||
use crate::index::SessionIndex;
|
||||
use crate::noise::{HANDSHAKE_MSG1_SIZE, HANDSHAKE_MSG2_SIZE, TAG_SIZE};
|
||||
|
||||
// ============================================================================
|
||||
// Constants
|
||||
// ============================================================================
|
||||
|
||||
/// Discriminator for encrypted frames (post-handshake data).
|
||||
pub const DISCRIMINATOR_ENCRYPTED: u8 = 0x00;
|
||||
|
||||
/// Discriminator for Noise IK message 1 (handshake initiation).
|
||||
pub const DISCRIMINATOR_MSG1: u8 = 0x01;
|
||||
|
||||
/// Discriminator for Noise IK message 2 (handshake response).
|
||||
pub const DISCRIMINATOR_MSG2: u8 = 0x02;
|
||||
|
||||
/// Size of Noise IK message 1 wire packet: discriminator + sender_idx + noise_msg1.
|
||||
pub const MSG1_WIRE_SIZE: usize = 1 + 4 + HANDSHAKE_MSG1_SIZE; // 87 bytes
|
||||
|
||||
/// Size of Noise IK message 2 wire packet: discriminator + sender_idx + receiver_idx + noise_msg2.
|
||||
pub const MSG2_WIRE_SIZE: usize = 1 + 4 + 4 + HANDSHAKE_MSG2_SIZE; // 42 bytes
|
||||
|
||||
/// Minimum size for encrypted frame: discriminator + receiver_idx + counter + tag.
|
||||
pub const ENCRYPTED_MIN_SIZE: usize = 1 + 4 + 8 + TAG_SIZE; // 29 bytes
|
||||
|
||||
// ============================================================================
|
||||
// Encrypted Frame Header
|
||||
// ============================================================================
|
||||
|
||||
/// Parsed encrypted frame header.
|
||||
///
|
||||
/// Wire format:
|
||||
/// ```text
|
||||
/// [0x00][receiver_idx:4 LE][counter:8 LE][ciphertext+tag]
|
||||
/// ```
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct EncryptedHeader {
|
||||
/// Session index chosen by the receiver (for O(1) lookup).
|
||||
pub receiver_idx: SessionIndex,
|
||||
/// Monotonic counter used as AEAD nonce.
|
||||
pub counter: u64,
|
||||
/// Offset where ciphertext begins in the original packet.
|
||||
pub ciphertext_offset: usize,
|
||||
}
|
||||
|
||||
impl EncryptedHeader {
|
||||
/// Parse an encrypted frame header from packet data.
|
||||
///
|
||||
/// Returns None if the packet is too short or has wrong discriminator.
|
||||
pub fn parse(data: &[u8]) -> Option<Self> {
|
||||
if data.len() < ENCRYPTED_MIN_SIZE {
|
||||
return None;
|
||||
}
|
||||
|
||||
if data[0] != DISCRIMINATOR_ENCRYPTED {
|
||||
return None;
|
||||
}
|
||||
|
||||
let receiver_idx = SessionIndex::from_le_bytes([data[1], data[2], data[3], data[4]]);
|
||||
let counter = u64::from_le_bytes([
|
||||
data[5], data[6], data[7], data[8], data[9], data[10], data[11], data[12],
|
||||
]);
|
||||
|
||||
Some(Self {
|
||||
receiver_idx,
|
||||
counter,
|
||||
ciphertext_offset: 13,
|
||||
})
|
||||
}
|
||||
|
||||
/// Get the ciphertext slice from the original packet.
|
||||
#[cfg(test)]
|
||||
pub fn ciphertext<'a>(&self, data: &'a [u8]) -> &'a [u8] {
|
||||
&data[self.ciphertext_offset..]
|
||||
}
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Msg1 Header
|
||||
// ============================================================================
|
||||
|
||||
/// Parsed Noise IK message 1 header.
|
||||
///
|
||||
/// Wire format:
|
||||
/// ```text
|
||||
/// [0x01][sender_idx:4 LE][noise_msg1:82]
|
||||
/// ```
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct Msg1Header {
|
||||
/// Session index chosen by the sender (becomes receiver_idx for responses).
|
||||
pub sender_idx: SessionIndex,
|
||||
/// Offset where Noise msg1 payload begins.
|
||||
pub noise_msg1_offset: usize,
|
||||
}
|
||||
|
||||
impl Msg1Header {
|
||||
/// Parse a msg1 header from packet data.
|
||||
///
|
||||
/// Returns None if the packet has wrong size or discriminator.
|
||||
pub fn parse(data: &[u8]) -> Option<Self> {
|
||||
if data.len() != MSG1_WIRE_SIZE {
|
||||
return None;
|
||||
}
|
||||
|
||||
if data[0] != DISCRIMINATOR_MSG1 {
|
||||
return None;
|
||||
}
|
||||
|
||||
let sender_idx = SessionIndex::from_le_bytes([data[1], data[2], data[3], data[4]]);
|
||||
|
||||
Some(Self {
|
||||
sender_idx,
|
||||
noise_msg1_offset: 5,
|
||||
})
|
||||
}
|
||||
|
||||
/// Get the Noise msg1 payload from the original packet.
|
||||
#[cfg(test)]
|
||||
pub fn noise_msg1<'a>(&self, data: &'a [u8]) -> &'a [u8] {
|
||||
&data[self.noise_msg1_offset..]
|
||||
}
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Msg2 Header
|
||||
// ============================================================================
|
||||
|
||||
/// Parsed Noise IK message 2 header.
|
||||
///
|
||||
/// Wire format:
|
||||
/// ```text
|
||||
/// [0x02][sender_idx:4 LE][receiver_idx:4 LE][noise_msg2:33]
|
||||
/// ```
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct Msg2Header {
|
||||
/// Session index chosen by the responder.
|
||||
pub sender_idx: SessionIndex,
|
||||
/// Echo of the initiator's sender_idx from msg1.
|
||||
pub receiver_idx: SessionIndex,
|
||||
/// Offset where Noise msg2 payload begins.
|
||||
pub noise_msg2_offset: usize,
|
||||
}
|
||||
|
||||
impl Msg2Header {
|
||||
/// Parse a msg2 header from packet data.
|
||||
///
|
||||
/// Returns None if the packet has wrong size or discriminator.
|
||||
pub fn parse(data: &[u8]) -> Option<Self> {
|
||||
if data.len() != MSG2_WIRE_SIZE {
|
||||
return None;
|
||||
}
|
||||
|
||||
if data[0] != DISCRIMINATOR_MSG2 {
|
||||
return None;
|
||||
}
|
||||
|
||||
let sender_idx = SessionIndex::from_le_bytes([data[1], data[2], data[3], data[4]]);
|
||||
let receiver_idx = SessionIndex::from_le_bytes([data[5], data[6], data[7], data[8]]);
|
||||
|
||||
Some(Self {
|
||||
sender_idx,
|
||||
receiver_idx,
|
||||
noise_msg2_offset: 9,
|
||||
})
|
||||
}
|
||||
|
||||
/// Get the Noise msg2 payload from the original packet.
|
||||
#[cfg(test)]
|
||||
pub fn noise_msg2<'a>(&self, data: &'a [u8]) -> &'a [u8] {
|
||||
&data[self.noise_msg2_offset..]
|
||||
}
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Serialization Helpers
|
||||
// ============================================================================
|
||||
|
||||
/// Build a wire-format msg1 packet.
|
||||
///
|
||||
/// Format: `[0x01][sender_idx:4 LE][noise_msg1:82]`
|
||||
pub fn build_msg1(sender_idx: SessionIndex, noise_msg1: &[u8]) -> Vec<u8> {
|
||||
debug_assert_eq!(noise_msg1.len(), HANDSHAKE_MSG1_SIZE);
|
||||
|
||||
let mut packet = Vec::with_capacity(MSG1_WIRE_SIZE);
|
||||
packet.push(DISCRIMINATOR_MSG1);
|
||||
packet.extend_from_slice(&sender_idx.to_le_bytes());
|
||||
packet.extend_from_slice(noise_msg1);
|
||||
packet
|
||||
}
|
||||
|
||||
/// Build a wire-format msg2 packet.
|
||||
///
|
||||
/// Format: `[0x02][sender_idx:4 LE][receiver_idx:4 LE][noise_msg2:33]`
|
||||
pub fn build_msg2(sender_idx: SessionIndex, receiver_idx: SessionIndex, noise_msg2: &[u8]) -> Vec<u8> {
|
||||
debug_assert_eq!(noise_msg2.len(), HANDSHAKE_MSG2_SIZE);
|
||||
|
||||
let mut packet = Vec::with_capacity(MSG2_WIRE_SIZE);
|
||||
packet.push(DISCRIMINATOR_MSG2);
|
||||
packet.extend_from_slice(&sender_idx.to_le_bytes());
|
||||
packet.extend_from_slice(&receiver_idx.to_le_bytes());
|
||||
packet.extend_from_slice(noise_msg2);
|
||||
packet
|
||||
}
|
||||
|
||||
/// Build a wire-format encrypted frame.
|
||||
///
|
||||
/// Format: `[0x00][receiver_idx:4 LE][counter:8 LE][ciphertext+tag]`
|
||||
pub fn build_encrypted(receiver_idx: SessionIndex, counter: u64, ciphertext: &[u8]) -> Vec<u8> {
|
||||
let mut packet = Vec::with_capacity(13 + ciphertext.len());
|
||||
packet.push(DISCRIMINATOR_ENCRYPTED);
|
||||
packet.extend_from_slice(&receiver_idx.to_le_bytes());
|
||||
packet.extend_from_slice(&counter.to_le_bytes());
|
||||
packet.extend_from_slice(ciphertext);
|
||||
packet
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Tests
|
||||
// ============================================================================
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn test_encrypted_header_parse() {
|
||||
// Build a valid encrypted frame
|
||||
let receiver_idx = SessionIndex::new(0x12345678);
|
||||
let counter = 42u64;
|
||||
let ciphertext = vec![0xaa; 32]; // 16 plaintext + 16 tag
|
||||
|
||||
let packet = build_encrypted(receiver_idx, counter, &ciphertext);
|
||||
|
||||
assert_eq!(packet.len(), 13 + 32);
|
||||
assert_eq!(packet[0], DISCRIMINATOR_ENCRYPTED);
|
||||
|
||||
// Parse it back
|
||||
let header = EncryptedHeader::parse(&packet).expect("should parse");
|
||||
assert_eq!(header.receiver_idx, receiver_idx);
|
||||
assert_eq!(header.counter, 42);
|
||||
assert_eq!(header.ciphertext_offset, 13);
|
||||
assert_eq!(header.ciphertext(&packet), &ciphertext[..]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_encrypted_header_too_short() {
|
||||
let packet = vec![0x00; 28]; // One byte too short
|
||||
assert!(EncryptedHeader::parse(&packet).is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_encrypted_header_wrong_discriminator() {
|
||||
let mut packet = vec![0x00; 30];
|
||||
packet[0] = 0x01; // Wrong discriminator
|
||||
assert!(EncryptedHeader::parse(&packet).is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_msg1_header_parse() {
|
||||
let sender_idx = SessionIndex::new(0xABCDEF01);
|
||||
let noise_msg1 = vec![0xbb; HANDSHAKE_MSG1_SIZE];
|
||||
|
||||
let packet = build_msg1(sender_idx, &noise_msg1);
|
||||
|
||||
assert_eq!(packet.len(), MSG1_WIRE_SIZE);
|
||||
assert_eq!(packet[0], DISCRIMINATOR_MSG1);
|
||||
|
||||
let header = Msg1Header::parse(&packet).expect("should parse");
|
||||
assert_eq!(header.sender_idx, sender_idx);
|
||||
assert_eq!(header.noise_msg1_offset, 5);
|
||||
assert_eq!(header.noise_msg1(&packet), &noise_msg1[..]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_msg1_header_wrong_size() {
|
||||
let packet = vec![0x01; 86]; // One byte too short
|
||||
assert!(Msg1Header::parse(&packet).is_none());
|
||||
|
||||
let packet = vec![0x01; 88]; // One byte too long
|
||||
assert!(Msg1Header::parse(&packet).is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_msg1_header_wrong_discriminator() {
|
||||
let mut packet = vec![0x00; MSG1_WIRE_SIZE];
|
||||
packet[0] = 0x02; // Wrong discriminator
|
||||
assert!(Msg1Header::parse(&packet).is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_msg2_header_parse() {
|
||||
let sender_idx = SessionIndex::new(0x11223344);
|
||||
let receiver_idx = SessionIndex::new(0x55667788);
|
||||
let noise_msg2 = vec![0xcc; HANDSHAKE_MSG2_SIZE];
|
||||
|
||||
let packet = build_msg2(sender_idx, receiver_idx, &noise_msg2);
|
||||
|
||||
assert_eq!(packet.len(), MSG2_WIRE_SIZE);
|
||||
assert_eq!(packet[0], DISCRIMINATOR_MSG2);
|
||||
|
||||
let header = Msg2Header::parse(&packet).expect("should parse");
|
||||
assert_eq!(header.sender_idx, sender_idx);
|
||||
assert_eq!(header.receiver_idx, receiver_idx);
|
||||
assert_eq!(header.noise_msg2_offset, 9);
|
||||
assert_eq!(header.noise_msg2(&packet), &noise_msg2[..]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_msg2_header_wrong_size() {
|
||||
let packet = vec![0x02; 41]; // One byte too short
|
||||
assert!(Msg2Header::parse(&packet).is_none());
|
||||
|
||||
let packet = vec![0x02; 43]; // One byte too long
|
||||
assert!(Msg2Header::parse(&packet).is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_msg2_header_wrong_discriminator() {
|
||||
let mut packet = vec![0x00; MSG2_WIRE_SIZE];
|
||||
packet[0] = 0x00; // Wrong discriminator
|
||||
assert!(Msg2Header::parse(&packet).is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_wire_sizes() {
|
||||
// Verify constants match spec
|
||||
assert_eq!(MSG1_WIRE_SIZE, 87); // 1 + 4 + 82
|
||||
assert_eq!(MSG2_WIRE_SIZE, 42); // 1 + 4 + 4 + 33
|
||||
assert_eq!(ENCRYPTED_MIN_SIZE, 29); // 1 + 4 + 8 + 16
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_roundtrip_indices() {
|
||||
// Test that indices survive the roundtrip correctly (endianness)
|
||||
let idx = SessionIndex::new(0xDEADBEEF);
|
||||
|
||||
let msg1 = build_msg1(idx, &[0u8; HANDSHAKE_MSG1_SIZE]);
|
||||
let parsed = Msg1Header::parse(&msg1).unwrap();
|
||||
assert_eq!(parsed.sender_idx.as_u32(), 0xDEADBEEF);
|
||||
|
||||
// Verify little-endian encoding
|
||||
assert_eq!(msg1[1], 0xEF);
|
||||
assert_eq!(msg1[2], 0xBE);
|
||||
assert_eq!(msg1[3], 0xAD);
|
||||
assert_eq!(msg1[4], 0xDE);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user