s2n_quic_dc/path/secret/map/
entry.rs

1// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
2// SPDX-License-Identifier: Apache-2.0
3
4use super::{
5    size_of::SizeOf,
6    status::{Dedup, IsRetired},
7    Map,
8};
9use crate::{
10    credentials::{self, Credentials},
11    packet::{secret_control as control, WireVersion},
12    path::secret::{
13        map::Epoch,
14        open, receiver,
15        schedule::{self, Initiator},
16        seal, sender,
17    },
18    stream::TransportFeatures,
19};
20use s2n_codec::EncoderBuffer;
21use s2n_quic_core::{dc, varint::VarInt};
22use std::{
23    any::Any,
24    fmt,
25    net::SocketAddr,
26    sync::{
27        atomic::{AtomicU32, AtomicU64, AtomicU8, Ordering},
28        Arc,
29    },
30    time::{Duration, Instant},
31};
32
33#[cfg(test)]
34mod tests;
35
36pub type ApplicationData = Arc<dyn Any + Send + Sync>;
37
38#[derive(Debug, thiserror::Error)]
39#[error("{inner}")]
40pub struct ApplicationDataError {
41    pub msg: &'static str,
42    #[source]
43    pub inner: Box<dyn std::error::Error + Send + Sync>,
44}
45
46#[derive(Debug)]
47pub struct Entry {
48    creation_time: Instant,
49    peer: SocketAddr,
50    secret: schedule::Secret,
51    retired: IsRetired,
52    accessed_at: AccessedAt,
53    sender: sender::State,
54    receiver: receiver::State,
55    parameters: dc::ApplicationParams,
56    // we store this as a u8 to allow the cleaner to separately "take" accessed for id and addr
57    // maps while not having two writes and wasting an extra byte of space.
58    accessed: AtomicU8,
59    application_data: Option<ApplicationData>,
60}
61
62impl SizeOf for Entry {
63    fn size(&self) -> usize {
64        let Entry {
65            creation_time,
66            peer,
67            secret,
68            retired,
69            sender,
70            receiver,
71            parameters,
72            accessed,
73            application_data,
74            accessed_at,
75        } = self;
76        creation_time.size()
77            + peer.size()
78            + secret.size()
79            + retired.size()
80            + sender.size()
81            + receiver.size()
82            + parameters.size()
83            + accessed.size()
84            + application_data.size()
85            + accessed_at.size()
86    }
87}
88
89impl SizeOf for Option<ApplicationData> {
90    fn size(&self) -> usize {
91        std::mem::size_of::<ApplicationData>()
92    }
93}
94
95impl SizeOf for ApplicationData {
96    fn size(&self) -> usize {
97        std::mem::size_of_val(self)
98    }
99}
100
101impl SizeOf for AtomicU8 {}
102impl SizeOf for AtomicU32 {}
103
104impl Entry {
105    pub fn new(
106        peer: SocketAddr,
107        secret: schedule::Secret,
108        sender: sender::State,
109        receiver: receiver::State,
110        parameters: dc::ApplicationParams,
111        // FIXME: remove unused parameter
112        _: Duration,
113        application_data: Option<ApplicationData>,
114    ) -> Self {
115        // clamp max datagram size to a well-known value
116        parameters
117            .max_datagram_size
118            .fetch_min(crate::stream::MAX_DATAGRAM_SIZE as _, Ordering::Relaxed);
119
120        Self {
121            creation_time: Instant::now(),
122            peer,
123            secret,
124            retired: Default::default(),
125            sender,
126            receiver,
127            parameters,
128            accessed: AtomicU8::new(0),
129            accessed_at: AccessedAt(AtomicU64::new(0)),
130            application_data,
131        }
132    }
133
134    #[cfg(any(test, feature = "testing"))]
135    #[allow(
136        clippy::unwrap_used,
137        reason = "test-support helper may panic to surface setup failures"
138    )]
139    pub fn fake(peer: SocketAddr, receiver: Option<receiver::State>) -> Arc<Entry> {
140        let receiver = receiver.unwrap_or_default();
141
142        let mut secret = [0; 32];
143        aws_lc_rs::rand::fill(&mut secret).unwrap();
144
145        Arc::new(Entry::new(
146            peer,
147            schedule::Secret::new(
148                schedule::Ciphersuite::AES_GCM_128_SHA256,
149                dc::SUPPORTED_VERSIONS[0],
150                s2n_quic_core::endpoint::Type::Client,
151                &secret,
152            ),
153            sender::State::new([0; control::TAG_LEN]),
154            receiver,
155            dc::testing::TEST_APPLICATION_PARAMS,
156            dc::testing::TEST_REHANDSHAKE_PERIOD,
157            None,
158        ))
159    }
160
161    pub fn peer(&self) -> &SocketAddr {
162        &self.peer
163    }
164
165    pub fn id(&self) -> &credentials::Id {
166        self.secret.id()
167    }
168
169    pub fn secret(&self) -> &schedule::Secret {
170        &self.secret
171    }
172
173    /// The epoch of the most recent access to this entry (via either id or address), as recorded
174    /// by [`Self::set_accessed_id`] / [`Self::set_accessed_addr`]. Returns epoch 0 if never
175    /// accessed.
176    pub fn accessed_at_epoch(&self) -> Epoch {
177        Epoch(self.accessed_at.accessed_at_epoch())
178    }
179
180    pub fn set_accessed_id(&self, epoch: Epoch) {
181        self.accessed_at.set_accessed(epoch);
182        self.accessed.fetch_or(0b10, Ordering::Relaxed);
183    }
184
185    pub fn set_accessed_addr(&self, epoch: Epoch) {
186        self.accessed_at.set_accessed(epoch);
187        self.accessed.fetch_or(0b01, Ordering::Relaxed);
188    }
189
190    pub fn take_accessed_id(&self) -> bool {
191        self.accessed.fetch_and(!0b10, Ordering::Relaxed) & 0b10 != 0
192    }
193
194    pub fn take_accessed_addr(&self) -> bool {
195        self.accessed.fetch_and(!0b01, Ordering::Relaxed) & 0b01 != 0
196    }
197
198    pub fn retire(&self, at_epoch: Epoch) {
199        self.retired.retire(at_epoch);
200    }
201
202    pub fn retired_at(&self) -> Option<Epoch> {
203        self.retired.retired_at()
204    }
205
206    pub fn uni_sealer(&self) -> (seal::Once, Credentials) {
207        let key_id = self.sender.next_key_id();
208        let credentials = Credentials {
209            id: *self.secret.id(),
210            key_id,
211        };
212        let sealer = self.secret.application_sealer(key_id);
213        let sealer = seal::Once::new(sealer);
214
215        (sealer, credentials)
216    }
217
218    pub fn uni_opener(
219        self: Arc<Self>,
220        map: Map,
221        credentials: &Credentials,
222        queue_id: Option<VarInt>,
223    ) -> open::Once {
224        let key_id = credentials.key_id;
225        let opener = self.secret.application_opener(key_id);
226        let dedup = Dedup::new(self, key_id, queue_id, map);
227        open::Once::new(opener, dedup)
228    }
229
230    pub fn bidi_local(&self, features: &TransportFeatures) -> Bidirectional {
231        let key_id = self.sender.next_key_id();
232        let initiator = Initiator::Local;
233
234        let application = ApplicationPair::new(
235            &self.secret,
236            key_id,
237            initiator,
238            // we don't need to dedup locally-initiated openers
239            Dedup::disabled(),
240        );
241
242        let control = if features.is_reliable() {
243            None
244        } else {
245            Some(ControlPair::new(&self.secret, key_id, initiator))
246        };
247
248        Bidirectional {
249            credentials: Credentials {
250                id: *self.secret.id(),
251                key_id,
252            },
253            application,
254            control,
255        }
256    }
257
258    pub fn bidi_remote(
259        self: &Arc<Self>,
260        map: Map,
261        credentials: &Credentials,
262        queue_id: Option<VarInt>,
263        features: &TransportFeatures,
264    ) -> Bidirectional {
265        let key_id = credentials.key_id;
266        let initiator = Initiator::Remote;
267
268        let application = ApplicationPair::new(
269            &self.secret,
270            key_id,
271            initiator,
272            // Remote application keys need to be de-duplicated
273            Dedup::new(self.clone(), key_id, queue_id, map),
274        );
275
276        let control = if features.is_reliable() {
277            None
278        } else {
279            Some(ControlPair::new(&self.secret, key_id, initiator))
280        };
281
282        Bidirectional {
283            credentials: *credentials,
284            application,
285            control,
286        }
287    }
288
289    pub fn parameters(&self) -> dc::ApplicationParams {
290        self.parameters.clone()
291    }
292
293    pub fn update_max_datagram_size(&self, mtu: u16) {
294        self.parameters
295            .max_datagram_size
296            .store(mtu, Ordering::Relaxed);
297    }
298
299    pub fn age(&self) -> Duration {
300        self.creation_time.elapsed()
301    }
302
303    pub fn creation_time(&self) -> Instant {
304        self.creation_time
305    }
306
307    pub fn receiver(&self) -> &receiver::State {
308        &self.receiver
309    }
310
311    pub fn sender(&self) -> &sender::State {
312        &self.sender
313    }
314
315    pub fn control_opener(&self) -> crate::crypto::awslc::open::control::Secret {
316        self.sender.control_secret(&self.secret)
317    }
318
319    pub fn control_sealer(&self) -> crate::crypto::awslc::seal::control::Secret {
320        self.secret.control_sealer()
321    }
322
323    pub fn application_data(&self) -> &Option<ApplicationData> {
324        &self.application_data
325    }
326
327    #[cfg(test)]
328    pub fn reset_sender_counter(&self) {
329        self.sender.reset_counter();
330    }
331}
332
333impl receiver::Error {
334    pub(super) fn to_packet<'buffer>(
335        self,
336        entry: &Entry,
337        credentials: &Credentials,
338        queue_id: Option<VarInt>,
339        buffer: &'buffer mut [u8; control::MAX_PACKET_SIZE],
340    ) -> &'buffer [u8] {
341        debug_assert_eq!(entry.secret.id(), &credentials.id);
342        let encoder = EncoderBuffer::new(&mut buffer[..]);
343        let length = match self {
344            receiver::Error::AlreadyExists => control::ReplayDetected {
345                wire_version: WireVersion::ZERO,
346                credential_id: credentials.id,
347                rejected_key_id: credentials.key_id,
348                queue_id,
349            }
350            .encode(encoder, &entry.control_sealer()),
351            receiver::Error::Unknown => control::StaleKey {
352                wire_version: WireVersion::ZERO,
353                credential_id: credentials.id,
354                min_key_id: entry.receiver.minimum_unseen_key_id(),
355                queue_id,
356            }
357            .encode(encoder, &entry.control_sealer()),
358        };
359        &buffer[..length]
360    }
361}
362
363pub struct Bidirectional {
364    pub credentials: Credentials,
365    pub application: ApplicationPair,
366    pub control: Option<ControlPair>,
367}
368
369pub struct ApplicationPair {
370    pub sealer: seal::Application,
371    pub opener: open::Application,
372}
373
374impl ApplicationPair {
375    pub fn new(
376        secret: &schedule::Secret,
377        key_id: VarInt,
378        initiator: Initiator,
379        dedup: Dedup,
380    ) -> Self {
381        let (sealer, sealer_ku, opener, opener_ku) = secret.application_pair(key_id, initiator);
382
383        let sealer = seal::Application::new(sealer, sealer_ku);
384
385        let opener = open::Application::new(opener, opener_ku, dedup);
386
387        Self { sealer, opener }
388    }
389}
390
391pub struct ControlPair {
392    pub sealer: seal::control::Stream,
393    pub opener: open::control::Stream,
394}
395
396impl ControlPair {
397    fn new(secret: &schedule::Secret, key_id: VarInt, initiator: Initiator) -> Self {
398        let (sealer, opener) = secret.control_pair(key_id, initiator);
399
400        Self { sealer, opener }
401    }
402}
403
404// Stores a epoch (similar to RetiredAt) for the last access.
405//
406// This gives a rough sense of how long ago the most recent access was. Epochs increment roughly
407// every minute (via cleaner thread).
408#[derive(Default)]
409struct AccessedAt(AtomicU64);
410
411impl fmt::Debug for AccessedAt {
412    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
413        f.debug_tuple("AccessedAt")
414            .field(&self.accessed_at_epoch())
415            .finish()
416    }
417}
418
419impl AccessedAt {
420    pub fn set_accessed(&self, at_epoch: Epoch) {
421        // Intentionally load before storing -- we want to avoid contention. In practice access
422        // likely involved some contention anyway (Arc ref-count, Mutex's on the maps) but good to
423        // avoid extra cost if we can.
424        if self.0.load(Ordering::Relaxed) < at_epoch.get() {
425            // It's OK if we lose a store, at worst this makes the access time about a minute
426            // wrong, and for frequently-accessed entries that'll quickly recover.
427            self.0.store(at_epoch.get(), Ordering::Relaxed);
428        }
429    }
430
431    pub fn accessed_at_epoch(&self) -> u64 {
432        self.0.load(Ordering::Relaxed)
433    }
434}
435
436impl SizeOf for AccessedAt {}
s2n_quic_dc/path/secret/map/entry.rs

s2n_quic_dc/path/secret/map/
entry.rs
