diff --git a/crates/xmtp_db/migrations/2026-06-09-000000-0000_add_epoch_entered_at_ns_to_groups/down.sql b/crates/xmtp_db/migrations/2026-06-09-000000-0000_add_epoch_entered_at_ns_to_groups/down.sql new file mode 100644 index 0000000000..1ad3eade10 --- /dev/null +++ b/crates/xmtp_db/migrations/2026-06-09-000000-0000_add_epoch_entered_at_ns_to_groups/down.sql @@ -0,0 +1,2 @@ +ALTER TABLE groups +DROP COLUMN epoch_entered_at_ns; diff --git a/crates/xmtp_db/migrations/2026-06-09-000000-0000_add_epoch_entered_at_ns_to_groups/up.sql b/crates/xmtp_db/migrations/2026-06-09-000000-0000_add_epoch_entered_at_ns_to_groups/up.sql new file mode 100644 index 0000000000..5ec6cbe029 --- /dev/null +++ b/crates/xmtp_db/migrations/2026-06-09-000000-0000_add_epoch_entered_at_ns_to_groups/up.sql @@ -0,0 +1,8 @@ +-- XIP-82: the delivery-service envelope timestamp of the message by +-- which this client entered the current MLS epoch (the epoch's commit, +-- or the welcome for members added at that epoch). NULL means the group +-- has not advanced epoch since this column landed; readers fall back to +-- created_at_ns (the initial epoch). Consumed by the external-commit +-- validator's expire_in_ns staleness bound. +ALTER TABLE groups +ADD COLUMN epoch_entered_at_ns BIGINT; diff --git a/crates/xmtp_db/src/encrypted_store/group.rs b/crates/xmtp_db/src/encrypted_store/group.rs index ddd6ac9f6f..3094a345ec 100644 --- a/crates/xmtp_db/src/encrypted_store/group.rs +++ b/crates/xmtp_db/src/encrypted_store/group.rs @@ -106,6 +106,13 @@ pub struct StoredGroup { /// NULL if the pending-remove didn't receive an update yet #[builder(default = None)] pub has_pending_leave_request: Option, + /// XIP-82: envelope timestamp (ns) of the message by which this + /// client entered the current MLS epoch — the epoch's commit, or + /// the welcome for members added at that epoch. NULL falls back to + /// `created_at_ns` (the initial epoch). Consumed by the + /// external-commit validator's `expire_in_ns` staleness bound. + #[builder(default = None)] + pub epoch_entered_at_ns: Option, //todo: store member role? } @@ -369,6 +376,16 @@ pub trait QueryGroup { fn get_groups_have_pending_leave_request(&self) -> Result>, crate::ConnectionError>; + + /// Records the envelope timestamp (ns) of the message by which this + /// client entered the current MLS epoch (XIP-82): written on every + /// successful commit merge and at welcome-join. Readers fall back + /// to `created_at_ns` when NULL (the initial epoch). + fn set_group_epoch_entered_at_ns( + &self, + group_id: &GroupId, + epoch_entered_at_ns: i64, + ) -> Result<(), StorageError>; } impl QueryGroup for &T @@ -556,6 +573,14 @@ where ) -> Result>, crate::ConnectionError> { (**self).get_groups_have_pending_leave_request() } + + fn set_group_epoch_entered_at_ns( + &self, + group_id: &GroupId, + epoch_entered_at_ns: i64, + ) -> Result<(), StorageError> { + (**self).set_group_epoch_entered_at_ns(group_id, epoch_entered_at_ns) + } } impl QueryGroup for DbConnection { @@ -1203,6 +1228,32 @@ impl QueryGroup for DbConnection { Ok(()) } + fn set_group_epoch_entered_at_ns( + &self, + group_id: &GroupId, + epoch_entered_at_ns: i64, + ) -> Result<(), StorageError> { + use crate::schema::groups::dsl; + // Monotonic: envelope time only moves forward over a group's + // life (commits merge in cursor order; a welcome re-entry + // postdates the removal it recovers from), so a smaller value + // here can only come from pathological reordering or + // corruption. Never move the epoch-entry point backward — a + // backdated value would widen the XIP-82 staleness window. + self.raw_query(|conn| { + diesel::update( + dsl::groups.find(group_id).filter( + dsl::epoch_entered_at_ns + .is_null() + .or(dsl::epoch_entered_at_ns.lt(epoch_entered_at_ns)), + ), + ) + .set(dsl::epoch_entered_at_ns.eq(Some(epoch_entered_at_ns))) + .execute(conn) + })?; + Ok(()) + } + #[xmtp_common::db_span] fn get_groups_have_pending_leave_request( &self, @@ -1588,6 +1639,37 @@ pub(crate) mod tests { .await } + #[xmtp_common::test] + async fn test_epoch_entered_at_ns_round_trip() { + with_connection(|conn| { + let test_group = generate_group(None); + test_group.store(&conn).unwrap(); + + // Fresh groups have no epoch-entry timestamp — readers fall + // back to created_at_ns (the initial epoch). + let fetched: StoredGroup = conn.fetch(&test_group.id).ok().flatten().unwrap(); + assert_eq!(fetched.epoch_entered_at_ns, None); + + conn.set_group_epoch_entered_at_ns(&test_group.id, 42_000) + .unwrap(); + let fetched: StoredGroup = conn.fetch(&test_group.id).ok().flatten().unwrap(); + assert_eq!(fetched.epoch_entered_at_ns, Some(42_000)); + + // Later epoch advances overwrite. + conn.set_group_epoch_entered_at_ns(&test_group.id, 43_000) + .unwrap(); + let fetched: StoredGroup = conn.fetch(&test_group.id).ok().flatten().unwrap(); + assert_eq!(fetched.epoch_entered_at_ns, Some(43_000)); + + // Monotonic: an out-of-order older timestamp never moves + // the epoch-entry point backward. + conn.set_group_epoch_entered_at_ns(&test_group.id, 42_500) + .unwrap(); + let fetched: StoredGroup = conn.fetch(&test_group.id).ok().flatten().unwrap(); + assert_eq!(fetched.epoch_entered_at_ns, Some(43_000)); + }) + } + #[xmtp_common::test] fn test_new_group_has_correct_purpose() { with_connection(|conn| { diff --git a/crates/xmtp_db/src/encrypted_store/icebox.rs b/crates/xmtp_db/src/encrypted_store/icebox.rs index a787397c81..b43d0e90aa 100644 --- a/crates/xmtp_db/src/encrypted_store/icebox.rs +++ b/crates/xmtp_db/src/encrypted_store/icebox.rs @@ -392,6 +392,7 @@ mod tests { commit_log_public_key: None, is_commit_log_forked: None, has_pending_leave_request: None, + epoch_entered_at_ns: None, }; group.store(conn).unwrap(); group_id diff --git a/crates/xmtp_db/src/encrypted_store/message_deletion.rs b/crates/xmtp_db/src/encrypted_store/message_deletion.rs index c0e7095362..7c17c1bf04 100644 --- a/crates/xmtp_db/src/encrypted_store/message_deletion.rs +++ b/crates/xmtp_db/src/encrypted_store/message_deletion.rs @@ -197,6 +197,7 @@ mod tests { commit_log_public_key: None, is_commit_log_forked: None, has_pending_leave_request: None, + epoch_entered_at_ns: None, } .store(conn) .unwrap(); diff --git a/crates/xmtp_db/src/encrypted_store/schema_gen.rs b/crates/xmtp_db/src/encrypted_store/schema_gen.rs index b2591b60b3..d4ebe0e47c 100644 --- a/crates/xmtp_db/src/encrypted_store/schema_gen.rs +++ b/crates/xmtp_db/src/encrypted_store/schema_gen.rs @@ -89,6 +89,7 @@ diesel::table! { commit_log_public_key -> Nullable, is_commit_log_forked -> Nullable, has_pending_leave_request -> Nullable, + epoch_entered_at_ns -> Nullable, } } diff --git a/crates/xmtp_db/src/mock.rs b/crates/xmtp_db/src/mock.rs index d9f287a931..c8acb2d956 100644 --- a/crates/xmtp_db/src/mock.rs +++ b/crates/xmtp_db/src/mock.rs @@ -253,6 +253,12 @@ mock! { fn get_groups_have_pending_leave_request( &self, ) -> Result>, crate::ConnectionError>; + + fn set_group_epoch_entered_at_ns( + &self, + group_id: &GroupId, + epoch_entered_at_ns: i64, + ) -> Result<(), StorageError>; } impl QueryGroupVersion for DbQuery { diff --git a/crates/xmtp_mls/src/groups/mls_sync.rs b/crates/xmtp_mls/src/groups/mls_sync.rs index 4f43aa9012..80b44cb2d1 100644 --- a/crates/xmtp_mls/src/groups/mls_sync.rs +++ b/crates/xmtp_mls/src/groups/mls_sync.rs @@ -974,6 +974,17 @@ where next_intent_state: IntentState::ToPublish, }); } + // XIP-82: record when (by envelope time) we entered the new + // epoch — our own commit echoed back from the delivery + // service is the epoch-start message for this client. + storage + .db() + .set_group_epoch_entered_at_ns(&self.group_id, envelope_timestamp_ns) + .map_err(|err| IntentResolutionError { + processing_error: err.into(), + next_intent_state: IntentState::Error, + })?; + Self::mark_readd_requests_as_responded( storage, &self.group_id, @@ -1144,25 +1155,145 @@ where let validated_commit = match &processed_message.content() { ProcessedMessageContent::StagedCommitMessage(staged_commit) => { - // OpenMLS already verified the framing signature against the - // sender's leaf during `process_message`, and `extract_message_sender` - // above asserted `Sender::Member` — so this match is exhaustive - // for the cases that reach here. - let committer_leaf_index = match processed_message.sender() { - openmls::prelude::Sender::Member(idx) => *idx, - _ => { + // Route by framing sender. `Sender::Member` is the standard + // member-authored-commit path (`from_staged_commit`). + // `Sender::NewMemberCommit` is the atomic external-commit + // / QR-invite join path (L-7 `from_external_commit`); the + // joiner is not yet a tree member, so the regular path's + // committer-leaf-index lookup does not apply. Both arms + // converge below at `merge_staged_commit_logged`. + let result = match processed_message.sender() { + openmls::prelude::Sender::Member(idx) => { + let committer_leaf_index = *idx; + ValidatedCommit::from_staged_commit( + &self.context, + staged_commit, + committer_leaf_index, + mls_group, + ) + .await + } + openmls::prelude::Sender::NewMemberCommit => { + // Source the metadata capability-aware, mirroring + // the proposal branch above. The authorization + // gates themselves (EXTERNAL_COMMIT_POLICY, the + // GROUP_MEMBERSHIP external-committer grant) are + // read inside `from_external_commit` from the + // group's pre-merge extensions. + let is_migrated = super::app_data::is_migrated_group(mls_group); + let extensions = mls_group.extensions(); + let immutable_metadata = if is_migrated { + match super::app_data::component_source::read_group_metadata_from_dict( + mls_group, + ) { + Ok(Some(seed)) => { + use xmtp_proto::xmtp::mls::message_contents::GroupMetadataV1 as GroupMetadataProto; + let proto = GroupMetadataProto { + conversation_type: seed.conversation_type, + creator_inbox_id: seed.creator_inbox_id, + creator_account_address: String::new(), + dm_members: seed.dm_members, + oneshot_message: seed.oneshot, + }; + match xmtp_mls_common::group_metadata::GroupMetadata::try_from( + proto, + ) { + Ok(m) => m, + Err(e) => { + self.maybe_update_cursor(&self.context.db(), envelope)?; + return Err(CommitValidationError::from(e).into()); + } + } + } + Ok(None) | Err(_) => { + self.maybe_update_cursor(&self.context.db(), envelope)?; + return Err(CommitValidationError::from( + xmtp_mls_common::group_metadata::GroupMetadataError::MissingExtension, + ) + .into()); + } + } + } else { + match extract_group_metadata(extensions) { + Ok(m) => m, + Err(e) => { + self.maybe_update_cursor(&self.context.db(), envelope)?; + return Err(CommitValidationError::from(e).into()); + } + } + }; + let mutable_metadata = match super::app_data::component_source::extract_group_mutable_metadata_capability_aware( + mls_group, + ) { + Ok(m) => m, + Err(e) => { + self.maybe_update_cursor(&self.context.db(), envelope)?; + return Err(CommitValidationError::from( + xmtp_mls_common::group_mutable_metadata::GroupMutableMetadataError::from(e), + ) + .into()); + } + }; + + // XIP-82 checks 8 + 9 are evaluated against + // delivery-service envelope timestamps — never a + // wall clock — so every member reaches the same + // verdict regardless of when it syncs this commit. + // Epoch start is tracked in the group row (written + // on every commit merge and at welcome-join); + // NULL means the group hasn't advanced epoch since + // the column landed, where created_at_ns (the + // initial epoch) is the right floor. + let epoch_started_at_ns = match self.context.db().find_group(&self.group_id) + { + Ok(Some(group)) => { + group.epoch_entered_at_ns.unwrap_or(group.created_at_ns) + } + Ok(None) => { + // No row for a group whose message we're + // processing — shouldn't happen. Zero makes + // the staleness age span the full envelope + // timestamp: fail toward rejection. + tracing::warn!( + inbox_id = self.context.inbox_id(), + group_id = %self.group_id, + "no stored group row while validating external commit" + ); + 0 + } + Err(e) => return Err(e.into()), + }; + // Clamp at zero before widening: a (corrupt or + // forged) negative i64 would otherwise wrap to an + // enormous u64 — for the epoch start that would + // saturate the staleness age to zero, silently + // disabling check 9. + let timestamps = + crate::groups::validated_commit::ExternalCommitTimestamps { + commit_envelope_ns: envelope_timestamp_ns.max(0) as u64, + epoch_started_at_ns: epoch_started_at_ns.max(0) as u64, + }; + ValidatedCommit::from_external_commit( + staged_commit, + mls_group, + processed_message.sender(), + &immutable_metadata, + &mutable_metadata, + timestamps, + ) + } + other => { + tracing::warn!( + inbox_id = self.context.inbox_id(), + group_id = %self.group_id, + ?other, + "rejecting commit from unsupported sender type" + ); return Err(GroupMessageProcessingError::CommitValidation( CommitValidationError::ActorNotMember, )); } }; - let result = ValidatedCommit::from_staged_commit( - &self.context, - staged_commit, - committer_leaf_index, - mls_group, - ) - .await; let validated_commit = match result { Err(e) if !e.is_retryable() => { @@ -1540,6 +1671,13 @@ where cursor.sequence_id as i64, )?; + // XIP-82: record when (by envelope time) we entered the + // new epoch — the external-commit validator measures the + // `expire_in_ns` staleness bound from here. + storage + .db() + .set_group_epoch_entered_at_ns(&self.group_id, envelope_timestamp_ns)?; + Self::mark_readd_requests_as_responded( storage, &self.group_id, @@ -2075,9 +2213,17 @@ where } self.load_mls_group_with_lock_async(async |mut mls_group| { - // ensure we are processing a private message + // Accept PrivateMessage (regular handshake/application messages) and + // PublicMessage (external commits from non-members — RFC 9420 §12.4.3.2). + // L-8's dispatch inside process_message_inner enforces the per-sender + // policy; anything else is rejected here at the framing level. + // + // The wildcard arm is structurally unreachable today + // (`ProtocolMessage` only has these two variants), but is kept as a + // defense-in-depth guard if openmls grows another variant. + #[allow(unreachable_patterns)] match &envelope.message { - ProtocolMessage::PrivateMessage(_) => (), + ProtocolMessage::PrivateMessage(_) | ProtocolMessage::PublicMessage(_) => (), other => { return Err(GroupMessageProcessingError::UnsupportedMessageType( discriminant(other), @@ -4283,6 +4429,13 @@ where // Extracts the message sender, but does not do any validation to ensure that the // installation_id is actually part of the inbox. +// +// For `Sender::Member`, the sender's leaf is already in the tree and we read +// the inbox-id / installation-id from there. For `Sender::NewMemberCommit` +// (external commit / atomic join — see L-7 `from_external_commit`), the +// joiner is not yet in the tree; their leaf only appears on the staged +// commit's update-path. We fall back to that path-leaf as the source of +// truth for the joiner's identity and signature key. fn extract_message_sender( openmls_group: &mut OpenMlsGroup, decrypted_message: &ProcessedMessage, @@ -4297,6 +4450,21 @@ fn extract_message_sender( return Ok((sender_inbox_id, member.signature_key)); } + // Atomic external-commit path: the joiner's leaf lives on the staged + // commit's update-path, not in the pre-commit tree. OpenMLS has already + // verified the framing signature against that leaf during + // `process_message`, so trusting it here is sound. + if matches!(decrypted_message.sender(), Sender::NewMemberCommit) + && let ProcessedMessageContent::StagedCommitMessage(staged_commit) = + decrypted_message.content() + && let Some(joiner_leaf) = staged_commit.update_path_leaf_node() + { + let basic_credential = BasicCredential::try_from(joiner_leaf.credential().clone())?; + let sender_inbox_id = parse_credential(basic_credential.identity())?; + let sender_installation_id = joiner_leaf.signature_key().as_slice().to_vec(); + return Ok((sender_inbox_id, sender_installation_id)); + } + let basic_credential = BasicCredential::try_from(decrypted_message.credential().clone())?; Err(GroupMessageProcessingError::InvalidSender { message_time_ns: message_created_ns, diff --git a/crates/xmtp_mls/src/groups/welcome_sync.rs b/crates/xmtp_mls/src/groups/welcome_sync.rs index 29a732bd73..f639ded7c3 100644 --- a/crates/xmtp_mls/src/groups/welcome_sync.rs +++ b/crates/xmtp_mls/src/groups/welcome_sync.rs @@ -624,6 +624,8 @@ mod tests { db.expect_update_responded_at_sequence_id() .returning(|_, _, _| Ok(())); db.expect_insert_or_replace_group().returning(Ok); + db.expect_set_group_epoch_entered_at_ns() + .returning(|_, _| Ok(())); }) .mem(mem) .build(); @@ -763,6 +765,8 @@ mod tests { .returning(|_id, _entity, _| Ok(vec![Cursor::v3_welcomes(0)])); db.expect_update_cursor().returning(|_, _, _| Ok(true)); db.expect_insert_or_replace_group().returning(Ok); + db.expect_set_group_epoch_entered_at_ns() + .returning(|_, _| Ok(())); }) .transaction_calls(|db: &mut MockDbQuery| { db.expect_update_cursor() diff --git a/crates/xmtp_mls/src/groups/welcomes/xmtp_welcome.rs b/crates/xmtp_mls/src/groups/welcomes/xmtp_welcome.rs index 8cf6e91a8f..fabe2c7184 100644 --- a/crates/xmtp_mls/src/groups/welcomes/xmtp_welcome.rs +++ b/crates/xmtp_mls/src/groups/welcomes/xmtp_welcome.rs @@ -513,6 +513,14 @@ where // For existing groups, this only updates the sequence_id (not membership_state). let stored_group = db.insert_or_replace_group(to_store)?; + // XIP-82: a welcome is the message by which this client enters + // the group's current epoch — the external-commit validator + // measures the `expire_in_ns` staleness bound from here. Set + // explicitly (not via the builder) so re-welcomes to an existing + // row update it too: `insert_or_replace_group` preserves most + // columns for existing groups. + db.set_group_epoch_entered_at_ns(&stored_group.id, welcome.timestamp())?; + StoredConsentRecord::stitch_dm_consent(&db, &stored_group)?; // Create a GroupUpdated payload diff --git a/crates/xmtp_mls/src/utils/bench/groups.rs b/crates/xmtp_mls/src/utils/bench/groups.rs index 44158a46d3..f4e69bc94e 100644 --- a/crates/xmtp_mls/src/utils/bench/groups.rs +++ b/crates/xmtp_mls/src/utils/bench/groups.rs @@ -121,6 +121,7 @@ pub async fn create_dm_with_consent( fork_details: Default::default(), commit_log_public_key: None, has_pending_leave_request: None, + epoch_entered_at_ns: None, installations_last_checked: now_ns(), is_commit_log_forked: None, last_message_ns: None,