xmtp_mls/messages/
decoded_message.rs

1use crate::groups::GroupError;
2use prost::Message;
3use xmtp_content_types::actions::{Actions, ActionsCodec};
4use xmtp_content_types::group_updated::GroupUpdatedCodec;
5use xmtp_content_types::intent::{Intent, IntentCodec};
6use xmtp_content_types::leave_request::LeaveRequestCodec;
7use xmtp_content_types::multi_remote_attachment::MultiRemoteAttachmentCodec;
8use xmtp_content_types::reaction::{LegacyReactionCodec, ReactionCodec};
9use xmtp_content_types::read_receipt::ReadReceiptCodec;
10use xmtp_content_types::remote_attachment::RemoteAttachmentCodec;
11use xmtp_content_types::reply::ReplyCodec;
12use xmtp_content_types::transaction_reference::TransactionReferenceCodec;
13use xmtp_content_types::wallet_send_calls::{WalletSendCalls, WalletSendCallsCodec};
14use xmtp_content_types::{CodecError, ContentCodec, compression};
15use xmtp_content_types::{
16    attachment::{Attachment, AttachmentCodec},
17    markdown::MarkdownCodec,
18    read_receipt::ReadReceipt,
19    remote_attachment::RemoteAttachment,
20    text::TextCodec,
21    transaction_reference::TransactionReference,
22};
23use xmtp_db::group_message::StoredGroupMessage;
24use xmtp_db::group_message::{DeliveryStatus, GroupMessageKind};
25use xmtp_proto::types::GroupId;
26use xmtp_proto::xmtp::mls::message_contents::{
27    ContentTypeId, EncodedContent, GroupUpdated,
28    content_types::{LeaveRequest, MultiRemoteAttachment, ReactionV2},
29};
30
31#[derive(Debug, Clone)]
32pub struct Reply {
33    // The original message that this reply is in reply to.
34    // This goes at most one level deep from the original message, and won't happen recursively if there are replies to replies to replies
35    pub in_reply_to: Option<Box<DecodedMessage>>,
36    pub content: Box<MessageBody>,
37    pub reference_id: String,
38}
39
40// Wrap text content in a struct to be consistent with other content types
41#[derive(Debug, Clone)]
42pub struct Text {
43    pub content: String,
44}
45
46// Wrap markdown content in a struct to be consistent with other content types
47#[derive(Debug, Clone)]
48pub struct Markdown {
49    pub content: String,
50}
51
52#[derive(Debug, Clone, PartialEq, Eq)]
53pub enum DeletedBy {
54    /// Deleted by the original sender
55    Sender,
56    /// Deleted by a super admin
57    Admin(String), // inbox_id of the admin who deleted the message
58}
59
60/// Why received content could not be decoded.
61#[derive(Debug, Clone, Copy, PartialEq, Eq)]
62pub enum ContentDecodeFailureKind {
63    /// The bytes are not a typed `EncodedContent`: protobuf parsing failed,
64    /// or the content type is absent or incomplete.
65    MalformedEnvelope,
66    /// A complete typed envelope whose compression, payload, or nested
67    /// content failed to decode.
68    CodecDecodeFailed,
69}
70
71/// The typed cause kept beside undecodable content.
72#[derive(Debug, Clone, PartialEq, Eq)]
73pub struct ContentDecodeFailure {
74    pub kind: ContentDecodeFailureKind,
75    pub message: String,
76}
77
78impl ContentDecodeFailure {
79    fn malformed(message: impl Into<String>) -> Self {
80        Self {
81            kind: ContentDecodeFailureKind::MalformedEnvelope,
82            message: message.into(),
83        }
84    }
85
86    fn codec(message: impl ToString) -> Self {
87        Self {
88            kind: ContentDecodeFailureKind::CodecDecodeFailed,
89            message: message.to_string(),
90        }
91    }
92}
93
94impl DeletedBy {
95    /// Who deleted a message: its sender, or the admin with inbox
96    /// `deleter_inbox_id`.
97    pub(crate) fn new(deleter_inbox_id: &str, sender_inbox_id: &str) -> Self {
98        if deleter_inbox_id == sender_inbox_id {
99            DeletedBy::Sender
100        } else {
101            DeletedBy::Admin(deleter_inbox_id.to_string())
102        }
103    }
104}
105
106impl From<ContentDecodeFailure> for CodecError {
107    fn from(failure: ContentDecodeFailure) -> Self {
108        match failure.kind {
109            ContentDecodeFailureKind::MalformedEnvelope => CodecError::InvalidContentType,
110            ContentDecodeFailureKind::CodecDecodeFailed => CodecError::Decode(failure.message),
111        }
112    }
113}
114
115/// A complete typed envelope that no core codec decodes. A host codec may.
116#[derive(Debug, Clone, PartialEq)]
117pub struct CustomContent {
118    /// The decompressed envelope: the input for a host codec.
119    pub encoded: EncodedContent,
120    /// The exact received serialization of this envelope, before decompression.
121    pub raw_bytes: Vec<u8>,
122}
123
124/// Received content the client could not decode, kept with its exact bytes.
125#[derive(Debug, Clone, PartialEq)]
126pub struct UndecodableContent {
127    /// The parsed envelope, when the bytes parsed as protobuf. This is
128    /// evidence, not codec input: `content` is still compressed when
129    /// `compression` is set, and its type can be incomplete.
130    pub encoded: Option<EncodedContent>,
131    /// The exact received bytes.
132    pub raw_bytes: Vec<u8>,
133    pub failure: ContentDecodeFailure,
134}
135
136#[derive(Debug, Clone)]
137pub enum MessageBody {
138    Text(Text),
139    Markdown(Markdown),
140    Reply(Reply),
141    Reaction(ReactionV2),
142    Attachment(Attachment),
143    RemoteAttachment(RemoteAttachment),
144    MultiRemoteAttachment(MultiRemoteAttachment),
145    TransactionReference(TransactionReference),
146    GroupUpdated(GroupUpdated),
147    ReadReceipt(ReadReceipt),
148    WalletSendCalls(WalletSendCalls),
149    Intent(Option<Intent>),
150    Actions(Option<Actions>),
151    LeaveRequest(LeaveRequest),
152    /// Placeholder for a message that has been deleted (shown in message lists)
153    DeletedMessage {
154        deleted_by: DeletedBy,
155    },
156    /// A complete typed envelope with no core codec. A nested custom body
157    /// keeps its own exact bytes.
158    Custom(CustomContent),
159    /// Content that failed to decode, with its exact bytes and typed cause.
160    /// A nested failure makes the outer body undecodable, so this variant
161    /// appears only at the top level.
162    Undecodable(UndecodableContent),
163}
164
165#[derive(Debug, Clone)]
166pub struct DecodedMessageMetadata {
167    // The message ID
168    pub id: Vec<u8>,
169    // The group ID
170    pub group_id: GroupId,
171    // The timestamp of the message in nanoseconds
172    pub sent_at_ns: i64,
173    // The kind of message
174    pub kind: GroupMessageKind,
175    // The installation ID of the sender
176    pub sender_installation_id: Vec<u8>,
177    // The inbox ID of the sender
178    pub sender_inbox_id: String,
179    // The delivery status of the message
180    pub delivery_status: DeliveryStatus,
181    /// The received content type identifier, exactly as the envelope carried
182    /// it. Absent when the bytes did not parse or carried no identifier; it
183    /// can be incomplete. The client never substitutes another type.
184    pub content_type: Option<ContentTypeId>,
185    // Time in nanoseconds the message was inserted into the database
186    pub inserted_at_ns: i64,
187    // Timestamp (in NS) after which the message must be deleted
188    pub expires_at_ns: Option<i64>,
189}
190
191#[derive(Debug, Clone)]
192pub struct DecodedMessage {
193    pub metadata: DecodedMessageMetadata,
194    // The content of the message
195    pub content: MessageBody,
196    // Fallback text for the message
197    pub fallback_text: Option<String>,
198    // A list of reactions
199    pub reactions: Vec<DecodedMessage>,
200    // The number of replies to the message available
201    pub num_replies: usize,
202}
203
204#[cfg(feature = "conformance")]
205#[path = "decode_counter.rs"]
206pub mod decode_counter;
207
208/// Maximum number of reply envelopes inside one message.
209const MAX_REPLY_NESTING_DEPTH: usize = 8;
210
211/// Strict decode of an in-memory envelope. A content failure is an error.
212/// `Custom` raw bytes are this value's serialization, which is the input the
213/// decoder consumed; use [`MessageBody::from_received_bytes`] for received
214/// bytes that must be kept exactly.
215impl TryFrom<EncodedContent> for MessageBody {
216    type Error = GroupError;
217
218    fn try_from(value: EncodedContent) -> Result<Self, Self::Error> {
219        let bytes = value.encode_to_vec();
220        Self::decode_bytes(&bytes, &mut compression::DecompressionBudget::new(), 0)
221            .map_err(|failure| GroupError::CodecError(failure.into()))
222    }
223}
224
225fn has_complete_type(content: &EncodedContent) -> Option<&ContentTypeId> {
226    content
227        .r#type
228        .as_ref()
229        .filter(|kind| !kind.authority_id.is_empty() && !kind.type_id.is_empty())
230}
231
232impl MessageBody {
233    /// Decode received content from its exact bytes. Every content failure
234    /// returns `Undecodable` with those bytes; this never fails.
235    // implements: CTYPE-008
236    pub fn from_received_bytes(raw: &[u8]) -> Self {
237        Self::from_received(EncodedContent::decode(raw).ok(), raw)
238    }
239
240    fn from_received(envelope: Option<EncodedContent>, raw: &[u8]) -> Self {
241        let mut budget = compression::DecompressionBudget::new();
242        let result = match envelope {
243            Some(envelope) => Self::decode_envelope(envelope, raw, &mut budget, 0),
244            None => Err(ContentDecodeFailure::malformed(
245                "bytes are not a serialized EncodedContent",
246            )),
247        };
248        result.unwrap_or_else(|failure| {
249            MessageBody::Undecodable(UndecodableContent {
250                encoded: EncodedContent::decode(raw).ok(),
251                raw_bytes: raw.to_vec(),
252                failure,
253            })
254        })
255    }
256
257    fn decode_bytes(
258        raw: &[u8],
259        budget: &mut compression::DecompressionBudget,
260        depth: usize,
261    ) -> Result<Self, ContentDecodeFailure> {
262        let envelope = EncodedContent::decode(raw).map_err(|error| {
263            ContentDecodeFailure::malformed(format!(
264                "bytes are not a serialized EncodedContent: {error}"
265            ))
266        })?;
267        Self::decode_envelope(envelope, raw, budget, depth)
268    }
269
270    fn decode_envelope(
271        envelope: EncodedContent,
272        raw: &[u8],
273        budget: &mut compression::DecompressionBudget,
274        depth: usize,
275    ) -> Result<Self, ContentDecodeFailure> {
276        // An untyped envelope is malformed before anything else is checked.
277        let malformed_type =
278            || ContentDecodeFailure::malformed("content type identifier is absent or incomplete");
279        has_complete_type(&envelope).ok_or_else(malformed_type)?;
280        // implements: CTYPE-024, CTYPE-025
281        let value = compression::decompress_with_budget(envelope, budget)
282            .map_err(ContentDecodeFailure::codec)?;
283        let content_type = has_complete_type(&value).ok_or_else(malformed_type)?;
284        fn decoded<T>(result: Result<T, CodecError>) -> Result<T, ContentDecodeFailure> {
285            result.map_err(ContentDecodeFailure::codec)
286        }
287
288        if content_type.authority_id != "xmtp.org" {
289            return match (
290                content_type.authority_id.as_str(),
291                content_type.type_id.as_str(),
292                content_type.version_major,
293            ) {
294                ("coinbase.com", IntentCodec::TYPE_ID, IntentCodec::MAJOR_VERSION) => Ok(
295                    MessageBody::Intent(Some(decoded(IntentCodec::decode(value))?)),
296                ),
297                ("coinbase.com", ActionsCodec::TYPE_ID, ActionsCodec::MAJOR_VERSION) => Ok(
298                    MessageBody::Actions(Some(decoded(ActionsCodec::decode(value))?)),
299                ),
300                _ => Ok(MessageBody::Custom(CustomContent {
301                    encoded: value,
302                    raw_bytes: raw.to_vec(),
303                })),
304            };
305        }
306
307        match (content_type.type_id.as_str(), content_type.version_major) {
308            (TextCodec::TYPE_ID, TextCodec::MAJOR_VERSION) => {
309                #[cfg(feature = "conformance")]
310                decode_counter::record(&value.content);
311                let text = decoded(TextCodec::decode(value))?;
312                Ok(MessageBody::Text(Text { content: text }))
313            }
314            (MarkdownCodec::TYPE_ID, MarkdownCodec::MAJOR_VERSION) => {
315                let markdown = decoded(MarkdownCodec::decode(value))?;
316                Ok(MessageBody::Markdown(Markdown { content: markdown }))
317            }
318            (AttachmentCodec::TYPE_ID, AttachmentCodec::MAJOR_VERSION) => {
319                let attachment = decoded(AttachmentCodec::decode(value))?;
320                Ok(MessageBody::Attachment(attachment))
321            }
322            (RemoteAttachmentCodec::TYPE_ID, RemoteAttachmentCodec::MAJOR_VERSION) => {
323                let remote_attachment = decoded(RemoteAttachmentCodec::decode(value))?;
324                Ok(MessageBody::RemoteAttachment(remote_attachment))
325            }
326            (ReplyCodec::TYPE_ID, ReplyCodec::MAJOR_VERSION) => {
327                if depth >= MAX_REPLY_NESTING_DEPTH {
328                    return Err(ContentDecodeFailure::codec(format!(
329                        "reply nesting exceeds {MAX_REPLY_NESTING_DEPTH} levels"
330                    )));
331                }
332                // The reference is sender-controlled and kept as received. A
333                // value that is not a message id resolves to no parent; the
334                // reply itself is valid content.
335                let reference_id = value
336                    .parameters
337                    .get("reference")
338                    .cloned()
339                    .unwrap_or_default();
340                // The decompressed outer content is the nested envelope's
341                // exact serialization. A nested failure fails the outer body.
342                // implements: CTYPE-027, CTYPE-028, CTYPE-029
343                let content =
344                    Self::decode_bytes(&value.content, budget, depth + 1).map_err(|failure| {
345                        ContentDecodeFailure::codec(format!(
346                            "nested reply content: {}",
347                            failure.message
348                        ))
349                    })?;
350                Ok(MessageBody::Reply(Reply {
351                    in_reply_to: None,
352                    content: Box::new(content),
353                    reference_id,
354                }))
355            }
356            (ReactionCodec::TYPE_ID, ReactionCodec::MAJOR_VERSION) => {
357                let reaction = decoded(ReactionCodec::decode(value))?;
358                Ok(MessageBody::Reaction(reaction))
359            }
360            (LegacyReactionCodec::TYPE_ID, LegacyReactionCodec::MAJOR_VERSION) => {
361                let reaction = decoded(LegacyReactionCodec::decode(value))?;
362                Ok(MessageBody::Reaction(reaction.into()))
363            }
364            (MultiRemoteAttachmentCodec::TYPE_ID, MultiRemoteAttachmentCodec::MAJOR_VERSION) => {
365                let multi_remote_attachment = decoded(MultiRemoteAttachmentCodec::decode(value))?;
366                Ok(MessageBody::MultiRemoteAttachment(multi_remote_attachment))
367            }
368            (TransactionReferenceCodec::TYPE_ID, TransactionReferenceCodec::MAJOR_VERSION) => {
369                let transaction_reference = decoded(TransactionReferenceCodec::decode(value))?;
370                Ok(MessageBody::TransactionReference(transaction_reference))
371            }
372            (GroupUpdatedCodec::TYPE_ID, GroupUpdatedCodec::MAJOR_VERSION) => {
373                let group_updated = decoded(GroupUpdatedCodec::decode(value))?;
374                Ok(MessageBody::GroupUpdated(group_updated))
375            }
376            (ReadReceiptCodec::TYPE_ID, ReadReceiptCodec::MAJOR_VERSION) => {
377                let read_receipt = decoded(ReadReceiptCodec::decode(value))?;
378                Ok(MessageBody::ReadReceipt(read_receipt))
379            }
380            (WalletSendCallsCodec::TYPE_ID, WalletSendCallsCodec::MAJOR_VERSION) => {
381                let wallet_send_calls = decoded(WalletSendCallsCodec::decode(value))?;
382                Ok(MessageBody::WalletSendCalls(wallet_send_calls))
383            }
384            (LeaveRequestCodec::TYPE_ID, LeaveRequestCodec::MAJOR_VERSION) => {
385                let leave_request = decoded(LeaveRequestCodec::decode(value))?;
386                Ok(MessageBody::LeaveRequest(leave_request))
387            }
388
389            _ => Ok(MessageBody::Custom(CustomContent {
390                encoded: value,
391                raw_bytes: raw.to_vec(),
392            })),
393        }
394    }
395}
396
397impl DecodedMessage {
398    /// The deletion item for an expired message: its identity and metadata,
399    /// shown as the deleted-message placeholder, with no body or fallback.
400    /// A body decoded from nothing would read as an empty message or as a
401    /// decode failure.
402    ///
403    /// `deleted_by` is the actor of the delete. Expiry cleanup has no actor.
404    /// Its callers pass `DeletedBy::Sender`: `DeletedBy` has no expiry
405    /// variant, and adding one changes the binding types. No admin removed
406    /// the message, and `Admin` needs an inbox id.
407    // implements: META-051
408    pub(crate) fn expired(value: StoredGroupMessage, deleted_by: DeletedBy) -> Self {
409        DecodedMessage {
410            metadata: DecodedMessageMetadata {
411                id: value.id,
412                group_id: value.group_id,
413                sent_at_ns: value.sent_at_ns,
414                kind: value.kind,
415                sender_installation_id: value.sender_installation_id,
416                sender_inbox_id: value.sender_inbox_id,
417                delivery_status: value.delivery_status,
418                content_type: Some(crate::messages::enrichment::deleted_message_content_type()),
419                inserted_at_ns: value.inserted_at_ns,
420                expires_at_ns: value.expire_at_ns,
421            },
422            content: MessageBody::DeletedMessage { deleted_by },
423            fallback_text: None,
424            reactions: Vec::new(),
425            num_replies: 0,
426        }
427    }
428}
429
430/// Decode a stored message from its exact bytes. Content failures are kept
431/// in the body, so every stored row produces a message.
432impl From<StoredGroupMessage> for DecodedMessage {
433    fn from(value: StoredGroupMessage) -> Self {
434        let raw = value.decrypted_message_bytes.as_slice();
435        let envelope = EncodedContent::decode(raw).ok();
436        let (content_type, fallback) = envelope
437            .as_ref()
438            .map(|envelope| (envelope.r#type.clone(), envelope.fallback.clone()))
439            .unwrap_or_default();
440        let content = MessageBody::from_received(envelope, raw);
441
442        let metadata = DecodedMessageMetadata {
443            id: value.id,
444            group_id: value.group_id,
445            sent_at_ns: value.sent_at_ns,
446            kind: value.kind,
447            sender_installation_id: value.sender_installation_id,
448            sender_inbox_id: value.sender_inbox_id,
449            delivery_status: value.delivery_status,
450            content_type,
451            inserted_at_ns: value.inserted_at_ns,
452            expires_at_ns: value.expire_at_ns,
453        };
454
455        DecodedMessage {
456            metadata,
457            content,
458            fallback_text: fallback,
459            // Enrichment fills reactions and reply counts.
460            reactions: Vec::new(),
461            num_replies: 0,
462        }
463    }
464}
465
466#[cfg(test)]
467mod tests {
468    use super::*;
469    use xmtp_content_types::{
470        actions::{Action, Actions},
471        attachment::Attachment,
472        compression::compress,
473        intent::Intent,
474        reaction::LegacyReaction,
475        reply::{Reply as EncodedReply, ReplyCodec},
476    };
477    use xmtp_proto::xmtp::mls::message_contents::Compression;
478
479    // verifies: CTYPE-024
480    #[xmtp_common::test(unwrap_try = true)]
481    async fn nested_content_decompressed_before_decode() {
482        let inner = compress(TextCodec::encode("nested text".into())?, Compression::Gzip)?;
483        let outer = ReplyCodec::encode(EncodedReply {
484            reference: "0102".into(),
485            reference_inbox_id: None,
486            content: inner,
487        })?;
488        let outer = compress(outer, Compression::Deflate)?;
489        let fields = crate::groups::QueryableContentFields::try_from(outer.clone())?;
490        assert_eq!(fields.reference_id, Some(vec![1, 2]));
491        let MessageBody::Reply(reply) = MessageBody::try_from(outer)? else {
492            panic!("expected reply");
493        };
494        let MessageBody::Text(text) = *reply.content else {
495            panic!("expected nested text");
496        };
497        assert_eq!(text.content, "nested text");
498    }
499
500    fn stored_message(content: EncodedContent) -> StoredGroupMessage {
501        StoredGroupMessage {
502            id: vec![1, 2, 3],
503            group_id: GroupId::ONE,
504            decrypted_message_bytes: content.encode_to_vec(),
505            sent_at_ns: 1,
506            kind: GroupMessageKind::Application,
507            sender_installation_id: vec![4],
508            sender_inbox_id: "inbox".into(),
509            delivery_status: DeliveryStatus::Published,
510            content_type: xmtp_db::group_message::ContentType::Text,
511            version_major: 1,
512            version_minor: 0,
513            authority_id: "xmtp.org".into(),
514            reference_id: None,
515            sequence_id: 1,
516            envelope_hash: None,
517            expiry_ns: None,
518            inserted_at_ns: 0,
519            expire_at_ns: None,
520            should_push: false,
521            idempotency_key: String::new(),
522        }
523    }
524
525    fn unknown_content() -> EncodedContent {
526        let mut content = TextCodec::encode("custom payload".into()).unwrap();
527        content.r#type.as_mut().unwrap().type_id = "custom".into();
528        content
529    }
530
531    // verifies: CTYPE-001, CTYPE-008
532    #[xmtp_common::test(unwrap_try = true)]
533    async fn standard_codecs_match_authority_for_all_types() {
534        macro_rules! check {
535            ($content:expr, $standard:pat) => {{
536                let mut standard = $content;
537                let id = standard.r#type.as_mut().expect("encoded type");
538                id.version_minor = 9;
539                let name = format!("{}/{}", id.authority_id, id.type_id);
540                let decoded = DecodedMessage::from(stored_message(standard.clone()));
541                assert!(matches!(decoded.content, $standard), "{name} did not decode");
542
543                if standard.r#type.as_ref().expect("encoded type").authority_id != "xmtp.org" {
544                    let mut wrong_standard = standard.clone();
545                    wrong_standard.r#type.as_mut().expect("encoded type").authority_id =
546                        "xmtp.org".into();
547                    let decoded = DecodedMessage::from(stored_message(wrong_standard.clone()));
548                    assert!(
549                        matches!(decoded.content, MessageBody::Custom(actual) if actual.encoded == wrong_standard),
550                        "{name} selected the wrong standard authority"
551                    );
552                }
553
554                let mut custom = standard;
555                custom.r#type.as_mut().expect("encoded type").authority_id =
556                    "custom.example".into();
557                let decoded = DecodedMessage::from(stored_message(custom.clone()));
558                assert!(
559                    matches!(decoded.content, MessageBody::Custom(actual) if actual.encoded == custom),
560                    "{name} selected a codec for custom.example"
561                );
562            }};
563        }
564
565        check!(TextCodec::encode("text".into())?, MessageBody::Text(_));
566        check!(
567            MarkdownCodec::encode("markdown".into())?,
568            MessageBody::Markdown(_)
569        );
570        check!(
571            AttachmentCodec::encode(Attachment {
572                filename: None,
573                mime_type: "text/plain".into(),
574                content: vec![1],
575            })?,
576            MessageBody::Attachment(_)
577        );
578        check!(
579            RemoteAttachmentCodec::encode(RemoteAttachment {
580                url: "https://example.com/file".into(),
581                content_digest: "ab".repeat(32),
582                secret: vec![1; 32],
583                salt: vec![2; 32],
584                nonce: vec![3; 12],
585                scheme: "https://".into(),
586                ..Default::default()
587            })?,
588            MessageBody::RemoteAttachment(_)
589        );
590        check!(
591            ReplyCodec::encode(EncodedReply {
592                reference: "0102".into(),
593                reference_inbox_id: None,
594                content: TextCodec::encode("reply".into())?,
595            })?,
596            MessageBody::Reply(_)
597        );
598        check!(
599            ReactionCodec::encode(ReactionV2::default())?,
600            MessageBody::Reaction(_)
601        );
602        check!(
603            LegacyReactionCodec::encode(LegacyReaction {
604                action: "added".into(),
605                reference: "0102".into(),
606                reference_inbox_id: None,
607                schema: "unicode".into(),
608                content: "👍".into(),
609            })?,
610            MessageBody::Reaction(_)
611        );
612        check!(
613            MultiRemoteAttachmentCodec::encode(MultiRemoteAttachment::default())?,
614            MessageBody::MultiRemoteAttachment(_)
615        );
616        check!(
617            TransactionReferenceCodec::encode(TransactionReference {
618                namespace: None,
619                network_id: "1".into(),
620                reference: "0x123".into(),
621                metadata: None,
622            })?,
623            MessageBody::TransactionReference(_)
624        );
625        check!(
626            GroupUpdatedCodec::encode(GroupUpdated::default())?,
627            MessageBody::GroupUpdated(_)
628        );
629        check!(
630            ReadReceiptCodec::encode(ReadReceipt {})?,
631            MessageBody::ReadReceipt(_)
632        );
633        check!(
634            WalletSendCallsCodec::encode(WalletSendCalls {
635                version: "1".into(),
636                chain_id: "1".into(),
637                from: "0x123".into(),
638                calls: vec![],
639                capabilities: None,
640            })?,
641            MessageBody::WalletSendCalls(_)
642        );
643        check!(
644            IntentCodec::encode(Intent {
645                id: "one".into(),
646                action_id: "two".into(),
647                metadata: None,
648            })?,
649            MessageBody::Intent(_)
650        );
651        check!(
652            ActionsCodec::encode(Actions {
653                id: "one".into(),
654                description: "actions".into(),
655                actions: vec![Action {
656                    id: "two".into(),
657                    label: "two".into(),
658                    image_url: None,
659                    style: None,
660                    expires_at: None,
661                }],
662                expires_at: None,
663            })?,
664            MessageBody::Actions(_)
665        );
666        check!(
667            LeaveRequestCodec::encode(LeaveRequest::default())?,
668            MessageBody::LeaveRequest(_)
669        );
670    }
671
672    // verifies: CTYPE-024
673    #[xmtp_common::test(unwrap_try = true)]
674    async fn compressed_custom_content_is_decompressed_at_both_levels() {
675        let inner = compress(unknown_content(), Compression::Gzip)?;
676        let MessageBody::Custom(top) = DecodedMessage::from(stored_message(inner.clone())).content
677        else {
678            panic!("expected top-level custom content");
679        };
680        assert_eq!(top.encoded.compression, None);
681        assert_eq!(top.encoded.content, b"custom payload");
682        assert_eq!(top.raw_bytes, inner.encode_to_vec());
683
684        let outer = ReplyCodec::encode(EncodedReply {
685            reference: "0102".into(),
686            reference_inbox_id: None,
687            content: inner,
688        })?;
689        let MessageBody::Reply(reply) = MessageBody::try_from(outer.clone())? else {
690            panic!("expected reply");
691        };
692        let MessageBody::Custom(nested) = *reply.content else {
693            panic!("expected nested custom content");
694        };
695        assert_eq!(nested.encoded.compression, None);
696        assert_eq!(nested.encoded.content, b"custom payload");
697        assert_eq!(nested.raw_bytes, outer.content);
698    }
699
700    fn nested_reply(mut content: EncodedContent, levels: usize, pad_to: usize) -> EncodedContent {
701        for _ in 0..levels {
702            let mut reply = ReplyCodec::encode(EncodedReply {
703                reference: "0102".into(),
704                reference_inbox_id: None,
705                content,
706            })
707            .unwrap();
708            if pad_to > reply.content.len() + 6 {
709                // An unknown protobuf field pads this reply without changing its body.
710                let padding = pad_to - reply.content.len() - 6;
711                reply.content.extend_from_slice(&[0xa2, 0x06]);
712                let mut length = padding;
713                while length >= 0x80 {
714                    reply.content.push((length as u8) | 0x80);
715                    length >>= 7;
716                }
717                reply.content.push(length as u8);
718                reply.content.resize(reply.content.len() + padding, 0);
719            }
720            content = compress(reply, Compression::Deflate).unwrap();
721        }
722        content
723    }
724
725    // verifies: CTYPE-008, CTYPE-025
726    #[xmtp_common::test(unwrap_try = true)]
727    async fn nested_compressed_reply_uses_one_budget() {
728        let attack = nested_reply(
729            TextCodec::encode("inner".into())?,
730            20,
731            compression::MAX_DECOMPRESSED_BYTES - 4096,
732        );
733        let bytes = attack.encode_to_vec();
734        let mut budget = compression::DecompressionBudget::new();
735        let failure = MessageBody::decode_bytes(&bytes, &mut budget, 0).unwrap_err();
736        assert_eq!(failure.kind, ContentDecodeFailureKind::CodecDecodeFailed);
737        assert!(failure.message.contains("decompressed content exceeds"));
738        assert!(budget.used() > 0);
739        assert!(budget.peak_capacity() <= compression::MAX_DECOMPRESSED_BYTES + 65_536);
740        eprintln!(
741            "attack peak decompressed capacity: {} bytes",
742            budget.peak_capacity()
743        );
744
745        let expected = EncodedContent::decode(bytes.as_slice())?;
746        // Decode from the exact bytes: re-encoding a parameter map can
747        // reorder its entries.
748        let decoded = DecodedMessage::from(stored_bytes(bytes.clone()));
749        let MessageBody::Undecodable(original) = decoded.content else {
750            panic!("expected preserved decode failure");
751        };
752        assert_eq!(original.raw_bytes, bytes);
753        assert_eq!(original.encoded, Some(expected));
754        assert_eq!(
755            original.failure.kind,
756            ContentDecodeFailureKind::CodecDecodeFailed
757        );
758    }
759
760    // verifies: CTYPE-008
761    #[xmtp_common::test(unwrap_try = true)]
762    async fn reply_depth_limit_preserves_message() {
763        let content = nested_reply(
764            TextCodec::encode("inner".into())?,
765            MAX_REPLY_NESTING_DEPTH + 1,
766            0,
767        );
768        let bytes = content.encode_to_vec();
769        let decoded = DecodedMessage::from(stored_bytes(bytes.clone()));
770        let MessageBody::Undecodable(original) = decoded.content else {
771            panic!("expected preserved decode failure");
772        };
773        assert_eq!(original.raw_bytes, bytes);
774        assert_eq!(original.encoded, Some(content));
775        assert!(original.failure.message.contains("reply nesting exceeds"));
776    }
777
778    // verifies: CTYPE-008, CTYPE-024
779    #[xmtp_common::test(unwrap_try = true)]
780    async fn compressed_decode_failure_preserves_message() {
781        let mut content = TextCodec::encode("unchanged".into())?;
782        content.compression = Some(99);
783        let bytes = content.encode_to_vec();
784        let message = stored_message(content.clone());
785        let decoded = DecodedMessage::from(message);
786        assert_eq!(decoded.metadata.id, vec![1, 2, 3]);
787        assert_eq!(decoded.metadata.content_type, content.r#type);
788        let MessageBody::Undecodable(undecodable) = decoded.content else {
789            panic!("expected original undecodable content");
790        };
791        assert_eq!(undecodable.raw_bytes, bytes);
792        assert_eq!(undecodable.encoded, Some(content));
793        assert_eq!(
794            undecodable.failure.kind,
795            ContentDecodeFailureKind::CodecDecodeFailed
796        );
797        assert!(undecodable.failure.message.contains("unknown compression"));
798    }
799
800    fn stored_bytes(bytes: Vec<u8>) -> StoredGroupMessage {
801        let mut message = stored_message(EncodedContent::default());
802        message.decrypted_message_bytes = bytes;
803        message
804    }
805
806    /// Append an unknown field (number 200, length-delimited) so the exact
807    /// serialization differs from the parsed value's re-encoding.
808    fn with_unknown_field(mut bytes: Vec<u8>) -> Vec<u8> {
809        bytes.extend_from_slice(&[0xc2, 0x0c, 0x03, b'x', b'y', b'z']);
810        bytes
811    }
812
813    // verifies: CTYPE-008, CTYPE-009
814    #[xmtp_common::test(unwrap_try = true)]
815    async fn malformed_bytes_are_kept_with_their_cause() {
816        let decoded = DecodedMessage::from(stored_bytes(vec![0xff]));
817        assert_eq!(decoded.metadata.id, vec![1, 2, 3]);
818        assert_eq!(decoded.metadata.content_type, None);
819        assert_eq!(decoded.fallback_text, None);
820        let MessageBody::Undecodable(undecodable) = decoded.content else {
821            panic!("expected undecodable content, got {:?}", decoded.content);
822        };
823        assert_eq!(undecodable.raw_bytes, vec![0xff]);
824        assert_eq!(undecodable.encoded, None);
825        assert_eq!(
826            undecodable.failure.kind,
827            ContentDecodeFailureKind::MalformedEnvelope
828        );
829    }
830
831    // verifies: CTYPE-008, CTYPE-009
832    #[xmtp_common::test(unwrap_try = true)]
833    async fn absent_or_partial_type_is_malformed_and_keeps_the_partial_identifier() {
834        let mut untyped = TextCodec::encode("text".into())?;
835        untyped.r#type = None;
836        untyped.fallback = Some("kept".into());
837        let mut partial = untyped.clone();
838        partial.r#type = Some(ContentTypeId {
839            authority_id: String::new(),
840            type_id: "text".into(),
841            version_major: 1,
842            version_minor: 0,
843        });
844        // Invalid compression does not change the cause of an untyped envelope.
845        let mut untyped_compressed = untyped.clone();
846        untyped_compressed.compression = Some(99);
847        let mut partial_compressed = partial.clone();
848        partial_compressed.compression = Some(99);
849        for content in [untyped, partial, untyped_compressed, partial_compressed] {
850            let bytes = content.encode_to_vec();
851            let decoded = DecodedMessage::from(stored_bytes(bytes.clone()));
852            assert_eq!(decoded.metadata.content_type, content.r#type);
853            assert_eq!(decoded.fallback_text.as_deref(), Some("kept"));
854            let MessageBody::Undecodable(undecodable) = decoded.content else {
855                panic!("expected undecodable content, got {:?}", decoded.content);
856            };
857            assert_eq!(undecodable.raw_bytes, bytes);
858            assert_eq!(undecodable.encoded, Some(content));
859            assert_eq!(
860                undecodable.failure.kind,
861                ContentDecodeFailureKind::MalformedEnvelope
862            );
863        }
864    }
865
866    // verifies: CTYPE-008, CTYPE-027
867    #[xmtp_common::test(unwrap_try = true)]
868    async fn exact_bytes_survive_at_both_levels() {
869        let nested_bytes = with_unknown_field(unknown_content().encode_to_vec());
870        let mut outer = ReplyCodec::encode(EncodedReply {
871            reference: "0102".into(),
872            reference_inbox_id: None,
873            content: unknown_content(),
874        })?;
875        outer.content = nested_bytes.clone();
876        let outer_bytes = with_unknown_field(outer.encode_to_vec());
877
878        let MessageBody::Custom(top) =
879            DecodedMessage::from(stored_bytes(nested_bytes.clone())).content
880        else {
881            panic!("expected custom content");
882        };
883        assert_eq!(top.raw_bytes, nested_bytes);
884        assert_ne!(top.encoded.encode_to_vec(), nested_bytes);
885
886        let MessageBody::Reply(reply) = DecodedMessage::from(stored_bytes(outer_bytes)).content
887        else {
888            panic!("expected reply");
889        };
890        assert_eq!(reply.reference_id, "0102");
891        let MessageBody::Custom(nested) = *reply.content else {
892            panic!("expected nested custom content");
893        };
894        assert_eq!(nested.raw_bytes, nested_bytes);
895        assert_eq!(nested.encoded.content, b"custom payload");
896    }
897
898    // verifies: CTYPE-008, CTYPE-029
899    #[xmtp_common::test(unwrap_try = true)]
900    async fn nested_failures_make_the_outer_reply_undecodable() {
901        let reply_with = |nested: Vec<u8>| {
902            let mut outer = ReplyCodec::encode(EncodedReply {
903                reference: "0102".into(),
904                reference_inbox_id: None,
905                content: TextCodec::encode("placeholder".into()).unwrap(),
906            })
907            .unwrap();
908            outer.content = nested;
909            outer.fallback = Some("outer fallback".into());
910            outer
911        };
912        let mut untyped = TextCodec::encode("nested".into())?;
913        untyped.r#type = None;
914        let mut bad_compression = TextCodec::encode("nested".into())?;
915        bad_compression.compression = Some(99);
916        let mut bad_text = TextCodec::encode("nested".into())?;
917        bad_text.content = vec![0xff, 0xfe];
918        let cases = [
919            ("malformed", vec![0xff]),
920            ("untyped", untyped.encode_to_vec()),
921            ("bad compression", bad_compression.encode_to_vec()),
922            ("failed standard", bad_text.encode_to_vec()),
923        ];
924        for (name, nested) in cases {
925            let outer = reply_with(nested);
926            let bytes = outer.encode_to_vec();
927            let decoded = DecodedMessage::from(stored_bytes(bytes.clone()));
928            assert_eq!(
929                decoded.fallback_text.as_deref(),
930                Some("outer fallback"),
931                "{name}"
932            );
933            assert_eq!(decoded.metadata.content_type, outer.r#type, "{name}");
934            let MessageBody::Undecodable(undecodable) = decoded.content else {
935                panic!(
936                    "{name}: expected undecodable outer reply, got {:?}",
937                    decoded.content
938                );
939            };
940            assert_eq!(undecodable.raw_bytes, bytes, "{name}");
941            assert_eq!(undecodable.encoded, Some(outer), "{name}");
942            assert_eq!(
943                undecodable.failure.kind,
944                ContentDecodeFailureKind::CodecDecodeFailed,
945                "{name}"
946            );
947            assert!(undecodable.failure.message.contains("nested"), "{name}");
948        }
949    }
950
951    #[xmtp_common::test(unwrap_try = true)]
952    async fn invalid_reply_reference_is_a_reply_with_no_parent() {
953        // A sender-controlled reference that is not a message id resolves to
954        // no parent; the reply itself is valid content and is never marked
955        // undecodable. The reference is kept as received.
956        for reference in ["not-valid-hex!@#", ""] {
957            let outer = ReplyCodec::encode(EncodedReply {
958                reference: reference.into(),
959                reference_inbox_id: None,
960                content: TextCodec::encode("reply".into())?,
961            })?;
962            let decoded = DecodedMessage::from(stored_bytes(outer.encode_to_vec()));
963            let MessageBody::Reply(reply) = decoded.content else {
964                panic!("{reference:?}: expected a reply, got {:?}", decoded.content);
965            };
966            assert!(reply.in_reply_to.is_none());
967            assert_eq!(reply.reference_id, reference);
968            assert!(
969                matches!(*reply.content, MessageBody::Text(ref text) if text.content == "reply")
970            );
971        }
972    }
973
974    // verifies: CTYPE-028
975    #[xmtp_common::test(unwrap_try = true)]
976    async fn nested_type_wins_over_the_outer_content_type_parameter() {
977        let mut outer = ReplyCodec::encode(EncodedReply {
978            reference: "0102".into(),
979            reference_inbox_id: None,
980            content: TextCodec::encode("placeholder".into())?,
981        })?;
982        assert_eq!(outer.parameters["contentType"], "xmtp.org/text:1.0");
983        outer.content = ReactionCodec::encode(ReactionV2 {
984            reference: "0102".into(),
985            content: "👍".into(),
986            ..Default::default()
987        })?
988        .encode_to_vec();
989        let MessageBody::Reply(reply) = MessageBody::try_from(outer)? else {
990            panic!("expected reply");
991        };
992        assert!(
993            matches!(*reply.content, MessageBody::Reaction(ref reaction) if reaction.content == "👍"),
994            "the outer contentType parameter selected the codec: {:?}",
995            reply.content
996        );
997    }
998
999    // verifies: CTYPE-009
1000    #[xmtp_common::test(unwrap_try = true)]
1001    async fn strict_decode_reports_failures_as_errors() {
1002        let mut untyped = TextCodec::encode("text".into())?;
1003        untyped.r#type = None;
1004        assert!(matches!(
1005            MessageBody::try_from(untyped),
1006            Err(GroupError::CodecError(CodecError::InvalidContentType))
1007        ));
1008        let mut bad_text = TextCodec::encode("text".into())?;
1009        bad_text.content = vec![0xff, 0xfe];
1010        assert!(matches!(
1011            MessageBody::try_from(bad_text),
1012            Err(GroupError::CodecError(CodecError::Decode(_)))
1013        ));
1014    }
1015}
xmtp_mls/messages/decoded_message.rs

xmtp_mls/messages/
decoded_message.rs
