use std::sync::atomic::{AtomicUsize, Ordering::Relaxed};
pub(crate) enum PreparedCryptoWork {
Open { work: CryptoWork, cipher: AeadKey },
Seal { work: OutboundCryptoWork, cipher: AeadKey },
}
struct PreparedSealWork {
work: OutboundCryptoWork,
cipher: AeadKey,
}
impl PreparedSealWork {
fn execute(self) -> CryptoCompletion {
execute_seal_crypto_work(self.work, self.cipher)
}
fn lane(&self) -> Lane {
self.work.reservation.lane
}
}
const DATAPLANE_AEAD_WORKER_FAIRNESS_PACKETS: usize = 8;
const DATAPLANE_AEAD_JOB_PACKETS: usize = 128;
impl PreparedCryptoWork {
pub(crate) fn open(work: CryptoWork, cipher: AeadKey) -> Self {
Self::Open { work, cipher }
}
pub(crate) fn seal(work: OutboundCryptoWork, cipher: AeadKey) -> Self {
Self::Seal { work, cipher }
}
fn lane(&self) -> Lane {
match self {
Self::Open { work, .. } => work.reservation.lane,
Self::Seal { work, .. } => work.reservation.lane,
}
}
}
enum PreparedCryptoJob {
OpenRun {
queued_at: Option<crate::perf_profile::TraceStamp>,
run: OpenCryptoOwnerRun,
cipher: AeadKey,
bulk_count: usize,
},
Seal {
queued_at: Option<crate::perf_profile::TraceStamp>,
work: Vec<PreparedSealWork>,
bulk_count: usize,
},
}
impl PreparedCryptoJob {
fn open_run(run: OpenCryptoOwnerRun, cipher: AeadKey) -> Self {
let bulk_count = run.bulk_count();
Self::OpenRun {
queued_at: crate::perf_profile::stamp(),
run,
cipher,
bulk_count,
}
}
fn seal(work: Vec<PreparedSealWork>, bulk_count: usize) -> Self {
Self::Seal {
queued_at: crate::perf_profile::stamp(),
work,
bulk_count,
}
}
fn queued_at(&self) -> Option<crate::perf_profile::TraceStamp> {
match self {
Self::OpenRun { queued_at, .. } | Self::Seal { queued_at, .. } => *queued_at,
}
}
fn len(&self) -> usize {
match self {
Self::OpenRun { run, .. } => run.len(),
Self::Seal { work, .. } => work.len(),
}
}
fn bulk_count(&self) -> usize {
match self {
Self::OpenRun { bulk_count, .. } | Self::Seal { bulk_count, .. } => *bulk_count,
}
}
fn execute_completion_batches(self) -> Vec<CryptoCompletionBatch> {
match self {
Self::OpenRun { run, cipher, .. } => execute_open_run_job(run, cipher),
Self::Seal { work, .. } => execute_seal_job(work),
}
}
}
struct PreparedOpenRunJobBuilder {
job_packets: usize,
run: Option<OpenCryptoOwnerRun>,
cipher: Option<AeadKey>,
}
impl PreparedOpenRunJobBuilder {
fn new() -> Self {
Self {
job_packets: DATAPLANE_AEAD_JOB_PACKETS,
run: None,
cipher: None,
}
}
fn push(
&mut self,
pool: &mut DataplaneAeadWorkerPool,
work: CryptoWork,
cipher: AeadKey,
) {
if !self.matches_run(&work, &cipher) {
self.flush(pool);
}
if self
.run
.as_ref()
.is_some_and(|run| run.len() >= self.job_packets)
{
self.flush(pool);
}
match &mut self.run {
Some(run) => run.push(work),
None => {
self.run = Some(OpenCryptoOwnerRun::new(work, self.job_packets));
self.cipher = Some(cipher);
}
}
}
fn flush(&mut self, pool: &mut DataplaneAeadWorkerPool) {
let Some(run) = self.run.take() else {
return;
};
let cipher = self
.cipher
.take()
.expect("open run cipher exists when work is non-empty");
pool.submit_open_run_job(run, cipher);
}
fn matches_run(&self, work: &CryptoWork, cipher: &AeadKey) -> bool {
let Some(run) = self.run.as_ref() else {
return true;
};
let Some(current_cipher) = self.cipher.as_ref() else {
return true;
};
Arc::ptr_eq(current_cipher, cipher) && run.matches(work)
}
}
struct PreparedSealJobBuilder {
job_packets: usize,
work: Vec<PreparedSealWork>,
bulk_count: usize,
}
impl PreparedSealJobBuilder {
fn new() -> Self {
Self {
job_packets: DATAPLANE_AEAD_JOB_PACKETS,
work: Vec::new(),
bulk_count: 0,
}
}
fn push(
&mut self,
pool: &mut DataplaneAeadWorkerPool,
work: PreparedSealWork,
) {
if work.lane() == Lane::Bulk {
self.bulk_count = self.bulk_count.saturating_add(1);
}
if self.work.capacity() == 0 {
self.work.reserve_exact(self.job_packets);
}
self.work.push(work);
if self.work.len() >= self.job_packets {
self.flush(pool);
}
}
fn flush(&mut self, pool: &mut DataplaneAeadWorkerPool) {
if self.work.is_empty() {
return;
}
let next = Vec::with_capacity(self.job_packets);
let work = std::mem::replace(&mut self.work, next);
let bulk_count = std::mem::take(&mut self.bulk_count);
pool.submit_seal_job(work, bulk_count);
}
}
#[derive(Clone, Debug)]
struct DataplaneAeadWorkerCounters {
in_flight: Arc<AtomicUsize>,
bulk_in_flight: Arc<AtomicUsize>,
}
impl DataplaneAeadWorkerCounters {
fn new() -> Self {
Self {
in_flight: Arc::new(AtomicUsize::new(0)),
bulk_in_flight: Arc::new(AtomicUsize::new(0)),
}
}
fn add(&self, count: usize, bulk_count: usize) {
self.in_flight.fetch_add(count, Relaxed);
if bulk_count > 0 {
self.bulk_in_flight.fetch_add(bulk_count, Relaxed);
}
}
fn finish(&self, count: usize, bulk_count: usize) {
self.in_flight.fetch_sub(count, Relaxed);
if bulk_count > 0 {
self.bulk_in_flight.fetch_sub(bulk_count, Relaxed);
}
}
}
#[derive(Debug)]
pub(crate) struct DataplaneAeadWorkerPool {
completion_tx: tokio::sync::mpsc::Sender<Vec<CryptoCompletionBatch>>,
completion_rx: tokio::sync::mpsc::Receiver<Vec<CryptoCompletionBatch>>,
completion_notify: Arc<tokio::sync::Notify>,
pending_completion_batches: VecDeque<CryptoCompletionBatch>,
counters: DataplaneAeadWorkerCounters,
max_in_flight: usize,
runtime: Option<tokio::runtime::Handle>,
tasks: tokio::task::JoinSet<()>,
}
impl DataplaneAeadWorkerPool {
pub(crate) fn new(max_in_flight: usize) -> Self {
let max_in_flight = max_in_flight.max(1);
let (completion_tx, completion_rx) = tokio::sync::mpsc::channel(max_in_flight);
Self {
completion_tx,
completion_rx,
completion_notify: Arc::new(tokio::sync::Notify::new()),
pending_completion_batches: VecDeque::new(),
counters: DataplaneAeadWorkerCounters::new(),
max_in_flight,
runtime: tokio::runtime::Handle::try_current().ok(),
tasks: tokio::task::JoinSet::new(),
}
}
pub(crate) fn completion_notify(&self) -> Arc<tokio::sync::Notify> {
Arc::clone(&self.completion_notify)
}
pub(crate) fn has_ready_completions(&self) -> bool {
!self.pending_completion_batches.is_empty()
|| !self.completion_rx.is_empty()
}
pub(crate) fn drain_completion_batches_into_sink<S>(
&mut self,
limit: usize,
sink: &mut S,
) -> usize
where
S: DataplaneCompletionSink,
{
self.drain_completion_batches_with(limit, |batch| {
sink.push_completion_batch(batch);
})
}
pub(crate) fn record_perf_depths(&self) {
if !crate::perf_profile::enabled() {
return;
}
crate::perf_profile::record_event_count(
crate::perf_profile::Event::DataplaneAeadInFlight,
self.counters.in_flight.load(Relaxed) as u64,
);
let pending_completion_depth = self
.pending_completion_batches
.iter()
.map(CryptoCompletionBatch::len)
.sum::<usize>();
let pending_completion_batches = self.pending_completion_batches.len();
let rx_queued_messages = self.completion_rx.len();
let completion_depth = pending_completion_depth.saturating_add(rx_queued_messages);
crate::perf_profile::record_event_count(
crate::perf_profile::Event::DataplaneAeadCompletionQueueDepth,
completion_depth as u64,
);
crate::perf_profile::record_dataplane_aead_completion_backlog(
rx_queued_messages,
pending_completion_batches,
pending_completion_depth,
);
}
fn finish_drained_completions(&self, count: usize, bulk_count: usize) {
self.counters.finish(count, bulk_count);
}
fn drain_completion_batch(
&mut self,
mut batch: CryptoCompletionBatch,
limit: usize,
push_batch: &mut impl FnMut(CryptoCompletionBatch),
) -> (usize, Option<CryptoCompletionBatch>) {
crate::perf_profile::record_dataplane_aead_completion_batch(batch.len());
let drained = batch.len().min(limit);
if drained == 0 {
return (0, Some(batch));
}
let pending = if drained < batch.len() {
let pending = batch.split_off(drained);
crate::perf_profile::record_dataplane_aead_completion_split(pending.len());
Some(pending)
} else {
None
};
let bulk_count = if batch.lane() == Lane::Bulk { drained } else { 0 };
self.finish_drained_completions(drained, bulk_count);
push_batch(batch);
(drained, pending)
}
fn drain_completion_batches_with(
&mut self,
limit: usize,
mut push_batch: impl FnMut(CryptoCompletionBatch),
) -> usize {
self.reap_finished_tasks();
let mut drained = 0usize;
while drained < limit {
if let Some(batch) = self.pending_completion_batches.pop_front() {
let (got, pending) = self.drain_completion_batch(
batch,
limit.saturating_sub(drained),
&mut push_batch,
);
drained = drained.saturating_add(got);
if let Some(pending) = pending {
self.pending_completion_batches.push_front(pending);
break;
}
continue;
}
let received = self.completion_rx.try_recv();
match received {
Ok(mut batches) => {
let mut batches = batches.drain(..);
while let Some(batch) = batches.next() {
if drained >= limit {
self.pending_completion_batches.push_back(batch);
self.pending_completion_batches.extend(batches);
break;
}
let (got, pending) = self.drain_completion_batch(
batch,
limit.saturating_sub(drained),
&mut push_batch,
);
drained = drained.saturating_add(got);
if let Some(pending) = pending {
self.pending_completion_batches.push_back(pending);
self.pending_completion_batches.extend(batches);
break;
}
}
}
Err(tokio::sync::mpsc::error::TryRecvError::Empty) => break,
Err(tokio::sync::mpsc::error::TryRecvError::Disconnected) => break,
}
}
self.reap_finished_tasks();
drained
}
fn available_capacity(&self) -> usize {
self.max_in_flight.saturating_sub(
self.counters.in_flight.load(Relaxed),
)
}
fn available_capacity_for_lane(&self, lane: Lane) -> usize {
let total_available = self.available_capacity();
if lane == Lane::Priority {
return total_available;
}
let bulk_limit =
self.max_in_flight
.saturating_sub(dataplane_aead_worker_priority_reserve(
self.max_in_flight,
));
let bulk_in_flight = self
.counters
.bulk_in_flight
.load(Relaxed);
bulk_limit.saturating_sub(bulk_in_flight).min(total_available)
}
fn submit_seal_job(
&mut self,
work: Vec<PreparedSealWork>,
bulk_count: usize,
) {
if work.is_empty() {
return;
}
let job = PreparedCryptoJob::seal(work, bulk_count);
self.submit_job(job);
}
fn submit_open_run_job(&mut self, run: OpenCryptoOwnerRun, cipher: AeadKey) {
if run.is_empty() {
return;
}
self.submit_job(PreparedCryptoJob::open_run(run, cipher));
}
fn submit_job(&mut self, job: PreparedCryptoJob) {
self.reap_finished_tasks();
let chunk_len = job.len();
let bulk_count = job.bulk_count();
self.counters.add(chunk_len, bulk_count);
let completion_tx = self.completion_tx.clone();
let completion_notify = Arc::clone(&self.completion_notify);
let counters = self.counters.clone();
let runtime = self
.runtime
.get_or_insert_with(tokio::runtime::Handle::current)
.clone();
self.tasks.spawn_on(
async move {
crate::perf_profile::record_since(
crate::perf_profile::Stage::DataplaneAeadWorkerQueueWait,
job.queued_at(),
);
let completions = job.execute_completion_batches();
if completions.is_empty() {
return;
}
let completed_count = completions
.iter()
.map(CryptoCompletionBatch::len)
.sum::<usize>();
let completed_bulk_count = completions
.iter()
.filter(|batch| batch.lane() == Lane::Bulk)
.map(CryptoCompletionBatch::len)
.sum::<usize>();
if send_completion_batches(completions, &completion_tx)
.await
.is_err()
{
counters.finish(completed_count, completed_bulk_count);
return;
}
completion_notify.notify_one();
},
&runtime,
);
crate::perf_profile::record_dataplane_aead_prepared_job(chunk_len);
}
fn reap_finished_tasks(&mut self) {
while let Some(result) = self.tasks.try_join_next() {
result.expect("dataplane AEAD task failed");
}
}
fn submit_prepared_chunk(
&mut self,
prepared: &mut Vec<PreparedCryptoWork>,
) {
if prepared.is_empty() {
return;
}
let mut open_jobs = PreparedOpenRunJobBuilder::new();
let mut seal_jobs = PreparedSealJobBuilder::new();
for work in prepared.drain(..) {
match work {
PreparedCryptoWork::Open { work, cipher } => {
open_jobs.push(self, work, cipher);
}
PreparedCryptoWork::Seal { work, cipher } => {
seal_jobs.push(self, PreparedSealWork { work, cipher });
}
}
}
open_jobs.flush(self);
seal_jobs.flush(self);
}
}
async fn send_completion_batches(
batches: Vec<CryptoCompletionBatch>,
completion_tx: &tokio::sync::mpsc::Sender<Vec<CryptoCompletionBatch>>,
) -> Result<(), ()> {
if batches.is_empty() {
return Ok(());
}
if crate::perf_profile::enabled() {
let completion_batch_count = batches.len();
let completion_packet_count = batches
.iter()
.map(CryptoCompletionBatch::len)
.sum::<usize>();
crate::perf_profile::record_dataplane_aead_completion_send(
1,
completion_batch_count,
completion_packet_count,
);
}
completion_tx.send(batches).await.map_err(|_| ())?;
Ok(())
}
fn execute_open_run_job(
mut run: OpenCryptoOwnerRun,
cipher: AeadKey,
) -> Vec<CryptoCompletionBatch> {
if run.is_empty() {
return Vec::new();
}
let _timer =
crate::perf_profile::Timer::start(crate::perf_profile::Stage::DataplaneAeadOpen);
for item in &mut run.items {
let OpenCryptoOwnerRunItem::Prepared(work) =
std::mem::replace(item, OpenCryptoOwnerRunItem::Vacant)
else {
panic!("open owner run executed twice");
};
*item = OpenCryptoOwnerRunItem::Completed(execute_open_crypto_work(work, &cipher));
}
vec![CryptoCompletionBatch::from_open_owner_run(run)]
}
fn execute_seal_job(work: Vec<PreparedSealWork>) -> Vec<CryptoCompletionBatch> {
if work.is_empty() {
return Vec::new();
}
let mut completions = Vec::with_capacity(work.len());
for work in work {
CryptoCompletionBatch::push_grouped(work.execute(), &mut completions);
}
completions
}
fn dataplane_aead_worker_priority_reserve(max_in_flight: usize) -> usize {
max_in_flight
.saturating_sub(DATAPLANE_AEAD_WORKER_FAIRNESS_PACKETS)
.min(DATAPLANE_AEAD_WORKER_FAIRNESS_PACKETS)
}
impl std::fmt::Debug for PreparedCryptoWork {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Open { work, .. } => f
.debug_struct("PreparedCryptoWork::Open")
.field("reservation", &work.reservation)
.finish_non_exhaustive(),
Self::Seal { work, .. } => f
.debug_struct("PreparedCryptoWork::Seal")
.field("reservation", &work.reservation)
.finish_non_exhaustive(),
}
}
}
fn failed_crypto_completion(
reservation: OwnerReservation,
kind: CryptoFailureKind,
) -> CryptoCompletion {
CryptoCompletion {
reservation,
result: CryptoResult::Failed(kind),
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum AeadHeader {
Fmp([u8; FMP_ESTABLISHED_HEADER_SIZE]),
Fsp([u8; FSP_HEADER_SIZE]),
}
impl AeadHeader {
fn as_aad(&self) -> &[u8] {
match self {
Self::Fmp(header) => header,
Self::Fsp(header) => header,
}
}
}
struct AeadOpenWork {
work: CryptoWork,
header: AeadHeader,
ciphertext_offset: usize,
}
fn execute_open_crypto_work(work: CryptoWork, cipher: &LessSafeKey) -> CryptoCompletion {
let reservation = work.reservation.clone();
match AeadOpenWork::from_crypto_work(work) {
Ok(work) => work.execute(cipher),
Err(_) => failed_crypto_completion(reservation, CryptoFailureKind::Open),
}
}
impl AeadOpenWork {
fn from_crypto_work(work: CryptoWork) -> Result<Self, WirePreflightError> {
let (header, ciphertext_offset, counter) = match work.packet.owner.protocol {
PacketProtocol::Fmp => {
let header = FmpWireHeader::parse(work.packet.payload.as_slice())?;
(
AeadHeader::Fmp(header.header_bytes()),
header.ciphertext_offset(),
header.counter(),
)
}
PacketProtocol::Fsp => {
let header = FspWireHeader::parse(work.packet.payload.as_slice())?;
(
AeadHeader::Fsp(header.header_bytes()),
header.ciphertext_offset(),
header.counter(),
)
}
};
if counter != work.packet.counter {
return Err(WirePreflightError::CounterMismatch);
}
Ok(Self {
work,
header,
ciphertext_offset,
})
}
fn execute(self, cipher: &LessSafeKey) -> CryptoCompletion {
let mut work = self;
let reservation = work.work.reservation;
let target = work.work.packet.output;
let header = work.header;
let source_wire_len = work.work.packet.payload.len();
let opened_len = match work
.work
.packet
.payload
.as_mut_slice()
.get_mut(work.ciphertext_offset..)
{
Some(ciphertext) => {
let nonce = aead_nonce(reservation.counter);
cipher
.open_in_place(nonce, Aad::from(header.as_aad()), ciphertext)
.map(|plaintext| plaintext.len())
.ok()
}
None => None,
};
let result = match opened_len {
Some(plaintext_len) => {
work.work
.packet
.payload
.truncate(work.ciphertext_offset + plaintext_len);
CryptoResult::Opened(PacketOutput {
owner: reservation.owner,
counter: reservation.counter,
ingress_seq: reservation.ingress_seq,
lane: reservation.lane,
target,
source_path: reservation.source_path.clone(),
previous_hop: reservation.previous_hop,
ce_flag: reservation.ce_flag,
path_mtu: reservation.path_mtu,
source_peer: reservation.source_peer,
path: reservation.output_path.clone(),
activity_tick: reservation.activity_tick,
fmp_timestamp_ms: reservation.fmp_timestamp_ms,
source_wire_len: Some(source_wire_len),
fsp_send_receipt: None,
send_token: reservation.send_token,
payload: work.work.packet.payload,
})
}
None => CryptoResult::Failed(CryptoFailureKind::Open),
};
CryptoCompletion {
reservation,
result,
}
}
}
struct AeadSealWork {
work: OutboundCryptoWork,
cipher: AeadKey,
post_seal: OutboundPostSeal,
aad_len: usize,
ciphertext_offset: usize,
}
fn execute_seal_crypto_work(work: OutboundCryptoWork, cipher: AeadKey) -> CryptoCompletion {
let reservation = work.reservation.clone();
let _timer = crate::perf_profile::Timer::start(crate::perf_profile::Stage::DataplaneAeadSeal);
match AeadSealWork::from_outbound_work(work, cipher) {
Ok(work) => work.execute(),
Err(_) => failed_crypto_completion(reservation, CryptoFailureKind::Seal),
}
}
impl AeadSealWork {
fn from_outbound_work(
mut work: OutboundCryptoWork,
cipher: AeadKey,
) -> Result<Self, WireBuildError> {
let inner_prefix = work.packet.crypto_plaintext_prefix(
work.reservation.fmp_timestamp_ms,
work.reservation.fsp_timestamp_ms,
)?;
let payload_len = u16::try_from(inner_prefix.len().saturating_add(work.packet.payload.len()))
.map_err(|_| WireBuildError::PayloadTooLarge)?;
let counter = work.reservation.counter;
let (header, coord_prefix, ciphertext_offset) =
match (work.packet.owner.protocol, work.packet.wire) {
(
PacketProtocol::Fmp,
OutboundWire::Fmp {
receiver_idx,
flags,
},
) => (
AeadHeader::Fmp(build_fmp_established_header(
receiver_idx,
counter,
flags,
payload_len,
)),
Vec::new(),
FMP_ESTABLISHED_HEADER_SIZE,
),
(PacketProtocol::Fsp, OutboundWire::Fsp { flags }) => {
let coord_prefix = std::mem::take(&mut work.packet.fsp_cleartext_prefix);
validate_fsp_cleartext_prefix(flags, &coord_prefix)?;
let ciphertext_offset = FSP_HEADER_SIZE + coord_prefix.len();
(
AeadHeader::Fsp(build_fsp_established_header(counter, flags, payload_len)?),
coord_prefix,
ciphertext_offset,
)
}
_ => return Err(WireBuildError::ProtocolMismatch),
};
let aad = header.as_aad();
let aad_len = aad.len();
let prefix_len = aad
.len()
.saturating_add(coord_prefix.len())
.saturating_add(inner_prefix.len());
if work.packet.payload.try_prepend_slices(
&[aad, coord_prefix.as_slice(), inner_prefix.as_slice()],
AEAD_TAG_SIZE,
) {
crate::perf_profile::record_event(crate::perf_profile::Event::DataplaneSealInPlace);
} else {
crate::perf_profile::record_event(crate::perf_profile::Event::DataplaneSealAllocated);
let plaintext = std::mem::take(&mut work.packet.payload);
let mut payload = Vec::with_capacity(
prefix_len
.saturating_add(plaintext.len())
.saturating_add(AEAD_TAG_SIZE),
);
payload.extend_from_slice(aad);
payload.extend_from_slice(&coord_prefix);
payload.extend_from_slice(&inner_prefix);
payload.extend_from_slice(plaintext.as_slice());
work.packet.payload = PacketBuffer::new(payload);
}
Ok(Self {
post_seal: work.packet.post_seal,
work,
cipher,
aad_len,
ciphertext_offset,
})
}
fn execute(self) -> CryptoCompletion {
let mut work = self;
let reservation = work.work.reservation;
let tag = if work.aad_len <= work.ciphertext_offset
&& work.ciphertext_offset <= work.work.packet.payload.len()
{
let nonce = aead_nonce(reservation.counter);
let (prefix, plaintext) = work
.work
.packet
.payload
.as_mut_slice()
.split_at_mut(work.ciphertext_offset);
let Some(aad) = prefix.get(..work.aad_len) else {
return CryptoCompletion {
reservation,
result: CryptoResult::Failed(CryptoFailureKind::Seal),
};
};
work.cipher
.seal_in_place_separate_tag(nonce, Aad::from(aad), plaintext)
.ok()
} else {
None
};
let result = match tag {
Some(tag) => {
work.work.packet.payload.extend_from_slice(tag.as_ref());
match work.post_seal {
OutboundPostSeal::Transport => CryptoResult::Sealed(PacketOutput {
owner: reservation.owner,
counter: reservation.counter,
ingress_seq: reservation.ingress_seq,
lane: reservation.lane,
target: OutputTarget::Transport,
source_path: reservation.source_path.clone(),
previous_hop: reservation.previous_hop,
ce_flag: reservation.ce_flag,
path_mtu: reservation.path_mtu,
source_peer: reservation.source_peer,
path: reservation.output_path.clone(),
activity_tick: reservation.activity_tick,
fmp_timestamp_ms: reservation.fmp_timestamp_ms,
source_wire_len: None,
fsp_send_receipt: work.work.packet.fsp_send_receipt,
send_token: reservation.send_token,
payload: work.work.packet.payload,
}),
OutboundPostSeal::FmpWrap(route) => {
let mut packet = route
.into_fmp_outbound(work.work.packet.class, work.work.packet.payload)
.with_fsp_send_receipt(DataplaneFspSendReceipt {
owner: reservation.owner,
counter: reservation.counter,
});
if let Some(send_token) = work.work.packet.send_token {
packet = packet.with_send_token(send_token);
}
if let Some(tick) = reservation.activity_tick {
packet = packet.with_activity_tick(tick);
}
CryptoResult::Outbound(packet)
}
}
}
None => CryptoResult::Failed(CryptoFailureKind::Seal),
};
CryptoCompletion {
reservation,
result,
}
}
}
fn validate_fsp_cleartext_prefix(flags: u8, prefix: &[u8]) -> Result<(), WireBuildError> {
if flags & crate::node::session_wire::FSP_FLAG_CP == 0 {
return if prefix.is_empty() {
Ok(())
} else {
Err(WireBuildError::BadFspCoords)
};
}
crate::node::session_wire::parse_encrypted_coords(prefix)
.map(|_| ())
.map_err(|_| WireBuildError::BadFspCoords)
}
fn aead_nonce(counter: u64) -> Nonce {
let mut nonce_bytes = [0u8; 12];
nonce_bytes[4..12].copy_from_slice(&counter.to_le_bytes());
Nonce::assume_unique_for_key(nonce_bytes)
}