struct ExpectedFmpReceipt<'a> {
source_addr: NodeAddr,
transport_id: TransportId,
remote_addr: &'a TransportAddr,
packet_timestamp_ms: u64,
packet_len: usize,
fmp_counter: u64,
inner_timestamp_ms: u32,
fmp_flags: u8,
}
fn assert_fmp_receipt(receipt: &DataplaneFmpIngressReceipt, expected: ExpectedFmpReceipt<'_>) {
let ExpectedFmpReceipt {
source_addr,
transport_id,
remote_addr,
packet_timestamp_ms,
packet_len,
fmp_counter,
inner_timestamp_ms,
fmp_flags,
} = expected;
assert_eq!(receipt.source_addr(), &source_addr);
assert_eq!(receipt.transport_id(), transport_id);
assert_eq!(receipt.remote_addr(), remote_addr);
assert_eq!(receipt.packet_timestamp_ms(), packet_timestamp_ms);
assert_eq!(receipt.packet_len(), packet_len);
assert_eq!(receipt.fmp_counter(), fmp_counter);
assert_eq!(receipt.inner_timestamp_ms(), inner_timestamp_ms);
assert_eq!(receipt.fmp_flags(), fmp_flags);
}
fn register_keyed_owner(
driver: &mut DataplaneTurnDriver,
owner: OwnerId,
config: OwnerConfig,
key: u8,
) {
driver.register_owner(owner, config);
driver
.owner_mut(owner)
.unwrap()
.set_crypto_keys(OwnerCryptoKeys::new(test_key(key), test_key(key)));
}
fn register_fmp_session_ingress_route(
routes: &mut DataplaneLiveRouteTable,
transport_id: TransportId,
receiver_idx: u32,
owner: OwnerId,
local_addr: NodeAddr,
) {
routes.register_fmp(
transport_id,
receiver_idx,
DataplaneIngressRoute::new(
owner,
1,
OutputTarget::SessionIngress { local_addr },
)
.with_class(PacketClass::Liveness),
);
}
fn register_fsp_session_payload_route(
routes: &mut DataplaneLiveRouteTable,
source_addr: NodeAddr,
owner: OwnerId,
local_addr: NodeAddr,
) {
routes.register_fsp(
source_addr,
DataplaneIngressRoute::new(
owner,
1,
OutputTarget::SessionPayload { local_addr },
)
.with_class(PacketClass::Liveness),
);
}
async fn pump_one_fmp_session_ingress_turn(
driver: &mut DataplaneTurnDriver,
routes: &mut DataplaneLiveRouteTable,
transport_id: TransportId,
remote_addr: TransportAddr,
wire: Vec<u8>,
timestamp_ms: u64,
) -> DataplaneLiveNodeTurn {
let mut raw_source = VecDeque::from([
DataplaneRawIngress::from_live_received(PacketProtocol::Fmp, ReceivedPacket::with_timestamp(
transport_id,
remote_addr,
crate::transport::PacketBuffer::new(wire),
timestamp_ms,
)),
]);
let (_endpoint_data_tx, mut endpoint_data_rx) = endpoint_data_batch_channel(1);
let (_tun_outbound_tx, mut tun_outbound_rx) = crate::upper::tun::tun_outbound_channel(1);
let (_tun_tx, tun_rx) = crate::upper::tun::write_channel();
let mut node = crate::Node::new(crate::Config::new()).expect("node");
let mut endpoint_io = node.attach_endpoint_data_io(8).expect("endpoint io");
let mut deferred_endpoint_data_batches = Vec::new();
let mut deferred_tun_packets = Vec::new();
let transports = HashMap::<TransportId, TransportHandle>::new();
let turn = pump_aead_live_node_route_table_turn(
driver,
AeadLiveRouteTableTurn {
raw_ingress: &mut raw_source,
routes,
raw_ingress_limit: 8,
endpoint_data_rx: &mut endpoint_data_rx,
endpoint_limit: 0,
tun_outbound_rx: &mut tun_outbound_rx,
tun_limit: 0,
deferred_endpoint_data_batches: &mut deferred_endpoint_data_batches,
deferred_tun_packets: &mut deferred_tun_packets,
endpoint_tx: &endpoint_io.event_tx,
transports: &transports,
crypto_limit: 8,
},
)
.await;
assert!(raw_source.is_empty());
assert!(deferred_endpoint_data_batches.is_empty());
assert!(deferred_tun_packets.is_empty());
assert!(tun_outbound_rx.try_recv().is_err());
assert!(tun_rx.try_recv_packet().is_err());
assert!(endpoint_io.event_rx.try_recv().is_err());
turn
}
#[tokio::test]
async fn live_node_session_ingress_reports_fmp_receipt_before_fast_fsp_delivery() {
let local_addr = NodeAddr::from_bytes([0xa1; 16]);
let source_identity = crate::Identity::generate();
let source_peer = PeerIdentity::from_pubkey_full(source_identity.pubkey_full());
let source_addr = *source_peer.node_addr();
let next_hop_peer =
PeerIdentity::from_pubkey_full(crate::Identity::generate().pubkey_full());
let next_hop = *next_hop_peer.node_addr();
let fmp_owner = OwnerId::fmp_node(next_hop);
let fsp_owner = OwnerId::fsp_node(source_addr);
let fmp_key = 0xa4;
let fsp_key = 0xa5;
let transport_id = TransportId::new(0xa6);
let remote_addr = TransportAddr::from_string("198.51.100.166:9000");
let fmp_timestamp = 166_001;
let fmp_inner_timestamp = 166_002_u32;
let fsp_inner_timestamp = 166_003_u32;
let fsp_inner_flags = 0x05;
let fmp_counter = 166;
let fsp_counter = 167;
let fmp_flags = crate::node::wire::FLAG_CE | crate::node::wire::FLAG_SP;
let endpoint_payload = b"fast-endpoint";
let fsp_inner = crate::node::session_wire::fsp_prepend_inner_header(
fsp_inner_timestamp,
crate::protocol::SessionMessageType::EndpointData.to_byte(),
fsp_inner_flags,
endpoint_payload,
);
let fsp_wire = fsp_encrypted_wire(
fsp_counter,
crate::node::session_wire::FSP_FLAG_CP,
&fsp_inner,
fsp_key,
);
let datagram = crate::protocol::SessionDatagram::new(source_addr, local_addr, fsp_wire)
.with_ttl(8)
.with_path_mtu(1280)
.encode();
let mut fmp_plaintext = fmp_inner_timestamp.to_le_bytes().to_vec();
fmp_plaintext.extend_from_slice(&datagram);
let fmp_wire = fmp_encrypted_wire(0xa7, fmp_counter, fmp_flags, &fmp_plaintext, fmp_key);
let fmp_wire_len = fmp_wire.len();
let mut driver = DataplaneTurnDriver::new(AdmissionConfig::new(4, 8));
register_keyed_owner(
&mut driver,
fmp_owner,
OwnerConfig::new(1, 8).with_source_peer(next_hop_peer),
fmp_key,
);
register_keyed_owner(
&mut driver,
fsp_owner,
OwnerConfig::new(1, 8).with_source_peer(source_peer),
fsp_key,
);
let mut routes = DataplaneLiveRouteTable::default();
register_fmp_session_ingress_route(&mut routes, transport_id, 0xa7, fmp_owner, local_addr);
register_fsp_session_payload_route(&mut routes, source_addr, fsp_owner, local_addr);
let turn = pump_one_fmp_session_ingress_turn(
&mut driver,
&mut routes,
transport_id,
remote_addr.clone(),
fmp_wire,
fmp_timestamp,
)
.await;
assert_eq!(turn.summary().raw_ingress_dropped(), 0);
assert_eq!(turn.summary().inbound_admitted(), 2);
assert_eq!(turn.summary().outputs_dropped(), 0);
let mut receipt_turn = turn.clone();
let fmp_ingress_receipts = receipt_turn.take_fmp_ingress_receipts();
assert_eq!(fmp_ingress_receipts.len(), 1);
assert!(turn.fmp_link_ingress().is_empty());
assert_fmp_receipt(
&fmp_ingress_receipts[0],
ExpectedFmpReceipt {
source_addr: next_hop,
transport_id,
remote_addr: &remote_addr,
packet_timestamp_ms: fmp_timestamp,
packet_len: fmp_wire_len,
fmp_counter,
inner_timestamp_ms: fmp_inner_timestamp,
fmp_flags,
},
);
assert!(driver
.owner_fsp_activity(fsp_owner)
.unwrap()
.current_epoch_confirmed());
assert_eq!(turn.fsp_session_ingress_count(), 1);
let session_ingress = turn
.fsp_authenticated_ingress
.sessions
.first()
.expect("session ingress");
let (
session_source_addr,
_session_source_peer,
session_previous_hop,
session_ce_flag,
_session_activity_tick,
session_timestamp_ms,
session_msg_type,
session_inner_flags,
session_plaintext,
) = session_ingress.clone().into_parts();
assert_eq!(session_source_addr, source_addr);
assert_eq!(session_previous_hop, next_hop);
assert!(session_ce_flag);
assert_eq!(session_timestamp_ms, fsp_inner_timestamp);
assert_eq!(
session_msg_type,
crate::protocol::SessionMessageType::EndpointData.to_byte()
);
assert_eq!(session_inner_flags, fsp_inner_flags);
assert_eq!(session_plaintext.as_slice(), fsp_inner.as_slice());
assert_eq!(
&session_plaintext.as_slice()[crate::node::session_wire::FSP_INNER_HEADER_SIZE..],
endpoint_payload
);
}
#[tokio::test]
async fn live_node_session_ingress_reports_fsp_coord_warmup_before_fsp_delivery() {
let local_addr = NodeAddr::from_bytes([0xc1; 16]);
let source_identity = crate::Identity::generate();
let source_peer = PeerIdentity::from_pubkey_full(source_identity.pubkey_full());
let source_addr = *source_peer.node_addr();
let next_hop_peer =
PeerIdentity::from_pubkey_full(crate::Identity::generate().pubkey_full());
let next_hop = *next_hop_peer.node_addr();
let root_addr = NodeAddr::from_bytes([0xcf; 16]);
let source_coords =
crate::tree::TreeCoordinate::from_addrs(vec![source_addr, root_addr]).unwrap();
let local_coords =
crate::tree::TreeCoordinate::from_addrs(vec![local_addr, root_addr]).unwrap();
let mut coords_prefix = Vec::new();
crate::protocol::encode_coords(&source_coords, &mut coords_prefix);
crate::protocol::encode_coords(&local_coords, &mut coords_prefix);
let fmp_owner = OwnerId::fmp_node(next_hop);
let fsp_owner = OwnerId::fsp_node(source_addr);
let fmp_key = 0xc4;
let fsp_key = 0xc5;
let transport_id = TransportId::new(0xc6);
let remote_addr = TransportAddr::from_string("198.51.100.198:9000");
let fmp_timestamp = 198_001;
let fmp_inner_timestamp = 198_002_u32;
let fsp_inner_timestamp = 198_003_u32;
let endpoint_payload = b"coord-warm-endpoint";
let fsp_inner = crate::node::session_wire::fsp_prepend_inner_header(
fsp_inner_timestamp,
crate::protocol::SessionMessageType::EndpointData.to_byte(),
0,
endpoint_payload,
);
let fsp_wire = fsp_encrypted_wire_with_coords(
198,
crate::node::session_wire::FSP_FLAG_CP,
&fsp_inner,
fsp_key,
&coords_prefix,
);
let datagram = crate::protocol::SessionDatagram::new(source_addr, local_addr, fsp_wire)
.with_ttl(8)
.with_path_mtu(1280)
.encode();
let mut fmp_plaintext = fmp_inner_timestamp.to_le_bytes().to_vec();
fmp_plaintext.extend_from_slice(&datagram);
let fmp_wire = fmp_encrypted_wire(0xc7, 199, 0, &fmp_plaintext, fmp_key);
let mut driver = DataplaneTurnDriver::new(AdmissionConfig::new(4, 8));
register_keyed_owner(
&mut driver,
fmp_owner,
OwnerConfig::new(1, 8).with_source_peer(next_hop_peer),
fmp_key,
);
register_keyed_owner(
&mut driver,
fsp_owner,
OwnerConfig::new(1, 8).with_source_peer(source_peer),
fsp_key,
);
assert_eq!(driver.record_fsp_decrypt_failure(fsp_owner), Some(1));
let mut routes = DataplaneLiveRouteTable::default();
register_fmp_session_ingress_route(&mut routes, transport_id, 0xc7, fmp_owner, local_addr);
register_fsp_session_payload_route(&mut routes, source_addr, fsp_owner, local_addr);
let turn = pump_one_fmp_session_ingress_turn(
&mut driver,
&mut routes,
transport_id,
remote_addr,
fmp_wire,
fmp_timestamp,
)
.await;
assert_eq!(turn.summary().raw_ingress_dropped(), 0);
assert_eq!(turn.summary().inbound_admitted(), 2);
assert_eq!(turn.fsp_coord_warmups().len(), 1);
let mut coord_cache = crate::cache::CoordCache::new(8, 1_000);
turn.fsp_coord_warmups()[0]
.clone()
.apply_to(&mut coord_cache, 10);
assert_eq!(coord_cache.get(&source_addr, 10), Some(&source_coords));
assert_eq!(coord_cache.get(&local_addr, 10), Some(&local_coords));
assert_eq!(turn.fsp_session_ingress_count(), 1);
assert!(turn.fsp_local_session_ingress().is_empty());
assert!(turn.fmp_link_ingress().is_empty());
let activity = driver.owner_fsp_activity(fsp_owner).unwrap();
assert!(activity.current_epoch_confirmed());
assert_eq!(activity.last_rx_age_ms(fmp_timestamp), Some(0));
assert_eq!(
driver.record_fsp_decrypt_failure(fsp_owner),
Some(1),
"authenticated FSP output collection should reset the owner failure streak"
);
}
#[tokio::test]
async fn live_node_session_ingress_keeps_fsp_handshake_on_local_session_path() {
let local_addr = NodeAddr::from_bytes([0xac; 16]);
let source_addr = NodeAddr::from_bytes([0xad; 16]);
let next_hop_peer =
PeerIdentity::from_pubkey_full(crate::Identity::generate().pubkey_full());
let next_hop = *next_hop_peer.node_addr();
let fmp_owner = OwnerId::fmp_node(next_hop);
let fmp_key = 0xaf;
let transport_id = TransportId::new(0xb0);
let remote_addr = TransportAddr::from_string("198.51.100.176:9000");
let fmp_timestamp = 176_001;
let fmp_inner_timestamp = 176_002_u32;
let fmp_counter = 176;
let fmp_flags = crate::node::wire::FLAG_CE;
let path_mtu = 1240;
let mut fsp_handshake =
crate::node::session_wire::build_fsp_handshake_prefix(
crate::node::session_wire::FSP_PHASE_MSG1,
4,
)
.to_vec();
fsp_handshake.extend_from_slice(b"msg1");
let datagram =
crate::protocol::SessionDatagram::new(source_addr, local_addr, fsp_handshake.clone())
.with_ttl(8)
.with_path_mtu(path_mtu)
.encode();
let mut fmp_plaintext = fmp_inner_timestamp.to_le_bytes().to_vec();
fmp_plaintext.extend_from_slice(&datagram);
let fmp_wire = fmp_encrypted_wire(0xb1, fmp_counter, fmp_flags, &fmp_plaintext, fmp_key);
let fmp_wire_len = fmp_wire.len();
let mut driver = DataplaneTurnDriver::new(AdmissionConfig::new(4, 8));
register_keyed_owner(
&mut driver,
fmp_owner,
OwnerConfig::new(1, 8).with_source_peer(next_hop_peer),
fmp_key,
);
let mut routes = DataplaneLiveRouteTable::default();
register_fmp_session_ingress_route(&mut routes, transport_id, 0xb1, fmp_owner, local_addr);
let turn = pump_one_fmp_session_ingress_turn(
&mut driver,
&mut routes,
transport_id,
remote_addr.clone(),
fmp_wire,
fmp_timestamp,
)
.await;
assert_eq!(turn.summary().raw_ingress_dropped(), 0);
assert_eq!(turn.summary().inbound_admitted(), 1);
assert_eq!(turn.summary().outputs_dropped(), 0);
let mut receipt_turn = turn.clone();
let fmp_ingress_receipts = receipt_turn.take_fmp_ingress_receipts();
assert_eq!(fmp_ingress_receipts.len(), 1);
assert!(turn.fmp_link_ingress().is_empty());
assert_eq!(turn.fsp_session_ingress_count(), 0);
assert_eq!(turn.fsp_local_session_ingress().len(), 1);
assert_fmp_receipt(
&fmp_ingress_receipts[0],
ExpectedFmpReceipt {
source_addr: next_hop,
transport_id,
remote_addr: &remote_addr,
packet_timestamp_ms: fmp_timestamp,
packet_len: fmp_wire_len,
fmp_counter,
inner_timestamp_ms: fmp_inner_timestamp,
fmp_flags,
},
);
let (local_source, local_previous_hop, local_ce, local_path_mtu, local_payload) = turn
.fsp_local_session_ingress()[0]
.clone()
.into_parts();
assert_eq!(local_source, source_addr);
assert_eq!(local_previous_hop, next_hop);
assert!(local_ce);
assert_eq!(local_path_mtu, path_mtu);
assert_eq!(local_payload.as_slice(), fsp_handshake.as_slice());
assert!(turn.raw_ingress_drops().is_empty());
assert!(turn.output_drops().is_empty());
}
#[tokio::test]
async fn live_node_session_ingress_reports_non_local_fmp_link_message() {
let local_addr = NodeAddr::from_bytes([0xb1; 16]);
let next_hop_peer =
PeerIdentity::from_pubkey_full(crate::Identity::generate().pubkey_full());
let next_hop = *next_hop_peer.node_addr();
let fmp_owner = OwnerId::fmp_node(next_hop);
let fmp_key = 0xb3;
let transport_id = TransportId::new(0xb4);
let remote_addr = TransportAddr::from_string("198.51.100.180:9000");
let fmp_timestamp = 180_001;
let fmp_inner_timestamp = 180_002_u32;
let fmp_counter = 180;
let fmp_flags = crate::node::wire::FLAG_SP;
let mut fmp_plaintext = fmp_inner_timestamp.to_le_bytes().to_vec();
fmp_plaintext.push(crate::protocol::LinkMessageType::Heartbeat.to_byte());
let fmp_wire = fmp_encrypted_wire(0xb5, fmp_counter, fmp_flags, &fmp_plaintext, fmp_key);
let fmp_wire_len = fmp_wire.len();
let mut driver = DataplaneTurnDriver::new(AdmissionConfig::new(4, 8));
register_keyed_owner(
&mut driver,
fmp_owner,
OwnerConfig::new(1, 8).with_source_peer(next_hop_peer),
fmp_key,
);
let mut routes = DataplaneLiveRouteTable::default();
register_fmp_session_ingress_route(&mut routes, transport_id, 0xb5, fmp_owner, local_addr);
let turn = pump_one_fmp_session_ingress_turn(
&mut driver,
&mut routes,
transport_id,
remote_addr.clone(),
fmp_wire,
fmp_timestamp,
)
.await;
assert_eq!(turn.summary().raw_ingress_dropped(), 0);
assert_eq!(turn.summary().inbound_admitted(), 1);
assert_eq!(turn.summary().outputs(), 0);
assert_eq!(turn.summary().outputs_dropped(), 0);
let mut receipt_turn = turn.clone();
assert!(receipt_turn.take_fmp_ingress_receipts().is_empty());
assert_eq!(turn.fmp_link_ingress().len(), 1);
let ingress = &turn.fmp_link_ingress()[0];
assert_eq!(
ingress.msg_type(),
Some(crate::protocol::LinkMessageType::Heartbeat.to_byte())
);
assert_eq!(ingress.payload(), &[] as &[u8]);
assert_fmp_receipt(
ingress.receipt(),
ExpectedFmpReceipt {
source_addr: next_hop,
transport_id,
remote_addr: &remote_addr,
packet_timestamp_ms: fmp_timestamp,
packet_len: fmp_wire_len,
fmp_counter,
inner_timestamp_ms: fmp_inner_timestamp,
fmp_flags,
},
);
assert!(turn.output_drops().is_empty());
}