diff --git a/CHANGELOG.md b/CHANGELOG.md index bddc7075..8b19731d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -190,6 +190,25 @@ archived by series under [docs/changelog/](docs/changelog/); see the Found while running it: a message that a direct stream took and then lost (a device that went away without closing the stream) is retried over direct carriers only and never handed to the mesh (#541). +- **`message_decryption_failed` says how old the frame is.** Two new fields, + `sealed_at_ms` (when the sender sealed the frame, sender clock, from the + cleartext envelope) and `session_established_at_ms` (when this device's + current 1:1 session with the sender was created or adopted, local clock), + both `null` when unknown. A frame sealed before the session existed cannot + have been sealed under it, so its failure is not evidence that the two sides + hold different sessions; an app's split-brain breaker can now leave it out. + The session time is recorded from this version on; a session established + earlier reports `null` until it is rebuilt. +- **Key packages advertise `ack_versions`.** A new capability list on the + key package payload, persisted per peer with the others. Entry 1 + (`ACK_UNDECRYPTABLE_V1`) says the sender reads `ack_status: + "undecryptable"` as "not delivered"; this version advertises it + unconditionally, and answers an undecryptable frame only to a peer that + advertised it, because an older SDK reads any ACK as a delivery. A peer + learns it from our next key package, which this SDK pushes on the first + discovery of each peer per launch (any inbound frame from them, a BLE + sighting, or a presence edge). Leaf devices do not advertise it. See + [Capability negotiation](docs/spec/capability-negotiation.md). ### Changed @@ -232,6 +251,40 @@ archived by series under [docs/changelog/](docs/changelog/); see the ### Fixed +- **A frame that can never decrypt gets an answer, and is not redelivered + forever.** A frame that failed with a spent ratchet generation or a dead + epoch, or that the pending-decryption queue gave up on, was left + unacknowledged so the sender would resend it re-sealed. A relay holding the + frame for an offline recipient read the silence as "not delivered yet" and + redelivered the same dead ciphertext on every connect, where it failed + again. The recipient now answers it with a delivery ACK carrying + `metadata["ack_status"] = "undecryptable"`, which a relay settles like any + ACK, but only when the frame arrived over the internet relay (these + failures are unauthenticated, so an answer on a mesh link would confirm + liveness to an injector in range) and the sender advertised `ack_versions: [1]` in its key + package (see Added). A sender that has not, or whose key package this + device has not seen since it upgraded, is answered with nothing, exactly as + before. The sender never reports such a message delivered: it re-seals the DM + under its current session and resends it once with the same id (verbatim + when the plaintext is gone, as for an outbox entry restored after a restart, + since a frame the recipient's queue dropped unattempted still decrypts); a + second answer for the same id resends nothing and leaves the entry to its + retry ladder, which ends in `message_failed` at max retries. Media chunks and frames whose envelope does not + parse keep their old handling. See + [Reserved metadata keys](docs/spec/wire-format.md#reserved-metadata-keys). +- **Frames the SDK sends on its own no longer push "new message" to an + offline peer.** A data-sync offer, a key package or a typing frame is the + same MLS ciphertext to the relay as a user's text, so a device that + reconnected or heard from a peer sent pushes with nothing behind them. + Every engine-originated frame now carries `metadata["no_alert"] = "1"` in + the cleartext envelope for the relay to honour. User messages (plain, rich, + forwarded, media), connection requests and accepts, group chat messages and + group invites do not. A document carried over the media path is the SDK's + own, so every one of its chunks carries it. A 1:1 Welcome carries it only + when no user message is waiting on the session, decided again on every + resend, and never for a session a document transfer opens. The key is + unsigned and receivers ignore it. See + [Reserved metadata keys](docs/spec/wire-format.md#reserved-metadata-keys). - **A message the mesh carried settles when its recipient answers.** A direct message to a peer no carrier could reach was handed to neighbours to carry, but registered no pending acknowledgement, and an acknowledgement with none diff --git a/bindings/react-native/src/types.ts b/bindings/react-native/src/types.ts index 5e331201..cc0d2ced 100644 --- a/bindings/react-native/src/types.ts +++ b/bindings/react-native/src/types.ts @@ -1343,6 +1343,19 @@ export interface MessageDecryptionFailedEvent extends BaseEvent { sender: string; code: DecryptionFailureCode; reason: string; + /** + * When the sender sealed the frame (sender clock, ms), from the cleartext + * envelope. Unauthenticated: a hint, never a security input. `null` unless + * this was a 1:1 decrypt attempt. + */ + sealed_at_ms: number | null; + /** + * When this device's current session with `sender` was created or adopted + * (local clock, ms); `null` when unknown. A frame sealed before it cannot + * have been sealed under this session, so its failure says nothing about + * whether the two sides hold the same session now. + */ + session_established_at_ms: number | null; } /** diff --git a/crates/offline-protocol-leaf/src/keypkg.rs b/crates/offline-protocol-leaf/src/keypkg.rs index a6167dbd..8f78efc6 100644 --- a/crates/offline-protocol-leaf/src/keypkg.rs +++ b/crates/offline-protocol-leaf/src/keypkg.rs @@ -131,6 +131,7 @@ pub(crate) fn payload( // the one that introduced the device at all. ctrl_versions: alloc::vec![CTRL_SIGN_V2], svc_versions: Vec::new(), + ack_versions: Vec::new(), nostr_pubkey: None, } } diff --git a/crates/offline-protocol-leaf/tests/phone_interop.rs b/crates/offline-protocol-leaf/tests/phone_interop.rs index 97b74a65..c704a2a3 100644 --- a/crates/offline-protocol-leaf/tests/phone_interop.rs +++ b/crates/offline-protocol-leaf/tests/phone_interop.rs @@ -509,6 +509,7 @@ fn a_key_package_that_claims_another_owner_is_refused() { data_versions: vec![], ctrl_versions: vec![offline_protocol_sealed::CTRL_SIGN_V2], svc_versions: Vec::new(), + ack_versions: Vec::new(), nostr_pubkey: None, }; @@ -1139,6 +1140,7 @@ fn a_peer_that_already_has_a_key_package_is_not_sent_another() { data_versions: vec![], ctrl_versions: vec![offline_protocol_sealed::CTRL_SIGN_V2], svc_versions: Vec::new(), + ack_versions: Vec::new(), nostr_pubkey: None, }; let frame = phone_control_frame( @@ -1723,6 +1725,7 @@ fn a_replayed_session_reset_does_not_tear_down_the_new_session() { data_versions: vec![], ctrl_versions: vec![offline_protocol_sealed::CTRL_SIGN_V2], svc_versions: Vec::new(), + ack_versions: Vec::new(), nostr_pubkey: None, }; let reset_frame = phone_control_frame( @@ -1822,6 +1825,7 @@ fn a_flood_of_strangers_cannot_displace_an_established_peer() { data_versions: vec![], ctrl_versions: vec![offline_protocol_sealed::CTRL_SIGN_V2], svc_versions: Vec::new(), + ack_versions: Vec::new(), nostr_pubkey: None, }; let frame = phone_control_frame( @@ -1854,6 +1858,7 @@ fn a_flood_of_strangers_cannot_displace_an_established_peer() { data_versions: vec![], ctrl_versions: vec![offline_protocol_sealed::CTRL_SIGN_V2], svc_versions: Vec::new(), + ack_versions: Vec::new(), nostr_pubkey: None, }; let frame = phone_control_frame( @@ -2280,6 +2285,7 @@ fn advertisement_body(user_id: &str) -> String { data_versions: vec![], ctrl_versions: vec![offline_protocol_sealed::CTRL_SIGN_V2], svc_versions: Vec::new(), + ack_versions: Vec::new(), nostr_pubkey: None, }; format!( @@ -3209,6 +3215,7 @@ fn a_reset_older_than_the_last_one_acted_on_is_refused() { data_versions: vec![], ctrl_versions: vec![offline_protocol_sealed::CTRL_SIGN_V2], svc_versions: Vec::new(), + ack_versions: Vec::new(), nostr_pubkey: None, }; phone_control_frame_at( @@ -3270,6 +3277,7 @@ fn a_spent_reset_stays_spent_across_a_power_cycle() { data_versions: vec![], ctrl_versions: vec![offline_protocol_sealed::CTRL_SIGN_V2], svc_versions: Vec::new(), + ack_versions: Vec::new(), nostr_pubkey: None, }; let reset_frame = phone_control_frame_at( @@ -3421,6 +3429,7 @@ fn a_session_reset_that_states_no_time_is_refused_its_teardown() { data_versions: vec![], ctrl_versions: vec![offline_protocol_sealed::CTRL_SIGN_V2], svc_versions: Vec::new(), + ack_versions: Vec::new(), nostr_pubkey: None, }; let content = format!( diff --git a/crates/offline-protocol-mls/src/manager.rs b/crates/offline-protocol-mls/src/manager.rs index a31b1454..c313a2ce 100644 --- a/crates/offline-protocol-mls/src/manager.rs +++ b/crates/offline-protocol-mls/src/manager.rs @@ -992,13 +992,40 @@ impl MlsManager { let key_type = StorageKeyType::ContactKeyPackage.as_str(); self.storage.delete(key_type, other_user_id)?; + self.stamp_session_established(&welcome.group_id); Ok(welcome) } /// Joins a session using a Welcome message. pub fn join_session(&self, welcome: &WelcomeMessage) -> Result { - self.session_manager.join_session(welcome) + let info = self.session_manager.join_session(welcome)?; + self.stamp_session_established(&welcome.group_id); + Ok(info) + } + + /// When the current 1:1 session with `other_user_id` was created or + /// adopted on this device (local clock, ms), or `None` when there is no + /// session or it predates the stamp. + /// + /// A frame sealed before this moment cannot have been sealed under this + /// session, so its failure to decrypt says nothing about whether the two + /// sides hold the same session now. + pub fn session_established_at_ms(&self, other_user_id: &str) -> Result> { + let session_id = self.session_manager.get_session_id(other_user_id)?; + Ok(self + .load_group_metadata(&session_id)? + .map(|metadata| metadata.created_at_ms)) + } + + /// Records the moment a session's group state was written, in the + /// group-metadata slot a 1:1 session otherwise leaves empty. Best effort: + /// the session exists either way, and a missing stamp only means + /// [`Self::session_established_at_ms`] answers `None`. + fn stamp_session_established(&self, session_id: &GroupId) { + if let Err(e) = self.save_group_metadata(session_id, &GroupMetadata::new(None)) { + warn!(session_id = %session_id, error = %e, "Failed to stamp session establishment time"); + } } /// Replaces an existing session with an incoming Welcome message. @@ -1032,7 +1059,7 @@ impl MlsManager { // Join using their Welcome. `join_session` adopts non-destructively // (stage-then-swap), so a retransmitted Welcome that re-stages is a safe // no-op rather than deleting and re-creating our existing session. - self.session_manager.join_session(welcome) + self.join_session(welcome) } /// Encrypts a message for a 1:1 session. @@ -1115,7 +1142,15 @@ impl MlsManager { } /// Deletes a 1:1 session. + /// + /// The establishment stamp goes first: an `Err` must mean the session is + /// still there, because callers clean up their own state for the peer + /// only once this succeeds. Deleting the session first and then failing + /// on the stamp left a peer recorded as confirmed with no session. pub fn delete_session(&self, other_user_id: &str) -> Result<()> { + let session_id = self.session_manager.get_session_id(other_user_id)?; + self.storage + .delete(StorageKeyType::GroupMetadata.as_str(), session_id.as_str())?; self.session_manager.delete_session(other_user_id) } @@ -3426,6 +3461,53 @@ mod tests { (alice, bob) } + use crate::storage::{StorageError, StorageResult}; + + /// Storage that refuses to delete a session's establishment stamp. + struct StampDeleteFails(InMemoryStorage); + + impl MlsStorage for StampDeleteFails { + fn store(&self, key_type: &str, key_id: &str, data: &[u8]) -> StorageResult<()> { + self.0.store(key_type, key_id, data) + } + fn load(&self, key_type: &str, key_id: &str) -> StorageResult>> { + self.0.load(key_type, key_id) + } + fn delete(&self, key_type: &str, key_id: &str) -> StorageResult<()> { + if key_type == StorageKeyType::GroupMetadata.as_str() { + return Err(StorageError::DeleteFailed("refused".to_string())); + } + self.0.delete(key_type, key_id) + } + fn list_keys(&self, key_type: &str) -> StorageResult> { + self.0.list_keys(key_type) + } + } + + /// The engine clears a peer's confirmed-session state only once + /// `delete_session` returns `Ok`, so an `Err` must leave the session + /// where it was, not half deleted. + #[test] + fn a_failed_delete_session_leaves_the_session_in_place() { + let storage: Arc = Arc::new(StampDeleteFails(InMemoryStorage::new())); + let (keys, address) = test_identity("alice"); + seed_identity(&storage, &keys); + let alice = MlsManager::new(address.to_string(), storage).unwrap(); + let bob = create_test_manager("bob"); + let bob_kp = bob.generate_key_package().unwrap(); + alice + .import_key_package(&addr("bob"), &bob_kp.key_package_data) + .unwrap(); + alice.create_session(&addr("bob")).unwrap(); + + assert!(alice.delete_session(&addr("bob")).is_err()); + assert!(alice.has_session(&addr("bob")).unwrap()); + assert!(alice + .session_established_at_ms(&addr("bob")) + .unwrap() + .is_some()); + } + #[test] fn test_session_decrypt_rejects_spoofed_sender() { let (alice, bob) = create_test_session(); diff --git a/crates/offline-protocol-sealed/src/keypackage.rs b/crates/offline-protocol-sealed/src/keypackage.rs index 7bf2f210..518dceca 100644 --- a/crates/offline-protocol-sealed/src/keypackage.rs +++ b/crates/offline-protocol-sealed/src/keypackage.rs @@ -184,6 +184,24 @@ pub struct KeyPackagePayload { #[serde(default)] pub svc_versions: Vec, + /// Delivery-acknowledgement statuses the sender reads (e.g. `[1]` for + /// [`ACK_UNDECRYPTABLE_V1`]). Absent on legacy nodes (`#[serde(default)]` + /// gives empty, meaning plain delivery ACKs only). + /// + /// Like `ctrl_versions` this says what the sender *accepts*: a peer + /// advertising entry 1 settles an ACK carrying + /// `ack_status = "undecryptable"` as "not delivered" (resend once). + /// One without it would read that ACK as a delivery, so a receiver answers + /// an undecryptable frame from such a peer with nothing at all. + /// + /// Trust boundary: a plaintext field like the lists above. Stripping it + /// returns the peer to unanswered undecryptable frames (the old + /// behaviour); forging it onto a legacy peer makes us send it ACKs it + /// reads as deliveries, which an attacker in that position could equally + /// forge directly. A feature negotiation, never a security control. + #[serde(default)] + pub ack_versions: Vec, + /// This install's Nostr public key (x-only, 64-char lowercase hex), so a /// peer can seal Nostr gift wraps to a key only this install holds. /// @@ -248,6 +266,11 @@ pub const CTRL_SIGN_V2: u8 = 2; /// service layer, with the MLS-authenticated sender as the requester. pub const SVC_SEALED_V1: u8 = 1; +/// Acknowledgement version advertised in [`KeyPackagePayload::ack_versions`]: +/// the sender reads `ack_status = "undecryptable"` on a delivery ACK as "the +/// recipient can never decrypt this frame" rather than as a delivery. +pub const ACK_UNDECRYPTABLE_V1: u8 = 1; + /// The conformance vectors for the key package payload. /// /// The chapter these pin is `docs/spec/capability-negotiation.md`, and they pin @@ -330,6 +353,7 @@ mod spec_vectors { ("data_versions", &got.data_versions), ("ctrl_versions", &got.ctrl_versions), ("svc_versions", &got.svc_versions), + ("ack_versions", &got.ack_versions), ] { assert_eq!(got, &u8s(&want[field]), "[{name}] {field}"); } @@ -368,6 +392,7 @@ mod spec_vectors { data_versions: Vec::new(), ctrl_versions: Vec::new(), svc_versions: Vec::new(), + ack_versions: Vec::new(), nostr_pubkey: None, }; let rendered = serde_json::to_string(&payload).expect("it serializes"); @@ -404,6 +429,7 @@ mod spec_vectors { "data_versions", "ctrl_versions", "svc_versions", + "ack_versions", "nostr_pubkey", "session_reset", ] { diff --git a/crates/offline-protocol-sealed/src/lib.rs b/crates/offline-protocol-sealed/src/lib.rs index a6d976cc..5c3e548d 100644 --- a/crates/offline-protocol-sealed/src/lib.rs +++ b/crates/offline-protocol-sealed/src/lib.rs @@ -83,7 +83,9 @@ pub use identity_assertion::{ encode_identity_assertion, parse_identity_assertion, IdentityAssertion, ED25519_SIGNATURE_LEN, IDENTITY_ASSERTION_MIN_LEN, }; -pub use keypackage::{KeyPackagePayload, CTRL_SIGN_V2, MLS_ENVELOPE_COMPACT_V1, SVC_SEALED_V1}; +pub use keypackage::{ + KeyPackagePayload, ACK_UNDECRYPTABLE_V1, CTRL_SIGN_V2, MLS_ENVELOPE_COMPACT_V1, SVC_SEALED_V1, +}; #[cfg(all(test, feature = "std"))] mod manifest_guard_tests { diff --git a/crates/offline-protocol-sealed/tests/data/key-package-v1.vectors.json b/crates/offline-protocol-sealed/tests/data/key-package-v1.vectors.json index ca433661..f6d53199 100644 --- a/crates/offline-protocol-sealed/tests/data/key-package-v1.vectors.json +++ b/crates/offline-protocol-sealed/tests/data/key-package-v1.vectors.json @@ -33,13 +33,14 @@ "data_versions": [], "ctrl_versions": [], "svc_versions": [], + "ack_versions": [], "nostr_pubkey": null }, "note": "Every absent list defaults to empty, and empty selects the floor. This is rule 2 of capability negotiation expressed as a parse: absence is never an error." }, { "name": "a fully capable peer", - "json": "{\"user_id\":\"off1abc\",\"key_package_data\":[],\"remaining_lifetime_ms\":2419200000,\"timestamp_ms\":1700000000000,\"session_reset\":true,\"wire_versions\":[1],\"env_versions\":[1],\"rich_versions\":[1],\"data_versions\":[1,2,3],\"ctrl_versions\":[2],\"svc_versions\":[1],\"nostr_pubkey\":\"abcd\"}", + "json": "{\"user_id\":\"off1abc\",\"key_package_data\":[],\"remaining_lifetime_ms\":2419200000,\"timestamp_ms\":1700000000000,\"session_reset\":true,\"wire_versions\":[1],\"env_versions\":[1],\"rich_versions\":[1],\"data_versions\":[1,2,3],\"ctrl_versions\":[2],\"svc_versions\":[1],\"ack_versions\":[1],\"nostr_pubkey\":\"abcd\"}", "expect": { "user_id": "off1abc", "key_package_data": [], @@ -66,6 +67,9 @@ "svc_versions": [ 1 ], + "ack_versions": [ + 1 + ], "nostr_pubkey": "abcd" }, "note": "data_versions entries are read independently, so [1,2,3] is not a single level." @@ -85,6 +89,7 @@ "data_versions": [], "ctrl_versions": [], "svc_versions": [], + "ack_versions": [], "nostr_pubkey": null }, "note": "An unknown field is ignored rather than refused. A decoder that rejects it cannot be sent a newer key package, which is the frame that carries every other capability." @@ -104,6 +109,7 @@ "data_versions": [], "ctrl_versions": [], "svc_versions": [], + "ack_versions": [], "nostr_pubkey": null }, "note": "Absent and null mean the same thing on the way in. On the way out the field is omitted rather than written null." @@ -111,7 +117,7 @@ ], "emission_is_not_pinned": { "note": "Illustrative only. A conforming peer parses this; it is not required to produce these bytes, and this file does not assert that it does.", - "example": "{\"user_id\":\"off1abc\",\"key_package_data\":[],\"remaining_lifetime_ms\":0,\"timestamp_ms\":0,\"session_reset\":false,\"wire_versions\":[],\"env_versions\":[],\"rich_versions\":[],\"data_versions\":[],\"ctrl_versions\":[],\"svc_versions\":[]}", + "example": "{\"user_id\":\"off1abc\",\"key_package_data\":[],\"remaining_lifetime_ms\":0,\"timestamp_ms\":0,\"session_reset\":false,\"wire_versions\":[],\"env_versions\":[],\"rich_versions\":[],\"data_versions\":[],\"ctrl_versions\":[],\"svc_versions\":[],\"ack_versions\":[]}", "nostr_pubkey_absent_is_omitted_not_null": true } } diff --git a/crates/offline-protocol/src/constants.rs b/crates/offline-protocol/src/constants.rs index ab4f07ac..7ab7b819 100644 --- a/crates/offline-protocol/src/constants.rs +++ b/crates/offline-protocol/src/constants.rs @@ -17,6 +17,28 @@ pub const ACK_HOP_COUNT_KEY: &str = "ack_hop_count"; /// Metadata key for ACK messages indicating the transport used. pub const ACK_TRANSPORT_KEY: &str = "ack_transport"; +/// Metadata key on an ACK that answers a frame without delivering it. Absent +/// on an ordinary delivery ACK. +pub const ACK_STATUS_KEY: &str = "ack_status"; + +/// [`ACK_STATUS_KEY`] value: the recipient can never decrypt the frame it +/// names (a spent ratchet generation, a dead epoch, or the pending-decryption +/// queue gave up on it). The sender resends the message once; it is +/// never `message_delivered`. Still a delivery ACK in shape, so a relay holding +/// the frame drops it. +pub const ACK_STATUS_UNDECRYPTABLE: &str = "undecryptable"; + +/// Metadata key marking a frame the SDK sent on its own: control traffic, +/// key packages, data-sync frames, receipts, typing, presence, and a 1:1 +/// Welcome no user message is waiting on. +/// +/// Set to `"1"` in the cleartext envelope so the relay can tell these from +/// the user's own messages (both are opaque `__MLS_ENC__` ciphertext to it) +/// and skip the "new message" push for an offline recipient. Frames built +/// for the application's user-content sends never carry it. Unsigned like +/// all metadata; receivers ignore it. +pub const NO_ALERT_KEY: &str = "no_alert"; + /// Maximum number of entries in the outbox before evicting oldest entries. /// This prevents unbounded memory growth when messages cannot be delivered. pub const MAX_OUTBOX_ENTRIES: usize = 500; diff --git a/crates/offline-protocol/src/events.rs b/crates/offline-protocol/src/events.rs index bb3d8cb4..8be290f2 100644 --- a/crates/offline-protocol/src/events.rs +++ b/crates/offline-protocol/src/events.rs @@ -736,6 +736,17 @@ pub enum Event { code: DecryptionFailureCode, /// Clear failure reason for application handling/logging. reason: String, + /// When the sender sealed the frame (sender clock, ms), read from the + /// cleartext envelope. Unauthenticated: a hint for local heuristics, + /// never for security decisions. `None` when the failure is not a 1:1 + /// decrypt attempt (pending-queue drop, unparseable envelope, group). + sealed_at_ms: Option, + /// When this device's current session with `sender` was created or + /// adopted (local clock, ms). `None` when there is no session, or it + /// was established by an SDK that did not record the time. A frame + /// sealed before this cannot have been sealed under this session, so + /// its failure says nothing about whether the two sides agree now. + session_established_at_ms: Option, }, /// Transport was switched by DORS. @@ -2036,6 +2047,29 @@ impl Event { sender, code, reason, + sealed_at_ms: None, + session_established_at_ms: None, + } + } + + /// Creates a MessageDecryptionFailed event for a failed 1:1 decrypt, + /// carrying the frame's seal time and the current session's + /// establishment time (see the variant's fields). + pub fn message_decryption_failed_for_frame( + message_id: MessageId, + sender: String, + code: DecryptionFailureCode, + reason: String, + sealed_at_ms: u64, + session_established_at_ms: Option, + ) -> Self { + Self::MessageDecryptionFailed { + message_id: message_id.as_str(), + sender, + code, + reason, + sealed_at_ms: Some(sealed_at_ms), + session_established_at_ms, } } @@ -3130,12 +3164,16 @@ impl fmt::Debug for Event { sender: _, code, reason, + sealed_at_ms, + session_established_at_ms, } => f .debug_struct("MessageDecryptionFailed") .field("message_id", message_id) .field("sender", &"[REDACTED]") .field("code", code) .field("reason", reason) + .field("sealed_at_ms", sealed_at_ms) + .field("session_established_at_ms", session_established_at_ms) .finish(), Self::TransportSwitched { from, to, reason } => f .debug_struct("TransportSwitched") diff --git a/crates/offline-protocol/src/group_mesh.rs b/crates/offline-protocol/src/group_mesh.rs index 499f8b20..f298eff8 100644 --- a/crates/offline-protocol/src/group_mesh.rs +++ b/crates/offline-protocol/src/group_mesh.rs @@ -3431,7 +3431,13 @@ impl OfflineProtocol { serde_json::to_string(&welcome_payload) .map_err(|e| Error::Serialization(format!("Serialize welcome: {}", e)))? ); - self.send_internal_message(invitee_user_id, welcome_content, MessagePriority::High)?; + // Alerting: being added to a group is something the invitee hears about. + self.send_internal_message_alerting( + invitee_user_id, + welcome_content, + MessagePriority::High, + true, + )?; // Send Commit to all existing members (excluding self and invitee) // so they can process it and advance their MLS epoch @@ -4185,6 +4191,7 @@ impl OfflineProtocol { reply_to_msg, payload_forward_info, priority, + true, )?; let member_count = succeeded_members.len() as u32; @@ -4858,7 +4865,7 @@ impl OfflineProtocol { // retry policy for registration is the `relay_register_pending` // tracker armed just below, not the message ACK ladder — which could // never settle this frame and would re-send it ~10 times. - match self.send_relay_hint_message(content, MessagePriority::Medium) { + match self.send_relay_hint_message(content, MessagePriority::Medium, false) { Ok(_) => { // Arm the ack correlation: only a `__GROUP_CREATED__` that // answers an outstanding registration may set `relay_synced`. @@ -5030,7 +5037,7 @@ impl OfflineProtocol { serde_json::to_string(payload) .map_err(|e| Error::Serialization(format!("Serialize relay broadcast: {}", e)))? ); - self.send_relay_hint_message(content, MessagePriority::Medium)?; + self.send_relay_hint_message(content, MessagePriority::Medium, true)?; Ok(()) } @@ -5041,6 +5048,10 @@ impl OfflineProtocol { /// broadcast — receivers then dedup/emit under that id (see /// [`GroupMlsMessagePayload::message_id`]). /// + /// `alert` is true for a chat message and false for a replication frame + /// (see [`Self::send_internal_message_alerting`]): both are + /// `__GRP_MLS_MSG__` ciphertext, so the caller is the only one who knows. + /// /// Returns `(frame ids, succeeded members, failed members)`; errors only /// on payload serialization. #[allow(clippy::too_many_arguments)] @@ -5054,6 +5065,7 @@ impl OfflineProtocol { reply_to: Option<&str>, forward_info: Option, priority: MessagePriority, + alert: bool, ) -> Result<(Vec, Vec, Vec)> { let msg_payload = GroupMlsMessagePayload { group_id: group_id.to_string(), @@ -5078,7 +5090,8 @@ impl OfflineProtocol { if member == &self_id { continue; } - match self.send_internal_message(member, base_content.clone(), priority) { + match self.send_internal_message_alerting(member, base_content.clone(), priority, alert) + { Ok(mid) => { message_ids.push(mid); succeeded_members.push(member.clone()); @@ -5243,6 +5256,7 @@ impl OfflineProtocol { None, None, MessagePriority::Low, + false, )?; if !failed.is_empty() { // Not an error: a member that could not be reached now has the @@ -5518,6 +5532,7 @@ impl OfflineProtocol { entry.reply_to.as_deref(), entry.forward_info.clone(), entry.priority, + true, ) { Ok((_ids, succeeded, _failed)) => reissued = succeeded, Err(e) => { @@ -6112,6 +6127,7 @@ impl OfflineProtocol { entry.reply_to.as_deref(), entry.forward_info.clone(), entry.priority, + true, ) { error!( group_id = %entry.group_id, diff --git a/crates/offline-protocol/src/protocol/blocking.rs b/crates/offline-protocol/src/protocol/blocking.rs index 03f71807..fc0b6ad4 100644 --- a/crates/offline-protocol/src/protocol/blocking.rs +++ b/crates/offline-protocol/src/protocol/blocking.rs @@ -144,6 +144,7 @@ impl OfflineProtocol { // would make "block, then unblock" a way to walk a peer's own // protection back, requestable by anyone who can get a user to do it. self.peer_ctrl_freshness.remove(user_id); + self.peer_undecryptable_ack.remove(user_id); self.peer_rich_payload.remove(user_id); self.peer_rich_attested.remove(user_id); // Advertised half only, as above: `svc_sealed_proved_peers` is a @@ -1445,6 +1446,7 @@ mod tests { data_versions: Vec::new(), ctrl_versions: Vec::new(), svc_versions: Vec::new(), + ack_versions: Vec::new(), nostr_pubkey: None, }; let content = serde_json::to_string(&reset_payload).unwrap(); diff --git a/crates/offline-protocol/src/protocol/decryption_queue.rs b/crates/offline-protocol/src/protocol/decryption_queue.rs index 9162a872..e37110c3 100644 --- a/crates/offline-protocol/src/protocol/decryption_queue.rs +++ b/crates/offline-protocol/src/protocol/decryption_queue.rs @@ -48,6 +48,9 @@ struct EntryRef { pub(crate) struct DroppedPendingMessage { pub(crate) message: Message, pub(crate) reason: &'static str, + /// Transport the frame arrived on, when known: the route its + /// `undecryptable` answer takes. + pub(crate) received_via: Option, } /// What [`PendingDecryptionQueue::enqueue_via`] did with the incoming frame, @@ -589,6 +592,7 @@ impl PendingDecryptionQueue { expired_entries.push(DroppedPendingMessage { message: expired.message, reason: DropReason::TtlExpired.as_str(), + received_via: expired.received_via, }); } } @@ -650,6 +654,7 @@ impl PendingDecryptionQueue { evicted.push(DroppedPendingMessage { message: expired.message, reason: DropReason::TtlExpired.as_str(), + received_via: expired.received_via, }); } else { self.global_order.pop_front(); @@ -709,6 +714,7 @@ impl PendingDecryptionQueue { refused: (admission == Admission::Refused).then(|| DroppedPendingMessage { message: message.clone(), reason: DropReason::OverflowDropNewest.as_str(), + received_via: arrival_transport, }), } } @@ -806,6 +812,7 @@ impl PendingDecryptionQueue { dropped.push(DroppedPendingMessage { message: evicted.message, reason: DropReason::OverflowDropOldest.as_str(), + received_via: evicted.received_via, }); evicted_any = true; } @@ -882,6 +889,7 @@ impl PendingDecryptionQueue { dropped.push(DroppedPendingMessage { message: evicted.message, reason: DropReason::OverflowDropOldest.as_str(), + received_via: evicted.received_via, }); } None => { @@ -930,6 +938,7 @@ impl PendingDecryptionQueue { Some(evicted) => dropped.push(DroppedPendingMessage { message: evicted.message, reason: DropReason::OverflowDropOldest.as_str(), + received_via: evicted.received_via, }), None => { self.record_eviction_failure( @@ -980,6 +989,7 @@ impl PendingDecryptionQueue { Some(evicted) => dropped.push(DroppedPendingMessage { message: evicted.message, reason: DropReason::OverflowDropOldest.as_str(), + received_via: evicted.received_via, }), None => { self.record_eviction_failure( diff --git a/crates/offline-protocol/src/protocol/message_dispatch.rs b/crates/offline-protocol/src/protocol/message_dispatch.rs index 99c63bf3..12b590e0 100644 --- a/crates/offline-protocol/src/protocol/message_dispatch.rs +++ b/crates/offline-protocol/src/protocol/message_dispatch.rs @@ -176,6 +176,25 @@ impl OfflineProtocol { self.peer_svc_sealed.remove(sender); } + // Record whether this peer reads an `undecryptable` ACK as "not + // delivered", so a frame of theirs that can never decrypt may be + // answered with one. Same shape as the capabilities above; not + // gated on our config, since what it gates is our reply to them. + // Forgetting a peer costs the old behaviour: no answer at all. + if payload + .ack_versions + .contains(&offline_protocol_sealed::ACK_UNDECRYPTABLE_V1) + { + if !self.peer_undecryptable_ack.contains(sender) + && self.peer_undecryptable_ack.len() >= MAX_KEY_PACKAGE_SENT_TO + { + self.peer_undecryptable_ack.clear(); + } + self.peer_undecryptable_ack.insert(sender.to_string()); + } else { + self.peer_undecryptable_ack.remove(sender); + } + // Record whether this peer replicates documents, so the data // layer may push `__DATA_V1__` frames to them. Same shape again: // gated by our own kill switch, removed when a fresh key package @@ -369,6 +388,7 @@ impl OfflineProtocol { &payload.data_versions, &payload.ctrl_versions, if signed { &payload.svc_versions } else { &[] }, + &payload.ack_versions, advertised_nostr_pubkey.as_deref(), ); if caps.is_any() { @@ -1273,9 +1293,10 @@ impl OfflineProtocol { // the *message* is not lost — Tier 2 re-seals // every resend against the peer's current // session, so the sender's next attempt carries - // a fresh generation that can decrypt. Withhold - // the ACK (the receive loop's `Deferred` arm) so - // the sender keeps that lever, exactly like the + // a fresh generation that can decrypt. Answer + // `undecryptable` rather than "delivered" (the + // receive loop's `Undecryptable` arm) so the + // sender pulls that lever, exactly like the // desync path above. // // What this must NOT do is re-key. Desync stays @@ -1292,7 +1313,7 @@ impl OfflineProtocol { sender = %sender, error = %e, error_code = session_state_error.code(), - "Failed to decrypt message; withholding ACK so the sender can resend" + "Failed to decrypt message; withholding the delivery ACK so the sender can re-seal" ); DecryptResult::Failed { sender: sender.to_string(), @@ -1469,9 +1490,10 @@ impl OfflineProtocol { } => { // The session exists but is out of epoch sync (the two sides // diverged). Trigger a rate-limited re-key to heal the channel - // for future traffic, and return Deferred so the receive loop - // withholds the ACK and unmarks the id — the sender's ACK was - // a lie before (message dropped, sender told "delivered"). + // for future traffic, and return Undecryptable so the receive + // loop unmarks the id and answers `undecryptable` instead of + // "delivered": the sender's ACK was a lie before (message + // dropped, sender told "delivered"). // // We deliberately do NOT enqueue: unlike the not-yet-ready // case, this ciphertext is sealed to the now-dead epoch and @@ -1482,7 +1504,7 @@ impl OfflineProtocol { // retries surface an honest undeliverable instead of silent // loss, which is strictly better than the lying ACK. self.schedule_session_rekey(&sender_owned); - Some(InternalMessageResult::Deferred) + Some(InternalMessageResult::Undecryptable) } DecryptResult::Failed { sender: sender_owned, @@ -1496,36 +1518,52 @@ impl OfflineProtocol { kind, MlsOperationContext::Receive, ); + // The app's split-brain heuristics need to tell a stale + // frame (sealed before this session existed) from a live + // one; only a 1:1 session has a single establishment time. + let session_established_at_ms = if group_id.starts_with("session:") { + self.mls_manager + .as_ref() + .and_then(|mls| mls.read().ok()) + .and_then(|manager| { + manager + .session_established_at_ms(&sender_owned) + .ok() + .flatten() + }) + } else { + None + }; if let Ok(state) = lock_shared_state(&self.shared_state) { // On the retriable path this event is **advisory, not - // terminal**, and fires once per failed attempt rather - // than once per message: the frame was not ACKed, so the - // sender resends and each resend that still fails - // reports again. It is bounded by the sender's ACK retry - // budget, after which the sender settles the message as - // an honest `MessageFailed`. + // terminal**: the frame is never delivery-ACKed (and is + // answered `undecryptable` when it came over the relay), + // so the sender re-seals the message under its current + // session, and a resend that still fails reports again. The sender's retry ladder settles it + // as an honest `MessageFailed` if it never decrypts. let reason = if retriable { format!( - "Failed to decrypt MLS message ({kind:?}); not acknowledged, so the sender's resend can still deliver it" + "Failed to decrypt MLS message ({kind:?}); not delivery-ACKed, so the sender can re-seal it" ) } else { format!("Failed to decrypt MLS message ({kind:?})") }; - state.emit_event(Event::message_decryption_failed( + state.emit_event(Event::message_decryption_failed_for_frame( message.id.clone(), sender_owned.clone(), Self::decryption_failure_code_from_kind(kind), reason, + encrypted.timestamp_ms, + session_established_at_ms, )); } if retriable { - // Reuses the Deferred atom (skip ACK + `unmark_seen`) - // without enqueueing: like a desync, this ciphertext is - // dead — OpenMLS consumed the ratchet generation on the + // Not enqueued: like a desync, this ciphertext is dead. + // OpenMLS consumed the ratchet generation on the // failed attempt, so a queued copy could never drain. // Recovery is the sender's re-sealed resend, not this // frame. - Some(InternalMessageResult::Deferred) + Some(InternalMessageResult::Undecryptable) } else { Some(InternalMessageResult::Consumed) } diff --git a/crates/offline-protocol/src/protocol/mod.rs b/crates/offline-protocol/src/protocol/mod.rs index 6d9c002b..7239e605 100644 --- a/crates/offline-protocol/src/protocol/mod.rs +++ b/crates/offline-protocol/src/protocol/mod.rs @@ -183,6 +183,12 @@ pub struct OfflineProtocol { /// lives on the `OutboxEntry` itself. See [`OutboxReseal`]. pending_reseal: HashMap, + /// Outbox ids already re-sealed and resent once in answer to an + /// `undecryptable` ACK. A second such answer leaves the entry to its retry + /// ladder, so two devices cannot trade undecryptable ACKs and resends at + /// network speed. Cleared with the outbox entry. + undecryptable_resent: HashSet, + /// Dedicated outbox for file chunk messages, separate from the main outbox /// to prevent large file transfers from evicting regular messages. media_outbox: HashMap, @@ -376,6 +382,14 @@ pub struct OfflineProtocol { /// [`CTRL_SIGN_V2`]: offline_protocol_sealed::CTRL_SIGN_V2 peer_ctrl_freshness: std::collections::HashSet, + /// Peers that read an `undecryptable` ACK as "not delivered" + /// ([`ACK_UNDECRYPTABLE_V1`] in `ack_versions`). Only these are answered + /// when a frame of theirs can never decrypt: an older SDK reads any ACK as + /// a delivery. Same lifecycle as `peer_ctrl_freshness` above. + /// + /// [`ACK_UNDECRYPTABLE_V1`]: offline_protocol_sealed::ACK_UNDECRYPTABLE_V1 + pub(crate) peer_undecryptable_ack: std::collections::HashSet, + /// Peers that have presented a control-frame signature over the /// freshness-bound payload, so their frames under the older one are /// refused. @@ -1176,6 +1190,7 @@ impl OfflineProtocol { shared_state: Arc::new(Mutex::new(SharedState::new())), outbox: HashMap::new(), pending_reseal: HashMap::new(), + undecryptable_resent: HashSet::new(), media_outbox: HashMap::new(), dm_unreachable_parks: HashMap::new(), unrouted_acks: VecDeque::new(), @@ -1193,6 +1208,7 @@ impl OfflineProtocol { encryption_capable_peers: std::collections::HashSet::new(), peer_compact_envelope: std::collections::HashSet::new(), peer_ctrl_freshness: std::collections::HashSet::new(), + peer_undecryptable_ack: std::collections::HashSet::new(), control_freshness_peers: std::collections::HashSet::new(), control_reset_watermark: std::collections::HashMap::new(), peer_rich_payload: std::collections::HashSet::new(), @@ -1401,6 +1417,7 @@ impl OfflineProtocol { let previous_outbox = self.outbox.clone(); let previous_peer_compact_envelope = self.peer_compact_envelope.clone(); let previous_peer_ctrl_freshness = self.peer_ctrl_freshness.clone(); + let previous_peer_undecryptable_ack = self.peer_undecryptable_ack.clone(); // Snapshotted with its siblings so a failed re-initialize restores it // too. It is a ratchet, so the direction that matters is that a // rollback must not *lose* it: a peer we hold to the newer payload has @@ -1572,6 +1589,7 @@ impl OfflineProtocol { self.outbox = previous_outbox; self.peer_compact_envelope = previous_peer_compact_envelope; self.peer_ctrl_freshness = previous_peer_ctrl_freshness; + self.peer_undecryptable_ack = previous_peer_undecryptable_ack; self.control_freshness_peers = previous_control_freshness_peers; self.control_reset_watermark = previous_control_reset_watermark; self.peer_rich_payload = previous_peer_rich_payload; @@ -3057,7 +3075,7 @@ impl OfflineProtocol { self.mark_encryption_capable(peer_id); // Send welcome message to peer - let welcome_sent = self.send_welcome_message(peer_id, &welcome)?; + let welcome_sent = self.send_welcome_message(peer_id, &welcome, false)?; // All operations succeeded, now safe to remove the key package self.pending_key_packages.remove(peer_id); diff --git a/crates/offline-protocol/src/protocol/pending_queue.rs b/crates/offline-protocol/src/protocol/pending_queue.rs index cd5e342c..52030b67 100644 --- a/crates/offline-protocol/src/protocol/pending_queue.rs +++ b/crates/offline-protocol/src/protocol/pending_queue.rs @@ -49,7 +49,9 @@ impl OfflineProtocol { self.delete_pending_decrypt_entries_from_storage( outcome.dropped.iter().map(|entry| &entry.message.id), ); - let events = Self::pending_drop_events(outcome.dropped.into_iter().chain(outcome.refused)); + let gone: Vec<_> = outcome.dropped.into_iter().chain(outcome.refused).collect(); + self.answer_dropped_pending(&gone); + let events = Self::pending_drop_events(gone); self.emit_pending_drop_events(events); if outcome.admitted { self.persist_pending_decrypt_entry( @@ -139,9 +141,10 @@ impl OfflineProtocol { /// so neither the file_id nor the text can be named here.) /// /// Under the deferred-ACK model this is **advisory, not terminal**: an - /// evicted frame was never ACKed, so a sender still retrying resends it, and - /// the resend re-enters the queue and can still complete once the session - /// confirms. The event says the message is *at risk*, not that it has + /// evicted frame is never delivery-ACKed (one that came over the relay is + /// answered `undecryptable`, see `answer_dropped_pending`), so a sender + /// still retrying resends it, and the resend re-enters the queue and can + /// still complete once the session confirms. The event says the message is *at risk*, not that it has /// failed — for media the terminal signal is `FileReceiveFailed`. It is /// emitted for text as well as for chunks: text drops used to be /// metrics-only, which left an app with no way to distinguish "the sender @@ -153,10 +156,29 @@ impl OfflineProtocol { // to mirror what the queue holds, and a record that outlived its entry // would be restored, re-dropped and re-reported on the next launch. self.delete_pending_decrypt_entries_from_storage(dropped.iter().map(|e| &e.message.id)); + self.answer_dropped_pending(&dropped); let events = Self::pending_drop_events(dropped); self.emit_pending_drop_events(events); } + /// The queue gave up on these frames, so those that came over the relay + /// are answered `undecryptable`: the relay drops its copy instead of + /// redelivering it into the same queue on every connect, and the sender + /// resends the message once (see `send_undecryptable_ack`, which also + /// holds the transport and capability gates). Media chunks are left alone: + /// their recovery is the transfer's own resend, not a re-seal. A frame + /// with no recorded arrival transport stays unanswered, as before. + fn answer_dropped_pending(&mut self, dropped: &[DroppedPendingMessage]) { + for entry in dropped { + if entry.message.content_type == ContentType::FileChunk { + continue; + } + if let Some(transport) = entry.received_via { + self.send_undecryptable_ack(&entry.message, transport); + } + } + } + /// Logs each dropped frame and builds its `PendingQueueDropped` event /// without emitting it. Split from the emit because the restore path has /// to defer its events until the event pipeline is live. @@ -398,40 +420,37 @@ impl OfflineProtocol { debug!(message_id = %msg.id, "Delayed message was consumed internally"); } InternalMessageResult::Deferred => { - // Still undecryptable during a drain. Not ACKed and the - // id is not re-marked, so the sender's resend still + // Still not decryptable during a drain. Not ACKed and + // the id is not re-marked, so the sender's resend still // recovers the message instead of being told // "delivered" for a frame we dropped. // - // Deliberately does NOT re-queue here, because each of - // the three ways a drained frame reaches this arm has - // already settled the question: - // - // - Session not ready (unexpected post-confirmation): - // `handle_encrypted_message` re-queued it itself - // before returning `Deferred`, so a re-queue here is - // redundant — `enqueue` is idempotent by id. - // - Epoch desync, and a hard crypto/transport failure - // (the routine case): both refuse to enqueue on the - // live path for a reason that holds just as well - // here — the ciphertext is sealed to a dead epoch, or - // the failed attempt already spent its ratchet - // generation, so no number of later drains can make - // this frame decrypt. - // - // Re-queuing the latter two used to look free, but the - // drain *removes* the entry before processing, so the - // re-enqueue missed `enqueue`'s idempotency check and - // re-stamped `received_at`. A frame that can never - // decrypt then had its TTL restarted by every drain, - // outliving its budget and re-reporting an advisory - // decrypt failure each time — including after the - // sender's re-sealed resend had already delivered the - // same id. Dropping it costs nothing the un-ACK does - // not already cover. + // Deliberately does NOT re-queue here: the only way a + // drained frame reaches this arm is session not ready + // (unexpected post-confirmation), and + // `handle_encrypted_message` re-queued it itself before + // returning `Deferred`, and `enqueue` is idempotent by id. + // (A queued frame parsed at receipt, so the unparseable + // shape cannot occur here.) + debug!( + message_id = %msg.id, + "Delayed message still not decryptable during drain" + ); + } + InternalMessageResult::Undecryptable => { + // Dead ciphertext (the drain's attempt spent the + // ratchet generation, or it is sealed to an epoch that + // is not ours), so the queue gives up on it. Not + // re-queued: a re-enqueue would miss `enqueue`'s + // idempotency check (the drain removed the entry) and + // restart the TTL of a frame that can never decrypt. + // Answered (relay arrivals only) so the sender re-seals. + if let Some(transport) = received_via { + self.send_undecryptable_ack(&msg, transport); + } debug!( message_id = %msg.id, - "Delayed message still undecryptable during drain; dropping the queued copy (recovery is the sender's resend)" + "Delayed message undecryptable during drain; dropping the queued copy" ); } InternalMessageResult::SecurityRejected => { diff --git a/crates/offline-protocol/src/protocol/receive.rs b/crates/offline-protocol/src/protocol/receive.rs index 66db014e..454f3c49 100644 --- a/crates/offline-protocol/src/protocol/receive.rs +++ b/crates/offline-protocol/src/protocol/receive.rs @@ -247,11 +247,9 @@ impl OfflineProtocol { // The message was not delivered, but the sender // can still recover it by resending: it could not // be decrypted *yet* (session not ready, so queued - // for delayed decryption), or it is undecryptable - // as it stands (epoch desync, or a hard crypto - // failure while `crypto_recovery_enabled`) and was - // dropped without queueing. See the three - // conditions on `InternalMessageResult::Deferred`. + // for delayed decryption), or its envelope did not + // parse. See the conditions on + // `InternalMessageResult::Deferred`. // // Do NOT send a delivery ACK and do NOT keep the id // dedup-marked: the message is not delivered, so the @@ -261,12 +259,20 @@ impl OfflineProtocol { // confirming drains it: the message is surfaced and // the id re-marked (see `process_pending_decryption`), // so the sender's next resend is then deduped + - // re-ACKed. For the other two, recovery is that - // resend itself — re-sealed against a live - // generation by Tier 2. + // re-ACKed. For an unparseable envelope, recovery + // is that resend itself. self.unmark_seen_persisted(&message.id); continue; } + InternalMessageResult::Undecryptable => { + // As `Deferred`, but answered: the sender's + // re-sealed resend must still re-enter + // processing, and whoever holds this copy + // must stop redelivering it. + self.unmark_seen_persisted(&message.id); + self.send_undecryptable_ack(&message, transport_used); + continue; + } InternalMessageResult::Decrypted(plaintext) => { was_decrypted = true; Self::apply_decrypted_content(&mut message, plaintext); diff --git a/crates/offline-protocol/src/protocol/send.rs b/crates/offline-protocol/src/protocol/send.rs index 419dad4b..23bf8205 100644 --- a/crates/offline-protocol/src/protocol/send.rs +++ b/crates/offline-protocol/src/protocol/send.rs @@ -18,7 +18,8 @@ use super::{ }; use super::{classify_transport_send_error, send_failure_token}; use crate::constants::{ - ACK_FOR_KEY, ACK_HOP_COUNT_KEY, ACK_TRANSPORT_KEY, MAX_FORWARD_COUNT, MAX_OUTBOX_ENTRIES, + ACK_FOR_KEY, ACK_HOP_COUNT_KEY, ACK_STATUS_KEY, ACK_STATUS_UNDECRYPTABLE, ACK_TRANSPORT_KEY, + MAX_FORWARD_COUNT, MAX_OUTBOX_ENTRIES, }; use crate::events::{DecryptionFailureCode, Event, PresenceStatus}; use crate::file_transfer::{FileChunk, OutboundTransferState}; @@ -345,7 +346,7 @@ impl OfflineProtocol { // Create and dispatch the forwarded message let mut message = - self.create_message(&recipient_str, final_content, Some(priority), None)?; + self.create_message_for_app(&recipient_str, final_content, Some(priority), None, None)?; message.forwarded_from = Some(forward_info); message.content_type = original_message.content_type; message.media_metadata = original_message.media_metadata.clone(); @@ -523,6 +524,21 @@ impl OfflineProtocol { recipient: &str, content: String, priority: MessagePriority, + ) -> Result { + self.send_internal_message_alerting(recipient, content, priority, false) + } + + /// [`Self::send_internal_message`], with `alert` choosing whether the + /// relay may push for the frame. `true` only for the control frames that + /// carry something the user acts on or reads: a connection request or + /// accept, and a group chat message. Everything else is + /// [`crate::constants::NO_ALERT_KEY`]-marked. + pub(crate) fn send_internal_message_alerting( + &mut self, + recipient: &str, + content: String, + priority: MessagePriority, + alert: bool, ) -> Result { { let state = lock_shared_state(&self.shared_state)?; @@ -531,7 +547,11 @@ impl OfflineProtocol { } } - let mut message = self.create_message(recipient, content, Some(priority), None)?; + let mut message = if alert { + self.create_message_for_app(recipient, content, Some(priority), None, None)? + } else { + self.create_message(recipient, content, Some(priority), None)? + }; self.sign_control_message(&mut message)?; self.dispatch_internal_message(message) } @@ -644,10 +664,14 @@ impl OfflineProtocol { /// sent over a forced transport. A hint frame is not user traffic, and /// announcing a switch for one would report a transport change the app's /// messages did not take. + /// + /// `alert` is false for a registration and true for a broadcast, which + /// carries a group chat message (see [`Self::send_internal_message_alerting`]). pub(crate) fn send_relay_hint_message( &mut self, content: String, priority: MessagePriority, + alert: bool, ) -> Result { { let state = lock_shared_state(&self.shared_state)?; @@ -657,7 +681,11 @@ impl OfflineProtocol { } let self_id = self.local_id.clone(); - let mut message = self.create_message(&self_id, content, Some(priority), None)?; + let mut message = if alert { + self.create_message_for_app(&self_id, content, Some(priority), None, None)? + } else { + self.create_message(&self_id, content, Some(priority), None)? + }; // Set before signing for clarity; the canonical payload covers only // sender/id/recipient/content, so ordering is not security-relevant. message.requires_ack = false; @@ -783,6 +811,11 @@ impl OfflineProtocol { /// to the engine, not to an application, so they always carry /// [`crate::ProtocolConfig::app_id`]. Only an application's own send /// can name another id, through [`Self::create_message_for_app`]. + /// + /// For the same reason every frame built here carries + /// [`crate::constants::NO_ALERT_KEY`]: nothing the user wrote is in it, + /// so the relay must not push "new message" for it. User content is + /// built through [`Self::create_message_for_app`], which does not mark. pub(super) fn create_message( &mut self, recipient: impl Into, @@ -790,7 +823,12 @@ impl OfflineProtocol { priority: Option, reply_to_msg: Option, ) -> Result { - self.create_message_for_app(recipient, content, priority, reply_to_msg, None) + let mut message = + self.create_message_for_app(recipient, content, priority, reply_to_msg, None)?; + message + .metadata + .insert(crate::constants::NO_ALERT_KEY.to_string(), "1".to_string()); + Ok(message) } /// Validates a per-send app id from the rich send surface. @@ -806,7 +844,8 @@ impl OfflineProtocol { } /// Like [`Self::create_message`], stamping `app_id` when given and the - /// configured id otherwise. + /// configured id otherwise, and without the no-alert marker: this is the + /// builder for user content, which the relay should push. pub(super) fn create_message_for_app( &mut self, recipient: impl Into, @@ -878,7 +917,10 @@ impl OfflineProtocol { return self.encrypt_bytes_confirmed_session(&mls, recipient, plaintext); } - self.ensure_session_establishment(&mls, recipient)?; + // Every caller is a text send, whose message queues behind the + // Welcome; the session-confirm marker also lands here, but only + // after adopting a session, so it never creates one. + self.ensure_session_establishment(&mls, recipient, true)?; // Only encrypt if session is confirmed (Welcome processed or successful decrypt). // Confirmation truth comes from persisted session state. @@ -910,10 +952,15 @@ impl OfflineProtocol { /// key package remains (expired). Returns `Err(SessionNotReady)` when no /// key package is available, or right after creating a session with /// `store_pending` enabled (the session cannot be confirmed yet). + /// + /// `user_send` is whether something the user sent prompted this, and so + /// whether a Welcome it sends may alert (see + /// [`Self::send_welcome_message`]). pub(super) fn ensure_session_establishment( &mut self, mls: &Arc>, recipient: &str, + user_send: bool, ) -> Result<()> { // Check for existing session (requires storage I/O via load_group) let has_session = { @@ -979,7 +1026,7 @@ impl OfflineProtocol { ); } - let welcome_sent = self.send_welcome_message(recipient, &welcome)?; + let welcome_sent = self.send_welcome_message(recipient, &welcome, user_send)?; debug!( recipient = %recipient, @@ -1681,7 +1728,7 @@ impl OfflineProtocol { // expected "created, awaiting confirmation" outcome; // the message queues regardless. if let Some(mls) = self.mls_manager.clone() { - match self.ensure_session_establishment(&mls, recipient) { + match self.ensure_session_establishment(&mls, recipient, true) { Ok(()) | Err(Error::SessionNotReady(_)) => {} Err(err) => { warn!( @@ -2668,7 +2715,13 @@ impl OfflineProtocol { // confirmed synchronously, so callers retry after // `secure_session_established`. if let Some(mls) = self.mls_manager.clone() { - self.ensure_session_establishment(&mls, &recipient_str)?; + // A document-layer transfer is the SDK's own: the + // Welcome it opens must not alert. + self.ensure_session_establishment( + &mls, + &recipient_str, + data_purpose.is_none(), + )?; } return Err(Error::SessionNotReady( self.establishment_state(&recipient_str)?, @@ -2939,6 +2992,14 @@ impl OfflineProtocol { content_type.to_string(), ); } + // A document-layer transfer is the SDK's own traffic, like the + // `__DATA_V1__` frames beside it. The purpose rides sealed in + // chunk 0 only, so every chunk is marked here, where it is known. + if data_purpose.is_some() { + message + .metadata + .insert(crate::constants::NO_ALERT_KEY.to_string(), "1".to_string()); + } self.outbound_media_chunks .insert(message.id.clone(), (file_id.to_string(), chunk_index)); @@ -3311,6 +3372,7 @@ impl OfflineProtocol { // staging is consumed at entry creation; this covers the id being torn // down before that ever happened. self.pending_reseal.remove(message_id); + self.undecryptable_resent.remove(message_id); // A receipt names an outbox entry; with the entry gone it names // nothing, and keeping it would only grow the map. self.custody_receipts.remove(message_id); @@ -4986,22 +5048,110 @@ impl OfflineProtocol { message: &Message, inbound_transport: TransportType, ) -> Result<()> { + self.send_ack_with_status(message, inbound_transport, None) + } + + /// Answers a frame that can never decrypt: a delivery ACK in shape (so a + /// relay holding the frame drops it) carrying + /// `ack_status = "undecryptable"`, which the sender settles without + /// `message_delivered` (see [`Self::handle_undecryptable_ack`]). Only for + /// a frame that asked for an answer; the plain ACK's rules otherwise apply. + /// + /// The one gate for every path that gives up on a frame (receive loop, + /// queue drain, queue eviction): only a sender that advertised + /// [`ACK_UNDECRYPTABLE_V1`](offline_protocol_sealed::ACK_UNDECRYPTABLE_V1) + /// is answered. An older SDK reads any ACK as a delivery, so it gets what + /// it always got (nothing) and keeps its retries. + /// + /// And only a frame that arrived over the internet relay. Every frame this + /// answers failed before anything authenticated it: garbage ciphertext in + /// a derivable session slot, or a parked frame whose sender is only a + /// claim. Answering on a mesh or peer-stream link would tell an injector + /// in radio range that the target is live and processing, which the + /// threat model's acknowledgement rule withholds. The relay already knows + /// the device is connected, and its mailbox is the one thing that + /// redelivers a dead frame forever, so it is the one carrier answered. + pub(super) fn send_undecryptable_ack( + &mut self, + message: &Message, + inbound_transport: TransportType, + ) { + if inbound_transport != TransportType::Internet + || !message.requires_ack + || !self + .peer_undecryptable_ack + .contains(message.sender.as_str()) + { + return; + } + let ack = match self.build_ack(message, inbound_transport, Some(ACK_STATUS_UNDECRYPTABLE)) { + Ok(ack) => ack, + Err(err) => { + error!( + message_id = %message.id, + error = %err, + "Failed to build undecryptable ACK" + ); + return; + } + }; + // Pinned to the relay, never the ACK ladder: `route_ack` falls back to + // the mesh and to whatever DORS picks when the relay is down, and a + // held ACK re-runs that same ladder later. Either would put this + // answer on a radio link, where R25 says it never goes. Dropped + // instead: the relay mailbox redelivers the frame on the next + // connect, and the live path answers it then. + let sent = self + .transport_manager + .can_address_via(TransportType::Internet, message.sender.as_str()) + && self + .transport_manager + .send_via_transport(&ack, TransportType::Internet) + .is_ok(); + if !sent { + debug!( + message_id = %message.id, + "Relay unavailable; undecryptable ACK dropped until the frame is redelivered" + ); + } + } + + fn send_ack_with_status( + &mut self, + message: &Message, + inbound_transport: TransportType, + status: Option<&str>, + ) -> Result<()> { + let ack = self.build_ack(message, inbound_transport, status)?; + self.route_ack(&ack, inbound_transport) + } + + /// Builds the acknowledgement for `message`, optionally carrying + /// [`ACK_STATUS_KEY`]. Routing is the caller's. + fn build_ack( + &self, + message: &Message, + inbound_transport: TransportType, + status: Option<&str>, + ) -> Result { let sender = UserId::new(&self.local_id)?; let recipient = message.sender.clone(); let app_id = AppId::new(&self.config.app_id)?; let ttl = TTL::new(self.config.initial_ttl).unwrap_or_else(|_| TTL::default()); - let ack_message = Message::builder(sender, recipient, app_id) + let mut builder = Message::builder(sender, recipient, app_id) .content(String::new()) .priority(MessagePriority::Low) .ttl(ttl) .requires_ack(false) .metadata(ACK_FOR_KEY, message.id.as_str()) .metadata(ACK_HOP_COUNT_KEY, message.hop_count.value().to_string()) - .metadata(ACK_TRANSPORT_KEY, Self::transport_label(inbound_transport)) - .build(); + .metadata(ACK_TRANSPORT_KEY, Self::transport_label(inbound_transport)); + if let Some(status) = status { + builder = builder.metadata(ACK_STATUS_KEY, status); + } - self.route_ack(&ack_message, inbound_transport) + Ok(builder.build()) } /// Gets an acknowledgement back to whoever sent us the message it answers. @@ -5226,6 +5376,14 @@ impl OfflineProtocol { return; } + // Answered, not delivered: never `message_delivered`. + if message.metadata.get(ACK_STATUS_KEY).map(String::as_str) + == Some(ACK_STATUS_UNDECRYPTABLE) + { + self.handle_undecryptable_ack(&message_id); + return; + } + // A delivery ack settles an outbound connection request: // the recipient provably received it, so a later stale // recipient_unreachable signal must not fire a false @@ -5324,6 +5482,46 @@ impl OfflineProtocol { } } + /// Settles an `undecryptable` answer: the recipient cannot decrypt the + /// copy it holds. An encrypted DM is resent once, same id: re-sealed under + /// the current session when its plaintext is still in memory (the flush + /// path re-seals), verbatim when it is not (an entry restored after a + /// restart). Verbatim is not futile: a frame the recipient's + /// pending-decryption queue gave up on (TTL or overflow) was never + /// attempted, so the same ciphertext decrypts once the session is up. + /// Failing it here instead would turn a recoverable drop into a terminal + /// one, and since a parked frame's sender is unauthenticated, anyone + /// able to fill the queue could trigger that. A second answer for an id + /// already resent leaves the entry to its retry ladder, which re-seals on + /// every attempt it can and ends in `message_failed` at max retries: that + /// bounds two devices trading answers and resends at network speed, and + /// keeps a split-brain pair's message deliverable once the session is + /// rebuilt. Anything else (no entry left, plaintext, media) is ignored, + /// which is exactly what the frame's failure used to cost it. + fn handle_undecryptable_ack(&mut self, message_id: &MessageId) { + let Some(entry) = self.outbox.get(message_id) else { + return; + }; + if !entry + .message + .content + .starts_with(internal_prefixes::ENCRYPTED) + { + return; + } + if !self.undecryptable_resent.insert(message_id.clone()) { + return; + } + let (message, attempt_count) = (entry.message.clone(), entry.attempt_count); + debug!( + message_id = %message_id, + "Recipient cannot decrypt this copy; resending once" + ); + self.retry_queue.remove(&message_id.as_str()); + self.ack_manager.remove_ack(message_id); + self.try_flush_send(message, attempt_count); + } + /// Whether `ack` comes from the peer the message it answers was addressed /// to — the one thing about an acknowledgement that is not simply asserted /// by whoever sent it. See [`Self::handle_ack_message`] for what that does @@ -5580,7 +5778,8 @@ impl OfflineProtocol { serde_json::to_string(&payload).map_err(|e| Error::Serialization(e.to_string()))?; let content = format!("{}{}", internal_prefixes::CONN_REQUEST, serialized); - let message_id = self.send_internal_message(recipient, content, MessagePriority::High)?; + let message_id = + self.send_internal_message_alerting(recipient, content, MessagePriority::High, true)?; self.track_pending_connection_request(&message_id, recipient); info!(recipient = %recipient, "Sent connection request"); Ok(message_id) @@ -5650,7 +5849,8 @@ impl OfflineProtocol { serde_json::to_string(&payload).map_err(|e| Error::Serialization(e.to_string()))?; let content = format!("{}{}", internal_prefixes::CONN_ACCEPT, serialized); - let message_id = self.send_internal_message(recipient, content, MessagePriority::High)?; + let message_id = + self.send_internal_message_alerting(recipient, content, MessagePriority::High, true)?; info!(recipient = %recipient, "Accepted connection request"); Ok(message_id) } @@ -5877,6 +6077,9 @@ impl OfflineProtocol { // publish, not what a peer may seal to us: withholding it would // downgrade the peer's traffic to the bootstrap key as a side // effect of a local setting. + // Read unconditionally: an `undecryptable` ACK is settled as "not + // delivered" whatever our config, so a peer may always send us one. + ack_versions: vec![offline_protocol_sealed::ACK_UNDECRYPTABLE_V1], nostr_pubkey: self.transport_manager.nostr_public_key(), } } diff --git a/crates/offline-protocol/src/protocol/session.rs b/crates/offline-protocol/src/protocol/session.rs index 3a2e51e8..757a1f8f 100644 --- a/crates/offline-protocol/src/protocol/session.rs +++ b/crates/offline-protocol/src/protocol/session.rs @@ -1113,10 +1113,14 @@ impl OfflineProtocol { // WELCOME LIFECYCLE // ======================================================================== + /// `user_send` is true when a user's own send created the session: their + /// message waits on this Welcome, so its first send must alert (see + /// [`Self::try_send_welcome`]). pub(super) fn send_welcome_message( &mut self, recipient: &str, welcome: &WelcomeMessage, + user_send: bool, ) -> Result { let serialized = serde_json::to_string(welcome).map_err(|e| Error::Serialization(e.to_string()))?; @@ -1127,7 +1131,7 @@ impl OfflineProtocol { let group_id = welcome.group_id.as_str().to_string(); self.upsert_welcome_lifecycle(recipient, &group_id, message, "welcome_created")?; - self.try_send_welcome(recipient, "welcome_initial_send") + self.try_send_welcome(recipient, "welcome_initial_send", user_send) } /// Parks a Welcome that currently has no transport carrier: refresh the @@ -1184,7 +1188,19 @@ impl OfflineProtocol { Ok(false) } - pub(super) fn try_send_welcome(&mut self, peer_id: &str, source_event: &str) -> Result { + /// The Welcome is the frame that wakes an offline peer for a first + /// message queued behind it, so the relay push is decided per send rather + /// than baked into the stored frame: it alerts when `user_send` (the + /// message is being queued right now) or when user content is already + /// waiting for the peer, and is otherwise + /// [`crate::constants::NO_ALERT_KEY`]-marked. Metadata is outside the + /// control signature, so changing it here leaves the frame verifiable. + pub(super) fn try_send_welcome( + &mut self, + peer_id: &str, + source_event: &str, + user_send: bool, + ) -> Result { let now = Utc::now(); let mut record = self .welcome_lifecycles @@ -1259,8 +1275,16 @@ impl OfflineProtocol { // path below and the peer's probe or decrypt is still what settles // it. A carried one waits the mesh confirmation timeout, since a // neighbour is a mesh link whatever the selector last picked. + let mut welcome = record.welcome_message.clone(); + if user_send || self.has_user_content_waiting_for(peer_id) { + welcome.metadata.remove(crate::constants::NO_ALERT_KEY); + } else { + welcome + .metadata + .insert(crate::constants::NO_ALERT_KEY.to_string(), "1".to_string()); + } let sent = self - .send_or_carry_handshake(&record.welcome_message) + .send_or_carry_handshake(&welcome) .map(|handed_to_mesh| { if handed_to_mesh > 0 { None @@ -1380,7 +1404,7 @@ impl OfflineProtocol { entry.expires_at = Utc::now() + ChronoDuration::seconds(WELCOME_LIFECYCLE_TTL_SECS); } - if let Err(err) = self.try_send_welcome(peer_id, source_event) { + if let Err(err) = self.try_send_welcome(peer_id, source_event, false) { warn!( peer_id = %peer_id, error = %err, @@ -1404,7 +1428,7 @@ impl OfflineProtocol { ); return; } - if let Err(err) = self.try_send_welcome(peer_id, source_event) { + if let Err(err) = self.try_send_welcome(peer_id, source_event, false) { warn!( peer_id = %peer_id, error = %err, @@ -1686,7 +1710,7 @@ impl OfflineProtocol { ); return; } - if let Err(err) = self.try_send_welcome(peer_id, source_event) { + if let Err(err) = self.try_send_welcome(peer_id, source_event, false) { warn!( peer_id = %peer_id, error = %err, @@ -2412,7 +2436,7 @@ impl OfflineProtocol { .collect(); for peer_id in due_peers { - if let Err(err) = self.try_send_welcome(&peer_id, "welcome_retry") { + if let Err(err) = self.try_send_welcome(&peer_id, "welcome_retry", false) { warn!( peer_id = %peer_id, error = %err, @@ -2429,4 +2453,20 @@ impl OfflineProtocol { .get(peer_id) .is_some_and(|record| matches!(record.state, WelcomeDeliveryState::Expired)) } + + /// Whether a message the user sent is waiting for `peer_id`: queued + /// behind session establishment, or on the outbox unmarked (an entry + /// restored from before the marker existed counts, the safe direction). + fn has_user_content_waiting_for(&self, peer_id: &str) -> bool { + self.pending_encrypted_messages + .get(peer_id) + .is_some_and(|queued| !queued.is_empty()) + || self.outbox.values().any(|entry| { + entry.message.recipient.as_str() == peer_id + && !entry + .message + .metadata + .contains_key(crate::constants::NO_ALERT_KEY) + }) + } } diff --git a/crates/offline-protocol/src/protocol/storage.rs b/crates/offline-protocol/src/protocol/storage.rs index 7a8e39e2..f8fe730c 100644 --- a/crates/offline-protocol/src/protocol/storage.rs +++ b/crates/offline-protocol/src/protocol/storage.rs @@ -3221,6 +3221,12 @@ impl OfflineProtocol { { self.peer_ctrl_freshness.insert(peer_id.clone()); } + if caps + .ack_versions + .contains(&offline_protocol_sealed::ACK_UNDECRYPTABLE_V1) + { + self.peer_undecryptable_ack.insert(peer_id.clone()); + } if self.config.encryption.rich_payload_enabled && caps.attested_rich_versions.contains(&RICH_PAYLOAD_V1) { diff --git a/crates/offline-protocol/src/protocol/tests/data_sync.rs b/crates/offline-protocol/src/protocol/tests/data_sync.rs index 8c4cbf2f..5f0723dd 100644 --- a/crates/offline-protocol/src/protocol/tests/data_sync.rs +++ b/crates/offline-protocol/src/protocol/tests/data_sync.rs @@ -5326,3 +5326,261 @@ fn a_carried_snapshot_outside_the_interest_is_refused_before_it_is_reassembled() "the same transfer must be admitted for a document inside the interest" ); } + +fn no_alert(message: &offline_protocol_core::Message) -> Option<&str> { + message + .metadata + .get(crate::constants::NO_ALERT_KEY) + .map(String::as_str) +} + +/// Take what `node` sent since the last call. +fn drain_sent(node: &Node) -> Vec { + let sent = node.transport.sent_messages(); + node.transport.clear_sent_messages(); + sent +} + +/// The relay sees an MLS-encrypted sync offer and a user's text as the same +/// opaque ciphertext, and pushed "new message" for both. Every frame the +/// SDK sends on its own carries the cleartext marker; nothing the user sent +/// does. +#[test] +fn frames_the_sdk_sends_itself_are_marked_no_alert_and_user_messages_are_not() { + let (mut alice, bob) = pair(); + let space = Node::space_for(&bob); + let bob_id = bob.address.clone(); + + write(&mut alice, &space, "notes", "title", "hello"); + alice.protocol.kick_data_sync(&space, "peer_rediscovered"); + let sync = drain_sent(&alice); + assert!(sync.len() >= 2, "a delta and an offer, got {}", sync.len()); + + alice.protocol.send_key_package_to(&bob_id, false).unwrap(); + alice + .protocol + .send_read_receipt( + &bob_id, + vec![offline_protocol_core::MessageId::new().as_str()], + ) + .unwrap(); + alice + .protocol + .send_typing_indicator(&bob_id, "chat", true) + .unwrap(); + alice.protocol.reject_connection_request(&bob_id).unwrap(); + let control = drain_sent(&alice); + assert_eq!(control.len(), 4); + + for message in sync.iter().chain(&control) { + assert_eq!(no_alert(message), Some("1"), "{:.40}", message.content); + } + + // User content: a plain DM, a sealed rich DM, a forward, and the + // connection request and accept the user is meant to hear about. + alice.protocol.peer_rich_payload.insert(bob_id.clone()); + let plain = alice + .protocol + .send_message(&bob_id, "plain", None, None::) + .unwrap(); + let rich = alice + .protocol + .send_message_with( + &bob_id, + "rich", + crate::SendMessageOptions { + reply_context: Some(offline_protocol_core::ReplyContext { + sender: offline_protocol_core::UserId::new(&bob_id).unwrap(), + text: "quoted".to_string(), + timestamp: None, + reply_media_label: None, + reply_content_type: Some("text".to_string()), + }), + ..Default::default() + }, + ) + .unwrap(); + // Forwarding takes a received message; the outbox copy stands in for one. + let mut original = alice.protocol.outbox[&plain].message.clone(); + original.content = "forwarded".to_string(); + alice + .protocol + .forward_message(&original, &bob_id, None) + .unwrap(); + alice + .protocol + .send_connection_request(&bob_id, "Alice", None, None) + .unwrap(); + alice + .protocol + .accept_connection_request(&bob_id, "Alice", None) + .unwrap(); + let user = drain_sent(&alice); + assert_eq!(user.len(), 5); + for message in &user { + assert_eq!(no_alert(message), None, "{:.40}", message.content); + } + assert!(user.iter().any(|m| m.id == rich)); +} + +/// The marker rides the outbox copy, so a resend, a re-seal against a new +/// epoch, and a restart all keep it. +#[test] +fn a_resent_or_restored_sync_frame_keeps_its_no_alert_marker() { + let (mut alice, bob) = pair(); + let space = Node::space_for(&bob); + + write(&mut alice, &space, "notes", "title", "hello"); + let frame = drain_sent(&alice).pop().expect("a delta"); + assert_eq!(no_alert(&frame), Some("1")); + + let mut resend = alice.protocol.outbox[&frame.id].message.clone(); + alice.protocol.reseal_for_resend_in_place(&mut resend); + assert_ne!( + resend.content, frame.content, + "the resend was not re-sealed" + ); + assert_eq!(no_alert(&resend), Some("1")); + + alice.restart(&bob.address); + let restored = &alice.protocol.outbox[&frame.id].message; + assert_eq!(no_alert(restored), Some("1")); +} + +/// A document carried over the media path is the SDK's own traffic like any +/// other sync frame, and it prefers the relay, so every chunk is marked. A +/// file the user sends is not. +#[test] +fn a_document_carried_as_media_is_marked_no_alert_and_a_users_file_is_not() { + let (mut alice, bob) = pair(); + // More than one chunk, so the marker is shown past chunk 0, the only one + // that carries the purpose. + let bytes = vec![7u8; 100 * 1024]; + let send = |alice: &mut Node, purpose| { + alice + .protocol + .send_media_inner( + bob.address.clone(), + bytes.clone(), + "notes".to_string(), + offline_protocol_core::ContentType::File, + crate::protocol::types::MediaSendOptions::default(), + purpose, + ) + .expect("the transfer starts"); + drain_sent(alice) + .into_iter() + .filter(|m| m.content_type == offline_protocol_core::ContentType::FileChunk) + .collect::>() + }; + + let snapshot = send( + &mut alice, + Some(crate::media_envelope::DataPurpose::Snapshot { + doc: "notes".to_string(), + }), + ); + assert!(snapshot.len() > 1, "{} chunks", snapshot.len()); + for chunk in &snapshot { + assert_eq!(no_alert(chunk), Some("1")); + } + + let file = send(&mut alice, None); + assert!(!file.is_empty()); + for chunk in &file { + assert_eq!(no_alert(chunk), None); + } +} + +/// Hand `node` a key package for `peer`, as one arriving over the wire would. +fn hold_key_package(node: &mut Node, peer: &str) { + let kp = crate::test_identity::manager_for(peer, Arc::new(InMemoryStorage::new())) + .get_or_create_key_package() + .unwrap(); + node.protocol.pending_key_packages.insert( + id(peer), + crate::protocol::types::ReceivedKeyPackage { + key_package_data: kp.key_package_data, + local_expires_at_ms: chrono::Utc::now().timestamp_millis() as u64 + 60_000, + }, + ); +} + +fn sent_welcome(node: &Node) -> offline_protocol_core::Message { + drain_sent(node) + .into_iter() + .rev() + .find(|m| { + m.content + .starts_with(crate::protocol::prefixes::internal_prefixes::WELCOME) + }) + .expect("a Welcome on the wire") +} + +/// A first DM to a new contact waits behind the Welcome, and the Welcome's +/// push is the only one the recipient gets for it. So a 1:1 Welcome alerts +/// exactly while user content waits on it, decided again on every send. +#[test] +fn a_welcome_alerts_only_while_a_user_message_waits_on_it() { + let (mut alice, _bob) = pair(); + + // A send that creates the session: the DM queues behind the Welcome. + hold_key_package(&mut alice, "carol"); + let carol = id("carol"); + alice + .protocol + .send_message(&carol, "hi", None, None::) + .unwrap(); + assert!(!alice.protocol.pending_encrypted_messages[&carol].is_empty()); + assert_eq!(no_alert(&sent_welcome(&alice)), None); + + // A retry alerts while the DM is still queued, and not once it is gone. + alice + .protocol + .try_send_welcome(&carol, "test_retry", false) + .unwrap(); + assert_eq!(no_alert(&sent_welcome(&alice)), None); + alice.protocol.pending_encrypted_messages.remove(&carol); + alice + .protocol + .try_send_welcome(&carol, "test_retry", false) + .unwrap(); + assert_eq!(no_alert(&sent_welcome(&alice)), Some("1")); + + // An establishment nothing the user wrote waits on is silent. + hold_key_package(&mut alice, "dave"); + alice + .protocol + .establish_secure_session(&id("dave")) + .unwrap() + .expect("a session was created"); + assert_eq!(no_alert(&sent_welcome(&alice)), Some("1")); + + // Media is never queued behind a session, so the path differs: the send + // opens the session and fails not-ready. A user's file alerts; a + // document the data layer carries is the SDK's own, so it is silent. + let mut open_with_media = |peer: &str, purpose| { + hold_key_package(&mut alice, peer); + let peer = id(peer); + alice.protocol.peer_data_sync.insert(peer.clone()); + alice.protocol.peer_data_media.insert(peer.clone()); + let err = alice + .protocol + .send_media_inner( + peer, + b"bytes".to_vec(), + "notes".to_string(), + offline_protocol_core::ContentType::File, + crate::protocol::types::MediaSendOptions::default(), + purpose, + ) + .expect_err("no confirmed session yet"); + assert!(matches!(err, crate::Error::SessionNotReady(_)), "{err}"); + sent_welcome(&alice) + }; + assert_eq!(no_alert(&open_with_media("erin", None)), None); + let snapshot = Some(crate::media_envelope::DataPurpose::Snapshot { + doc: "notes".to_string(), + }); + assert_eq!(no_alert(&open_with_media("frank", snapshot)), Some("1")); +} diff --git a/crates/offline-protocol/src/protocol/tests/data_sync_group.rs b/crates/offline-protocol/src/protocol/tests/data_sync_group.rs index 076b4b2c..18803a7e 100644 --- a/crates/offline-protocol/src/protocol/tests/data_sync_group.rs +++ b/crates/offline-protocol/src/protocol/tests/data_sync_group.rs @@ -2493,3 +2493,58 @@ fn pass_group_frames(alice: &mut Member, bob: &mut Member, carol: &mut Member) - } carried } + +/// A group chat message and a replication frame are the same +/// `__GRP_MLS_MSG__` ciphertext to the relay; only the replication frame may +/// carry the no-alert marker. An invite alerts the invitee; the commit that +/// tells the rest of the roster does not. +#[test] +fn only_group_frames_the_sdk_sends_itself_are_marked_no_alert() { + let (mut alice, _bob, _carol, group) = trio(); + let no_alert = |m: &offline_protocol_core::Message| { + m.metadata.get(crate::constants::NO_ALERT_KEY).cloned() + }; + + write(&mut alice, &group, "notes", "title", "hello"); + let replication = alice.transport.sent_messages(); + alice.transport.clear_sent_messages(); + assert_eq!(replication.len(), 2); + assert!(replication + .iter() + .all(|m| no_alert(m).as_deref() == Some("1"))); + + alice + .protocol + .send_group_message(&group, "hi all", None, None) + .unwrap(); + let chat = alice.transport.sent_messages(); + alice.transport.clear_sent_messages(); + assert_eq!(chat.len(), 2); + assert!(chat.iter().all(|m| no_alert(m).is_none())); + + let dave = id("dave"); + let kp = crate::test_identity::manager_for("dave", Arc::new(InMemoryStorage::new())) + .get_or_create_key_package() + .unwrap(); + alice.protocol.pending_key_packages.insert( + dave.clone(), + crate::protocol::types::ReceivedKeyPackage { + key_package_data: kp.key_package_data, + local_expires_at_ms: chrono::Utc::now().timestamp_millis() as u64 + 60_000, + }, + ); + alice.protocol.invite_to_group(&group, &dave).unwrap(); + let invite = alice.transport.sent_messages(); + let welcome = invite + .iter() + .find(|m| m.content.starts_with(internal_prefixes::GROUP_MLS_WELCOME)) + .expect("a group Welcome to the invitee"); + assert_eq!(welcome.recipient.as_str(), dave); + assert_eq!(no_alert(welcome), None); + let commits: Vec<_> = invite + .iter() + .filter(|m| m.content.starts_with(internal_prefixes::GROUP_MLS_COMMIT)) + .collect(); + assert!(!commits.is_empty()); + assert!(commits.iter().all(|m| no_alert(m).as_deref() == Some("1"))); +} diff --git a/crates/offline-protocol/src/protocol/tests/mod.rs b/crates/offline-protocol/src/protocol/tests/mod.rs index 5949cda3..f79e5bba 100644 --- a/crates/offline-protocol/src/protocol/tests/mod.rs +++ b/crates/offline-protocol/src/protocol/tests/mod.rs @@ -13,7 +13,9 @@ mod per_send_app_id; mod service_seal; use super::*; -use crate::constants::{ACK_FOR_KEY, ACK_HOP_COUNT_KEY, ACK_TRANSPORT_KEY}; +use crate::constants::{ + ACK_FOR_KEY, ACK_HOP_COUNT_KEY, ACK_STATUS_KEY, ACK_STATUS_UNDECRYPTABLE, ACK_TRANSPORT_KEY, +}; use crate::events::{DecryptionFailureCode, PresenceSource, PresenceStatus, SecurityWarningCode}; use crate::mls_observability::{ DecryptionFailureKind, MlsErrorCategory, MlsLifecycleEvent, MlsOperationContext, @@ -1878,6 +1880,7 @@ fn feed_key_package(protocol: &mut OfflineProtocol, sender: &str, wire_versions: data_versions: Vec::new(), ctrl_versions: Vec::new(), svc_versions: Vec::new(), + ack_versions: Vec::new(), nostr_pubkey: None, }; let content = format!( @@ -1959,6 +1962,7 @@ fn feed_key_package_with_env(protocol: &mut OfflineProtocol, sender: &str, env_v data_versions: Vec::new(), ctrl_versions: Vec::new(), svc_versions: Vec::new(), + ack_versions: Vec::new(), nostr_pubkey: None, }; let content = format!( @@ -2156,6 +2160,7 @@ pub(crate) fn feed_key_package_with_capabilities( data_versions, ctrl_versions: Vec::new(), svc_versions: Vec::new(), + ack_versions: Vec::new(), nostr_pubkey: None, }; let content = format!( @@ -2276,6 +2281,7 @@ fn feed_key_package_with_caps( data_versions: Vec::new(), ctrl_versions: Vec::new(), svc_versions: Vec::new(), + ack_versions: Vec::new(), nostr_pubkey: None, }; let content = format!( @@ -2419,6 +2425,7 @@ fn key_package_message_with_nostr_pubkey( data_versions: Vec::new(), ctrl_versions: Vec::new(), svc_versions: Vec::new(), + ack_versions: Vec::new(), nostr_pubkey: nostr_pubkey.map(str::to_string), }; let content = format!( @@ -2755,6 +2762,7 @@ fn feed_key_package_with_data( data_versions, ctrl_versions: Vec::new(), svc_versions: Vec::new(), + ack_versions: Vec::new(), nostr_pubkey: None, }; let content = format!( @@ -3122,6 +3130,56 @@ fn peer_capability_unblock_clears_record() { ); } +/// `ack_versions` is learned from a key package, dropped by one that stops +/// advertising it, and survives a restart like the other capabilities. +#[test] +fn peer_undecryptable_ack_capability_is_learned_and_survives_restart() { + let storage = Arc::new(InMemoryStorage::new()); + let mut protocol = protocol_with_mls_storage(storage.clone()); + let feed = |protocol: &mut OfflineProtocol, sender: &str, ack_versions: Vec| { + let payload = KeyPackagePayload { + user_id: sender.to_string(), + key_package_data: vec![1, 2, 3, 4], + remaining_lifetime_ms: 30 * 24 * 60 * 60 * 1000, + timestamp_ms: 0, + session_reset: false, + wire_versions: Vec::new(), + env_versions: Vec::new(), + rich_versions: Vec::new(), + data_versions: Vec::new(), + ctrl_versions: Vec::new(), + svc_versions: Vec::new(), + ack_versions, + nostr_pubkey: None, + }; + let content = format!( + "{}{}", + internal_prefixes::KEY_PACKAGE, + serde_json::to_string(&payload).unwrap() + ); + protocol.process_internal_message(&signed_frame(sender, &id("user123"), &content)); + }; + feed( + &mut protocol, + &id("peer"), + vec![offline_protocol_sealed::ACK_UNDECRYPTABLE_V1], + ); + feed(&mut protocol, &id("legacy"), Vec::new()); + assert!(protocol.peer_undecryptable_ack.contains(&id("peer"))); + assert!(!protocol.peer_undecryptable_ack.contains(&id("legacy"))); + + let mut restarted = protocol_with_mls_storage(storage.clone()); + assert!(restarted.peer_undecryptable_ack.contains(&id("peer"))); + assert!(!restarted.peer_undecryptable_ack.contains(&id("legacy"))); + + // A downgrade drops it in memory and on disk. + feed(&mut restarted, &id("peer"), Vec::new()); + assert!(!restarted.peer_undecryptable_ack.contains(&id("peer"))); + assert!(!protocol_with_mls_storage(storage) + .peer_undecryptable_ack + .contains(&id("peer"))); +} + #[test] fn peer_capability_restore_prunes_overflow() { // A durable store inflated past the cap (e.g. by a forged-sender flood @@ -3136,6 +3194,7 @@ fn peer_capability_restore_prunes_overflow() { data_versions: Vec::new(), ctrl_versions: Vec::new(), svc_versions: Vec::new(), + ack_versions: Vec::new(), nostr_pubkey: None, }; let encoded = serde_json::to_vec(&caps).unwrap(); @@ -3278,6 +3337,7 @@ fn peer_capability_restore_prefers_session_peers() { data_versions: Vec::new(), ctrl_versions: Vec::new(), svc_versions: Vec::new(), + ack_versions: Vec::new(), nostr_pubkey: None, }; let encoded = serde_json::to_vec(&caps).unwrap(); @@ -3886,6 +3946,7 @@ fn test_process_internal_message_key_package() { data_versions: Vec::new(), ctrl_versions: Vec::new(), svc_versions: Vec::new(), + ack_versions: Vec::new(), nostr_pubkey: None, }; let content = format!( @@ -13150,7 +13211,7 @@ fn test_welcome_no_carrier_ticks_keep_lifecycle_alive() { // Far more no-carrier attempts than max_retries — none may expire the // Welcome, and the retry budget must not accumulate. for _ in 0..10 { - let _ = protocol.try_send_welcome(&id("bob"), "test_no_carrier_tick"); + let _ = protocol.try_send_welcome(&id("bob"), "test_no_carrier_tick", false); } let lifecycle = protocol.welcome_lifecycles.get(&id("bob")).unwrap(); @@ -13220,7 +13281,7 @@ fn test_no_carrier_welcome_parks_without_churn() { None::, ); for _ in 0..10 { - let _ = protocol.try_send_welcome(&id("bob"), "test_no_carrier_tick"); + let _ = protocol.try_send_welcome(&id("bob"), "test_no_carrier_tick", false); } let lifecycle = protocol.welcome_lifecycles.get(&id("bob")).unwrap(); @@ -13386,7 +13447,7 @@ fn test_welcome_sends_when_carrier_appears() { .add_transport(TransportType::BLE, Box::new(mock)); protocol - .try_send_welcome(&id("bob"), "test_carrier_appeared") + .try_send_welcome(&id("bob"), "test_carrier_appeared", false) .unwrap(); let lifecycle = protocol.welcome_lifecycles.get(&id("bob")).unwrap(); @@ -14403,11 +14464,11 @@ fn test_desync_dm_withholds_ack_and_triggers_rekey() { bob.initialize_mls_for_test(Arc::new(InMemoryStorage::new())) .unwrap(); - let bob_transport = MockTransport::new(TransportType::BLE); + let bob_transport = MockTransport::new(TransportType::Internet); bob_transport.start().unwrap(); let bob_handle = bob_transport.clone(); bob.transport_manager_mut() - .add_transport(TransportType::BLE, Box::new(bob_transport)); + .add_transport(TransportType::Internet, Box::new(bob_transport)); let received: Arc>> = Arc::new(Mutex::new(Vec::new())); let received_handle = Arc::clone(&received); @@ -14417,6 +14478,7 @@ fn test_desync_dm_withholds_ack_and_triggers_rekey() { } }); bob.start().unwrap(); + bob.peer_undecryptable_ack.insert(id("alice")); // Establish bob's session with alice via a real Welcome. let alice_manager = @@ -14468,11 +14530,11 @@ fn test_desync_dm_withholds_ack_and_triggers_rekey() { while bob.receive_message().is_some() {} let sent = bob_handle.sent_messages(); - let acks = sent - .iter() - .filter(|m| m.metadata.get(ACK_FOR_KEY) == Some(&msg_id_str)) - .count(); - assert_eq!(acks, 0, "a desync DM must NOT be delivery-ACKed"); + assert_eq!( + acks_for(&sent, &msg_id_str), + (0, 1), + "a desync DM must NOT be delivery-ACKed, only answered undecryptable" + ); assert!( received.lock().unwrap().is_empty(), @@ -14512,11 +14574,11 @@ fn hard_failure_pair( bob.initialize_mls_for_test(Arc::new(InMemoryStorage::new())) .unwrap(); - let bob_transport = MockTransport::new(TransportType::BLE); + let bob_transport = MockTransport::new(TransportType::Internet); bob_transport.start().unwrap(); let bob_handle = bob_transport.clone(); bob.transport_manager_mut() - .add_transport(TransportType::BLE, Box::new(bob_transport)); + .add_transport(TransportType::Internet, Box::new(bob_transport)); bob.start().unwrap(); let alice_manager = @@ -14540,6 +14602,10 @@ fn hard_failure_pair( ); bob_handle.queue_message(welcome_wire); while bob.receive_message().is_some() {} + // Alice's SDK reads an `undecryptable` answer; see + // `test_undecryptable_frame_from_peer_without_capability_is_not_answered` + // for one that does not. + bob.peer_undecryptable_ack.insert(id("alice")); (bob, bob_handle, alice_manager) } @@ -14566,6 +14632,21 @@ fn encrypted_wire( wire } +/// The ACKs `sent` holds for `msg_id`, split into (delivery, undecryptable). +fn acks_for(sent: &[Message], msg_id: &str) -> (usize, usize) { + let answers: Vec<_> = sent + .iter() + .filter(|m| m.metadata.get(ACK_FOR_KEY).map(String::as_str) == Some(msg_id)) + .collect(); + let undecryptable = answers + .iter() + .filter(|m| { + m.metadata.get(ACK_STATUS_KEY).map(String::as_str) == Some(ACK_STATUS_UNDECRYPTABLE) + }) + .count(); + (answers.len() - undecryptable, undecryptable) +} + /// A structurally valid envelope for the live session whose MLS ciphertext is /// garbage — an AEAD/parse failure, NOT an epoch desync and NOT session-not-ready. fn corrupt_ciphertext_for(alice_manager: &MlsManager) -> offline_protocol_mls::EncryptedMessage { @@ -14593,19 +14674,29 @@ fn test_hard_decrypt_failure_withholds_ack_and_does_not_rekey() { let received: Arc>> = Arc::new(Mutex::new(Vec::new())); let received_handle = Arc::clone(&received); - let failures: Arc>> = Arc::new(Mutex::new(Vec::new())); + #[allow(clippy::type_complexity)] + let failures: Arc, Option)>>> = + Arc::new(Mutex::new(Vec::new())); let failures_handle = Arc::clone(&failures); bob.on_event(move |event| match event { Event::MessageReceived { content, .. } => { received_handle.lock().unwrap().push(content.clone()) } - Event::MessageDecryptionFailed { reason, .. } => { - failures_handle.lock().unwrap().push(reason.clone()) - } + Event::MessageDecryptionFailed { + reason, + sealed_at_ms, + session_established_at_ms, + .. + } => failures_handle.lock().unwrap().push(( + reason.clone(), + sealed_at_ms, + session_established_at_ms, + )), _ => {} }); - let wire = encrypted_wire(&corrupt_ciphertext_for(&alice_manager), None); + let corrupt = corrupt_ciphertext_for(&alice_manager); + let wire = encrypted_wire(&corrupt, None); assert!(wire.requires_ack, "delivery ACK must be in play"); let msg_id = wire.id.clone(); let msg_id_str = msg_id.as_str().to_string(); @@ -14616,11 +14707,9 @@ fn test_hard_decrypt_failure_withholds_ack_and_does_not_rekey() { let sent = bob_handle.sent_messages(); assert_eq!( - sent.iter() - .filter(|m| m.metadata.get(ACK_FOR_KEY) == Some(&msg_id_str)) - .count(), - 0, - "a hard decrypt failure must NOT be delivery-ACKed" + acks_for(&sent, &msg_id_str), + (0, 1), + "a hard decrypt failure must NOT be delivery-ACKed, only answered undecryptable once" ); assert_eq!( sent.iter() @@ -14644,9 +14733,80 @@ fn test_hard_decrypt_failure_withholds_ack_and_does_not_rekey() { ); let got = failures.lock().unwrap(); assert!( - got.iter().any(|r| r.contains("not acknowledged")), + got.iter().any(|(r, _, _)| r.contains("not delivery-ACKed")), "the failure event must read as advisory, not terminal, got {got:?}" ); + // The frame-age fields the app's split-brain breaker compares. + let (_, sealed_at_ms, session_established_at_ms) = &got[0]; + assert_eq!(*sealed_at_ms, Some(corrupt.timestamp_ms)); + assert!( + session_established_at_ms.is_some(), + "a session joined by this SDK records when it was established" + ); +} + +/// A sender that never advertised `ack_versions` gets exactly what 0.28 gave +/// it: no answer at all (an older SDK reads any ACK as a delivery), the id +/// unmarked so its re-sealed retry still lands, nothing queued. +#[test] +fn test_undecryptable_frame_from_peer_without_capability_is_not_answered() { + let (mut bob, bob_handle, alice_manager) = hard_failure_pair(true); + bob.peer_undecryptable_ack.remove(&id("alice")); + + let wire = encrypted_wire(&corrupt_ciphertext_for(&alice_manager), None); + let msg_id = wire.id.clone(); + bob_handle.clear_sent_messages(); + bob_handle.queue_message(wire.clone()); + while bob.receive_message().is_some() {} + + assert_eq!( + acks_for(&bob_handle.sent_messages(), &msg_id.as_str()), + (0, 0) + ); + assert!(!bob.deduplicator.is_duplicate(&msg_id)); + assert!(!bob.pending_queue.contains_peer("alice")); + + // The queue paths go through the same gate. + bob.enqueue_pending_decryption_via("alice", &wire, Some(TransportType::Internet)); + bob.process_pending_decryption("alice"); + assert_eq!( + acks_for(&bob_handle.sent_messages(), &msg_id.as_str()), + (0, 0) + ); +} + +/// The frames answered `undecryptable` failed before anything authenticated +/// them, so an answer on a mesh link would confirm to an injector in radio +/// range that the target is live. Only a relay arrival is answered; the same +/// frame over Bluetooth LE, live or drained from the queue, gets nothing. +#[test] +fn test_undecryptable_frame_off_the_relay_is_not_answered() { + let (mut bob, internet_handle, alice_manager) = hard_failure_pair(true); + let ble = MockTransport::new(TransportType::BLE); + ble.start().unwrap(); + let ble_handle = ble.clone(); + bob.transport_manager_mut() + .add_transport(TransportType::BLE, Box::new(ble)); + + let wire = encrypted_wire(&corrupt_ciphertext_for(&alice_manager), None); + let msg_id = wire.id.as_str().to_string(); + let answers = || { + let (internet, ble) = ( + acks_for(&internet_handle.sent_messages(), &msg_id), + acks_for(&ble_handle.sent_messages(), &msg_id), + ); + (internet.0 + ble.0, internet.1 + ble.1) + }; + internet_handle.clear_sent_messages(); + + ble_handle.queue_message(wire.clone()); + while bob.receive_message().is_some() {} + assert_eq!(answers(), (0, 0), "a BLE arrival is never answered"); + assert!(!bob.deduplicator.is_duplicate(&wire.id)); + + bob.enqueue_pending_decryption_via("alice", &wire, Some(TransportType::BLE)); + bob.process_pending_decryption("alice"); + assert_eq!(answers(), (0, 0), "nor is one drained from the queue"); } /// The recovery half: because the ACK was withheld and the id unmarked, the @@ -14669,13 +14829,8 @@ fn test_hard_decrypt_failure_recovers_on_resealed_resend() { let corrupt = encrypted_wire(&corrupt_ciphertext_for(&alice_manager), None); let msg_id = corrupt.id.clone(); let msg_id_str = msg_id.as_str().to_string(); - let acks = |handle: &MockTransport| -> usize { - handle - .sent_messages() - .into_iter() - .filter(|m| m.metadata.get(ACK_FOR_KEY) == Some(&msg_id_str)) - .count() - }; + let acks = + |handle: &MockTransport| -> usize { acks_for(&handle.sent_messages(), &msg_id_str).0 }; bob_handle.clear_sent_messages(); bob_handle.queue_message(corrupt); @@ -14771,20 +14926,16 @@ fn test_drained_message_that_hard_fails_is_not_acked() { // Queue it with a recorded arrival transport, so the drain would genuinely // be able to ACK it — with `received_via: None` the ACK is a no-op and this // test would pass vacuously. - bob.enqueue_pending_decryption_via("alice", &wire, Some(TransportType::BLE)); + bob.enqueue_pending_decryption_via("alice", &wire, Some(TransportType::Internet)); assert_eq!(bob.pending_queue.peer_queue_len("alice"), 1); bob_handle.clear_sent_messages(); bob.process_pending_decryption("alice"); assert_eq!( - bob_handle - .sent_messages() - .iter() - .filter(|m| m.metadata.get(ACK_FOR_KEY) == Some(&msg_id_str)) - .count(), - 0, - "a message that hard-fails on drain must NOT be ACKed" + acks_for(&bob_handle.sent_messages(), &msg_id_str), + (0, 1), + "a message that hard-fails on drain must NOT be ACKed, only answered undecryptable" ); assert!( !bob.deduplicator.is_duplicate(&msg_id), @@ -15148,6 +15299,216 @@ fn test_reseal_is_noop_without_provenance() { ); } +/// An alice with a confirmed session to a bare bob manager, one encrypted DM +/// in her outbox, and her transport handle cleared. Events land in the vec. +fn alice_with_sealed_dm() -> ( + OfflineProtocol, + MockTransport, + MlsManager, + MessageId, + Arc>>, +) { + let mut alice_config = create_test_config_for_user("alice"); + alice_config.encryption.enabled = true; + alice_config.encryption.store_pending = true; + alice_config.encryption.compact_envelope_enabled = false; + let mut alice = OfflineProtocol::new(alice_config).unwrap(); + alice + .initialize_mls_for_test(Arc::new(InMemoryStorage::new())) + .unwrap(); + let alice_transport = MockTransport::new(TransportType::BLE); + alice_transport.start().unwrap(); + let alice_handle = alice_transport.clone(); + alice + .transport_manager_mut() + .add_transport(TransportType::BLE, Box::new(alice_transport)); + let events: Arc>> = Arc::new(Mutex::new(Vec::new())); + let events_handle = Arc::clone(&events); + alice.on_event(move |event| events_handle.lock().unwrap().push(event)); + alice.start().unwrap(); + + let bob_manager = + crate::test_identity::manager_for("bob", Arc::new(crate::mls::InMemoryStorage::new())); + let bob_kp = bob_manager.get_or_create_key_package().unwrap(); + let welcome = { + let mgr = alice.mls_manager.as_ref().unwrap().read().unwrap(); + mgr.import_key_package(&id("bob"), &bob_kp.key_package_data) + .unwrap(); + mgr.create_session(&id("bob")).unwrap() + }; + bob_manager.join_session(&welcome).unwrap(); + alice.confirmed_sessions.insert(id("bob")); + + let msg_id = alice + .send_message(&id("bob"), "hello", None, None::) + .unwrap(); + assert!(alice.outbox.get(&msg_id).unwrap().reseal.is_some()); + alice_handle.clear_sent_messages(); + events.lock().unwrap().clear(); + (alice, alice_handle, bob_manager, msg_id, events) +} + +/// Bob's `undecryptable` answer to `msg_id`. +fn undecryptable_ack_from_bob(alice: &OfflineProtocol, msg_id: &MessageId) -> Message { + Message::builder( + UserId::new(id("bob")).unwrap(), + UserId::new(alice.local_id.clone()).unwrap(), + AppId::new("test-app").unwrap(), + ) + .content(String::new()) + .requires_ack(false) + .metadata(ACK_FOR_KEY, msg_id.as_str()) + .metadata(ACK_STATUS_KEY, ACK_STATUS_UNDECRYPTABLE) + .build() +} + +fn delivered_or_failed(events: &Mutex>, msg_id: &MessageId) -> (usize, usize) { + let events = events.lock().unwrap(); + let id = msg_id.as_str(); + let delivered = events + .iter() + .filter(|e| matches!(e, Event::MessageDelivered { message_id, .. } if *message_id == id)) + .count(); + let failed = events + .iter() + .filter(|e| matches!(e, Event::MessageFailed { message_id, .. } if *message_id == id)) + .count(); + (delivered, failed) +} + +/// An `undecryptable` answer is not a delivery: the sender re-seals the DM +/// under its current session and resends it once, same id, fresh ciphertext +/// the recipient can open. A second answer for the same id resends nothing, so +/// two devices cannot ping-pong; the entry stays for its retry ladder, and a +/// real delivery ACK still settles it. +#[test] +fn test_undecryptable_ack_reseals_once_and_never_reports_delivered() { + let (mut alice, alice_handle, bob_manager, msg_id, events) = alice_with_sealed_dm(); + let original = alice.outbox.get(&msg_id).unwrap().message.content.clone(); + + alice.handle_ack_message(&undecryptable_ack_from_bob(&alice, &msg_id)); + + let resent: Vec = alice_handle + .sent_messages() + .into_iter() + .filter(|m| m.id == msg_id) + .collect(); + assert_eq!(resent.len(), 1, "exactly one re-sealed resend"); + assert_ne!( + resent[0].content, original, + "the resend carries new ciphertext" + ); + let enc: offline_protocol_mls::EncryptedMessage = serde_json::from_str( + resent[0] + .content + .strip_prefix(internal_prefixes::ENCRYPTED) + .unwrap(), + ) + .unwrap(); + assert_eq!( + bob_manager + .decrypt_from_user(&enc, &id("alice")) + .unwrap() + .as_deref(), + Some(&b"hello"[..]) + ); + assert_eq!(delivered_or_failed(&events, &msg_id), (0, 0)); + + // The resend failed too: no second resend, still not delivered. + alice_handle.clear_sent_messages(); + alice.handle_ack_message(&undecryptable_ack_from_bob(&alice, &msg_id)); + assert!( + alice_handle.sent_messages().iter().all(|m| m.id != msg_id), + "a second undecryptable answer must not resend (no ping-pong)" + ); + assert_eq!(delivered_or_failed(&events, &msg_id), (0, 0)); + assert!( + alice.outbox.contains_key(&msg_id), + "left to the retry ladder, not dropped" + ); + + // A genuine delivery ACK still settles it. + let mut delivered = undecryptable_ack_from_bob(&alice, &msg_id); + delivered.metadata.remove(ACK_STATUS_KEY); + alice.handle_ack_message(&delivered); + assert_eq!(delivered_or_failed(&events, &msg_id), (1, 0)); +} + +/// A relay arrival answered while the relay is down (a drain, a prune, an +/// eviction) must not fall back to the ACK ladder's mesh or DORS routes: the +/// answer goes over the relay or not at all, and it is not held for a later +/// carrier edge, which would re-run the same ladder. +#[test] +fn test_undecryptable_ack_is_never_rerouted_off_the_relay() { + let (mut bob, internet_handle, alice_manager) = hard_failure_pair(true); + let ble = MockTransport::new(TransportType::BLE); + ble.start().unwrap(); + let ble_handle = ble.clone(); + bob.transport_manager_mut() + .add_transport(TransportType::BLE, Box::new(ble)); + + let wire = encrypted_wire(&corrupt_ciphertext_for(&alice_manager), None); + let msg_id = wire.id.as_str().to_string(); + internet_handle.clear_sent_messages(); + ble_handle.clear_sent_messages(); + + bob.enqueue_pending_decryption_via("alice", &wire, Some(TransportType::Internet)); + internet_handle.set_status(offline_protocol_transport::TransportStatus::Unavailable); + bob.process_pending_decryption("alice"); + + assert_eq!(acks_for(&ble_handle.sent_messages(), &msg_id), (0, 0)); + assert_eq!(acks_for(&internet_handle.sent_messages(), &msg_id), (0, 0)); + assert!( + bob.unrouted_acks.is_empty(), + "nothing held to re-route later" + ); + + // With the relay back, the redelivered frame is answered over it. + internet_handle.set_status(offline_protocol_transport::TransportStatus::Available); + internet_handle.queue_message(wire); + while bob.receive_message().is_some() {} + assert_eq!(acks_for(&internet_handle.sent_messages(), &msg_id), (0, 1)); + assert_eq!(acks_for(&ble_handle.sent_messages(), &msg_id), (0, 0)); +} + +/// Without the plaintext (a restored entry) the one resend replays the +/// ciphertext verbatim and the entry stays for its retry ladder: never +/// delivered, never failed outright. +#[test] +fn test_undecryptable_ack_without_plaintext_replays_once_and_keeps_the_entry() { + // An entry restored after a restart has no plaintext to re-seal. The + // answer may come from a pending-queue drop, whose ciphertext was never + // attempted and still decrypts, so the entry is replayed rather than + // failed. + let (mut alice, alice_handle, _bob_manager, msg_id, events) = alice_with_sealed_dm(); + alice.outbox.get_mut(&msg_id).unwrap().reseal = None; + let original = alice.outbox.get(&msg_id).unwrap().message.content.clone(); + + alice.handle_ack_message(&undecryptable_ack_from_bob(&alice, &msg_id)); + + let resent: Vec = alice_handle + .sent_messages() + .into_iter() + .filter(|m| m.id == msg_id) + .collect(); + assert_eq!(resent.len(), 1, "exactly one resend"); + assert_eq!(resent[0].content, original, "replayed verbatim"); + assert_eq!(delivered_or_failed(&events, &msg_id), (0, 0)); + assert!(alice.outbox.contains_key(&msg_id), "not failed"); + + alice_handle.clear_sent_messages(); + alice.handle_ack_message(&undecryptable_ack_from_bob(&alice, &msg_id)); + assert!( + alice_handle.sent_messages().iter().all(|m| m.id != msg_id), + "a second answer must not resend" + ); + assert_eq!(delivered_or_failed(&events, &msg_id), (0, 0)); + assert!( + alice.outbox.contains_key(&msg_id), + "left to the retry ladder" + ); +} + /// Like [`make_encrypted_protocol`], but takes the **production** identity /// path: the address is derived from a freshly minted identity key rather than /// forced to the profile string, so the instance runs as a real `off1…` @@ -15278,6 +15639,7 @@ fn feed_session_reset_key_package( data_versions: Vec::new(), ctrl_versions: Vec::new(), svc_versions: Vec::new(), + ack_versions: Vec::new(), nostr_pubkey: None, }; let content = format!( @@ -16339,11 +16701,11 @@ fn test_evicted_pending_message_recovers_on_resend_after_session_ready() { bob.initialize_mls_for_test(Arc::new(InMemoryStorage::new())) .unwrap(); - let bob_transport = MockTransport::new(TransportType::BLE); + let bob_transport = MockTransport::new(TransportType::Internet); bob_transport.start().unwrap(); let bob_handle = bob_transport.clone(); bob.transport_manager_mut() - .add_transport(TransportType::BLE, Box::new(bob_transport)); + .add_transport(TransportType::Internet, Box::new(bob_transport)); let received: Arc>> = Arc::new(Mutex::new(Vec::new())); let received_handle = Arc::clone(&received); @@ -16353,6 +16715,7 @@ fn test_evicted_pending_message_recovers_on_resend_after_session_ready() { } }); bob.start().unwrap(); + bob.peer_undecryptable_ack.insert(id("alice")); let alice_manager = crate::test_identity::manager_for("alice", Arc::new(crate::mls::InMemoryStorage::new())); @@ -16393,13 +16756,8 @@ fn test_evicted_pending_message_recovers_on_resend_after_session_ready() { let evictor_wire = wire(&encrypted_evictor); let evicted_id = evicted_wire.id.clone(); let evicted_id_str = evicted_id.as_str().to_string(); - let acks_for_evicted = |handle: &MockTransport| -> usize { - handle - .sent_messages() - .into_iter() - .filter(|m| m.metadata.get(ACK_FOR_KEY) == Some(&evicted_id_str)) - .count() - }; + let acks_for_evicted = + |handle: &MockTransport| -> usize { acks_for(&handle.sent_messages(), &evicted_id_str).0 }; // 1) Evicted message arrives, session not ready: queued, unmarked, not ACKed. bob_handle.queue_message(evicted_wire.clone()); @@ -16423,6 +16781,11 @@ fn test_evicted_pending_message_recovers_on_resend_after_session_ready() { 0, "the evicted message was never ACKed — that is what keeps recovery possible" ); + assert_eq!( + acks_for(&bob_handle.sent_messages(), &evicted_id_str), + (0, 1), + "the queue gave up on it, so it is answered undecryptable exactly once" + ); // 3) Welcome establishes the session and drains the queue. Only the evictor // surfaces; the evicted message is provably gone (never delivered). @@ -18312,7 +18675,7 @@ fn test_encrypted_message_decryption_failure_emits_app_error_event() { message_id, sender, code, - reason, + reason, .. } if message_id == &message.id.as_str() && sender == &id("sender123") && code == &DecryptionFailureCode::NotInitialized @@ -18357,7 +18720,7 @@ fn test_invalid_encrypted_payload_emits_app_error_event_and_is_deferred() { message_id, sender, code, - reason, + reason, .. } if message_id == &message.id.as_str() && sender == &id("sender123") && code == &DecryptionFailureCode::InvalidPayload @@ -18705,6 +19068,7 @@ fn test_pending_queue_overflow_emits_pending_queue_dropped_for_text() { sender, code, reason, + .. } = event { dropped_clone @@ -18965,6 +19329,7 @@ fn test_pending_decrypt_record_past_persisted_max_age_is_dropped_with_event() { sender, code, reason, + .. } => Some((message_id.clone(), sender.clone(), *code, reason.clone())), _ => None, }) @@ -19950,6 +20315,64 @@ fn test_pending_queue_ttl_expiration_is_deterministic_and_monotonic() { ); } +/// A frame the pending queue gives up on (TTL) is answered `undecryptable` +/// exactly once when its sender advertised the status, and not at all otherwise, so a relay holding it drops it instead of redelivering it +/// into the same queue on every connect. +#[test] +fn test_pending_queue_ttl_drop_answers_undecryptable_once() { + let mut config = create_test_config(); + config.encryption.enabled = true; + config.encryption.pending_queue.pending_ttl_ms = 1_000; + let mut protocol = OfflineProtocol::new(config).unwrap(); + let transport = MockTransport::new(TransportType::Internet); + transport.start().unwrap(); + let handle = transport.clone(); + protocol + .transport_manager_mut() + .add_transport(TransportType::Internet, Box::new(transport)); + protocol.start().unwrap(); + protocol.peer_undecryptable_ack.insert(id("sender123")); + + let msg = pending_test_message(&id("sender123"), "parked"); + let legacy = pending_test_message(&id("legacy"), "parked"); + protocol.enqueue_pending_decryption_via("sender123", &msg, Some(TransportType::Internet)); + protocol.enqueue_pending_decryption_via("legacy", &legacy, Some(TransportType::Internet)); + let past = Instant::now() - Duration::from_millis(2_000); + protocol + .pending_queue + .set_front_received_at("sender123", past); + protocol.pending_queue.set_front_received_at("legacy", past); + handle.clear_sent_messages(); + + assert_eq!( + protocol.prune_expired_pending_global_front(Instant::now(), 10), + 2 + ); + assert_eq!( + protocol.prune_expired_pending_global_front(Instant::now(), 10), + 0 + ); + assert!( + handle + .sent_messages() + .iter() + .all(|m| m.metadata.get(ACK_FOR_KEY) != Some(&legacy.id.as_str())), + "a sender that never advertised the status gets no answer" + ); + + let answer = handle + .sent_messages() + .into_iter() + .filter(|m| m.metadata.get(ACK_FOR_KEY) == Some(&msg.id.as_str())) + .collect::>(); + assert_eq!(answer.len(), 1, "exactly one answer for the dropped frame"); + assert_eq!( + answer[0].metadata.get(ACK_STATUS_KEY).map(String::as_str), + Some(ACK_STATUS_UNDECRYPTABLE) + ); + assert!(!answer[0].requires_ack && answer[0].content.is_empty()); +} + #[test] fn test_pending_messages_replay_decrypt_after_session_readiness() { let mut bob_config = create_test_config_for_user("bob"); @@ -20310,6 +20733,7 @@ fn test_lamport_clock_merge_on_internal_message() { data_versions: Vec::new(), ctrl_versions: Vec::new(), svc_versions: Vec::new(), + ack_versions: Vec::new(), nostr_pubkey: None, }; let content = format!( @@ -20522,6 +20946,7 @@ fn test_key_package_remaining_lifetime_ms() { data_versions: Vec::new(), ctrl_versions: Vec::new(), svc_versions: Vec::new(), + ack_versions: Vec::new(), nostr_pubkey: None, }; let content = format!( @@ -20608,6 +21033,7 @@ fn test_peer_key_package_persisted_and_restored_after_restart() { data_versions: Vec::new(), ctrl_versions: Vec::new(), svc_versions: Vec::new(), + ack_versions: Vec::new(), nostr_pubkey: None, }; let content = format!( @@ -20697,6 +21123,7 @@ fn test_pending_key_packages_capped_evicts_soonest_to_expire() { data_versions: Vec::new(), ctrl_versions: Vec::new(), svc_versions: Vec::new(), + ack_versions: Vec::new(), nostr_pubkey: None, }; let content = format!( @@ -20749,6 +21176,7 @@ fn test_received_key_package_lifetime_is_clamped() { data_versions: Vec::new(), ctrl_versions: Vec::new(), svc_versions: Vec::new(), + ack_versions: Vec::new(), nostr_pubkey: None, }; let content = format!( @@ -22300,6 +22728,7 @@ fn test_establish_secure_session_loads_from_storage_after_restart() { data_versions: Vec::new(), ctrl_versions: Vec::new(), svc_versions: Vec::new(), + ack_versions: Vec::new(), nostr_pubkey: None, }; let content = format!( @@ -39872,6 +40301,7 @@ fn key_package_frame_body(sender: &str, session_reset: bool, ctrl_versions: Vec< data_versions: Vec::new(), ctrl_versions, svc_versions: Vec::new(), + ack_versions: Vec::new(), nostr_pubkey: None, }; format!( @@ -41631,6 +42061,7 @@ fn the_ordinary_handshake_starts_a_latency_without_any_session_missing() { data_versions: Vec::new(), ctrl_versions: Vec::new(), svc_versions: Vec::new(), + ack_versions: Vec::new(), nostr_pubkey: None, }; let content = format!( diff --git a/crates/offline-protocol/src/protocol/tests/service_seal.rs b/crates/offline-protocol/src/protocol/tests/service_seal.rs index f4bffa8d..87cef35c 100644 --- a/crates/offline-protocol/src/protocol/tests/service_seal.rs +++ b/crates/offline-protocol/src/protocol/tests/service_seal.rs @@ -325,6 +325,7 @@ fn key_package_body(sender: &str, svc_versions: Vec) -> String { data_versions: Vec::new(), ctrl_versions: Vec::new(), svc_versions, + ack_versions: Vec::new(), nostr_pubkey: None, }; serde_json::to_string(&payload).unwrap() diff --git a/crates/offline-protocol/src/protocol/types.rs b/crates/offline-protocol/src/protocol/types.rs index 129fbcc7..90d9d774 100644 --- a/crates/offline-protocol/src/protocol/types.rs +++ b/crates/offline-protocol/src/protocol/types.rs @@ -1501,6 +1501,14 @@ pub(crate) struct PeerCapabilities { #[serde(default, skip_serializing_if = "Vec::is_empty")] pub(crate) svc_versions: Vec, + /// Delivery-ACK statuses the peer reads, from + /// [`KeyPackagePayload::ack_versions`]. Persisted for the same reason as + /// the lists above: a peer met before a restart would otherwise get no + /// `undecryptable` answer until its next key package, and a relay mailbox + /// would keep redelivering its dead frames meanwhile. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub(crate) ack_versions: Vec, + /// The peer's Nostr public key, from [`KeyPackagePayload::nostr_pubkey`]. /// /// Persisted for the same reason as the capability lists: the cached key @@ -1546,6 +1554,7 @@ impl PeerCapabilities { data_versions: &[u8], ctrl_versions: &[u8], svc_versions: &[u8], + ack_versions: &[u8], nostr_pubkey: Option<&str>, ) -> Self { Self { @@ -1574,6 +1583,11 @@ impl PeerCapabilities { .copied() .take(MAX_PERSISTED_CAPABILITY_VERSIONS) .collect(), + ack_versions: ack_versions + .iter() + .copied() + .take(MAX_PERSISTED_CAPABILITY_VERSIONS) + .collect(), attested_rich_versions: Vec::new(), attested_data_versions: Vec::new(), nostr_pubkey: nostr_pubkey.and_then(normalize_nostr_pubkey), @@ -1588,6 +1602,7 @@ impl PeerCapabilities { || !self.data_versions.is_empty() || !self.ctrl_versions.is_empty() || !self.svc_versions.is_empty() + || !self.ack_versions.is_empty() || !self.attested_rich_versions.is_empty() || !self.attested_data_versions.is_empty() || self.nostr_pubkey.is_some() @@ -1679,8 +1694,7 @@ pub(crate) enum InternalMessageResult { /// loop responds by unmarking the id (so a resend re-enters processing /// instead of hitting the duplicate re-ACK path) and skipping the ACK. /// - /// Four conditions produce it, differing in whether the *frame* is worth - /// keeping: + /// Two conditions produce it: /// /// - **Session not ready**: the MLS session/group is not established yet, /// so the frame is queued for delayed decryption @@ -1689,24 +1703,29 @@ pub(crate) enum InternalMessageResult { /// (`process_pending_decryption`), which also sends the deferred delivery /// ACK directly on the recorded arrival transport (so a sender that gave /// up before the session confirmed still learns of delivery). - /// - **Epoch desync**: the frame is sealed to a dead epoch, so it is *not* - /// queued (it could never drain) and a rate-limited re-key is triggered. - /// - **Crypto/transport failure** with `crypto_recovery_enabled`: OpenMLS - /// consumed the ratchet generation on the failed attempt, so the frame is - /// likewise not queued — and no re-key is triggered, which stays - /// desync-only. /// - **Envelope parse failure** with `crypto_recovery_enabled`: the /// `__MLS_ENC__` payload did not parse in any envelope form, so there is - /// no ciphertext to decrypt. Not queued either — an unparseable frame can - /// never become parseable. - /// - /// In the latter three, recovery is the sender's *resend* rather than this - /// frame: Tier 2 re-seals each resend of an encrypted DM against a live - /// generation, and a message that stays undeliverable settles as an honest - /// `MessageFailed` instead of a lying "delivered". See - /// `docs/state-machines/delivery-and-acks.md` for the deferred-acknowledgement - /// atom and the decrypt-failure classification. + /// no ciphertext to decrypt. Not queued (an unparseable frame can never + /// become parseable) and not answered, since for an injector silence + /// reveals less than an answer. Recovery is the sender's resend. + /// + /// See `docs/state-machines/delivery-and-acks.md` for the + /// deferred-acknowledgement atom and the decrypt-failure classification. Deferred, + /// The frame can never decrypt, though the message behind it may: an epoch + /// desync (sealed to an epoch that is not ours, and a re-key is scheduled) + /// or a crypto/transport failure with `crypto_recovery_enabled` (OpenMLS + /// spent the ratchet generation on the failed attempt). Not queued. + /// + /// Handled like `Deferred` (id unmarked, no delivery ACK) except that + /// the frame is answered with an `undecryptable` ACK + /// ([`crate::constants::ACK_STATUS_UNDECRYPTABLE`]) when it arrived over + /// the internet relay and its sender advertised that it reads one; any + /// other frame gets no answer. Silence left a relay + /// holding the frame to redeliver it on every connect, failing again each + /// time; the answer lets the relay drop it and tells the sender to resend + /// the message once, re-sealed under its current session when it can. + Undecryptable, /// Message was decrypted, here's the plaintext. Decrypted(String), } diff --git a/crates/offline-protocol/src/telemetry/record.rs b/crates/offline-protocol/src/telemetry/record.rs index 714eecee..d60b1b6e 100644 --- a/crates/offline-protocol/src/telemetry/record.rs +++ b/crates/offline-protocol/src/telemetry/record.rs @@ -436,6 +436,8 @@ pub(crate) mod tests { sender: String::new(), code: crate::events::DecryptionFailureCode::Unknown, reason: String::new(), + sealed_at_ms: None, + session_established_at_ms: None, }, Event::TransportSwitched { from: None, diff --git a/crates/offline-protocol/src/telemetry/scrub_event.rs b/crates/offline-protocol/src/telemetry/scrub_event.rs index 14cb9687..a741fc90 100644 --- a/crates/offline-protocol/src/telemetry/scrub_event.rs +++ b/crates/offline-protocol/src/telemetry/scrub_event.rs @@ -294,6 +294,8 @@ fn scrub_in_place(event: &mut Event, scrubber: &Scrubber) { message_id: _, code: _, reason: _, + sealed_at_ms: _, + session_established_at_ms: _, } => { hash_string(sender, scrubber); } @@ -1386,6 +1388,8 @@ mod tests { sender: "alice".into(), code: DecryptionFailureCode::InvalidCiphertext, reason: "bad mac".into(), + sealed_at_ms: None, + session_established_at_ms: None, }; let scrubbed = scrub_event(&event, &scrubber_enabled()).into_owned(); match scrubbed { diff --git a/docs/message-delivery.md b/docs/message-delivery.md index 8a17efa0..4a6c2c8a 100644 --- a/docs/message-delivery.md +++ b/docs/message-delivery.md @@ -503,13 +503,23 @@ The delivery system emits events at each stage of the message lifecycle: | `MessageDelivered` | ACK received from recipient | `message_id`, `latency_ms`, `hop_count`, `transport` | | `MessageFailed` | Terminal failure (max ACK retries, outbox lifetime or capacity exceeded) | `message_id`, `reason`, `retry_count` | | `MediaResendRequired` | Interrupted outbound media transfer detected at `start()`; app must re-supply the bytes via `send_media` with the same `file_id` | `file_id`, `recipient`, `file_name`, `file_size` | -| `MessageDecryptionFailed` | An inbound encrypted message failed to decrypt on this attempt; **advisory, not terminal** (see below) | `message_id`, `sender`, `code`, `reason` | +| `MessageDecryptionFailed` | An inbound encrypted message failed to decrypt on this attempt; **advisory, not terminal** (see below) | `message_id`, `sender`, `code`, `reason`, `sealed_at_ms`, `session_established_at_ms` | `MessageDecryptionFailed` is the one receiver-side event on this list, and it -does not settle anything. A message that fails to decrypt is not delivery-ACKed -(so the sender keeps retrying, and each resend of a DM is re-sealed against the -current session), which means the event fires once per failed *attempt* rather -than once per message — bounded by the sender's ACK retry budget. Treat it as +does not settle anything. A message that fails to decrypt is not delivery-ACKed. +A frame that can never decrypt (spent ratchet generation, dead epoch, or one the +pending-decryption queue gave up on) is answered with an `ack_status: +"undecryptable"` ACK instead, when it arrived over the internet relay and its +sender advertised `ack_versions` entry 1 (otherwise it is left unanswered, as +before): the sender re-seals the DM once under its current session and resends +it with the same id (verbatim when the plaintext is gone, as for a restored +outbox entry); it never emits `MessageDelivered` for it, and a second such +answer for the same id leaves the entry to its retry ladder. The event +therefore fires once per failed *attempt* rather than once per message, bounded +by the sender's ACK retry budget. `sealed_at_ms` (sender clock, from the +envelope) and `session_established_at_ms` (local clock) let an app tell a stale +frame (sealed before the current session existed) from one that says something +about the session now. Treat it as "this attempt did not decrypt"; the terminal signal is `MessageFailed` on the sender, or `FileReceiveFailed` for media. See [Crypto-Failure Recovery](mls-integration.md#crypto-failure-recovery). diff --git a/docs/mls-integration.md b/docs/mls-integration.md index 4a72c6b6..08dc18ed 100644 --- a/docs/mls-integration.md +++ b/docs/mls-integration.md @@ -1313,10 +1313,12 @@ for dedup and ACK correlation. Tier 1 makes the failure honest; Tier 2 is what makes the message actually arrive. What is deliberately **excluded from the re-key**: failures that are *not* an -epoch mismatch — an AEAD/authentication failure, a discarded past ratchet -generation, a malformed frame — withhold the ACK exactly like an epoch mismatch -(Tier 1 and Tier 2 both apply, so the sender's re-sealed resend still recovers -the message), but they never trigger a re-key. Widening the *re-key* trigger to +epoch mismatch (an AEAD/authentication failure, a discarded past ratchet +generation, a malformed frame). They are answered exactly like an epoch +mismatch: never a delivery ACK, and an `undecryptable` ACK when the frame came +over the relay, so the sender re-seals the message once and the relay drops the +dead frame. Tier 2 still applies to every retry. But they never trigger a +re-key. Widening the *re-key* trigger to cover them would turn every malformed frame into a session teardown, which is an unbounded churn vector; withholding an ACK carries no such cost, because the recovery it enables is the sender's own retry. diff --git a/docs/security/threat-model.md b/docs/security/threat-model.md index 2cab6c79..012d64b4 100644 --- a/docs/security/threat-model.md +++ b/docs/security/threat-model.md @@ -202,7 +202,14 @@ The rule the protocol settled on: resolves as a policy refusal is never acknowledged, because the arrival path withheld the acknowledgement already. - A frame that failed for a **recoverable** reason gets no acknowledgement, so - the sender's resend is the recovery path. + the sender's resend is the recovery path. One exception: a frame that can + never decrypt (a dead epoch, a spent ratchet generation, or one the + pending-decryption queue gave up on) is answered `ack_status: + "undecryptable"` when it arrived over the **internet relay** and its sender + advertised that it reads the status. These failures happen before anything + is authenticated, so the answer is withheld on every other carrier, where an + injector in radio range would read it. See + [R25](#r25-the-relay-learns-which-of-its-frames-a-device-could-not-decrypt). Both refusals are permanent, so what separates them is neither authentication nor recoverability: a **policy** refusal is a statement about a *frame*, while a @@ -1080,6 +1087,27 @@ the 5000 most recent it tracks). A device that declines to carry same flood from a direct neighbour or over the relay was possible before and is unchanged. +### R25. The relay learns which of its frames a device could not decrypt + +A frame that can never decrypt is answered with an `undecryptable` ACK when it +arrived over the internet relay, so the relay drops its mailbox copy instead +of redelivering it on every connect. Every such failure happens before +authentication, so anyone able to inject on the relay ingest path, under a +claimed sender that advertised the status, gets an answer for a garbage frame +in that sender's derivable session slot. + +**Why it stands:** silence left a relay redelivering dead ciphertext forever, +and each redelivery failed again on the recipient. The relay is the only +carrier with a mailbox that does that. + +**What bounds it:** the answer is addressed to the claimed sender and routed +by the relay, which already knows the device is connected (A3), so it adds the +fact that one named frame was processed and failed, not liveness. It is never +sent on a mesh or peer-stream link, where an A1 or A2 neighbour would read it. +It goes only to a sender whose key package advertised `ack_versions` entry 1, +and the sender reacts with one resend of a message it already holds, never a +delivery. + ## Network egress Until 0.26 the Rust crates opened no socket: every byte that left a device diff --git a/docs/spec/capability-negotiation.md b/docs/spec/capability-negotiation.md index 1cf20d7c..5ace0326 100644 --- a/docs/spec/capability-negotiation.md +++ b/docs/spec/capability-negotiation.md @@ -13,6 +13,7 @@ Peers advertise what they can parse in the key package payload, the body of a | `data_versions` | End to end | Whether we may send `__DATA_V1__` document sync frames, and which document encoding they carry. Entry 1 is 1:1 replication; entry 2 additionally means the peer intercepts these frames inside a *group* ciphertext; entry 3 additionally means the peer speaks the blob-fetch frames and routes a data-purposed media transfer into its document layer; entry 4 additionally means the peer reads the removals a version offer carries; entry 5 additionally means it answers inside the interest an offer declares; entry 6 additionally means it carries attachment bytes inside a group; entry 7 additionally means it parses the custody receipt, so a custodian may answer its deposits | No replication with that peer | | `ctrl_versions` | End to end | Which control-frame signing payload we build for this peer. Entry 2 means the peer verifies `offline-ctrl-v2`, which binds the frame's timestamp | Build `offline-ctrl-v1`, which states no freshness | | `svc_versions` | End to end | Whether a service request or response may travel sealed inside `__MLS_ENC__` toward this peer. Entry 1 means the peer routes a decrypted `__SVC_REQ__` or `__SVC_RESP__` into its service layer, with the MLS-authenticated sender as the requester | Signed plaintext service frames | +| `ack_versions` | End to end | Which delivery-ACK statuses we may send this peer. Entry 1 means the peer reads `ack_status: "undecryptable"` as "this frame can never decrypt" (it resends the message once, re-sealed when it can) rather than as a delivery | Never answer an undecryptable frame; leave it unacknowledged | | `nostr_pubkey` | End to end | Which key metadata is sealed to on the Nostr path | Seal to the publicly computable key | The key package payload also carries `user_id`, the MLS key package itself, a @@ -152,6 +153,13 @@ member, which today is entries 2 and 6, read independently. It opens the package from that peer overrides it in both directions. Absence of an attestation means "no information" and MUST NOT be read as a downgrade. +`ack_versions` persists with the rest. A peer that has not advertised entry 1, +or whose advertisement this device has not yet received, is sent no +`undecryptable` answer at all: an older sender reads any ACK as a delivery, so +answering it would mark as delivered a message its recipient could not read. +The cost of not knowing is the older behaviour, an unanswered frame that a relay +mailbox keeps redelivering until the sender's next key package arrives. + `wire_versions` is **hop-local** and deliberately in-memory only. It describes what the next hop decodes, and it is re-exchanged on connect. @@ -289,6 +297,7 @@ A minimal leaf advertises little and works fully: | `data_versions` | Empty | No document replication with this peer | | `ctrl_versions` | `[2]`, always | Not optional, unlike every other row: a leaf verifies the freshness-bound payload on every control frame except `__MLS_KEY_PKG__`, so a peer that never learned this and kept sending the older one would have its Welcome refused and never complete a pairing | | `svc_versions` | Empty | A leaf has no service registry; peers send it signed plaintext service frames, which it never receives in practice | +| `ack_versions` | Empty | Peers leave a frame the leaf can never decrypt unanswered rather than answering `undecryptable` | | `nostr_pubkey` | Absent unless the device is reachable over that carrier | Peers seal to the publicly computable key | Two consequences follow from the universal rules above, and both are easy to diff --git a/docs/spec/wire-format.md b/docs/spec/wire-format.md index 0a6c464b..fd1f3712 100644 --- a/docs/spec/wire-format.md +++ b/docs/spec/wire-format.md @@ -339,9 +339,11 @@ observed on the wire: |-----|---------| | `ack_for` | This message acknowledges the named message id | | `ack_hop_count` | Hop count observed by the acknowledging party | +| `ack_status` | On an `ack_for` frame: `"undecryptable"` when the recipient can never decrypt the named frame (spent ratchet generation, dead epoch, or the pending-decryption queue gave up on it). The sender resends the message once (re-sealed under its current session when it still holds the plaintext, verbatim otherwise) and never reports it delivered. Sent only for a frame that arrived over the internet relay, and only to a peer that advertised `ack_versions` entry 1 ([Capability negotiation](capability-negotiation.md)), since an SDK that does not know it reads it as a delivery ACK. Absent on a delivery ACK | | `ack_transport` | Transport the acknowledged message arrived on | | `transport_preference` | Requested transport for this message | | `original_content_type` | Pre-chunking content type of a file transfer | +| `no_alert` | `"1"` on a frame the engine sent on its own (control traffic, key packages, data-sync frames including a document carried over the media path, receipts, typing, presence, and a 1:1 Welcome no user message waits on); a relay skips the "new message" push for it. Absent on user content, connection requests and accepts, group chat messages and group invites. Unsigned; receivers ignore it | | `__custody` | The class a depositor asserts for its own frame when it offers it into custody; engine-written, unsigned, and stripped by every device that forwards the frame, so it travels one hop. See [Custody](custody.md) | | `__ctrl_sig` | Base64 Ed25519 signature over the control-message canonical payload | | `__ctrl_pk` | Base64 Ed25519 public key of the signer, 32 raw bytes | diff --git a/docs/state-machines/delivery-and-acks.md b/docs/state-machines/delivery-and-acks.md index 3d1949bb..ddc0743f 100644 --- a/docs/state-machines/delivery-and-acks.md +++ b/docs/state-machines/delivery-and-acks.md @@ -91,7 +91,8 @@ deduplicator keeps identifiers for 7 days by default, persisted across restarts: the sender's redelivery window (`REDELIVERY_WINDOW_MS`), the longest a sender re-sends one identifier. With a shorter window a late re-send is not a duplicate. It reaches the ratchet, fails to decrypt because that generation is -spent, and is never acknowledged, so the sender re-sends it on every reconnect. +spent, and is answered `undecryptable` when it came over the relay: the +sender resends it once, re-sealed when it still holds the plaintext. **An acknowledgement nothing can carry is held, not dropped.** It is owed from the moment the frame is accepted, and the usual reason no route takes it is a @@ -193,9 +194,9 @@ flowchart TD P -->|envelope unparseable| D1[Deferred: no ack, no enqueue] P -->|no| E{Epoch disagreement?
WrongEpoch / NoPastEpochData} E -->|yes| SD[SessionDesync] - SD --> D2[Deferred: no ack, no enqueue,
+ schedule re-key] + SD --> D2[Undecryptable: undecryptable ack,
no enqueue, + schedule re-key] E -->|no| C{Session established
but decrypt failed?} - C -->|AEAD / corrupt / ratchet| D3[Deferred: no ack, no enqueue,
NO re-key] + C -->|AEAD / corrupt / ratchet| D3[Undecryptable: undecryptable ack,
no enqueue, NO re-key] C -->|session not ready| D4[Deferred: no ack, ENQUEUE] C -->|policy refusal that can never
become decryptable| K[Consumed: ack, drop] ``` @@ -205,14 +206,33 @@ flowchart TD | Class | Acknowledged | Enqueued | Re-key | Why | |-------|--------------|----------|--------|-----| | Session not ready | no | **yes** | no | It will become decryptable when the session arrives | -| Session desync (epoch fork) | no | no | **yes** | Ciphertext is sealed to a dead epoch and can never drain | -| Crypto failure (AEAD, corrupt, ratchet generation) | no | no | **no** | The attempt spent the generation; a queued copy could never drain | -| Transport failure | no | no | no | Recoverable by resend | +| Session desync (epoch fork) | `undecryptable` | no | **yes** | Ciphertext is sealed to a dead epoch and can never drain | +| Crypto failure (AEAD, corrupt, ratchet generation) | `undecryptable` | no | **no** | The attempt spent the generation; a queued copy could never drain | +| Transport failure | `undecryptable` | no | no | Classified with crypto failures: the attempt may have spent the generation, so recovery is the sender's re-sealed resend | | Envelope parse failure | no | no | no | Unparseable now is unparseable forever; the resend is the fix | | Policy refusal (commit not authorized) | **yes** | no | no | Can never become decryptable, so retries are pure waste | | Security refusal (identity mismatch, foreign session slot) | **no**, identifier unmarked | no | no | An acknowledgement confirms to an injector that the target is live | | Post-decrypt failure (empty, non-UTF-8, malformed plaintext) | **yes** | no | no | The generation is spent and a re-seal would produce the same malformed plaintext | +An `undecryptable` answer is a delivery ACK in shape (`ack_for`, empty content, +`requires_ack: false`) with `ack_status: "undecryptable"`, so a relay holding the +frame drops it instead of redelivering the same dead ciphertext on every +connect. It goes only to a frame that arrived over the internet relay, and +only to a sender that advertised `ack_versions` entry 1. Every failure in these +rows happens before anything is authenticated, so on a mesh or peer-stream +link the answer would confirm liveness to an injector in range (the +security-refusal row's leak); the relay already knows the device is connected +([R25](../security/threat-model.md#r25-the-relay-learns-which-of-its-frames-a-device-could-not-decrypt)). +Any other frame or sender is answered with nothing, the identifier unmarked, as before the +status existed, because it would read the answer as a delivery. The sender never reports it delivered: it re-seals an encrypted DM +under its current session and resends it once with the same identifier, and +ignores a second answer for the same identifier, leaving the entry to its retry +ladder. When the plaintext is gone (an entry restored after a restart) the one +resend is verbatim rather than a failure: a frame the pending-decryption queue +dropped was never attempted, so the same ciphertext still decrypts, and failing +it would let anyone who can fill that queue (a parked frame's sender is +unauthenticated) turn a delay into a lost message. + The policy-refusal and security-refusal rows are the two halves of "can never become decryptable", and they are deliberately opposite. Both refusals are permanent, but a policy refusal is a statement about a **frame** while a @@ -268,9 +288,10 @@ acknowledgement there would burn the sender's whole retry budget to no purpose. ## The drain has its own rules -A queued frame that hard-fails when the drain retries it is `Deferred`: no -acknowledgement, no re-mark, and **the queued copy is dropped rather than -re-enqueued**. +A queued frame that hard-fails when the drain retries it is `Undecryptable`: +an `undecryptable` answer, no re-mark, and **the queued copy is dropped rather +than re-enqueued**. A frame the queue evicts (overflow or time-to-live) is +answered `undecryptable` too, when it arrived over the relay. That last part is not an optimization. The drain **removes** the entry before processing it, so a re-enqueue misses the idempotency check and re-stamps the diff --git a/tools/spec-vectors/generate.py b/tools/spec-vectors/generate.py index 9c51efe1..940bd929 100644 --- a/tools/spec-vectors/generate.py +++ b/tools/spec-vectors/generate.py @@ -1269,6 +1269,7 @@ def build_key_package_vectors() -> dict: "data_versions": [], "ctrl_versions": [], "svc_versions": [], + "ack_versions": [], "nostr_pubkey": None, }, "note": "Every absent list defaults to empty, and empty selects " @@ -1281,7 +1282,7 @@ def build_key_package_vectors() -> dict: '"remaining_lifetime_ms":2419200000,"timestamp_ms":1700000000000,' '"session_reset":true,"wire_versions":[1],"env_versions":[1],' '"rich_versions":[1],"data_versions":[1,2,3],"ctrl_versions":[2],' - '"svc_versions":[1],"nostr_pubkey":"abcd"}', + '"svc_versions":[1],"ack_versions":[1],"nostr_pubkey":"abcd"}', "expect": { "user_id": "off1abc", "key_package_data": [], @@ -1294,6 +1295,7 @@ def build_key_package_vectors() -> dict: "data_versions": [1, 2, 3], "ctrl_versions": [2], "svc_versions": [1], + "ack_versions": [1], "nostr_pubkey": "abcd", }, "note": "data_versions entries are read independently, so [1,2,3] " @@ -1315,6 +1317,7 @@ def build_key_package_vectors() -> dict: "data_versions": [], "ctrl_versions": [], "svc_versions": [], + "ack_versions": [], "nostr_pubkey": None, }, "note": "An unknown field is ignored rather than refused. A " @@ -1337,6 +1340,7 @@ def build_key_package_vectors() -> dict: "data_versions": [], "ctrl_versions": [], "svc_versions": [], + "ack_versions": [], "nostr_pubkey": None, }, "note": "Absent and null mean the same thing on the way in. On " @@ -1350,7 +1354,7 @@ def build_key_package_vectors() -> dict: "example": '{"user_id":"off1abc","key_package_data":[],' '"remaining_lifetime_ms":0,"timestamp_ms":0,"session_reset":false,' '"wire_versions":[],"env_versions":[],"rich_versions":[],' - '"data_versions":[],"ctrl_versions":[],"svc_versions":[]}', + '"data_versions":[],"ctrl_versions":[],"svc_versions":[],"ack_versions":[]}', "nostr_pubkey_absent_is_omitted_not_null": True, }, }