zerodds_discovery/security/
stateless.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright 2026 ZeroDDS Contributors
3//! Builtin endpoint `DCPSParticipantStatelessMessage` — DDS-Security 1.2
4//! §7.4.4 + §10.3.4.
5//!
6//! Wire profile:
7//! - Reliability: BestEffort (Spec §7.5.3 — "stateless").
8//! - Durability:  Volatile.
9//! - Topic type:  `ParticipantGenericMessage` (Spec §7.5.5), encoded with
10//!   a 4-byte CDR-LE encapsulation header + XCDR1 body via
11//!   [`security_runtime::builtin_topics::encode_generic_message`].
12//! - EntityIds:   `BUILTIN_PARTICIPANT_STATELESS_MESSAGE_{WRITER,READER}`.
13//!
14//! We do **not** use [`zerodds_rtps::ReliableWriter`], because stateless
15//! has no receive status, no HEARTBEATs and no AckNack loop per spec.
16//! Instead: a simple multi-reader fan-out list; each `write()` produces
17//! one DATA datagram per [`ReaderProxy`].
18//!
19//! C3.4-c scope: no plugin pipeline logic in the reader. The reader
20//! decodes the `ParticipantGenericMessage` and passes it to the caller —
21//! the auth plugin hook (Spec §10.3.4.1) is installed on top by the DCPS
22//! layer.
23
24extern crate alloc;
25
26use alloc::rc::Rc;
27use alloc::vec::Vec;
28
29use zerodds_rtps::datagram::{ParsedSubmessage, decode_datagram, encode_data_datagram};
30use zerodds_rtps::error::WireError;
31use zerodds_rtps::fragment_assembler::{AssemblerCaps, FragmentAssembler};
32use zerodds_rtps::header::RtpsHeader;
33use zerodds_rtps::message_builder::OutboundDatagram;
34use zerodds_rtps::reader_proxy::ReaderProxy;
35use zerodds_rtps::submessages::{DataFragSubmessage, DataSubmessage};
36use zerodds_rtps::wire_types::{EntityId, Guid, GuidPrefix, SequenceNumber, VendorId};
37use zerodds_rtps::writer_proxy::WriterProxy;
38
39use zerodds_security::error::{SecurityError, SecurityErrorKind, SecurityResult};
40use zerodds_security::generic_message::ParticipantGenericMessage;
41
42use crate::security::codec::{decode_generic_message, encode_generic_message};
43
44/// Stateless message writer (Spec §7.4.4 + §10.3.4).
45///
46/// Maintains multi-reader fan-out for the
47/// `BUILTIN_PARTICIPANT_STATELESS_MESSAGE_WRITER` endpoint. Each
48/// `write()` produces one [`OutboundDatagram`] per registered
49/// [`ReaderProxy`] — no cache, no resend, no HEARTBEAT
50/// (stateless = no receive status).
51#[derive(Debug)]
52pub struct StatelessMessageWriter {
53    guid: Guid,
54    vendor_id: VendorId,
55    next_sn: i64,
56    reader_proxies: Vec<ReaderProxy>,
57}
58
59impl StatelessMessageWriter {
60    /// Creates a writer for the local participant.
61    #[must_use]
62    pub fn new(participant_prefix: GuidPrefix, vendor_id: VendorId) -> Self {
63        Self {
64            guid: Guid::new(
65                participant_prefix,
66                EntityId::BUILTIN_PARTICIPANT_STATELESS_MESSAGE_WRITER,
67            ),
68            vendor_id,
69            next_sn: 1,
70            reader_proxies: Vec::new(),
71        }
72    }
73
74    /// GUID of the writer.
75    #[must_use]
76    pub fn guid(&self) -> Guid {
77        self.guid
78    }
79
80    /// Read-only slice of the registered reader proxies.
81    #[must_use]
82    pub fn reader_proxies(&self) -> &[ReaderProxy] {
83        &self.reader_proxies
84    }
85
86    /// Number of registered reader proxies.
87    #[must_use]
88    pub fn reader_proxy_count(&self) -> usize {
89        self.reader_proxies.len()
90    }
91
92    /// Adds a reader proxy (idempotent: same GUID overwrites).
93    pub fn add_reader_proxy(&mut self, proxy: ReaderProxy) {
94        let guid = proxy.remote_reader_guid;
95        if let Some(idx) = self
96            .reader_proxies
97            .iter()
98            .position(|p| p.remote_reader_guid == guid)
99        {
100            self.reader_proxies[idx] = proxy;
101        } else {
102            self.reader_proxies.push(proxy);
103        }
104    }
105
106    /// Removes a reader proxy. Returns it if present.
107    pub fn remove_reader_proxy(&mut self, guid: Guid) -> Option<ReaderProxy> {
108        let idx = self
109            .reader_proxies
110            .iter()
111            .position(|p| p.remote_reader_guid == guid)?;
112        Some(self.reader_proxies.remove(idx))
113    }
114
115    /// Sends a `ParticipantGenericMessage` to all reader proxies.
116    ///
117    /// Returns one datagram per proxy (or empty if none are registered).
118    ///
119    /// # Errors
120    /// `WireError::ValueOutOfRange` on sequence-number overflow or
121    /// `WireError::*` from the DATA encoding pipeline.
122    pub fn write(
123        &mut self,
124        msg: &ParticipantGenericMessage,
125    ) -> Result<Vec<OutboundDatagram>, WireError> {
126        if self.reader_proxies.is_empty() {
127            return Ok(Vec::new());
128        }
129        let payload = encode_generic_message(msg);
130        let sn = SequenceNumber(self.next_sn);
131        self.next_sn = self
132            .next_sn
133            .checked_add(1)
134            .ok_or(WireError::ValueOutOfRange {
135                message: "stateless writer sequence overflow",
136            })?;
137
138        let mut out = Vec::with_capacity(self.reader_proxies.len());
139        for proxy in &self.reader_proxies {
140            let data = DataSubmessage {
141                extra_flags: 0,
142                reader_id: proxy.remote_reader_guid.entity_id,
143                writer_id: self.guid.entity_id,
144                writer_sn: sn,
145                inline_qos: None,
146                key_flag: false,
147                non_standard_flag: false,
148                serialized_payload: payload.clone().into(),
149            };
150            let header = RtpsHeader::new(self.vendor_id, self.guid.prefix);
151            let bytes = encode_data_datagram(header, &[data])?;
152            // Target = unicast locators of the proxy (we ignore multicast
153            // for stateless: the auth handshake is always point-to-point).
154            let targets = Rc::new(proxy.unicast_locators.clone());
155            out.push(OutboundDatagram { bytes, targets });
156        }
157        Ok(out)
158    }
159}
160
161/// Stateless message reader (Spec §7.4.4 + §10.3.4).
162///
163/// Decodes incoming DATA submessages addressed to the
164/// `BUILTIN_PARTICIPANT_STATELESS_MESSAGE_READER`.
165/// Stateless: no history cache, no ACKNACK, no heartbeat state.
166/// A known writer proxy is optional for source-authenticity checks —
167/// the reader always delivers the message, the auth plugin hook decides
168/// based on the `source_guid` field.
169#[derive(Debug)]
170pub struct StatelessMessageReader {
171    guid: Guid,
172    #[allow(dead_code)]
173    vendor_id: VendorId,
174    /// Known writer proxies. The reader does not use them for filtering
175    /// (stateless accepts from any writer), but for caller diagnostics
176    /// (`writer_proxy_count`).
177    writer_proxies: Vec<WriterProxy>,
178    /// Fragment reassembly for LARGE stateless messages: cyclone/FastDDS
179    /// RTPS-fragment the HandshakeReply/Final (DATA_FRAG), because with
180    /// c.id-PEM + c.perm-p7s + c.pdata it easily exceeds the MTU.
181    frag: FragmentAssembler,
182}
183
184impl StatelessMessageReader {
185    /// Creates a reader for the local participant.
186    #[must_use]
187    pub fn new(participant_prefix: GuidPrefix, vendor_id: VendorId) -> Self {
188        Self {
189            guid: Guid::new(
190                participant_prefix,
191                EntityId::BUILTIN_PARTICIPANT_STATELESS_MESSAGE_READER,
192            ),
193            vendor_id,
194            writer_proxies: Vec::new(),
195            frag: FragmentAssembler::new(AssemblerCaps::default()),
196        }
197    }
198
199    /// GUID of the reader.
200    #[must_use]
201    pub fn guid(&self) -> Guid {
202        self.guid
203    }
204
205    /// Number of registered writer proxies.
206    #[must_use]
207    pub fn writer_proxy_count(&self) -> usize {
208        self.writer_proxies.len()
209    }
210
211    /// Read-only slice of the registered writer proxies.
212    #[must_use]
213    pub fn writer_proxies(&self) -> &[WriterProxy] {
214        &self.writer_proxies
215    }
216
217    /// Adds a writer proxy (idempotent).
218    pub fn add_writer_proxy(&mut self, proxy: WriterProxy) {
219        let guid = proxy.remote_writer_guid;
220        if let Some(idx) = self
221            .writer_proxies
222            .iter()
223            .position(|p| p.remote_writer_guid == guid)
224        {
225            self.writer_proxies[idx] = proxy;
226        } else {
227            self.writer_proxies.push(proxy);
228        }
229    }
230
231    /// Removes a writer proxy.
232    pub fn remove_writer_proxy(&mut self, guid: Guid) -> Option<WriterProxy> {
233        let idx = self
234            .writer_proxies
235            .iter()
236            .position(|p| p.remote_writer_guid == guid)?;
237        Some(self.writer_proxies.remove(idx))
238    }
239
240    /// Processes an incoming DATA submessage and decodes it into a
241    /// `ParticipantGenericMessage`.
242    ///
243    /// # Errors
244    /// `BadArgument` if the encapsulation/CDR decode fails.
245    pub fn handle_data(
246        &mut self,
247        data: &DataSubmessage,
248    ) -> SecurityResult<ParticipantGenericMessage> {
249        decode_generic_message(&data.serialized_payload)
250    }
251
252    /// Processes an incoming DATA_FRAG submessage. Large stateless
253    /// messages (HandshakeReply/Final with cert/permissions) are
254    /// RTPS-fragmented cross-vendor. Returns the decoded
255    /// `ParticipantGenericMessage` once all fragments are present
256    /// (otherwise empty — best-effort, no NACK).
257    ///
258    /// # Errors
259    /// `BadArgument` if the reassembled generic-message body does not parse.
260    pub fn handle_data_frag(
261        &mut self,
262        df: &DataFragSubmessage,
263    ) -> SecurityResult<Vec<ParticipantGenericMessage>> {
264        match self.frag.insert(df) {
265            Some(completed) => Ok(alloc::vec![decode_generic_message(&completed.payload)?]),
266            None => Ok(Vec::new()),
267        }
268    }
269
270    /// Processes a complete RTPS datagram. Returns all decoded stateless
271    /// messages from this datagram.
272    ///
273    /// # Errors
274    /// - `BadArgument` if the datagram does not parse (wire decoder
275    ///   error) or a relevant DATA submessage has a corrupt
276    ///   generic-message body.
277    pub fn handle_datagram(
278        &mut self,
279        datagram: &[u8],
280    ) -> SecurityResult<Vec<ParticipantGenericMessage>> {
281        let parsed = decode_datagram(datagram).map_err(|_| {
282            SecurityError::new(
283                SecurityErrorKind::BadArgument,
284                "stateless reader: wire decode failed",
285            )
286        })?;
287        let mut out = Vec::new();
288        for sub in parsed.submessages {
289            if let ParsedSubmessage::Data(d) = sub {
290                if d.reader_id == self.guid.entity_id
291                    || d.writer_id == EntityId::BUILTIN_PARTICIPANT_STATELESS_MESSAGE_WRITER
292                {
293                    out.push(decode_generic_message(&d.serialized_payload)?);
294                }
295            }
296        }
297        Ok(out)
298    }
299}
300
301#[cfg(test)]
302#[allow(
303    clippy::expect_used,
304    clippy::unwrap_used,
305    clippy::panic,
306    clippy::unreachable
307)]
308mod tests {
309    use super::*;
310    use zerodds_rtps::wire_types::Locator;
311    use zerodds_security::generic_message::{MessageIdentity, class_id};
312    use zerodds_security::token::DataHolder;
313
314    fn sample_msg(seq: i64) -> ParticipantGenericMessage {
315        ParticipantGenericMessage {
316            message_identity: MessageIdentity {
317                source_guid: [0xAA; 16],
318                sequence_number: seq,
319            },
320            related_message_identity: MessageIdentity::default(),
321            destination_participant_key: [0xBB; 16],
322            destination_endpoint_key: [0; 16],
323            source_endpoint_key: [0xCC; 16],
324            message_class_id: class_id::AUTH_REQUEST.into(),
325            message_data: alloc::vec![DataHolder::new("DDS:Auth:PKI-DH:1.2+AuthReq")],
326        }
327    }
328
329    fn local_prefix() -> GuidPrefix {
330        GuidPrefix::from_bytes([1; 12])
331    }
332
333    fn remote_prefix() -> GuidPrefix {
334        GuidPrefix::from_bytes([2; 12])
335    }
336
337    #[test]
338    fn writer_has_expected_entity_id() {
339        let w = StatelessMessageWriter::new(local_prefix(), VendorId::ZERODDS);
340        assert_eq!(
341            w.guid().entity_id,
342            EntityId::BUILTIN_PARTICIPANT_STATELESS_MESSAGE_WRITER
343        );
344        assert_eq!(w.guid().prefix, local_prefix());
345    }
346
347    #[test]
348    fn reader_has_expected_entity_id() {
349        let r = StatelessMessageReader::new(local_prefix(), VendorId::ZERODDS);
350        assert_eq!(
351            r.guid().entity_id,
352            EntityId::BUILTIN_PARTICIPANT_STATELESS_MESSAGE_READER
353        );
354    }
355
356    #[test]
357    fn write_without_proxies_returns_empty() {
358        let mut w = StatelessMessageWriter::new(local_prefix(), VendorId::ZERODDS);
359        let dgs = w.write(&sample_msg(1)).unwrap();
360        assert!(dgs.is_empty(), "no proxies → no fan-out");
361    }
362
363    #[test]
364    fn write_to_one_proxy_produces_one_datagram() {
365        let mut w = StatelessMessageWriter::new(local_prefix(), VendorId::ZERODDS);
366        let remote = Guid::new(
367            remote_prefix(),
368            EntityId::BUILTIN_PARTICIPANT_STATELESS_MESSAGE_READER,
369        );
370        w.add_reader_proxy(ReaderProxy::new(
371            remote,
372            alloc::vec![Locator::udp_v4([127, 0, 0, 1], 7411)],
373            alloc::vec![],
374            false,
375        ));
376        let dgs = w.write(&sample_msg(1)).unwrap();
377        assert_eq!(dgs.len(), 1);
378        assert_eq!(dgs[0].targets.len(), 1);
379    }
380
381    #[test]
382    fn write_to_two_proxies_produces_two_datagrams() {
383        let mut w = StatelessMessageWriter::new(local_prefix(), VendorId::ZERODDS);
384        let remote_a = Guid::new(
385            GuidPrefix::from_bytes([2; 12]),
386            EntityId::BUILTIN_PARTICIPANT_STATELESS_MESSAGE_READER,
387        );
388        let remote_b = Guid::new(
389            GuidPrefix::from_bytes([3; 12]),
390            EntityId::BUILTIN_PARTICIPANT_STATELESS_MESSAGE_READER,
391        );
392        w.add_reader_proxy(ReaderProxy::new(
393            remote_a,
394            alloc::vec![Locator::udp_v4([127, 0, 0, 1], 7411)],
395            alloc::vec![],
396            false,
397        ));
398        w.add_reader_proxy(ReaderProxy::new(
399            remote_b,
400            alloc::vec![Locator::udp_v4([127, 0, 0, 1], 7412)],
401            alloc::vec![],
402            false,
403        ));
404        assert_eq!(w.reader_proxy_count(), 2);
405        let dgs = w.write(&sample_msg(1)).unwrap();
406        assert_eq!(dgs.len(), 2);
407    }
408
409    #[test]
410    fn add_reader_proxy_is_idempotent() {
411        let mut w = StatelessMessageWriter::new(local_prefix(), VendorId::ZERODDS);
412        let remote = Guid::new(
413            remote_prefix(),
414            EntityId::BUILTIN_PARTICIPANT_STATELESS_MESSAGE_READER,
415        );
416        w.add_reader_proxy(ReaderProxy::new(
417            remote,
418            alloc::vec![Locator::udp_v4([127, 0, 0, 1], 7411)],
419            alloc::vec![],
420            false,
421        ));
422        w.add_reader_proxy(ReaderProxy::new(
423            remote,
424            alloc::vec![Locator::udp_v4([127, 0, 0, 1], 7411)],
425            alloc::vec![],
426            false,
427        ));
428        assert_eq!(w.reader_proxy_count(), 1);
429    }
430
431    #[test]
432    fn remove_reader_proxy_returns_proxy() {
433        let mut w = StatelessMessageWriter::new(local_prefix(), VendorId::ZERODDS);
434        let remote = Guid::new(
435            remote_prefix(),
436            EntityId::BUILTIN_PARTICIPANT_STATELESS_MESSAGE_READER,
437        );
438        w.add_reader_proxy(ReaderProxy::new(
439            remote,
440            alloc::vec![],
441            alloc::vec![],
442            false,
443        ));
444        let removed = w.remove_reader_proxy(remote);
445        assert!(removed.is_some());
446        assert_eq!(w.reader_proxy_count(), 0);
447        assert!(w.remove_reader_proxy(remote).is_none());
448    }
449
450    #[test]
451    fn write_increments_sequence_number() {
452        let mut w = StatelessMessageWriter::new(local_prefix(), VendorId::ZERODDS);
453        let remote = Guid::new(
454            remote_prefix(),
455            EntityId::BUILTIN_PARTICIPANT_STATELESS_MESSAGE_READER,
456        );
457        w.add_reader_proxy(ReaderProxy::new(
458            remote,
459            alloc::vec![Locator::udp_v4([127, 0, 0, 1], 7411)],
460            alloc::vec![],
461            false,
462        ));
463        let dg1 = w.write(&sample_msg(1)).unwrap()[0].clone();
464        let dg2 = w.write(&sample_msg(2)).unwrap()[0].clone();
465        // Decode the SN from the wire bytes
466        let p1 = decode_datagram(&dg1.bytes).unwrap();
467        let p2 = decode_datagram(&dg2.bytes).unwrap();
468        let sn1 = match &p1.submessages[0] {
469            ParsedSubmessage::Data(d) => d.writer_sn,
470            _ => unreachable!(),
471        };
472        let sn2 = match &p2.submessages[0] {
473            ParsedSubmessage::Data(d) => d.writer_sn,
474            _ => unreachable!(),
475        };
476        assert_eq!(sn1, SequenceNumber(1));
477        assert_eq!(sn2, SequenceNumber(2));
478    }
479
480    #[test]
481    fn write_carries_writer_entity_id_on_wire() {
482        let mut w = StatelessMessageWriter::new(local_prefix(), VendorId::ZERODDS);
483        let remote = Guid::new(
484            remote_prefix(),
485            EntityId::BUILTIN_PARTICIPANT_STATELESS_MESSAGE_READER,
486        );
487        w.add_reader_proxy(ReaderProxy::new(
488            remote,
489            alloc::vec![Locator::udp_v4([127, 0, 0, 1], 7411)],
490            alloc::vec![],
491            false,
492        ));
493        let dgs = w.write(&sample_msg(1)).unwrap();
494        let parsed = decode_datagram(&dgs[0].bytes).unwrap();
495        match &parsed.submessages[0] {
496            ParsedSubmessage::Data(d) => {
497                assert_eq!(
498                    d.writer_id,
499                    EntityId::BUILTIN_PARTICIPANT_STATELESS_MESSAGE_WRITER
500                );
501                assert_eq!(
502                    d.reader_id,
503                    EntityId::BUILTIN_PARTICIPANT_STATELESS_MESSAGE_READER
504                );
505            }
506            _ => panic!("expected DATA"),
507        }
508    }
509
510    #[test]
511    fn reader_handle_data_decodes_generic_message() {
512        let mut r = StatelessMessageReader::new(local_prefix(), VendorId::ZERODDS);
513        let msg = sample_msg(42);
514        let payload = encode_generic_message(&msg);
515        let data = DataSubmessage {
516            extra_flags: 0,
517            reader_id: EntityId::BUILTIN_PARTICIPANT_STATELESS_MESSAGE_READER,
518            writer_id: EntityId::BUILTIN_PARTICIPANT_STATELESS_MESSAGE_WRITER,
519            writer_sn: SequenceNumber(1),
520            inline_qos: None,
521            key_flag: false,
522            non_standard_flag: false,
523            serialized_payload: payload.into(),
524        };
525        let decoded = r.handle_data(&data).unwrap();
526        assert_eq!(decoded, msg);
527    }
528
529    #[test]
530    fn reader_handle_data_rejects_corrupt_payload() {
531        let mut r = StatelessMessageReader::new(local_prefix(), VendorId::ZERODDS);
532        let data = DataSubmessage {
533            extra_flags: 0,
534            reader_id: EntityId::BUILTIN_PARTICIPANT_STATELESS_MESSAGE_READER,
535            writer_id: EntityId::BUILTIN_PARTICIPANT_STATELESS_MESSAGE_WRITER,
536            writer_sn: SequenceNumber(1),
537            inline_qos: None,
538            key_flag: false,
539            non_standard_flag: false,
540            serialized_payload: alloc::vec![0x00, 0x99, 0, 0].into(),
541        };
542        let err = r.handle_data(&data).unwrap_err();
543        assert_eq!(err.kind, SecurityErrorKind::BadArgument);
544    }
545
546    #[test]
547    fn reader_writer_proxy_management() {
548        let mut r = StatelessMessageReader::new(local_prefix(), VendorId::ZERODDS);
549        let remote = Guid::new(
550            remote_prefix(),
551            EntityId::BUILTIN_PARTICIPANT_STATELESS_MESSAGE_WRITER,
552        );
553        r.add_writer_proxy(WriterProxy::new(
554            remote,
555            alloc::vec![Locator::udp_v4([127, 0, 0, 1], 7411)],
556            alloc::vec![],
557            false,
558        ));
559        // Idempotency
560        r.add_writer_proxy(WriterProxy::new(
561            remote,
562            alloc::vec![],
563            alloc::vec![],
564            false,
565        ));
566        assert_eq!(r.writer_proxy_count(), 1);
567        assert!(r.remove_writer_proxy(remote).is_some());
568        assert_eq!(r.writer_proxy_count(), 0);
569    }
570
571    #[test]
572    fn end_to_end_writer_to_reader_loopback() {
573        // Writer builds the datagram, reader decodes it back.
574        let mut w = StatelessMessageWriter::new(local_prefix(), VendorId::ZERODDS);
575        let mut r = StatelessMessageReader::new(remote_prefix(), VendorId::ZERODDS);
576        let remote_reader_guid = Guid::new(
577            remote_prefix(),
578            EntityId::BUILTIN_PARTICIPANT_STATELESS_MESSAGE_READER,
579        );
580        w.add_reader_proxy(ReaderProxy::new(
581            remote_reader_guid,
582            alloc::vec![Locator::udp_v4([127, 0, 0, 1], 7411)],
583            alloc::vec![],
584            false,
585        ));
586        let msg = sample_msg(7);
587        let dgs = w.write(&msg).unwrap();
588        let decoded = r.handle_datagram(&dgs[0].bytes).unwrap();
589        assert_eq!(decoded.len(), 1);
590        assert_eq!(decoded[0], msg);
591    }
592
593    #[test]
594    fn reader_handle_datagram_rejects_invalid_magic() {
595        let mut r = StatelessMessageReader::new(local_prefix(), VendorId::ZERODDS);
596        let err = r.handle_datagram(&[0u8; 24]).unwrap_err();
597        assert_eq!(err.kind, SecurityErrorKind::BadArgument);
598    }
599}
zerodds_discovery/security/stateless.rs

zerodds_discovery/security/
stateless.rs
