zerodds_discovery/security/
volatile_secure.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright 2026 ZeroDDS Contributors
3//! Builtin endpoint `DCPSParticipantVolatileMessageSecure` — DDS-Security
4//! 1.2 §7.4.5 + §10.5.4.
5//!
6//! Wire profile:
7//! - Reliability: Reliable (Spec §7.5.4 Tab.20).
8//! - Durability:  Volatile (KEEP_LAST 1 per spec, we conservatively use
9//!   KEEP_LAST 16 for the re-send window).
10//! - Topic type:  `ParticipantGenericMessage` (Spec §7.5.5).
11//! - EntityIds:   `BUILTIN_PARTICIPANT_VOLATILE_MESSAGE_SECURE_{WRITER,READER}`.
12//!
13//! Wrapper around [`zerodds_rtps::reliable_writer::ReliableWriter`] +
14//! [`zerodds_rtps::reliable_reader::ReliableReader`] with fixed EntityIds —
15//! analogous to the SEDP endpoints, only with a different topic-type codec.
16
17extern crate alloc;
18use alloc::vec::Vec;
19use core::time::Duration;
20
21use zerodds_rtps::error::WireError;
22use zerodds_rtps::fragment_assembler::AssemblerCaps;
23use zerodds_rtps::history_cache::HistoryKind;
24use zerodds_rtps::message_builder::{DEFAULT_MTU, OutboundDatagram};
25use zerodds_rtps::reader_proxy::ReaderProxy;
26use zerodds_rtps::reliable_reader::{
27    DEFAULT_HEARTBEAT_RESPONSE_DELAY, ReliableReader, ReliableReaderConfig,
28};
29use zerodds_rtps::reliable_writer::{DEFAULT_FRAGMENT_SIZE, ReliableWriter, ReliableWriterConfig};
30use zerodds_rtps::submessages::{
31    DataFragSubmessage, DataSubmessage, GapSubmessage, HeartbeatSubmessage, NackFragSubmessage,
32};
33use zerodds_rtps::wire_types::{EntityId, Guid, GuidPrefix, SequenceNumber, VendorId};
34use zerodds_rtps::writer_proxy::WriterProxy;
35
36use zerodds_security::error::{SecurityError, SecurityErrorKind, SecurityResult};
37use zerodds_security::generic_message::ParticipantGenericMessage;
38
39use crate::security::codec::{decode_generic_message, encode_generic_message};
40
41/// Default history depth (the spec says KEEP_LAST 1 — we conservatively
42/// use 16, so that short crypto-token bursts during onboarding do not
43/// drop individual tokens).
44pub const VOLATILE_SECURE_DEFAULT_DEPTH: usize = 16;
45
46/// Default HEARTBEAT period. Shorter than the SEDP default — we want
47/// fast crypto-token delivery during auth onboarding.
48pub const VOLATILE_SECURE_HEARTBEAT_PERIOD: Duration = Duration::from_millis(250);
49
50/// Reader cache depth (analogous to the SEDP reader 256).
51pub const VOLATILE_SECURE_READER_CAPACITY: usize = 64;
52
53/// Writer for `DCPSParticipantVolatileMessageSecure`.
54#[derive(Debug)]
55pub struct VolatileSecureMessageWriter {
56    inner: ReliableWriter,
57}
58
59impl VolatileSecureMessageWriter {
60    /// Creates a writer for the local participant.
61    #[must_use]
62    pub fn new(participant_prefix: GuidPrefix, vendor_id: VendorId) -> Self {
63        let guid = Guid::new(
64            participant_prefix,
65            EntityId::BUILTIN_PARTICIPANT_VOLATILE_MESSAGE_SECURE_WRITER,
66        );
67        let mut inner = ReliableWriter::new(ReliableWriterConfig {
68            guid,
69            vendor_id,
70            reader_proxies: Vec::new(),
71            max_samples: VOLATILE_SECURE_DEFAULT_DEPTH,
72            history_kind: HistoryKind::KeepLast {
73                depth: VOLATILE_SECURE_DEFAULT_DEPTH,
74            },
75            heartbeat_period: VOLATILE_SECURE_HEARTBEAT_PERIOD,
76            fragment_size: DEFAULT_FRAGMENT_SIZE,
77            mtu: DEFAULT_MTU,
78        });
79        // cyclones filtered VolatileSecure-Reader matcht per voller GUID:
80        // without INFO_DST, cyclone resolves dst=0:0:0:ff0202c4 -> no match
81        // -> wn->last_seq haengt -> maxseq-0-base-N-Deadlock (Trace-belegt).
82        inner.set_emit_info_dst(true);
83        Self { inner }
84    }
85
86    /// GUID of the writer.
87    #[must_use]
88    pub fn guid(&self) -> Guid {
89        self.inner.guid()
90    }
91
92    /// Number of registered reader proxies.
93    #[must_use]
94    pub fn reader_proxy_count(&self) -> usize {
95        self.inner.reader_proxy_count()
96    }
97
98    /// Read-only access to the ReliableWriter (tests/diagnostics).
99    #[must_use]
100    pub fn inner(&self) -> &ReliableWriter {
101        &self.inner
102    }
103
104    /// Adds a reader proxy.
105    pub fn add_reader_proxy(&mut self, proxy: ReaderProxy) {
106        self.inner.add_reader_proxy(proxy);
107    }
108
109    /// Removes a reader proxy.
110    pub fn remove_reader_proxy(&mut self, guid: Guid) -> Option<ReaderProxy> {
111        self.inner.remove_reader_proxy(guid)
112    }
113
114    /// Sends a `ParticipantGenericMessage`. Returns one datagram per
115    /// reader proxy.
116    ///
117    /// # Errors
118    /// `WireError` from the reliable writer (cache overflow on
119    /// `KeepAll`, sequence overflow).
120    pub fn write(
121        &mut self,
122        msg: &ParticipantGenericMessage,
123    ) -> Result<Vec<OutboundDatagram>, WireError> {
124        let payload = encode_generic_message(msg);
125        self.inner.write(&payload)
126    }
127
128    /// Write + piggyback HEARTBEAT in ONE operation (RTPS 2.5 §8.4.15.5).
129    ///
130    /// Security tokens (Kx, per-endpoint crypto) are reliable and
131    /// time-critical: the remote must NACK a loss IMMEDIATELY, not only at
132    /// the periodic HEARTBEAT (`VOLATILE_SECURE_HEARTBEAT_PERIOD` =
133    /// 250ms). Cross-vendor, the token otherwise falls behind Cyclone's
134    /// AUTHOK `delete_pending_match` and is never applied. The piggyback
135    /// HEARTBEAT (final_flag=false) forces the prompt ACKNACK response —
136    /// per-send, so robust for late joiners / rediscovery (no global
137    /// state).
138    ///
139    /// # Errors
140    /// Wire encode errors.
141    pub fn write_with_heartbeat(
142        &mut self,
143        msg: &ParticipantGenericMessage,
144        now: Duration,
145    ) -> Result<Vec<OutboundDatagram>, WireError> {
146        let payload = encode_generic_message(msg);
147        self.inner.write_with_heartbeat(&payload, now)
148    }
149
150    /// Tick (HEARTBEAT + resends).
151    ///
152    /// # Errors
153    /// Wire encode errors.
154    pub fn tick(&mut self, now: Duration) -> Result<Vec<OutboundDatagram>, WireError> {
155        self.inner.tick(now)
156    }
157
158    /// Dispatch of an ACKNACK from the remote reader.
159    pub fn handle_acknack(
160        &mut self,
161        src_guid: Guid,
162        base: SequenceNumber,
163        requested: impl IntoIterator<Item = SequenceNumber>,
164    ) {
165        self.inner.handle_acknack(src_guid, base, requested);
166    }
167
168    /// Dispatch of a NACK_FRAG from the remote reader.
169    pub fn handle_nackfrag(&mut self, src_guid: Guid, nf: &NackFragSubmessage) {
170        self.inner.handle_nackfrag(src_guid, nf);
171    }
172}
173
174/// Reader for `DCPSParticipantVolatileMessageSecure`.
175#[derive(Debug)]
176pub struct VolatileSecureMessageReader {
177    inner: ReliableReader,
178}
179
180impl VolatileSecureMessageReader {
181    /// Creates a reader for the local participant.
182    #[must_use]
183    pub fn new(participant_prefix: GuidPrefix, vendor_id: VendorId) -> Self {
184        let guid = Guid::new(
185            participant_prefix,
186            EntityId::BUILTIN_PARTICIPANT_VOLATILE_MESSAGE_SECURE_READER,
187        );
188        Self {
189            inner: ReliableReader::new(ReliableReaderConfig {
190                guid,
191                vendor_id,
192                writer_proxies: Vec::new(),
193                max_samples_per_proxy: VOLATILE_SECURE_READER_CAPACITY,
194                heartbeat_response_delay: DEFAULT_HEARTBEAT_RESPONSE_DELAY,
195                assembler_caps: AssemblerCaps::default(),
196            }),
197        }
198    }
199
200    /// GUID of the reader.
201    #[must_use]
202    pub fn guid(&self) -> Guid {
203        self.inner.guid()
204    }
205
206    /// Number of registered writer proxies.
207    #[must_use]
208    pub fn writer_proxy_count(&self) -> usize {
209        self.inner.writer_proxy_count()
210    }
211
212    /// Read-only access to the ReliableReader (tests/diagnostics).
213    #[must_use]
214    pub fn inner(&self) -> &ReliableReader {
215        &self.inner
216    }
217
218    /// Adds a writer proxy.
219    pub fn add_writer_proxy(&mut self, proxy: WriterProxy) {
220        self.inner.add_writer_proxy(proxy);
221    }
222
223    /// Removes a writer proxy.
224    pub fn remove_writer_proxy(&mut self, guid: Guid) -> Option<WriterProxy> {
225        self.inner.remove_writer_proxy(guid)
226    }
227
228    /// Processes an incoming DATA submessage and returns decoded
229    /// generic messages.
230    ///
231    /// # Errors
232    /// `BadArgument` if the encapsulation/CDR decode fails.
233    pub fn handle_data(
234        &mut self,
235        source_prefix: GuidPrefix,
236        data: &DataSubmessage,
237    ) -> SecurityResult<Vec<ParticipantGenericMessage>> {
238        let samples = self.inner.handle_data(source_prefix, data, None);
239        decode_samples(samples.into_iter().map(|s| s.payload))
240    }
241
242    /// Process DATA_FRAG.
243    ///
244    /// # Errors
245    /// see [`handle_data`](Self::handle_data).
246    pub fn handle_data_frag(
247        &mut self,
248        source_prefix: GuidPrefix,
249        df: &DataFragSubmessage,
250        now: Duration,
251    ) -> SecurityResult<Vec<ParticipantGenericMessage>> {
252        let samples = self.inner.handle_data_frag(source_prefix, df, now, None);
253        decode_samples(samples.into_iter().map(|s| s.payload))
254    }
255
256    /// Process GAP.
257    ///
258    /// # Errors
259    /// see [`handle_data`](Self::handle_data).
260    pub fn handle_gap(
261        &mut self,
262        source_prefix: GuidPrefix,
263        gap: &GapSubmessage,
264    ) -> SecurityResult<Vec<ParticipantGenericMessage>> {
265        let samples = self.inner.handle_gap(source_prefix, gap);
266        decode_samples(samples.into_iter().map(|s| s.payload))
267    }
268
269    /// HEARTBEAT verarbeiten.
270    pub fn handle_heartbeat(
271        &mut self,
272        source_prefix: GuidPrefix,
273        hb: &HeartbeatSubmessage,
274        now: Duration,
275    ) {
276        self.inner.handle_heartbeat(source_prefix, hb, now);
277    }
278
279    /// Tick (ACKNACK / NACK_FRAG outbound).
280    ///
281    /// # Errors
282    /// Wire encode errors.
283    pub fn tick_outbound(&mut self, now: Duration) -> Result<Vec<OutboundDatagram>, WireError> {
284        self.inner.tick_outbound(now)
285    }
286}
287
288fn decode_samples<B, I>(payloads: I) -> SecurityResult<Vec<ParticipantGenericMessage>>
289where
290    B: AsRef<[u8]>,
291    I: IntoIterator<Item = B>,
292{
293    let mut out = Vec::new();
294    for p in payloads {
295        // Propagate the original SecurityError directly — the detail string
296        // in the codec is already descriptive enough, no topic-specific
297        // wrapper needed (Spec §7.5.5 makes no per-topic distinction at the
298        // codec level).
299        out.push(decode_generic_message(p.as_ref())?);
300    }
301    Ok(out)
302}
303
304// Marker that both imports above are permanently used.
305const _: Option<SecurityErrorKind> = None;
306const _: Option<SecurityError> = None;
307
308#[cfg(test)]
309#[allow(clippy::expect_used, clippy::unwrap_used)]
310mod tests {
311    use super::*;
312    use zerodds_rtps::wire_types::Locator;
313    use zerodds_security::generic_message::{MessageIdentity, class_id};
314    use zerodds_security::token::DataHolder;
315
316    fn local_prefix() -> GuidPrefix {
317        GuidPrefix::from_bytes([1; 12])
318    }
319    fn remote_prefix() -> GuidPrefix {
320        GuidPrefix::from_bytes([2; 12])
321    }
322
323    fn sample_msg() -> ParticipantGenericMessage {
324        ParticipantGenericMessage {
325            message_identity: MessageIdentity {
326                source_guid: [0xAA; 16],
327                sequence_number: 1,
328            },
329            related_message_identity: MessageIdentity::default(),
330            destination_participant_key: [0xBB; 16],
331            destination_endpoint_key: [0; 16],
332            source_endpoint_key: [0xCC; 16],
333            message_class_id: class_id::PARTICIPANT_CRYPTO_TOKENS.into(),
334            message_data: alloc::vec![DataHolder::new("DDS:Crypto:AES-GCM-GMAC")],
335        }
336    }
337
338    #[test]
339    fn writer_has_expected_entity_id() {
340        let w = VolatileSecureMessageWriter::new(local_prefix(), VendorId::ZERODDS);
341        assert_eq!(
342            w.guid().entity_id,
343            EntityId::BUILTIN_PARTICIPANT_VOLATILE_MESSAGE_SECURE_WRITER
344        );
345    }
346
347    #[test]
348    fn reader_has_expected_entity_id() {
349        let r = VolatileSecureMessageReader::new(local_prefix(), VendorId::ZERODDS);
350        assert_eq!(
351            r.guid().entity_id,
352            EntityId::BUILTIN_PARTICIPANT_VOLATILE_MESSAGE_SECURE_READER
353        );
354    }
355
356    #[test]
357    fn writer_starts_with_zero_proxies() {
358        let w = VolatileSecureMessageWriter::new(local_prefix(), VendorId::ZERODDS);
359        assert_eq!(w.reader_proxy_count(), 0);
360    }
361
362    #[test]
363    fn reader_starts_with_zero_proxies() {
364        let r = VolatileSecureMessageReader::new(local_prefix(), VendorId::ZERODDS);
365        assert_eq!(r.writer_proxy_count(), 0);
366    }
367
368    #[test]
369    fn write_without_proxies_returns_empty_datagrams() {
370        let mut w = VolatileSecureMessageWriter::new(local_prefix(), VendorId::ZERODDS);
371        let dgs = w.write(&sample_msg()).unwrap();
372        assert!(dgs.is_empty());
373    }
374
375    #[test]
376    fn write_with_one_proxy_produces_one_datagram() {
377        let mut w = VolatileSecureMessageWriter::new(local_prefix(), VendorId::ZERODDS);
378        let remote = Guid::new(
379            remote_prefix(),
380            EntityId::BUILTIN_PARTICIPANT_VOLATILE_MESSAGE_SECURE_READER,
381        );
382        w.add_reader_proxy(ReaderProxy::new(
383            remote,
384            alloc::vec![Locator::udp_v4([127, 0, 0, 1], 7411)],
385            alloc::vec![],
386            true,
387        ));
388        let dgs = w.write(&sample_msg()).unwrap();
389        assert_eq!(dgs.len(), 1);
390    }
391
392    #[test]
393    fn write_with_heartbeat_piggybacks_heartbeat_submessage() {
394        use zerodds_rtps::datagram::{ParsedSubmessage, decode_datagram};
395        let mut w = VolatileSecureMessageWriter::new(local_prefix(), VendorId::ZERODDS);
396        let remote = Guid::new(
397            remote_prefix(),
398            EntityId::BUILTIN_PARTICIPANT_VOLATILE_MESSAGE_SECURE_READER,
399        );
400        w.add_reader_proxy(ReaderProxy::new(
401            remote,
402            alloc::vec![Locator::udp_v4([127, 0, 0, 1], 7411)],
403            alloc::vec![],
404            true,
405        ));
406        // §8.4.15.5: token delivery must be prompt. A plain `write()`
407        // makes the reader wait until the periodic HEARTBEAT (250ms) before
408        // it NACKs a loss — cross-vendor the endpoint crypto token thereby
409        // falls behind Cyclone's AUTHOK match. A piggyback HEARTBEAT
410        // directly with the DATA solves this per-send (robust for late joiners).
411        let dgs = w
412            .write_with_heartbeat(&sample_msg(), Duration::from_millis(0))
413            .unwrap();
414        assert_eq!(
415            dgs.len(),
416            1,
417            "DATA + piggyback HEARTBEAT share one datagram"
418        );
419        let parsed = decode_datagram(&dgs[0].bytes).expect("decode");
420        assert!(
421            parsed
422                .submessages
423                .iter()
424                .any(|s| matches!(s, ParsedSubmessage::Data(_))),
425            "DATA submessage missing"
426        );
427        assert!(
428            parsed
429                .submessages
430                .iter()
431                .any(|s| matches!(s, ParsedSubmessage::Heartbeat(_))),
432            "piggyback HEARTBEAT missing — Volatile token would otherwise wait 250ms for periodic HB"
433        );
434    }
435
436    #[test]
437    fn add_remove_reader_proxy_roundtrip() {
438        let mut w = VolatileSecureMessageWriter::new(local_prefix(), VendorId::ZERODDS);
439        let remote = Guid::new(
440            remote_prefix(),
441            EntityId::BUILTIN_PARTICIPANT_VOLATILE_MESSAGE_SECURE_READER,
442        );
443        w.add_reader_proxy(ReaderProxy::new(remote, alloc::vec![], alloc::vec![], true));
444        assert_eq!(w.reader_proxy_count(), 1);
445        assert!(w.remove_reader_proxy(remote).is_some());
446        assert_eq!(w.reader_proxy_count(), 0);
447    }
448
449    #[test]
450    fn add_remove_writer_proxy_roundtrip() {
451        let mut r = VolatileSecureMessageReader::new(local_prefix(), VendorId::ZERODDS);
452        let remote = Guid::new(
453            remote_prefix(),
454            EntityId::BUILTIN_PARTICIPANT_VOLATILE_MESSAGE_SECURE_WRITER,
455        );
456        r.add_writer_proxy(WriterProxy::new(
457            remote,
458            alloc::vec![Locator::udp_v4([127, 0, 0, 1], 7411)],
459            alloc::vec![],
460            true,
461        ));
462        assert_eq!(r.writer_proxy_count(), 1);
463        assert!(r.remove_writer_proxy(remote).is_some());
464        assert_eq!(r.writer_proxy_count(), 0);
465    }
466
467    #[test]
468    fn reader_decodes_data_with_known_writer() {
469        let mut r = VolatileSecureMessageReader::new(local_prefix(), VendorId::ZERODDS);
470        let remote = Guid::new(
471            remote_prefix(),
472            EntityId::BUILTIN_PARTICIPANT_VOLATILE_MESSAGE_SECURE_WRITER,
473        );
474        r.add_writer_proxy(WriterProxy::new(
475            remote,
476            alloc::vec![Locator::udp_v4([127, 0, 0, 1], 7411)],
477            alloc::vec![],
478            true,
479        ));
480        let msg = sample_msg();
481        let payload = encode_generic_message(&msg);
482        let data = DataSubmessage {
483            extra_flags: 0,
484            reader_id: EntityId::BUILTIN_PARTICIPANT_VOLATILE_MESSAGE_SECURE_READER,
485            writer_id: EntityId::BUILTIN_PARTICIPANT_VOLATILE_MESSAGE_SECURE_WRITER,
486            writer_sn: SequenceNumber(1),
487            inline_qos: None,
488            key_flag: false,
489            non_standard_flag: false,
490            serialized_payload: payload.into(),
491        };
492        let out = r.handle_data(remote_prefix(), &data).unwrap();
493        assert_eq!(out.len(), 1);
494        assert_eq!(out[0], msg);
495    }
496
497    #[test]
498    fn reader_drops_data_from_unknown_writer() {
499        let mut r = VolatileSecureMessageReader::new(local_prefix(), VendorId::ZERODDS);
500        // No writer proxy registered → handle_data must return empty
501        let msg = sample_msg();
502        let payload = encode_generic_message(&msg);
503        let data = DataSubmessage {
504            extra_flags: 0,
505            reader_id: EntityId::BUILTIN_PARTICIPANT_VOLATILE_MESSAGE_SECURE_READER,
506            writer_id: EntityId::BUILTIN_PARTICIPANT_VOLATILE_MESSAGE_SECURE_WRITER,
507            writer_sn: SequenceNumber(1),
508            inline_qos: None,
509            key_flag: false,
510            non_standard_flag: false,
511            serialized_payload: payload.into(),
512        };
513        let out = r.handle_data(remote_prefix(), &data).unwrap();
514        assert!(out.is_empty());
515    }
516
517    #[test]
518    fn reader_rejects_corrupt_payload() {
519        let mut r = VolatileSecureMessageReader::new(local_prefix(), VendorId::ZERODDS);
520        let remote = Guid::new(
521            remote_prefix(),
522            EntityId::BUILTIN_PARTICIPANT_VOLATILE_MESSAGE_SECURE_WRITER,
523        );
524        r.add_writer_proxy(WriterProxy::new(remote, alloc::vec![], alloc::vec![], true));
525        let data = DataSubmessage {
526            extra_flags: 0,
527            reader_id: EntityId::BUILTIN_PARTICIPANT_VOLATILE_MESSAGE_SECURE_READER,
528            writer_id: EntityId::BUILTIN_PARTICIPANT_VOLATILE_MESSAGE_SECURE_WRITER,
529            writer_sn: SequenceNumber(1),
530            inline_qos: None,
531            key_flag: false,
532            non_standard_flag: false,
533            serialized_payload: alloc::vec![0x00, 0x99, 0, 0].into(),
534        };
535        let err = r.handle_data(remote_prefix(), &data).unwrap_err();
536        assert_eq!(err.kind, SecurityErrorKind::BadArgument);
537    }
538}
zerodds_discovery/security/volatile_secure.rs

zerodds_discovery/security/
volatile_secure.rs
