Scope authenticated delivery cleanup

This commit is contained in:
jack
2026-07-25 23:49:42 +02:00
parent 26a7d3aaee
commit 460c97a810
12 changed files with 684 additions and 73 deletions
+34
View File
@@ -426,6 +426,40 @@ final class ConversationStore: ObservableObject {
@discardableResult
func setDeliveryStatus(_ status: DeliveryStatus, forMessageID messageID: String) -> Bool {
guard let ids = conversationIDsByMessageID[messageID] else { return false }
return applyDeliveryStatus(status, forMessageID: messageID, among: ids)
}
/// Applies an authenticated delivery/read receipt only to the supplied
/// direct-conversation aliases. A colliding message ID in another peer's
/// conversation (or a public timeline) must not inherit the receipt.
///
/// Stable and ephemeral aliases can temporarily hold the same message
/// instance during handoff. The shared helper republishes every targeted
/// alias even when the first mutation already changed that instance.
@discardableResult
func setDeliveryStatus(
_ status: DeliveryStatus,
forMessageID messageID: String,
inDirectPeerAliases peerIDs: Set<PeerID>
) -> Bool {
guard !peerIDs.isEmpty,
let indexedIDs = conversationIDsByMessageID[messageID] else {
return false
}
let allowedIDs = Set(peerIDs.map { ConversationID.directPeer($0) })
return applyDeliveryStatus(
status,
forMessageID: messageID,
among: indexedIDs.intersection(allowedIDs)
)
}
private func applyDeliveryStatus(
_ status: DeliveryStatus,
forMessageID messageID: String,
among ids: Set<ConversationID>
) -> Bool {
guard !ids.isEmpty else { return false }
var applied = false
var skipped: [ConversationID] = []
for id in ids {
@@ -106,13 +106,14 @@ final class BridgeCourierService: ObservableObject {
dedupKey: String?,
operationID: UUID?
)] = []
/// Message IDs already published as drops (sender-side dedup) and drop
/// event IDs already handled (multi-relay dedup). Both persist across
/// relaunches: relays hold drops for the full 24h NIP-40 window and the
/// persisted outbox keeps re-depositing, so in-memory-only dedup meant
/// every relaunch republished the same message as a fresh drop and every
/// gateway relaunch re-delivered the whole backlog (field-verified
/// amplification storm). Entries age out with the 24h drop window.
/// Opaque recipient/message keys already published as drops (sender-side
/// dedup) and drop event IDs already handled (multi-relay dedup). Both
/// persist across relaunches: relays hold drops for the full 24h NIP-40
/// window and the persisted outbox keeps re-depositing, so in-memory-only
/// dedup meant every relaunch republished the same message as a fresh drop
/// and every gateway relaunch re-delivered the whole backlog
/// (field-verified amplification storm). Entries age out with the 24h drop
/// window.
private var publishedDropKeys: ExpiringIDSet
private var seenDropEventIDs: ExpiringIDSet
private var subscriptionOpen = false
@@ -126,7 +127,7 @@ final class BridgeCourierService: ObservableObject {
}
/// Sender operations queued locally or awaiting relay confirmation.
/// The per-attempt ID prevents a stale pre-wipe callback from completing
/// a newer attempt for the same message.
/// a newer attempt for the same recipient/message pair.
private var activeDropOperations: [String: ActiveDropOperation] = [:]
/// Held-envelope publishes have no sender message ID, but still need an
/// in-flight identity: repeated refreshes inside the relay-OK wait window
@@ -214,11 +215,46 @@ final class BridgeCourierService: ObservableObject {
// MARK: - Sender role
/// Stable, opaque sender-side dedup key. Recipient scope prevents one
/// conversation's colliding message ID from suppressing another's drop,
/// while hashing keeps recipient keys out of the persisted snapshot.
private static func senderDropKey(
messageID: String,
recipientNoiseKey: Data
) -> String {
var material = Data("bitchat-bridge-drop-dedup-v2".utf8)
appendLengthPrefixed(Data(messageID.utf8), to: &material)
appendLengthPrefixed(recipientNoiseKey, to: &material)
return "v2:\(material.sha256Hex())"
}
private static func appendLengthPrefixed(_ value: Data, to output: inout Data) {
let length = UInt32(value.count)
output.append(UInt8((length >> 24) & 0xFF))
output.append(UInt8((length >> 16) & 0xFF))
output.append(UInt8((length >> 8) & 0xFF))
output.append(UInt8(length & 0xFF))
output.append(value)
}
/// Previous releases persisted raw message IDs without recipient scope.
/// They remain conservative wildcards for their original 24-hour
/// lifetime: assigning one to a recipient would be guesswork and could
/// republish the original drop. New acceptances persist only v2 keys.
private func wasPublished(
legacyMessageID: String,
dedupKey: String,
now date: Date
) -> Bool {
publishedDropKeys.contains(dedupKey, now: date)
|| publishedDropKeys.contains(legacyMessageID, now: date)
}
/// Parallel-deposit a sealed copy of an outbound private message as a
/// relay drop. Called by the message router alongside physical courier
/// deposits; idempotent per message ID. Completion becomes true only
/// after a real relay acceptance arrives, which is when the router may
/// show "carried".
/// deposits; idempotent per recipient/message pair. Completion becomes
/// true only after a real relay acceptance arrives, which is when the
/// router may show "carried".
func depositDrop(
content: String,
messageID: String,
@@ -229,9 +265,18 @@ final class BridgeCourierService: ObservableObject {
completion(false)
return
}
guard !publishedDropKeys.contains(messageID, now: now()),
activeDropOperations[messageID] == nil,
!rejectedDropKeys.contains(messageID, now: now()) else {
let date = now()
let dedupKey = Self.senderDropKey(
messageID: messageID,
recipientNoiseKey: recipientNoiseKey
)
guard !wasPublished(
legacyMessageID: messageID,
dedupKey: dedupKey,
now: date
),
activeDropOperations[dedupKey] == nil,
!rejectedDropKeys.contains(dedupKey, now: date) else {
completion(false)
return
}
@@ -244,13 +289,13 @@ final class BridgeCourierService: ObservableObject {
// of the sealing); suppress it in-memory so the retry sweep does not
// churn, but never persist it as a published drop.
guard let encoded = envelope.encode(), encoded.count <= Limits.maxDropEnvelopeBytes else {
rejectedDropKeys.insert(messageID, now: now())
rejectedDropKeys.insert(dedupKey, now: date)
completion(false)
return
}
let operationID = UUID()
activeDropOperations[messageID] = ActiveDropOperation(id: operationID, completion: completion)
publishDrop(envelope, messageID: messageID, operationID: operationID)
activeDropOperations[dedupKey] = ActiveDropOperation(id: operationID, completion: completion)
publishDrop(envelope, dedupKey: dedupKey, operationID: operationID)
}
/// Publishes held envelopes (mail we carry for others) as drops,
@@ -272,13 +317,14 @@ final class BridgeCourierService: ObservableObject {
}
}
/// Publishes a drop, or queues it when relays are down. `messageID` is the
/// sender-side dedup key (nil for held/relayed envelopes we don't track);
/// it rides the pending queue so an evicted or failed drop can release its
/// in-flight slot. Completion reports actual NIP-01 relay acceptance.
/// Publishes a drop, or queues it when relays are down. `dedupKey` is the
/// opaque sender-side recipient/message key (nil for held/relayed
/// envelopes we don't track); it rides the pending queue so an evicted or
/// failed drop can release its in-flight slot. Completion reports actual
/// NIP-01 relay acceptance.
private func publishDrop(
_ envelope: CourierEnvelope,
messageID: String? = nil,
dedupKey: String? = nil,
operationID: UUID? = nil,
untrackedCompletion: (@MainActor (Bool) -> Void)? = nil
) {
@@ -286,7 +332,7 @@ final class BridgeCourierService: ObservableObject {
encoded.count <= Limits.maxDropEnvelopeBytes,
!envelope.isExpired else {
finishPublish(
messageID: messageID,
dedupKey: dedupKey,
operationID: operationID,
succeeded: false,
untrackedCompletion: untrackedCompletion
@@ -297,15 +343,15 @@ final class BridgeCourierService: ObservableObject {
// Held mail remains in CourierStore and has no sender operation to
// recover after an in-memory queue loss. Leave its cooldown unset
// and let the next connected refresh offer it again.
guard messageID != nil else {
guard dedupKey != nil else {
untrackedCompletion?(false)
return
}
pendingDrops.append((envelope, messageID, operationID))
pendingDrops.append((envelope, dedupKey, operationID))
while pendingDrops.count > Limits.maxPendingDrops {
let evicted = pendingDrops.removeFirst()
finishPublish(
messageID: evicted.dedupKey,
dedupKey: evicted.dedupKey,
operationID: evicted.operationID,
succeeded: false
)
@@ -321,7 +367,7 @@ final class BridgeCourierService: ObservableObject {
) else {
SecureLogger.error("📦🌉 Failed to compose courier drop", category: .encryption)
finishPublish(
messageID: messageID,
dedupKey: dedupKey,
operationID: operationID,
succeeded: false,
untrackedCompletion: untrackedCompletion
@@ -331,7 +377,7 @@ final class BridgeCourierService: ObservableObject {
guard let publishEvent else {
SecureLogger.error("📦🌉 Courier drop publisher is not configured", category: .session)
finishPublish(
messageID: messageID,
dedupKey: dedupKey,
operationID: operationID,
succeeded: false,
untrackedCompletion: untrackedCompletion
@@ -341,7 +387,7 @@ final class BridgeCourierService: ObservableObject {
publishEvent(event) { [weak self] succeeded in
guard let self else { return }
guard self.finishPublish(
messageID: messageID,
dedupKey: dedupKey,
operationID: operationID,
succeeded: succeeded,
untrackedCompletion: untrackedCompletion
@@ -356,23 +402,23 @@ final class BridgeCourierService: ObservableObject {
@discardableResult
private func finishPublish(
messageID: String?,
dedupKey: String?,
operationID: UUID?,
succeeded: Bool,
untrackedCompletion: (@MainActor (Bool) -> Void)? = nil
) -> Bool {
guard let messageID else {
guard let dedupKey else {
untrackedCompletion?(succeeded)
return true
}
// Missing/mismatched means this callback was duplicated, invalidated
// by panic wipe, or belongs to an older attempt for the same key.
guard let operationID,
let operation = activeDropOperations[messageID],
let operation = activeDropOperations[dedupKey],
operation.id == operationID else { return false }
activeDropOperations.removeValue(forKey: messageID)
activeDropOperations.removeValue(forKey: dedupKey)
if succeeded {
publishedDropKeys.insert(messageID, now: now())
publishedDropKeys.insert(dedupKey, now: now())
persistDedup()
}
operation.completion(succeeded)
@@ -387,7 +433,7 @@ final class BridgeCourierService: ObservableObject {
for item in queued {
publishDrop(
item.envelope,
messageID: item.dedupKey,
dedupKey: item.dedupKey,
operationID: item.operationID
)
}
@@ -64,11 +64,11 @@ struct ExpiringIDSet {
/// fresh drop (fresh throwaway seal, undeduplicatable downstream) and every
/// gateway relaunch re-delivered the whole backlog. Field-verified: ~20
/// copies of one DM delivered in 40ms fed the storm behind a permanent
/// device freeze. Persisting both sides caps this at one drop per message
/// ID per 24h regardless of relaunch count.
/// device freeze. Persisting both sides caps this at one drop per
/// recipient/message pair per 24h regardless of relaunch count.
///
/// Contents are opaque IDs (message UUIDs, relay event IDs) no plaintext,
/// no peer identities so until-first-unlock protection matches
/// Contents are opaque hashes and relay event IDs no plaintext or peer
/// identities so until-first-unlock protection matches
/// `NostrProcessedEventStore`, and the file must load during a
/// locked-background restoration relaunch. Wiped on panic with the rest of
/// the courier state.
+38 -27
View File
@@ -34,6 +34,11 @@ struct CourierDirectory {
final class MessageRouter {
typealias QueuedMessage = MessageOutboxStore.QueuedMessage
private struct PeerMessageKey: Hashable {
let peerID: PeerID
let messageID: String
}
private let transports: [Transport]
private let now: () -> Date
private let courierDirectory: CourierDirectory
@@ -104,15 +109,17 @@ final class MessageRouter {
}
private var bridgeSweepTask: Task<Void, Never>?
private var bridgeDepositsInFlight = Set<String>()
private var bridgeDepositsInFlight = Set<PeerMessageKey>()
private var outbox: [PeerID: [QueuedMessage]] = [:]
/// IDs whose latest router-owned transmission used an already-established
/// secure session and still awaits an ack. This deliberately excludes
/// messages handed to BLE while a handshake is pending: BLE owns those
/// sends and drains its queue after authentication, so retrying them here
/// would duplicate every normal first-handshake DM.
private var securelyTransmittedMessageIDs = Set<String>()
/// Peer/message pairs whose latest router-owned transmission used an
/// already-established secure session and still await an ack. Peer scope
/// is required because message IDs are not globally unique across direct
/// conversations. This deliberately excludes messages handed to BLE while
/// a handshake is pending: BLE owns those sends and drains its queue after
/// authentication, so retrying them here would duplicate every normal
/// first-handshake DM.
private var secureTransmissions = Set<PeerMessageKey>()
// Outbox limits to prevent unbounded memory growth
private static let maxMessagesPerPeer = 100
@@ -205,7 +212,7 @@ final class MessageRouter {
// replacement handshake will retry this same message ID, which
// receivers deduplicate.
enqueue(message, for: peerID)
securelyTransmittedMessageIDs.insert(messageID)
secureTransmissions.insert(PeerMessageKey(peerID: peerID, messageID: messageID))
SecureLogger.debug("Routing PM via \(type(of: transport)) (connected) to \(peerID.id.prefix(8))… id=\(messageID.prefix(8))", category: .session)
transport.sendPrivateMessage(content, to: peerID, recipientNickname: recipientNickname, messageID: messageID)
return
@@ -229,7 +236,7 @@ final class MessageRouter {
// away.
SecureLogger.debug("Routing PM via \(type(of: transport)) (connected, no secure session) to \(peerID.id.prefix(8))… id=\(messageID.prefix(8))", category: .session)
enqueue(message, for: peerID)
securelyTransmittedMessageIDs.remove(messageID)
secureTransmissions.remove(PeerMessageKey(peerID: peerID, messageID: messageID))
transport.sendPrivateMessage(content, to: peerID, recipientNickname: recipientNickname, messageID: messageID)
attemptCourierDeposit(messageID: messageID, for: peerID)
return
@@ -365,11 +372,12 @@ final class MessageRouter {
for peerID: PeerID,
recipientKey: Data
) {
let inFlightKey = PeerMessageKey(peerID: peerID, messageID: message.messageID)
guard let bridgeCourierDeposit,
bridgeDepositsInFlight.insert(message.messageID).inserted else { return }
bridgeDepositsInFlight.insert(inFlightKey).inserted else { return }
bridgeCourierDeposit(message.content, message.messageID, recipientKey) { [weak self] succeeded in
guard let self else { return }
self.bridgeDepositsInFlight.remove(message.messageID)
self.bridgeDepositsInFlight.remove(inFlightKey)
// A direct delivery may have cleared the outbox while the relay
// relay confirmation was in flight; do not regress its UI state.
guard succeeded, self.queuedMessage(message.messageID, for: peerID) != nil else { return }
@@ -401,7 +409,7 @@ final class MessageRouter {
/// retry state.
func markDelivered(_ messageID: String, from peerIDs: Set<PeerID>) {
guard !peerIDs.isEmpty else { return }
clearRetainedMessage(messageID, allowedPeerIDs: peerIDs)
_ = markDelivered(messageID, for: Array(peerIDs))
}
private func clearRetainedMessage(
@@ -418,11 +426,10 @@ final class MessageRouter {
outbox[peerID] = filtered.isEmpty ? nil : filtered
cleared = true
}
if !outbox.values.contains(where: { queue in
queue.contains { $0.messageID == messageID }
}) {
securelyTransmittedMessageIDs.remove(messageID)
let matchingSecureTransmissions = secureTransmissions.filter {
$0.messageID == messageID
}
secureTransmissions.subtract(matchingSecureTransmissions)
// The durable snapshot may still be hidden by protected data. Record
// the ack even when this cold-load view cannot find the message, then
// persist the current view so the store retains a removal tombstone.
@@ -449,10 +456,8 @@ final class MessageRouter {
outbox[peerID] = filtered.isEmpty ? nil : filtered
cleared = true
}
if !outbox.values.contains(where: { queue in
queue.contains { $0.messageID == messageID }
}) {
securelyTransmittedMessageIDs.remove(messageID)
for peerID in peerIDs {
secureTransmissions.remove(PeerMessageKey(peerID: peerID, messageID: messageID))
}
// Preserve the scoped ack even when protected data hides the durable
// queue during a cold launch.
@@ -514,7 +519,7 @@ final class MessageRouter {
}
private func dropMessage(_ messageID: String, for peerID: PeerID) {
securelyTransmittedMessageIDs.remove(messageID)
secureTransmissions.remove(PeerMessageKey(peerID: peerID, messageID: messageID))
metrics?.record(.outboxDropped)
onMessageDropped?(messageID, peerID)
}
@@ -558,7 +563,7 @@ final class MessageRouter {
/// Panic wipe: forget queued mail on disk and in memory.
func wipeOutbox() {
outbox.removeAll()
securelyTransmittedMessageIDs.removeAll()
secureTransmissions.removeAll()
outboxStore?.wipe()
}
@@ -592,7 +597,7 @@ final class MessageRouter {
/// A peer restart can leave that local session looking usable until the
/// replacement handshake arrives; the first ciphertext is then
/// undecryptable remotely. Normal pre-handshake sends are intentionally
/// absent from `securelyTransmittedMessageIDs` because BLE already queues
/// absent from `secureTransmissions` because BLE already queues
/// and drains them when authentication completes.
func retrySecurePrivateMessagesAfterAuthentication(for peerIDAliases: [PeerID]) {
typealias Candidate = (
@@ -617,7 +622,8 @@ final class MessageRouter {
}
for (queueOrder, message) in queued.enumerated() {
guard securelyTransmittedMessageIDs.contains(message.messageID) else { continue }
let key = PeerMessageKey(peerID: peerID, messageID: message.messageID)
guard secureTransmissions.contains(key) else { continue }
candidates.append((
peerID: peerID,
message: message,
@@ -647,8 +653,9 @@ final class MessageRouter {
for candidate in candidates {
let peerID = candidate.peerID
let message = candidate.message
let key = PeerMessageKey(peerID: peerID, messageID: message.messageID)
guard retriedMessageIDs.insert(message.messageID).inserted,
securelyTransmittedMessageIDs.contains(message.messageID),
secureTransmissions.contains(key),
queuedMessage(message.messageID, for: peerID) != nil,
let transport = connectedTransport(for: peerID),
transport.canDeliverSecurely(to: peerID) else {
@@ -732,7 +739,9 @@ final class MessageRouter {
continue
}
SecureLogger.debug("Outbox -> \(type(of: transport)) (connected) for \(peerID.id.prefix(8))… id=\(message.messageID.prefix(8))", category: .session)
securelyTransmittedMessageIDs.insert(message.messageID)
secureTransmissions.insert(
PeerMessageKey(peerID: peerID, messageID: message.messageID)
)
transport.sendPrivateMessage(message.content, to: peerID, recipientNickname: message.nickname, messageID: message.messageID)
metrics?.record(.outboxResent)
outboxChanged = incrementSendAttemptsIfQueued(message.messageID, for: peerID) || outboxChanged
@@ -748,7 +757,9 @@ final class MessageRouter {
// preserve. Retention stays bounded by the 24h outbox TTL
// and the per-peer FIFO cap.
SecureLogger.debug("Outbox -> \(type(of: transport)) (connected, no secure session) for \(peerID.id.prefix(8))… id=\(message.messageID.prefix(8))", category: .session)
securelyTransmittedMessageIDs.remove(message.messageID)
secureTransmissions.remove(
PeerMessageKey(peerID: peerID, messageID: message.messageID)
)
transport.sendPrivateMessage(message.content, to: peerID, recipientNickname: message.nickname, messageID: message.messageID)
metrics?.record(.outboxResent)
} else if let transport = reachableTransport(for: peerID) {
@@ -97,7 +97,7 @@ enum EncryptionStatus: Equatable {
case noiseHandshaking // Currently establishing
case noiseSecured // Established but not verified
case noiseVerified // Established and verified
var icon: String? { // Made optional to hide icon when no handshake
switch self {
case .none:
@@ -660,7 +660,7 @@ final class NoiseEncryptionService {
guard let packetData = packet.toBinaryDataForSigning() else {
return nil
}
// Sign with the noise private key (converted to Ed25519 for signing)
guard let signature = signData(packetData) else {
return nil
@@ -919,7 +919,7 @@ final class NoiseEncryptionService {
}
return try sessionManager.encrypt(data, for: peerID)
}
/// Decrypt data from a specific peer
func decrypt(_ data: Data, from peerID: PeerID) throws -> Data {
try decryptWithSessionGeneration(data, from: peerID).plaintext
@@ -936,7 +936,7 @@ final class NoiseEncryptionService {
let isStandardCiphertext = NoiseSecurityValidator.validateCiphertextSize(data)
let isAdmittedCiphertext = isStandardCiphertext
|| NoiseSecurityValidator.validatePrivateFileCiphertextSize(data)
// A quarantined transport is deliberately unavailable for outbound
// state, but remains receive-only until the responder proves identity
// or the bounded rollback restores it.
@@ -21,6 +21,14 @@ protocol ChatDeliveryContext: AnyObject {
/// message is unknown or no copy changed.
@discardableResult
func setDeliveryStatus(_ status: DeliveryStatus, forMessageID messageID: String) -> Bool
/// Applies an authenticated receipt only to the direct conversations
/// represented by the supplied peer aliases.
@discardableResult
func setDeliveryStatus(
_ status: DeliveryStatus,
forMessageID messageID: String,
inDirectPeerAliases peerIDs: Set<PeerID>
) -> Bool
/// Current delivery status of the message in whichever conversation holds it.
func deliveryStatus(forMessageID messageID: String) -> DeliveryStatus?
/// Message IDs across all direct conversations (read-receipt pruning).
@@ -47,6 +55,19 @@ extension ChatViewModel: ChatDeliveryContext {
conversations.setDeliveryStatus(status, forMessageID: messageID)
}
@discardableResult
func setDeliveryStatus(
_ status: DeliveryStatus,
forMessageID messageID: String,
inDirectPeerAliases peerIDs: Set<PeerID>
) -> Bool {
conversations.setDeliveryStatus(
status,
forMessageID: messageID,
inDirectPeerAliases: peerIDs
)
}
func deliveryStatus(forMessageID messageID: String) -> DeliveryStatus? {
conversations.deliveryStatus(forMessageID: messageID)
}
@@ -68,6 +89,12 @@ extension ChatViewModel: ChatDeliveryContext {
func markMessageDelivered(_ messageID: String, from peerIDs: Set<PeerID>) {
messageRouter.markDelivered(messageID, from: peerIDs)
// The caller has already bound this receipt to one of our outgoing
// conversations. Release the matching media retry only after that
// peer-scoped validation and accepted delivery-status transition.
mediaTransferCoordinator.confirmPrivateMediaDelivery(
messageID: messageID
)
}
func isOutgoingPrivateMessage(_ messageID: String, toAny peerIDs: Set<PeerID>) -> Bool {
@@ -157,7 +184,11 @@ final class ChatDeliveryCoordinator {
}
guard !peerIDAliases.isEmpty,
context.isOutgoingPrivateMessage(messageID, toAny: peerIDAliases),
context.setDeliveryStatus(status, forMessageID: messageID) else {
context.setDeliveryStatus(
status,
forMessageID: messageID,
inDirectPeerAliases: peerIDAliases
) else {
return false
}
context.markMessageDelivered(messageID, from: peerIDAliases)