use super::*;
use crate::PeerIdentity;
use crate::transport::{ReceivedPacket, TransportAddr, TransportId};
use ring::aead::UnboundKey;
fn mover() -> Dataplane {
Dataplane::new(AdmissionConfig::new(4, 8))
}
fn test_aead_worker_pool(max_in_flight: usize) -> DataplaneAeadWorkerPool {
static RUNTIME: std::sync::OnceLock<tokio::runtime::Runtime> = std::sync::OnceLock::new();
let runtime = RUNTIME.get_or_init(|| {
tokio::runtime::Builder::new_multi_thread()
.worker_threads(2)
.enable_all()
.build()
.expect("test AEAD runtime")
});
let _guard = runtime.enter();
DataplaneAeadWorkerPool::new(max_in_flight)
}
fn endpoint_payloads(payloads: Vec<Vec<u8>>) -> Vec<EndpointDataPayload> {
payloads
.into_iter()
.map(|payload| {
EndpointDataPayload::from_packet_payload(payload)
.expect("test endpoint payload should fit FSP endpoint data")
})
.collect()
}
fn route_endpoint_payloads(
route: &DataplaneEndpointDataRoute,
payloads: Vec<Vec<u8>>,
) -> DataplaneEndpointDataBatchRoute {
route.route_payloads(
endpoint_payloads(payloads),
ActivityTick::new(crate::time::now_ms()),
)
}
#[derive(Clone, Debug, Default, Eq, PartialEq)]
struct DataplaneTurn {
dispatched: usize,
retired: Vec<PacketOutput>,
drops: Vec<PacketDrop>,
}
impl DataplaneTurn {
fn dispatched(&self) -> usize {
self.dispatched
}
fn retired(&self) -> &[PacketOutput] {
&self.retired
}
fn drops(&self) -> &[PacketDrop] {
&self.drops
}
fn outputs(&self) -> Vec<&PacketOutput> {
self.retired.iter().collect()
}
}
fn execute_test_prepared_crypto_work(work: PreparedCryptoWork) -> CryptoCompletion {
let mut pool = test_aead_worker_pool(1);
let mut prepared = vec![work];
pool.submit_prepared_chunk(&mut prepared);
let mut completions = drain_worker_pool_completions(&mut pool, 1);
assert_eq!(completions.len(), 1);
completions.pop().unwrap()
}
fn complete_test_open_work(work: CryptoWork, key: u8) -> CryptoCompletion {
execute_test_prepared_crypto_work(PreparedCryptoWork::open(work, test_key(key)))
}
fn complete_test_seal_work(work: OutboundCryptoWork, key: u8) -> CryptoCompletion {
execute_test_prepared_crypto_work(PreparedCryptoWork::seal(work, test_key(key)))
}
fn dispatch_available(mover: &mut Dataplane, limit: usize) -> Vec<CryptoWork> {
capture_prepared_work(mover, limit)
.into_iter()
.map(|prepared| match prepared {
PreparedCryptoWork::Open { work, .. } => work,
PreparedCryptoWork::Seal { work, .. } => {
panic!("unexpected outbound work while capturing inbound: {work:?}")
}
})
.collect()
}
fn dispatch_outbound_available(
mover: &mut Dataplane,
limit: usize,
) -> Vec<OutboundCryptoWork> {
capture_prepared_work(mover, limit)
.into_iter()
.map(|prepared| match prepared {
PreparedCryptoWork::Seal { work, .. } => work,
PreparedCryptoWork::Open { work, .. } => {
panic!("unexpected inbound work while capturing outbound: {work:?}")
}
})
.collect()
}
fn capture_prepared_work(mover: &mut Dataplane, limit: usize) -> Vec<PreparedCryptoWork> {
seed_missing_test_owner_keys(mover);
let mut prepared_work = Vec::new();
let mut completion_batches = Vec::new();
let pool = test_aead_worker_pool(
limit
.saturating_add(DATAPLANE_AEAD_WORKER_FAIRNESS_PACKETS)
.max(1),
);
mover.prepare_aead_available_into(
limit,
&mut prepared_work,
&mut completion_batches,
&pool,
);
assert!(completion_batches.is_empty());
prepared_work
}
fn seed_missing_test_owner_keys(mover: &mut Dataplane) {
let key = test_key(0);
for shard in &mut mover.shards {
for owner in shard.owners.values_mut() {
if owner.crypto_keys.is_none() {
owner.set_crypto_keys(OwnerCryptoKeys::new(key.clone(), key.clone()));
}
}
}
}
fn drain_worker_pool_completions(
pool: &mut DataplaneAeadWorkerPool,
expected: usize,
) -> Vec<CryptoCompletion> {
let mut completions = Vec::new();
let mut batches = Vec::new();
for _ in 0..100 {
pool.drain_completion_batches_into_sink(
expected.saturating_sub(completions.len()),
&mut batches,
);
for batch in batches.drain(..) {
completions.extend(batch.into_completions());
}
if completions.len() >= expected {
break;
}
std::thread::sleep(std::time::Duration::from_millis(1));
}
completions
}
fn drain_worker_pool_completion_batches(
pool: &mut DataplaneAeadWorkerPool,
expected: usize,
) -> Vec<CryptoCompletionBatch> {
let mut batches = Vec::new();
for _ in 0..100 {
let drained = batches.iter().map(CryptoCompletionBatch::len).sum::<usize>();
pool.drain_completion_batches_into_sink(
expected.saturating_sub(drained),
&mut batches,
);
if batches.iter().map(CryptoCompletionBatch::len).sum::<usize>() >= expected {
break;
}
std::thread::sleep(std::time::Duration::from_millis(1));
}
batches
}
fn run_aead_available(mover: &mut Dataplane, limit: usize) -> DataplaneTurn {
let mut prepared_work = Vec::new();
let mut completion_batches = Vec::new();
let mut retired = Vec::new();
let mut outbound_packets = Vec::new();
let mut fsp_authenticated_ingress = DataplaneFspAuthenticatedIngress::default();
let mut drops = Vec::new();
let mut pool = test_aead_worker_pool(
limit
.saturating_add(DATAPLANE_AEAD_WORKER_FAIRNESS_PACKETS)
.max(1),
);
let capacity_before = pool.available_capacity();
let dispatched = mover.run_aead_available_into(
limit,
DataplaneAeadRunBuffers::new(
&mut prepared_work,
&mut completion_batches,
&mut retired,
&mut outbound_packets,
&mut fsp_authenticated_ingress,
&mut drops,
),
&mut pool,
false,
);
let worker_dispatched = capacity_before.saturating_sub(pool.available_capacity());
let mut worker_batches =
drain_worker_pool_completion_batches(&mut pool, worker_dispatched);
assert_eq!(
worker_batches
.iter()
.map(CryptoCompletionBatch::len)
.sum::<usize>(),
worker_dispatched
);
mover.queue_completion_batches(&mut worker_batches);
mover.retire_queued_completions_into(
limit,
&mut DataplaneRetiredOutputSink::new(
&mut retired,
&mut outbound_packets,
&mut fsp_authenticated_ingress,
),
false,
);
drops.append(&mut mover.drain_drops());
assert!(outbound_packets.is_empty());
assert!(fsp_authenticated_ingress.is_empty());
DataplaneTurn {
dispatched,
retired,
drops,
}
}
fn drain_worker_pool_into_driver(
driver: &mut DataplaneTurnDriver,
mut summary: DataplaneRuntimeSummary,
pool: &mut DataplaneAeadWorkerPool,
expected: usize,
compact_endpoint_data: bool,
) -> DataplaneRuntimeSummary {
let mut drained = 0usize;
for _ in 0..100 {
drained = drained.saturating_add(
pool.drain_completion_batches_into_sink(
expected.saturating_sub(drained),
&mut driver.mover,
),
);
if drained >= expected {
break;
}
std::thread::sleep(std::time::Duration::from_millis(1));
}
assert_eq!(drained, expected);
summary.completions = summary.completions.saturating_add(drained);
driver.retire_queued_completed_aead_outputs(expected, compact_endpoint_data);
driver.admit_retired_outbound_packets(summary)
}
fn collect_test_aead_outputs(
driver: &mut DataplaneTurnDriver,
mut summary: DataplaneRuntimeSummary,
limit: usize,
compact_endpoint_data: bool,
) -> DataplaneRuntimeSummary {
let mut pool = test_aead_worker_pool(
limit
.saturating_add(DATAPLANE_AEAD_WORKER_FAIRNESS_PACKETS)
.max(1),
);
let mut remaining = limit;
loop {
let dispatched_before = summary.dispatched;
let capacity_before = pool.available_capacity();
summary = driver.collect_aead_outputs(
summary,
remaining,
&mut pool,
compact_endpoint_data,
);
let dispatched = summary.dispatched.saturating_sub(dispatched_before);
let worker_dispatched = capacity_before.saturating_sub(pool.available_capacity());
remaining = remaining.saturating_sub(dispatched);
if dispatched == 0 {
break;
}
summary = drain_worker_pool_into_driver(
driver,
summary,
&mut pool,
worker_dispatched,
compact_endpoint_data,
);
}
summary.outputs = driver.outputs.len();
summary.drops = driver.drops.len();
summary
}
fn collect_test_live_session_outputs<R>(
driver: &mut DataplaneTurnDriver,
mut summary: DataplaneRuntimeSummary,
router: &mut R,
limit: usize,
compact_endpoint_data: bool,
deferred_raw_ingress: &mut std::collections::VecDeque<(DataplaneRawIngress, u8)>,
) -> DataplaneRuntimeSummary
where
R: DataplaneIngressRouter,
{
let mut pool = test_aead_worker_pool(
limit
.saturating_add(DATAPLANE_AEAD_WORKER_FAIRNESS_PACKETS)
.max(1),
);
let mut remaining = limit;
loop {
let dispatched_before = summary.dispatched;
let capacity_before = pool.available_capacity();
summary = driver.collect_live_session_outputs(
summary,
router,
remaining,
&mut pool,
compact_endpoint_data,
deferred_raw_ingress,
);
let dispatched = summary.dispatched.saturating_sub(dispatched_before);
let worker_dispatched = capacity_before.saturating_sub(pool.available_capacity());
remaining = remaining.saturating_sub(dispatched);
if dispatched == 0 {
break;
}
summary = drain_worker_pool_into_driver(
driver,
summary,
&mut pool,
worker_dispatched,
compact_endpoint_data,
);
}
summary.outputs = driver.outputs.len();
summary.drops = driver.drops.len();
summary
}
fn run_aead_completion_turn<I>(
driver: &mut DataplaneTurnDriver,
completions: I,
limit: usize,
) -> DataplaneRuntimeTurn<'_>
where
I: IntoIterator<Item = CryptoCompletion>,
{
driver.reset_turn_buffers();
let mut completion_work = completions.into_iter().collect::<Vec<_>>();
let queued = completion_work.len();
driver.completion_batches.clear();
CryptoCompletionBatch::drain_completion_vec_into_batches(
&mut completion_work,
&mut driver.completion_batches,
);
let mut summary = DataplaneRuntimeSummary::default();
summary.completions = summary.completions.saturating_add(queued);
driver
.mover
.queue_completion_batches(&mut driver.completion_batches);
driver.retire_queued_completed_aead_outputs(queued, false);
let summary = driver.admit_retired_outbound_packets(summary);
let summary = collect_test_aead_outputs(driver, summary, limit, false);
DataplaneRuntimeTurn {
summary,
raw_ingress_drops: &driver.raw_ingress_drops,
output_drops: &driver.output_drops,
outputs: &driver.outputs,
drops: &driver.drops,
}
}
async fn wait_for_live_worker_completion(live_node: &DataplaneLiveNode) {
let notify = live_node.completion_notify();
tokio::time::timeout(std::time::Duration::from_secs(1), notify.notified())
.await
.expect("live dataplane worker completion");
}
fn run_aead_classified_turn<I, O>(
driver: &mut DataplaneTurnDriver,
inbound: I,
outbound: O,
limit: usize,
) -> DataplaneRuntimeTurn<'_>
where
I: IntoIterator<Item = SocketPacket>,
O: IntoIterator<Item = OutboundPacket>,
{
driver.reset_turn_buffers();
let mut summary = DataplaneRuntimeSummary::default();
for packet in inbound {
driver.admit_socket_packet(packet, &mut summary);
}
for packet in outbound {
driver.admit_outbound_packet(packet, &mut summary);
}
finish_test_aead_turn(driver, summary, limit)
}
fn run_aead_classified_output_turn<'a, I, O, S>(
driver: &'a mut DataplaneTurnDriver,
inbound: I,
outbound: O,
sink: &mut S,
limit: usize,
) -> DataplaneRuntimeTurn<'a>
where
I: IntoIterator<Item = SocketPacket>,
O: IntoIterator<Item = OutboundPacket>,
S: DataplaneOutputSink,
{
driver.reset_turn_buffers();
let mut summary = DataplaneRuntimeSummary::default();
for packet in inbound {
driver.admit_socket_packet(packet, &mut summary);
}
for packet in outbound {
driver.admit_outbound_packet(packet, &mut summary);
}
finish_test_aead_output_turn(driver, summary, sink, limit)
}
fn admit_test_raw_ingress_packet<R>(
driver: &mut DataplaneTurnDriver,
packet: DataplaneRawIngress,
router: &mut R,
summary: &mut DataplaneRuntimeSummary,
) where
R: DataplaneIngressRouter,
{
let mut deferred_raw_ingress = std::collections::VecDeque::new();
let Some(socket_packet) = DataplaneTurnDriver::raw_ingress_socket_packet(
packet,
router,
summary,
&mut driver.raw_ingress_drops,
&mut deferred_raw_ingress,
0,
) else {
return;
};
driver.admit_socket_packet(socket_packet, summary);
}
fn run_aead_raw_ingress_turn<'a, I, O, R>(
driver: &'a mut DataplaneTurnDriver,
inbound: I,
router: &mut R,
outbound: O,
limit: usize,
) -> DataplaneRuntimeTurn<'a>
where
I: IntoIterator<Item = DataplaneRawIngress>,
O: IntoIterator<Item = OutboundPacket>,
R: DataplaneIngressRouter,
{
driver.reset_turn_buffers();
let mut summary = DataplaneRuntimeSummary::default();
for packet in inbound {
admit_test_raw_ingress_packet(driver, packet, router, &mut summary);
}
for packet in outbound {
driver.admit_outbound_packet(packet, &mut summary);
}
finish_test_aead_turn(driver, summary, limit)
}
struct AeadOutputCompletionTurn<'a, RI, R, S> {
completions: &'a mut VecDeque<CryptoCompletion>,
completion_limit: usize,
raw_ingress: &'a mut RI,
router: &'a mut R,
raw_ingress_limit: usize,
outbound: &'a mut VecDeque<OutboundPacket>,
outbound_limit: usize,
sink: &'a mut S,
crypto_limit: usize,
}
fn pump_aead_output_completion_turn<'a, RI, R, S>(
driver: &'a mut DataplaneTurnDriver,
request: AeadOutputCompletionTurn<'_, RI, R, S>,
) -> DataplaneRuntimeTurn<'a>
where
RI: DataplaneRawIngressSource,
R: DataplaneIngressRouter,
S: DataplaneOutputSink,
{
let AeadOutputCompletionTurn {
completions,
completion_limit,
raw_ingress,
router,
raw_ingress_limit,
outbound,
outbound_limit,
sink,
crypto_limit,
} = request;
driver.reset_turn_buffers();
let mut summary = DataplaneRuntimeSummary::default();
let queued = drain_test_completions_into_mover(driver, completions, completion_limit);
summary.completions = summary.completions.saturating_add(queued);
driver.retire_queued_completed_aead_outputs(completion_limit, false);
summary = driver.admit_retired_outbound_packets(summary);
raw_ingress.drain_raw_ingress(raw_ingress_limit, |packet| {
admit_test_raw_ingress_packet(driver, packet, router, &mut summary);
});
let outbound_count = outbound_limit.min(outbound.len());
for packet in outbound.drain(..outbound_count) {
driver.admit_outbound_packet(packet, &mut summary);
}
summary = collect_test_aead_outputs(driver, summary, crypto_limit, false);
driver.send_collected_outputs(summary, sink)
}
struct AeadLiveRouteTableTurn<'a, RI> {
raw_ingress: &'a mut RI,
routes: &'a mut DataplaneLiveRouteTable,
raw_ingress_limit: usize,
endpoint_data_rx: &'a mut EndpointDataBatchRx,
endpoint_limit: usize,
tun_outbound_rx: &'a mut TunOutboundRx,
tun_limit: usize,
deferred_endpoint_data_batches: &'a mut Vec<NodeEndpointDataBatch>,
deferred_tun_packets: &'a mut Vec<Vec<u8>>,
endpoint_tx: &'a EndpointEventSender,
transports: &'a HashMap<TransportId, TransportHandle>,
crypto_limit: usize,
}
async fn pump_aead_live_node_route_table_turn<RI>(
driver: &mut DataplaneTurnDriver,
request: AeadLiveRouteTableTurn<'_, RI>,
) -> DataplaneLiveNodeTurn
where
RI: DataplaneRawIngressSource,
{
let mut completions = VecDeque::<CryptoCompletion>::new();
pump_aead_live_node_route_table_turn_with_completions(
driver,
AeadLiveRouteTableCompletionTurn {
completions: &mut completions,
completion_limit: 0,
route: request,
},
)
.await
}
struct AeadLiveRouteTableCompletionTurn<'a, RI> {
completions: &'a mut VecDeque<CryptoCompletion>,
completion_limit: usize,
route: AeadLiveRouteTableTurn<'a, RI>,
}
async fn pump_aead_live_node_route_table_turn_with_completions<RI>(
driver: &mut DataplaneTurnDriver,
request: AeadLiveRouteTableCompletionTurn<'_, RI>,
) -> DataplaneLiveNodeTurn
where
RI: DataplaneRawIngressSource,
{
let AeadLiveRouteTableCompletionTurn {
completions,
completion_limit,
route,
} = request;
let AeadLiveRouteTableTurn {
raw_ingress,
routes,
raw_ingress_limit,
endpoint_data_rx,
endpoint_limit,
tun_outbound_rx,
tun_limit,
deferred_endpoint_data_batches,
deferred_tun_packets,
endpoint_tx,
transports,
crypto_limit,
} = route;
let transport_send_batch_packets = 8;
let mut pool = test_aead_worker_pool(
crypto_limit
.saturating_add(DATAPLANE_AEAD_WORKER_FAIRNESS_PACKETS)
.max(1),
);
let mut deferred_raw_ingress = std::collections::VecDeque::new();
let mut summary = start_test_aead_completion_turn(
driver,
completions,
completion_limit,
endpoint_tx.direct_sink().is_some(),
);
let mut aggregate: Option<DataplaneLiveNodeTurn> = None;
loop {
let capacity_before = pool.available_capacity();
let report = driver
.pump_aead_live_node_route_table_turn_after_completion_with_firsts(
DataplaneLivePumpRequest {
summary,
crypto_worker: &mut pool,
fast_ingress: None,
raw_ingress,
routes,
raw_ingress_limit,
endpoint_data_rx,
endpoint_limit,
tun_outbound_rx,
tun_limit,
outbound_firsts: DataplaneLiveOutboundFirsts::default(),
deferred_endpoint_data_batches,
deferred_tun_packets,
deferred_raw_ingress: &mut deferred_raw_ingress,
endpoint_tx,
transports,
crypto_limit,
transport_send_batch_packets,
},
)
.await;
let dispatched = report.summary().dispatched();
let worker_dispatched = capacity_before.saturating_sub(pool.available_capacity());
if let Some(aggregate) = aggregate.as_mut() {
aggregate.absorb(report);
} else {
aggregate = Some(report);
}
if dispatched == 0 || worker_dispatched == 0 {
break;
}
for _ in 0..100 {
if pool.has_ready_completions() {
break;
}
std::thread::sleep(std::time::Duration::from_millis(1));
}
summary = driver.start_aead_completion_turn(
&mut pool,
crypto_limit,
endpoint_tx.direct_sink().is_some(),
);
assert_eq!(summary.completions(), worker_dispatched);
}
aggregate.expect("live test turn")
}
fn finish_test_aead_turn(
driver: &mut DataplaneTurnDriver,
summary: DataplaneRuntimeSummary,
limit: usize,
) -> DataplaneRuntimeTurn<'_> {
let summary = collect_test_aead_outputs(driver, summary, limit, false);
DataplaneRuntimeTurn {
summary,
raw_ingress_drops: &driver.raw_ingress_drops,
output_drops: &driver.output_drops,
outputs: &driver.outputs,
drops: &driver.drops,
}
}
fn finish_test_aead_output_turn<'a, S>(
driver: &'a mut DataplaneTurnDriver,
summary: DataplaneRuntimeSummary,
sink: &mut S,
limit: usize,
) -> DataplaneRuntimeTurn<'a>
where
S: DataplaneOutputSink,
{
let summary = collect_test_aead_outputs(driver, summary, limit, false);
driver.send_collected_outputs(summary, sink)
}
fn drain_test_completions_into_mover(
driver: &mut DataplaneTurnDriver,
completions: &mut VecDeque<CryptoCompletion>,
limit: usize,
) -> usize {
driver.completion_batches.clear();
let drained = limit.min(completions.len());
for completion in completions.drain(..drained) {
CryptoCompletionBatch::push_grouped(completion, &mut driver.completion_batches);
}
driver
.mover
.queue_completion_batches(&mut driver.completion_batches);
drained
}
fn start_test_aead_completion_turn(
driver: &mut DataplaneTurnDriver,
completions: &mut VecDeque<CryptoCompletion>,
completion_limit: usize,
compact_endpoint_data: bool,
) -> DataplaneRuntimeSummary {
driver.reset_turn_buffers();
let completion_limit = driver.completion_drain_limit(completion_limit);
let queued = drain_test_completions_into_mover(driver, completions, completion_limit);
let mut summary = DataplaneRuntimeSummary::default();
summary.completions = queued;
driver.retire_queued_completed_aead_outputs(completion_limit, compact_endpoint_data);
driver.admit_retired_outbound_packets(summary)
}
fn test_node_addr(id: u64) -> NodeAddr {
let mut bytes = [0u8; 16];
bytes[8..16].copy_from_slice(&id.to_be_bytes());
NodeAddr::from_bytes(bytes)
}
fn fmp_owner(id: u64) -> OwnerId {
OwnerId::fmp_node(test_node_addr(id))
}
fn fsp_owner(id: u64) -> OwnerId {
OwnerId::fsp_node(test_node_addr(id))
}
fn test_receiver_idx(owner: OwnerId) -> u32 {
let node_addr = owner.node_addr();
let bytes: [u8; 4] = node_addr.as_bytes()[12..16]
.try_into()
.expect("test owner embeds receiver index");
u32::from_be_bytes(bytes)
}
fn live_path(id: u32) -> TransportPath {
let port = 10_000 + id % 50_000;
let remote_addr = format!("198.51.100.1:{port}");
TransportPath::live(TransportId::new(id), TransportAddr::from_string(&remote_addr))
}
fn tun_ipv6_packet(dest_addr: NodeAddr, len: usize) -> Vec<u8> {
assert!(len >= 40);
let mut packet = vec![0u8; len];
packet[0] = 0x60;
packet[6] = 17;
let dest = crate::FipsAddress::from_node_addr(&dest_addr);
packet[24..40].copy_from_slice(dest.as_bytes());
packet
}
fn packet(
owner: OwnerId,
generation: u64,
counter: u64,
class: PacketClass,
output: OutputTarget,
) -> SocketPacket {
SocketPacket::new(
owner,
generation,
counter,
class,
output,
PacketBuffer::new(vec![counter as u8]),
)
}
fn fmp_socket_packet(
owner: OwnerId,
generation: u64,
output: OutputTarget,
payload: Vec<u8>,
) -> Result<SocketPacket, WirePreflightError> {
let payload = PacketBuffer::new(payload);
let header = FmpWireHeader::parse(payload.as_slice())?;
Ok(SocketPacket::new(
owner,
generation,
header.counter(),
PacketClass::Bulk,
output,
payload,
)
.with_wire_flags(header.flags()))
}
fn fsp_socket_packet(
owner: OwnerId,
generation: u64,
output: OutputTarget,
payload: Vec<u8>,
) -> Result<SocketPacket, WirePreflightError> {
let payload = PacketBuffer::new(payload);
let header = FspWireHeader::parse(payload.as_slice())?;
Ok(SocketPacket::new(
owner,
generation,
header.counter(),
PacketClass::Bulk,
output,
payload,
)
.with_wire_flags(header.flags()))
}
fn fmp_wire(receiver_idx: u32, counter: u64, flags: u8) -> Vec<u8> {
let mut data = vec![0u8; FMP_ESTABLISHED_HEADER_SIZE + 16];
data[0] = (FMP_VERSION << 4) | FMP_PHASE_ESTABLISHED;
data[1] = flags;
data[4..8].copy_from_slice(&receiver_idx.to_le_bytes());
data[8..16].copy_from_slice(&counter.to_le_bytes());
data
}
fn fsp_wire(counter: u64, flags: u8) -> Vec<u8> {
let mut data = vec![0u8; FSP_HEADER_SIZE + 16];
data[0] = (FSP_VERSION << 4) | FSP_PHASE_ESTABLISHED;
data[1] = flags;
data[4..12].copy_from_slice(&counter.to_le_bytes());
data
}
fn transport_output(
owner: OwnerId,
counter: u64,
ingress_seq: u64,
transport_id: TransportId,
remote_addr: TransportAddr,
payload: Vec<u8>,
) -> PacketOutput {
PacketOutput {
owner,
counter,
ingress_seq,
lane: Lane::Bulk,
target: OutputTarget::Transport,
source_path: None,
previous_hop: None,
ce_flag: false,
path_mtu: u16::MAX,
source_peer: None,
path: Some(TransportPath::live(transport_id, remote_addr)),
activity_tick: None,
source_wire_len: None,
fmp_timestamp_ms: None,
fsp_send_receipt: None,
send_token: None,
payload: PacketBuffer::new(payload),
}
}
fn test_cipher(byte: u8) -> LessSafeKey {
let key = [byte; 32];
let unbound = UnboundKey::new(&ring::aead::CHACHA20_POLY1305, &key).unwrap();
LessSafeKey::new(unbound)
}
fn test_key(byte: u8) -> AeadKey {
Arc::new(test_cipher(byte))
}
fn unstarted_udp_transport(transport_id: TransportId) -> TransportHandle {
let (packet_tx, _packet_rx) = crate::transport::packet_channel(4);
TransportHandle::Udp(crate::transport::udp::UdpTransport::new(
transport_id,
None,
crate::config::UdpConfig {
bind_addr: Some("127.0.0.1:0".to_string()),
..Default::default()
},
packet_tx,
))
}
fn fmp_encrypted_wire(
receiver_idx: u32,
counter: u64,
flags: u8,
plaintext: &[u8],
key: u8,
) -> Vec<u8> {
let mut data = fmp_wire(receiver_idx, counter, flags);
data.truncate(FMP_ESTABLISHED_HEADER_SIZE);
let mut ciphertext = plaintext.to_vec();
test_cipher(key)
.seal_in_place_append_tag(
aead_nonce(counter),
Aad::from(&data[..FMP_ESTABLISHED_HEADER_SIZE]),
&mut ciphertext,
)
.unwrap();
data.extend_from_slice(&ciphertext);
data
}
fn fsp_encrypted_wire(counter: u64, flags: u8, plaintext: &[u8], key: u8) -> Vec<u8> {
fsp_encrypted_wire_with_coords(
counter,
flags,
plaintext,
key,
&empty_fsp_coords_prefix(),
)
}
fn fsp_encrypted_wire_with_coords(
counter: u64,
flags: u8,
plaintext: &[u8],
key: u8,
coords_prefix: &[u8],
) -> Vec<u8> {
let mut data = fsp_wire(counter, flags);
data.truncate(FSP_HEADER_SIZE);
let mut ciphertext = plaintext.to_vec();
test_cipher(key)
.seal_in_place_append_tag(
aead_nonce(counter),
Aad::from(&data[..FSP_HEADER_SIZE]),
&mut ciphertext,
)
.unwrap();
if flags & crate::node::session_wire::FSP_FLAG_CP != 0 {
data.extend_from_slice(coords_prefix);
}
data.extend_from_slice(&ciphertext);
data
}
fn encrypted_fmp_packet(
owner: OwnerId,
generation: u64,
counter: u64,
class: PacketClass,
output: OutputTarget,
key: u8,
) -> SocketPacket {
SocketPacket::new(
owner,
generation,
counter,
class,
output,
PacketBuffer::new(fmp_encrypted_wire(
test_receiver_idx(owner),
counter,
0,
&[counter as u8],
key,
)),
)
}
fn retire_completion(
mover: &mut Dataplane,
completion: CryptoCompletion,
) -> Vec<PacketOutput> {
let mut retired = Vec::new();
let mut completions = vec![CryptoCompletionBatch::from_completion(completion)];
mover.queue_completion_batches(&mut completions);
retire_queued_completions_to_outputs(mover, 1, &mut retired);
retired
}
fn retire_queued_completions_to_outputs(
mover: &mut Dataplane,
limit: usize,
retired: &mut Vec<PacketOutput>,
) -> usize {
let mut outbound_packets = Vec::new();
let mut fsp_authenticated_ingress = DataplaneFspAuthenticatedIngress::default();
let retired_count = mover.retire_queued_completions_into(
limit,
&mut DataplaneRetiredOutputSink::new(
retired,
&mut outbound_packets,
&mut fsp_authenticated_ingress,
),
false,
);
assert!(outbound_packets.is_empty());
assert!(fsp_authenticated_ingress.is_empty());
retired_count
}
fn empty_fsp_coords_prefix() -> Vec<u8> {
let mut prefix = Vec::with_capacity(2 * std::mem::size_of::<u16>());
prefix.extend_from_slice(&0u16.to_le_bytes());
prefix.extend_from_slice(&0u16.to_le_bytes());
prefix
}
fn open_sealed_output(output: &PacketOutput, key: u8) -> Vec<u8> {
match output.owner.protocol {
PacketProtocol::Fmp => open_fmp_wire_payload(output.payload.as_slice(), key),
PacketProtocol::Fsp => open_fsp_wire_payload(output.payload.as_slice(), key),
}
}
fn open_fmp_wire_payload(payload: &[u8], key: u8) -> Vec<u8> {
let header = FmpWireHeader::parse(payload).unwrap();
let aad = header.header_bytes();
open_wire_payload(
payload,
key,
header.counter(),
&aad,
header.ciphertext_offset(),
)
}
fn open_fsp_wire_payload(payload: &[u8], key: u8) -> Vec<u8> {
let header = FspWireHeader::parse(payload).unwrap();
let aad = header.header_bytes();
open_wire_payload(
payload,
key,
header.counter(),
&aad,
header.ciphertext_offset(),
)
}
fn open_wire_payload(
payload: &[u8],
key: u8,
counter: u64,
aad: &[u8],
ciphertext_offset: usize,
) -> Vec<u8> {
let mut ciphertext = payload[ciphertext_offset..].to_vec();
let plaintext_len = test_cipher(key)
.open_in_place(aead_nonce(counter), Aad::from(aad), &mut ciphertext)
.unwrap()
.len();
ciphertext.truncate(plaintext_len);
ciphertext
}
fn outbound_packet(
owner: OwnerId,
generation: u64,
class: PacketClass,
payload: &[u8],
) -> OutboundPacket {
match owner.protocol {
PacketProtocol::Fmp => OutboundPacket::fmp(
owner,
generation,
class,
test_receiver_idx(owner),
0,
PacketBuffer::new(payload.to_vec()),
),
PacketProtocol::Fsp => {
OutboundPacket::fsp(owner, generation, class, 0, PacketBuffer::new(payload.to_vec()))
}
}
}