use super::*;
#[tokio::test]
async fn test_session_direct_peer_handshake() {
let edges = vec![(0, 1)];
let mut nodes = run_tree_test(2, &edges, false).await;
verify_tree_convergence(&nodes);
populate_all_coord_caches(&mut nodes);
let node0_addr = *nodes[0].node.node_addr();
let node1_addr = *nodes[1].node.node_addr();
let node1_pubkey = nodes[1].node.identity().pubkey_full();
nodes[0]
.node
.initiate_session(node1_addr, node1_pubkey)
.await
.expect("initiate_session failed");
assert_eq!(nodes[0].node.session_count(), 1);
assert!(
nodes[0]
.node
.get_session(&node1_addr)
.unwrap()
.state()
.is_initiating()
);
tokio::time::sleep(Duration::from_millis(20)).await;
let count = process_available_packets(&mut nodes).await;
assert!(count > 0, "Expected SessionSetup packet to arrive");
assert_eq!(nodes[1].node.session_count(), 1);
assert!(
nodes[1]
.node
.get_session(&node0_addr)
.unwrap()
.state()
.is_awaiting_msg3()
);
tokio::time::sleep(Duration::from_millis(20)).await;
let count = process_available_packets(&mut nodes).await;
assert!(count > 0, "Expected SessionAck packet to arrive");
assert!(
nodes[0]
.node
.get_session(&node1_addr)
.unwrap()
.state()
.is_established()
);
tokio::time::sleep(Duration::from_millis(20)).await;
let count = process_available_packets(&mut nodes).await;
assert!(count > 0, "Expected SessionMsg3 packet to arrive");
assert!(
nodes[1]
.node
.get_session(&node0_addr)
.unwrap()
.state()
.is_established()
);
cleanup_nodes(&mut nodes).await;
}
#[tokio::test]
async fn test_session_direct_peer_data_transfer() {
let edges = vec![(0, 1)];
let mut nodes = run_tree_test(2, &edges, false).await;
verify_tree_convergence(&nodes);
populate_all_coord_caches(&mut nodes);
let node0_addr = *nodes[0].node.node_addr();
let node1_addr = *nodes[1].node.node_addr();
let node1_pubkey = nodes[1].node.identity().pubkey_full();
nodes[0]
.node
.initiate_session(node1_addr, node1_pubkey)
.await
.unwrap();
tokio::time::sleep(Duration::from_millis(20)).await;
process_available_packets(&mut nodes).await; tokio::time::sleep(Duration::from_millis(20)).await;
process_available_packets(&mut nodes).await; tokio::time::sleep(Duration::from_millis(20)).await;
process_available_packets(&mut nodes).await;
assert!(
nodes[0]
.node
.get_session(&node1_addr)
.unwrap()
.state()
.is_established()
);
assert!(
nodes[1]
.node
.get_session(&node0_addr)
.unwrap()
.state()
.is_established()
);
let test_data = b"Hello, FIPS session!";
nodes[0]
.node
.send_session_data(&node1_addr, 0, 0, test_data)
.await
.expect("send_session_data failed");
tokio::time::sleep(Duration::from_millis(20)).await;
let count = process_available_packets(&mut nodes).await;
assert!(count > 0, "Expected encrypted data to arrive");
cleanup_nodes(&mut nodes).await;
}
#[tokio::test]
async fn test_endpoint_data_flushes_after_session_establishment() {
let edges = vec![(0, 1)];
let mut nodes = run_tree_test(2, &edges, false).await;
verify_tree_convergence(&nodes);
populate_all_coord_caches(&mut nodes);
let mut node0_endpoint = nodes[0]
.node
.attach_endpoint_data_io(8)
.expect("node 0 endpoint data I/O should attach");
let mut node1_endpoint = nodes[1]
.node
.attach_endpoint_data_io(8)
.expect("node 1 endpoint data I/O should attach");
let node0_addr = *nodes[0].node.node_addr();
let node1_addr = *nodes[1].node.node_addr();
let node0_identity = PeerIdentity::from_pubkey_full(nodes[0].node.identity().pubkey_full());
let node1_identity = PeerIdentity::from_pubkey_full(nodes[1].node.identity().pubkey_full());
nodes[0]
.node
.send_endpoint_data(node1_identity, b"ping".to_vec())
.await
.expect("endpoint data should queue behind session establishment");
for _ in 0..10 {
tokio::time::sleep(Duration::from_millis(20)).await;
process_available_packets(&mut nodes).await;
}
let event = tokio::time::timeout(Duration::from_secs(1), node1_endpoint.event_rx.recv())
.await
.expect("endpoint data event should not time out")
.expect("endpoint data event should arrive");
match event {
NodeEndpointEvent::Data {
source_peer,
payload,
..
} => {
assert_eq!(*source_peer.node_addr(), node0_addr);
assert_eq!(source_peer.npub(), nodes[0].node.npub());
assert_eq!(payload, b"ping");
}
NodeEndpointEvent::DataBatch { .. } => panic!("expected single endpoint data event"),
}
nodes[1]
.node
.send_endpoint_data(node0_identity, b"pong".to_vec())
.await
.expect("reply data should send");
tokio::time::sleep(Duration::from_millis(20)).await;
process_available_packets(&mut nodes).await;
let event = tokio::time::timeout(Duration::from_secs(1), node0_endpoint.event_rx.recv())
.await
.expect("endpoint data event should not time out")
.expect("endpoint data event should arrive");
match event {
NodeEndpointEvent::Data {
source_peer,
payload,
..
} => {
assert_eq!(*source_peer.node_addr(), node1_addr);
assert_eq!(source_peer.npub(), nodes[1].node.npub());
assert_eq!(payload, b"pong");
}
NodeEndpointEvent::DataBatch { .. } => panic!("expected single endpoint data event"),
}
cleanup_nodes(&mut nodes).await;
}
#[tokio::test]
async fn test_endpoint_data_routes_through_non_endpoint_transit_node() {
let edges = vec![(0, 1), (1, 2)];
let mut nodes = run_tree_test(3, &edges, false).await;
verify_tree_convergence(&nodes);
populate_all_coord_caches(&mut nodes);
let mut alice_endpoint = nodes[0]
.node
.attach_endpoint_data_io(8)
.expect("alice endpoint data I/O should attach");
let mut transit_endpoint = nodes[1]
.node
.attach_endpoint_data_io(8)
.expect("transit endpoint data I/O should attach");
let mut bob_endpoint = nodes[2]
.node
.attach_endpoint_data_io(8)
.expect("bob endpoint data I/O should attach");
let alice_addr = *nodes[0].node.node_addr();
let bob_addr = *nodes[2].node.node_addr();
let alice_identity = PeerIdentity::from_pubkey_full(nodes[0].node.identity().pubkey_full());
let bob_identity = PeerIdentity::from_pubkey_full(nodes[2].node.identity().pubkey_full());
nodes[0]
.node
.send_endpoint_data(bob_identity, b"alice-to-bob".to_vec())
.await
.expect("alice endpoint data should send");
drain_to_quiescence(&mut nodes).await;
let event = tokio::time::timeout(Duration::from_secs(1), bob_endpoint.event_rx.recv())
.await
.expect("bob endpoint data should not time out")
.expect("bob endpoint data should arrive");
match event {
NodeEndpointEvent::Data {
source_peer,
payload,
..
} => {
assert_eq!(*source_peer.node_addr(), alice_addr);
assert_eq!(source_peer.npub(), nodes[0].node.npub());
assert_eq!(payload, b"alice-to-bob");
}
NodeEndpointEvent::DataBatch { .. } => panic!("expected single endpoint data event"),
}
assert!(
nodes[1].node.get_session(&alice_addr).is_none(),
"transit node must not create an app endpoint session for Alice"
);
assert!(
nodes[1].node.get_session(&bob_addr).is_none(),
"transit node must not create an app endpoint session for Bob"
);
assert!(
transit_endpoint.event_rx.try_recv().is_err(),
"transit node must not receive app endpoint data"
);
nodes[2]
.node
.send_endpoint_data(alice_identity, b"bob-to-alice".to_vec())
.await
.expect("bob endpoint data should send");
drain_to_quiescence(&mut nodes).await;
let event = tokio::time::timeout(Duration::from_secs(1), alice_endpoint.event_rx.recv())
.await
.expect("alice endpoint data should not time out")
.expect("alice endpoint data should arrive");
match event {
NodeEndpointEvent::Data {
source_peer,
payload,
..
} => {
assert_eq!(*source_peer.node_addr(), bob_addr);
assert_eq!(source_peer.npub(), nodes[2].node.npub());
assert_eq!(payload, b"bob-to-alice");
}
NodeEndpointEvent::DataBatch { .. } => panic!("expected single endpoint data event"),
}
assert!(
transit_endpoint.event_rx.try_recv().is_err(),
"transit node must stay outside the app endpoint flow"
);
cleanup_nodes(&mut nodes).await;
}
#[tokio::test]
async fn test_endpoint_data_reply_learned_first_contact_routes_via_intermediary() {
let edges = vec![(0, 1), (1, 2)];
let mut nodes = run_tree_test(3, &edges, false).await;
verify_tree_convergence(&nodes);
for node in &mut nodes {
node.node.config.node.routing.mode = RoutingMode::ReplyLearned;
}
let mut transit_endpoint = nodes[1]
.node
.attach_endpoint_data_io(8)
.expect("transit endpoint data I/O should attach");
let mut bob_endpoint = nodes[2]
.node
.attach_endpoint_data_io(8)
.expect("bob endpoint data I/O should attach");
let alice_addr = *nodes[0].node.node_addr();
let bob_addr = *nodes[2].node.node_addr();
let bob_identity = PeerIdentity::from_pubkey_full(nodes[2].node.identity().pubkey_full());
nodes[0]
.node
.send_endpoint_data(bob_identity, b"first-contact".to_vec())
.await
.expect("alice endpoint data should queue and trigger discovery");
for _ in 0..120 {
drain_to_quiescence(&mut nodes).await;
if let Ok(event) = bob_endpoint.event_rx.try_recv() {
match event {
NodeEndpointEvent::Data {
source_peer,
payload,
..
} => {
assert_eq!(*source_peer.node_addr(), alice_addr);
assert_eq!(source_peer.npub(), nodes[0].node.npub());
assert_eq!(payload, b"first-contact");
}
NodeEndpointEvent::DataBatch { .. } => {
panic!("expected single endpoint data event")
}
}
assert!(
nodes[1].node.get_session(&alice_addr).is_none(),
"transit node must not create an app endpoint session for Alice"
);
assert!(
nodes[1].node.get_session(&bob_addr).is_none(),
"transit node must not create an app endpoint session for Bob"
);
assert!(
transit_endpoint.event_rx.try_recv().is_err(),
"transit node must not receive app endpoint data"
);
cleanup_nodes(&mut nodes).await;
return;
}
tokio::time::sleep(Duration::from_millis(25)).await;
}
cleanup_nodes(&mut nodes).await;
panic!("reply-learned first-contact endpoint data did not reach Bob");
}