mirror of
https://github.com/permissionlesstech/bitchat.git
synced 2026-07-25 03:05:19 +00:00
Verify Nostr inbound signatures once, off the main actor
Every inbound relay event was Schnorr-verified TWICE, both times on the main actor: once in NostrRelayManager.handleParsedMessage and again in each NostrInboundPipeline / GeoPresenceTracker handler. Each verification re-serializes the event JSON, hashes it, and runs a secp256k1 Schnorr verify, so busy geohashes paid double crypto on the UI thread. Geohash gift-wrap NIP-17 decryption (two ECDH+ChaCha layers) also ran on the main actor. Changes: - NostrRelayManager now owns a serial off-main inbound pipeline (AsyncStream + single consumer task): frames are parsed and verified in arrival order off the main actor, preserving per-subscription delivery order. This is the single verification point for the whole inbound path; downstream handlers only ever see verified events. - Dedup stays two-phase and unpoisonable: a cheap main-actor duplicate LOOKUP runs before verification (duplicate fan-in from several relays never pays for crypto — previously every duplicate was verified), and events are RECORDED as seen only after the signature verifies, so a forged-signature copy can never suppress the genuine event. - NostrInboundPipeline and GeoPresenceTracker drop their redundant re-verification; geohash gift-wrap decryption moves off the main actor following the existing account-mailbox Task.detached pattern (atomic main-actor check-and-record, then decrypt off-main, then hop back for state updates). - Tests: relay-level coverage for tampered gift wraps and for in-order delivery of back-to-back frames; pipeline-level tampered-signature tests move to the relay boundary where the invariant now lives; the mock relay connection queues frames emitted before receive re-arms. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
@@ -213,8 +213,23 @@ final class NostrRelayManager: ObservableObject {
|
|||||||
// Bump generation to invalidate scheduled reconnects when we reset/disconnect
|
// Bump generation to invalidate scheduled reconnects when we reset/disconnect
|
||||||
private var connectionGeneration: Int = 0
|
private var connectionGeneration: Int = 0
|
||||||
|
|
||||||
|
// Serial off-main inbound pipeline: raw socket frames are parsed and
|
||||||
|
// Schnorr-verified in arrival order OFF the main actor (this is the single
|
||||||
|
// signature verification for the whole inbound path — downstream handlers
|
||||||
|
// receive only verified events), then hop back to the main actor for dedup
|
||||||
|
// recording and handler dispatch. A single consumer task preserves
|
||||||
|
// per-subscription arrival order.
|
||||||
|
private struct InboundFrame {
|
||||||
|
let message: URLSessionWebSocketTask.Message
|
||||||
|
let relayUrl: String
|
||||||
|
}
|
||||||
|
private let inboundContinuation: AsyncStream<InboundFrame>.Continuation
|
||||||
|
private var inboundTask: Task<Void, Never>?
|
||||||
|
|
||||||
init() {
|
init() {
|
||||||
self.dependencies = .live()
|
self.dependencies = .live()
|
||||||
|
let (inboundStream, inboundContinuation) = AsyncStream<InboundFrame>.makeStream()
|
||||||
|
self.inboundContinuation = inboundContinuation
|
||||||
hasMutualFavorites = dependencies.hasMutualFavorites()
|
hasMutualFavorites = dependencies.hasMutualFavorites()
|
||||||
hasLocationPermission = dependencies.hasLocationPermission()
|
hasLocationPermission = dependencies.hasLocationPermission()
|
||||||
applyDefaultRelayPolicy(force: true)
|
applyDefaultRelayPolicy(force: true)
|
||||||
@@ -238,10 +253,13 @@ final class NostrRelayManager: ObservableObject {
|
|||||||
self.applyDefaultRelayPolicy()
|
self.applyDefaultRelayPolicy()
|
||||||
}
|
}
|
||||||
.store(in: &cancellables)
|
.store(in: &cancellables)
|
||||||
|
startInboundPipeline(consuming: inboundStream)
|
||||||
}
|
}
|
||||||
|
|
||||||
internal init(dependencies: NostrRelayManagerDependencies) {
|
internal init(dependencies: NostrRelayManagerDependencies) {
|
||||||
self.dependencies = dependencies
|
self.dependencies = dependencies
|
||||||
|
let (inboundStream, inboundContinuation) = AsyncStream<InboundFrame>.makeStream()
|
||||||
|
self.inboundContinuation = inboundContinuation
|
||||||
hasMutualFavorites = dependencies.hasMutualFavorites()
|
hasMutualFavorites = dependencies.hasMutualFavorites()
|
||||||
hasLocationPermission = dependencies.hasLocationPermission()
|
hasLocationPermission = dependencies.hasLocationPermission()
|
||||||
applyDefaultRelayPolicy(force: true)
|
applyDefaultRelayPolicy(force: true)
|
||||||
@@ -265,6 +283,54 @@ final class NostrRelayManager: ObservableObject {
|
|||||||
self.applyDefaultRelayPolicy()
|
self.applyDefaultRelayPolicy()
|
||||||
}
|
}
|
||||||
.store(in: &cancellables)
|
.store(in: &cancellables)
|
||||||
|
startInboundPipeline(consuming: inboundStream)
|
||||||
|
}
|
||||||
|
|
||||||
|
deinit {
|
||||||
|
inboundContinuation.finish()
|
||||||
|
inboundTask?.cancel()
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Starts the single consumer task behind `inboundContinuation`.
|
||||||
|
///
|
||||||
|
/// Ordering is deliberate and security/performance-critical:
|
||||||
|
/// 1. `precheckInboundEvent` (main hop): per-relay stats plus a cheap
|
||||||
|
/// duplicate LOOKUP — duplicate fan-in from multiple relays dominates
|
||||||
|
/// real traffic and must never pay for Schnorr verification.
|
||||||
|
/// 2. `isValidSignature()` runs here, off the main actor — the ONLY
|
||||||
|
/// signature verification on the inbound path (JSON re-serialization +
|
||||||
|
/// SHA-256 + secp256k1 Schnorr per event).
|
||||||
|
/// 3. `deliverVerifiedInboundEvent` (main hop): authoritative
|
||||||
|
/// check-and-RECORD plus handler dispatch. Recording only after
|
||||||
|
/// verification means a forged-signature copy can never poison the
|
||||||
|
/// dedup cache and suppress the genuine event.
|
||||||
|
private func startInboundPipeline(consuming stream: AsyncStream<InboundFrame>) {
|
||||||
|
inboundTask = Task.detached(priority: .userInitiated) { [weak self] in
|
||||||
|
for await frame in stream {
|
||||||
|
guard let parsed = ParsedInbound(frame.message) else { continue }
|
||||||
|
guard let self else { return }
|
||||||
|
switch parsed {
|
||||||
|
case .event(let subId, let event):
|
||||||
|
guard await self.precheckInboundEvent(
|
||||||
|
subscriptionID: subId,
|
||||||
|
eventID: event.id,
|
||||||
|
relayUrl: frame.relayUrl
|
||||||
|
) else {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
guard event.isValidSignature() else {
|
||||||
|
SecureLogger.warning(
|
||||||
|
"⚠️ Dropped invalid Nostr event id=\(event.id.prefix(16))… sub=\(subId) relay=\(frame.relayUrl)",
|
||||||
|
category: .session
|
||||||
|
)
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
await self.deliverVerifiedInboundEvent(subscriptionID: subId, event: event, from: frame.relayUrl)
|
||||||
|
case .eose, .ok, .notice:
|
||||||
|
await self.handleParsedMessage(parsed, from: frame.relayUrl)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Connect to all configured relays
|
/// Connect to all configured relays
|
||||||
@@ -958,13 +1024,9 @@ final class NostrRelayManager: ObservableObject {
|
|||||||
|
|
||||||
switch result {
|
switch result {
|
||||||
case .success(let message):
|
case .success(let message):
|
||||||
// Parse off-main to reduce UI jank, then hop back for state updates
|
// Hand the raw frame to the serial inbound pipeline: parsing
|
||||||
Task.detached(priority: .utility) {
|
// and signature verification run off-main, in arrival order.
|
||||||
guard let parsed = ParsedInbound(message) else { return }
|
self.inboundContinuation.yield(InboundFrame(message: message, relayUrl: relayUrl))
|
||||||
await MainActor.run {
|
|
||||||
self.handleParsedMessage(parsed, from: relayUrl)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// Continue receiving
|
// Continue receiving
|
||||||
Task { @MainActor in
|
Task { @MainActor in
|
||||||
@@ -983,35 +1045,55 @@ final class NostrRelayManager: ObservableObject {
|
|||||||
// Note: declared at file scope below to avoid MainActor isolation inside this class
|
// Note: declared at file scope below to avoid MainActor isolation inside this class
|
||||||
// and keep parsing off the main actor.
|
// and keep parsing off the main actor.
|
||||||
|
|
||||||
// Handle parsed message on MainActor (state updates and handlers)
|
/// First main-actor hop for an inbound EVENT: per-relay stats plus a cheap
|
||||||
|
/// duplicate LOOKUP (no recording) so duplicate fan-in from multiple
|
||||||
|
/// relays never pays for Schnorr verification. Recording happens only
|
||||||
|
/// after the signature verifies (`deliverVerifiedInboundEvent`), so a
|
||||||
|
/// forged-signature copy can never poison the dedup cache and suppress
|
||||||
|
/// the genuine event.
|
||||||
|
private func precheckInboundEvent(subscriptionID: String, eventID: String, relayUrl: String) -> Bool {
|
||||||
|
if let index = relays.firstIndex(where: { $0.url == relayUrl }) {
|
||||||
|
relays[index].messagesReceived += 1
|
||||||
|
}
|
||||||
|
guard !eventID.isEmpty else { return true }
|
||||||
|
let key = InboundEventKey(subscriptionID: subscriptionID, eventID: eventID)
|
||||||
|
if recentInboundEventKeys.contains(key) {
|
||||||
|
recordDuplicateInboundEventDrop(subscriptionID: subscriptionID)
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Second main-actor hop, after off-main signature verification:
|
||||||
|
/// authoritative check-and-record (the serial pipeline means the same
|
||||||
|
/// event is never in flight twice, but the record must stay atomic with
|
||||||
|
/// delivery) and handler dispatch.
|
||||||
|
private func deliverVerifiedInboundEvent(subscriptionID subId: String, event: NostrEvent, from relayUrl: String) {
|
||||||
|
guard shouldDeliverInboundEvent(subscriptionID: subId, eventID: event.id) else {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if event.kind != 1059 {
|
||||||
|
// Per-event logging floods dev builds in busy geohashes; sample it.
|
||||||
|
inboundEventLogCount += 1
|
||||||
|
if inboundEventLogCount == 1 || inboundEventLogCount.isMultiple(of: TransportConfig.nostrInboundEventLogInterval) {
|
||||||
|
SecureLogger.debug("📥 Event #\(inboundEventLogCount) kind=\(event.kind) id=\(event.id.prefix(16))… relay=\(relayUrl)", category: .session)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if let handler = self.messageHandlers[subId] {
|
||||||
|
handler(event)
|
||||||
|
} else {
|
||||||
|
SecureLogger.warning("⚠️ No handler for subscription \(subId)", category: .session)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Handle parsed non-EVENT messages on MainActor (state updates and handlers)
|
||||||
private func handleParsedMessage(_ parsed: ParsedInbound, from relayUrl: String) {
|
private func handleParsedMessage(_ parsed: ParsedInbound, from relayUrl: String) {
|
||||||
switch parsed {
|
switch parsed {
|
||||||
case .event(let subId, let event):
|
case .event:
|
||||||
if let index = self.relays.firstIndex(where: { $0.url == relayUrl }) {
|
// Events flow through the serial inbound pipeline (precheck →
|
||||||
self.relays[index].messagesReceived += 1
|
// off-main signature verification → deliverVerifiedInboundEvent)
|
||||||
}
|
// and never reach this fallback.
|
||||||
guard event.isValidSignature() else {
|
assertionFailure("inbound EVENT bypassed the verified pipeline")
|
||||||
SecureLogger.warning(
|
|
||||||
"⚠️ Dropped invalid Nostr event id=\(event.id.prefix(16))… sub=\(subId) relay=\(relayUrl)",
|
|
||||||
category: .session
|
|
||||||
)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
guard shouldDeliverInboundEvent(subscriptionID: subId, eventID: event.id) else {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
if event.kind != 1059 {
|
|
||||||
// Per-event logging floods dev builds in busy geohashes; sample it.
|
|
||||||
inboundEventLogCount += 1
|
|
||||||
if inboundEventLogCount == 1 || inboundEventLogCount.isMultiple(of: TransportConfig.nostrInboundEventLogInterval) {
|
|
||||||
SecureLogger.debug("📥 Event #\(inboundEventLogCount) kind=\(event.kind) id=\(event.id.prefix(16))… relay=\(relayUrl)", category: .session)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
if let handler = self.messageHandlers[subId] {
|
|
||||||
handler(event)
|
|
||||||
} else {
|
|
||||||
SecureLogger.warning("⚠️ No handler for subscription \(subId)", category: .session)
|
|
||||||
}
|
|
||||||
case .eose(let subId):
|
case .eose(let subId):
|
||||||
if var tracker = eoseTrackers[subId] {
|
if var tracker = eoseTrackers[subId] {
|
||||||
tracker.pendingRelays.remove(relayUrl)
|
tracker.pendingRelays.remove(relayUrl)
|
||||||
|
|||||||
@@ -103,7 +103,8 @@ final class GeoPresenceTracker {
|
|||||||
else {
|
else {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
guard event.isValidSignature() else { return }
|
// The signature was already verified (exactly once, off the main
|
||||||
|
// actor) by NostrRelayManager before delivery.
|
||||||
guard shouldProcessGeoSamplingEvent(event.id) else { return }
|
guard shouldProcessGeoSamplingEvent(event.id) else { return }
|
||||||
|
|
||||||
let existingCount = context.geoParticipantCount(for: gh)
|
let existingCount = context.geoParticipantCount(for: gh)
|
||||||
|
|||||||
@@ -75,15 +75,17 @@ extension ChatViewModel: NostrInboundPipelineContext {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// The inbound Nostr hot path: raw relay events in, chat messages / Noise
|
/// The inbound Nostr hot path: verified relay events in, chat messages /
|
||||||
/// payloads out. Pure transformation plus dedup — no relay lifecycle.
|
/// Noise payloads out. Pure transformation plus dedup — no relay lifecycle.
|
||||||
///
|
///
|
||||||
/// Ordering is deliberate and performance-critical: cheap rejects (kind,
|
/// Every event arriving here already had its Schnorr signature verified
|
||||||
/// dedup lookup) run BEFORE Schnorr signature verification because duplicates
|
/// exactly once, off the main actor, by `NostrRelayManager`'s serial inbound
|
||||||
/// dominate real relay traffic; events are recorded only AFTER verification so
|
/// pipeline (which records events into its own dedup cache only AFTER
|
||||||
/// a forged-signature copy can never poison the dedup set; gift-wrap
|
/// verification, so forged copies can't suppress genuine events). This
|
||||||
/// verification for the account mailbox runs off-main with an atomic
|
/// pipeline therefore never re-verifies; it keeps its own event-ID dedup
|
||||||
/// main-actor check-and-record.
|
/// (cheap main-actor lookups) and moves NIP-17 gift-wrap decryption — two
|
||||||
|
/// ECDH+ChaCha layers — off the main actor with an atomic main-actor
|
||||||
|
/// check-and-record.
|
||||||
final class NostrInboundPipeline {
|
final class NostrInboundPipeline {
|
||||||
private weak var context: (any NostrInboundPipelineContext)?
|
private weak var context: (any NostrInboundPipelineContext)?
|
||||||
private let presence: GeoPresenceTracker
|
private let presence: GeoPresenceTracker
|
||||||
@@ -97,17 +99,15 @@ final class NostrInboundPipeline {
|
|||||||
@MainActor
|
@MainActor
|
||||||
func subscribeNostrEvent(_ event: NostrEvent) {
|
func subscribeNostrEvent(_ event: NostrEvent) {
|
||||||
guard let context else { return }
|
guard let context else { return }
|
||||||
// Cheap rejects (kind, dedup lookup) before Schnorr verification —
|
// Cheap rejects (kind, dedup lookup) — duplicates dominate real
|
||||||
// duplicates dominate real traffic and must not pay for crypto.
|
// traffic. The signature was already verified (exactly once, off the
|
||||||
// Only verified events are recorded, so a forged-signature copy can
|
// main actor) by NostrRelayManager before delivery.
|
||||||
// never poison the dedup set and suppress the genuine event.
|
|
||||||
guard (event.kind == NostrProtocol.EventKind.ephemeralEvent.rawValue
|
guard (event.kind == NostrProtocol.EventKind.ephemeralEvent.rawValue
|
||||||
|| event.kind == NostrProtocol.EventKind.geohashPresence.rawValue),
|
|| event.kind == NostrProtocol.EventKind.geohashPresence.rawValue),
|
||||||
!context.hasProcessedNostrEvent(event.id)
|
!context.hasProcessedNostrEvent(event.id)
|
||||||
else {
|
else {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
guard event.isValidSignature() else { return }
|
|
||||||
|
|
||||||
context.recordProcessedNostrEvent(event.id)
|
context.recordProcessedNostrEvent(event.id)
|
||||||
|
|
||||||
@@ -176,15 +176,14 @@ final class NostrInboundPipeline {
|
|||||||
@MainActor
|
@MainActor
|
||||||
func handleNostrEvent(_ event: NostrEvent) {
|
func handleNostrEvent(_ event: NostrEvent) {
|
||||||
guard let context else { return }
|
guard let context else { return }
|
||||||
// Cheap rejects (kind, dedup lookup) before Schnorr verification —
|
// Cheap rejects (kind, dedup lookup) — the signature was already
|
||||||
// duplicates dominate real traffic and must not pay for crypto.
|
// verified (exactly once, off the main actor) by NostrRelayManager.
|
||||||
guard (event.kind == NostrProtocol.EventKind.ephemeralEvent.rawValue
|
guard (event.kind == NostrProtocol.EventKind.ephemeralEvent.rawValue
|
||||||
|| event.kind == NostrProtocol.EventKind.geohashPresence.rawValue)
|
|| event.kind == NostrProtocol.EventKind.geohashPresence.rawValue)
|
||||||
else {
|
else {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
if context.hasProcessedNostrEvent(event.id) { return }
|
if context.hasProcessedNostrEvent(event.id) { return }
|
||||||
guard event.isValidSignature() else { return }
|
|
||||||
context.recordProcessedNostrEvent(event.id)
|
context.recordProcessedNostrEvent(event.id)
|
||||||
|
|
||||||
// Sampled: fires for every geo event and floods dev logs in busy geohashes.
|
// Sampled: fires for every geo event and floods dev logs in busy geohashes.
|
||||||
@@ -264,104 +263,102 @@ final class NostrInboundPipeline {
|
|||||||
@MainActor
|
@MainActor
|
||||||
func subscribeGiftWrap(_ giftWrap: NostrEvent, id: NostrIdentity) {
|
func subscribeGiftWrap(_ giftWrap: NostrEvent, id: NostrIdentity) {
|
||||||
guard let context else { return }
|
guard let context else { return }
|
||||||
// Dedup lookup before Schnorr verification; record only after it passes.
|
// Cheap dedup pre-check only; processGeohashGiftWrap does the
|
||||||
|
// authoritative main-actor check-and-record before the off-main
|
||||||
|
// NIP-17 unwrap. The outer signature was already verified (exactly
|
||||||
|
// once, off the main actor) by NostrRelayManager.
|
||||||
guard !context.hasProcessedNostrEvent(giftWrap.id) else { return }
|
guard !context.hasProcessedNostrEvent(giftWrap.id) else { return }
|
||||||
guard giftWrap.isValidSignature() else { return }
|
|
||||||
context.recordProcessedNostrEvent(giftWrap.id)
|
|
||||||
|
|
||||||
guard let (content, senderPubkey, rumorTs) = try? NostrProtocol.decryptPrivateMessage(
|
Task.detached(priority: .userInitiated) { [weak self] in
|
||||||
giftWrap: giftWrap,
|
await self?.processGeohashGiftWrap(giftWrap, id: id, verbose: false)
|
||||||
recipientIdentity: id
|
|
||||||
),
|
|
||||||
let packet = Self.decodeEmbeddedBitChatPacket(from: content),
|
|
||||||
packet.type == MessageType.noiseEncrypted.rawValue,
|
|
||||||
let noisePayload = NoisePayload.decode(packet.payload)
|
|
||||||
else {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
let messageTimestamp = Date(timeIntervalSince1970: TimeInterval(rumorTs))
|
|
||||||
let convKey = PeerID(nostr_: senderPubkey)
|
|
||||||
context.registerNostrKeyMapping(senderPubkey, for: convKey)
|
|
||||||
|
|
||||||
switch noisePayload.type {
|
|
||||||
case .privateMessage:
|
|
||||||
context.handlePrivateMessage(
|
|
||||||
noisePayload,
|
|
||||||
senderPubkey: senderPubkey,
|
|
||||||
convKey: convKey,
|
|
||||||
id: id,
|
|
||||||
messageTimestamp: messageTimestamp
|
|
||||||
)
|
|
||||||
case .delivered:
|
|
||||||
context.handleDelivered(noisePayload, senderPubkey: senderPubkey, convKey: convKey)
|
|
||||||
case .readReceipt:
|
|
||||||
context.handleReadReceipt(noisePayload, senderPubkey: senderPubkey, convKey: convKey)
|
|
||||||
case .verifyChallenge, .verifyResponse:
|
|
||||||
break
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@MainActor
|
@MainActor
|
||||||
func handleGiftWrap(_ giftWrap: NostrEvent, id: NostrIdentity) {
|
func handleGiftWrap(_ giftWrap: NostrEvent, id: NostrIdentity) {
|
||||||
guard let context else { return }
|
guard let context else { return }
|
||||||
// Dedup lookup before Schnorr verification; record only after it passes.
|
// Cheap dedup pre-check only; see subscribeGiftWrap.
|
||||||
if context.hasProcessedNostrEvent(giftWrap.id) {
|
if context.hasProcessedNostrEvent(giftWrap.id) {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
guard giftWrap.isValidSignature() else { return }
|
|
||||||
context.recordProcessedNostrEvent(giftWrap.id)
|
Task.detached(priority: .userInitiated) { [weak self] in
|
||||||
|
await self?.processGeohashGiftWrap(giftWrap, id: id, verbose: true)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Geohash-DM gift wrap ingest. The NIP-17 unwrap (two ECDH+ChaCha
|
||||||
|
/// layers) runs off the main actor; results hop back for state updates.
|
||||||
|
/// `verbose` keeps `handleGiftWrap`'s decrypt logging without adding it
|
||||||
|
/// to the sampling path.
|
||||||
|
private func processGeohashGiftWrap(_ giftWrap: NostrEvent, id: NostrIdentity, verbose: Bool) async {
|
||||||
|
guard let context else { return }
|
||||||
|
// Authoritative check-and-record, atomic on the main actor so two
|
||||||
|
// concurrent detached tasks can't both process the same event.
|
||||||
|
let alreadyProcessed: Bool = await MainActor.run {
|
||||||
|
if context.hasProcessedNostrEvent(giftWrap.id) { return true }
|
||||||
|
context.recordProcessedNostrEvent(giftWrap.id)
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
if alreadyProcessed { return }
|
||||||
|
|
||||||
guard let (content, senderPubkey, rumorTs) = try? NostrProtocol.decryptPrivateMessage(
|
guard let (content, senderPubkey, rumorTs) = try? NostrProtocol.decryptPrivateMessage(
|
||||||
giftWrap: giftWrap,
|
giftWrap: giftWrap,
|
||||||
recipientIdentity: id
|
recipientIdentity: id
|
||||||
) else {
|
) else {
|
||||||
SecureLogger.warning("GeoDM: failed decrypt giftWrap id=\(giftWrap.id.prefix(8))…", category: .session)
|
if verbose {
|
||||||
|
SecureLogger.warning("GeoDM: failed decrypt giftWrap id=\(giftWrap.id.prefix(8))…", category: .session)
|
||||||
|
}
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
SecureLogger.debug(
|
if verbose {
|
||||||
"GeoDM: decrypted gift-wrap id=\(giftWrap.id.prefix(16))... from=\(senderPubkey.prefix(8))...",
|
SecureLogger.debug(
|
||||||
category: .session
|
"GeoDM: decrypted gift-wrap id=\(giftWrap.id.prefix(16))... from=\(senderPubkey.prefix(8))...",
|
||||||
)
|
category: .session
|
||||||
|
|
||||||
guard let packet = Self.decodeEmbeddedBitChatPacket(from: content),
|
|
||||||
packet.type == MessageType.noiseEncrypted.rawValue,
|
|
||||||
let payload = NoisePayload.decode(packet.payload)
|
|
||||||
else {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
let convKey = PeerID(nostr_: senderPubkey)
|
|
||||||
context.registerNostrKeyMapping(senderPubkey, for: convKey)
|
|
||||||
|
|
||||||
switch payload.type {
|
|
||||||
case .privateMessage:
|
|
||||||
let messageTimestamp = Date(timeIntervalSince1970: TimeInterval(rumorTs))
|
|
||||||
context.handlePrivateMessage(
|
|
||||||
payload,
|
|
||||||
senderPubkey: senderPubkey,
|
|
||||||
convKey: convKey,
|
|
||||||
id: id,
|
|
||||||
messageTimestamp: messageTimestamp
|
|
||||||
)
|
)
|
||||||
case .delivered:
|
}
|
||||||
context.handleDelivered(payload, senderPubkey: senderPubkey, convKey: convKey)
|
|
||||||
case .readReceipt:
|
await MainActor.run {
|
||||||
context.handleReadReceipt(payload, senderPubkey: senderPubkey, convKey: convKey)
|
guard let packet = Self.decodeEmbeddedBitChatPacket(from: content),
|
||||||
case .verifyChallenge, .verifyResponse:
|
packet.type == MessageType.noiseEncrypted.rawValue,
|
||||||
break
|
let payload = NoisePayload.decode(packet.payload)
|
||||||
|
else {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
let convKey = PeerID(nostr_: senderPubkey)
|
||||||
|
context.registerNostrKeyMapping(senderPubkey, for: convKey)
|
||||||
|
|
||||||
|
switch payload.type {
|
||||||
|
case .privateMessage:
|
||||||
|
let messageTimestamp = Date(timeIntervalSince1970: TimeInterval(rumorTs))
|
||||||
|
context.handlePrivateMessage(
|
||||||
|
payload,
|
||||||
|
senderPubkey: senderPubkey,
|
||||||
|
convKey: convKey,
|
||||||
|
id: id,
|
||||||
|
messageTimestamp: messageTimestamp
|
||||||
|
)
|
||||||
|
case .delivered:
|
||||||
|
context.handleDelivered(payload, senderPubkey: senderPubkey, convKey: convKey)
|
||||||
|
case .readReceipt:
|
||||||
|
context.handleReadReceipt(payload, senderPubkey: senderPubkey, convKey: convKey)
|
||||||
|
case .verifyChallenge, .verifyResponse:
|
||||||
|
break
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@MainActor
|
@MainActor
|
||||||
func handleNostrMessage(_ giftWrap: NostrEvent) {
|
func handleNostrMessage(_ giftWrap: NostrEvent) {
|
||||||
guard let context else { return }
|
guard let context else { return }
|
||||||
// Cheap dedup pre-check only; Schnorr verification runs off-main in
|
// Cheap dedup pre-check only; processNostrMessage does the
|
||||||
// processNostrMessage, which then does the authoritative
|
// authoritative check-and-record before the off-main NIP-17 unwrap.
|
||||||
// check-and-record. Recording stays after verification so a
|
// The outer signature was already verified (exactly once, off the
|
||||||
// forged-signature copy can never poison the dedup set and suppress
|
// main actor) by NostrRelayManager, and only verified events are
|
||||||
// the genuine event.
|
// recorded, so a forged-signature copy can never poison the dedup
|
||||||
|
// set and suppress the genuine event.
|
||||||
if context.hasProcessedNostrEvent(giftWrap.id) { return }
|
if context.hasProcessedNostrEvent(giftWrap.id) { return }
|
||||||
|
|
||||||
Task.detached(priority: .userInitiated) { [weak self] in
|
Task.detached(priority: .userInitiated) { [weak self] in
|
||||||
@@ -370,7 +367,6 @@ final class NostrInboundPipeline {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func processNostrMessage(_ giftWrap: NostrEvent) async {
|
func processNostrMessage(_ giftWrap: NostrEvent) async {
|
||||||
guard giftWrap.isValidSignature() else { return }
|
|
||||||
guard let context else { return }
|
guard let context else { return }
|
||||||
// Authoritative check-and-record, atomic on the main actor so two
|
// Authoritative check-and-record, atomic on the main actor so two
|
||||||
// concurrent detached tasks can't both process the same event.
|
// concurrent detached tasks can't both process the same event.
|
||||||
|
|||||||
@@ -382,10 +382,12 @@ struct ChatNostrCoordinatorContextTests {
|
|||||||
|
|
||||||
coordinator.inbound.handleGiftWrap(giftWrap, id: recipient)
|
coordinator.inbound.handleGiftWrap(giftWrap, id: recipient)
|
||||||
|
|
||||||
|
// The NIP-17 unwrap runs off the main actor; wait for the hop back.
|
||||||
let convKey = PeerID(nostr_: sender.publicKeyHex)
|
let convKey = PeerID(nostr_: sender.publicKeyHex)
|
||||||
|
let routed = await TestHelpers.waitUntil({ context.handledPrivateMessages.count == 1 })
|
||||||
|
#expect(routed)
|
||||||
#expect(context.recordedNostrEventIDs == [giftWrap.id])
|
#expect(context.recordedNostrEventIDs == [giftWrap.id])
|
||||||
#expect(context.nostrKeyMapping[convKey] == sender.publicKeyHex)
|
#expect(context.nostrKeyMapping[convKey] == sender.publicKeyHex)
|
||||||
#expect(context.handledPrivateMessages.count == 1)
|
|
||||||
#expect(context.handledPrivateMessages.first?.senderPubkey == sender.publicKeyHex)
|
#expect(context.handledPrivateMessages.first?.senderPubkey == sender.publicKeyHex)
|
||||||
#expect(context.handledPrivateMessages.first?.convKey == convKey)
|
#expect(context.handledPrivateMessages.first?.convKey == convKey)
|
||||||
|
|
||||||
@@ -398,30 +400,37 @@ struct ChatNostrCoordinatorContextTests {
|
|||||||
|
|
||||||
// The same gift wrap is dropped on replay.
|
// The same gift wrap is dropped on replay.
|
||||||
coordinator.inbound.handleGiftWrap(giftWrap, id: recipient)
|
coordinator.inbound.handleGiftWrap(giftWrap, id: recipient)
|
||||||
|
await drainMainQueue()
|
||||||
#expect(context.recordedNostrEventIDs == [giftWrap.id])
|
#expect(context.recordedNostrEventIDs == [giftWrap.id])
|
||||||
#expect(context.handledPrivateMessages.count == 1)
|
#expect(context.handledPrivateMessages.count == 1)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// NOTE: Inbound Schnorr signature verification (and the forged-copy
|
||||||
|
// dedup-poisoning invariant) is enforced once, off the main actor, at the
|
||||||
|
// relay boundary — see NostrRelayManagerTests
|
||||||
|
// `test_receiveEvent_invalidSignatureDoesNotPoisonDuplicateCache` and
|
||||||
|
// `test_receiveGiftWrap_tamperedSignatureIsDroppedAndDoesNotPoisonDedup`.
|
||||||
|
// The inbound pipeline only ever sees verified events.
|
||||||
|
|
||||||
@Test @MainActor
|
@Test @MainActor
|
||||||
func processNostrMessage_invalidSignatureDoesNotPoisonDedup() async throws {
|
func processNostrMessage_duplicateDeliveryProcessesOnce() async throws {
|
||||||
let context = MockChatNostrContext()
|
let context = MockChatNostrContext()
|
||||||
let coordinator = ChatNostrCoordinator(context: context)
|
let coordinator = ChatNostrCoordinator(context: context)
|
||||||
|
|
||||||
let recipient = try NostrIdentity.generate()
|
let recipient = try NostrIdentity.generate()
|
||||||
let sender = try NostrIdentity.generate()
|
let sender = try NostrIdentity.generate()
|
||||||
|
context.nostrIdentity = recipient
|
||||||
let giftWrap = try NostrProtocol.createPrivateMessage(
|
let giftWrap = try NostrProtocol.createPrivateMessage(
|
||||||
content: "verify:noop",
|
content: "verify:noop",
|
||||||
recipientPubkey: recipient.publicKeyHex,
|
recipientPubkey: recipient.publicKeyHex,
|
||||||
senderIdentity: sender
|
senderIdentity: sender
|
||||||
)
|
)
|
||||||
var invalidGiftWrap = giftWrap
|
|
||||||
invalidGiftWrap.sig = String(repeating: "0", count: 128)
|
|
||||||
|
|
||||||
// A forged-signature copy is rejected WITHOUT entering the dedup set...
|
// Fan-in of the same (already verified) gift wrap from several relays
|
||||||
await coordinator.inbound.processNostrMessage(invalidGiftWrap)
|
// records and processes exactly once.
|
||||||
#expect(context.recordedNostrEventIDs.isEmpty)
|
await coordinator.inbound.processNostrMessage(giftWrap)
|
||||||
|
#expect(context.recordedNostrEventIDs == [giftWrap.id])
|
||||||
|
|
||||||
// ...so the genuine event with the same ID still processes and records.
|
|
||||||
await coordinator.inbound.processNostrMessage(giftWrap)
|
await coordinator.inbound.processNostrMessage(giftWrap)
|
||||||
#expect(context.recordedNostrEventIDs == [giftWrap.id])
|
#expect(context.recordedNostrEventIDs == [giftWrap.id])
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -352,31 +352,11 @@ struct ChatViewModelNostrExtensionTests {
|
|||||||
#expect(!viewModel.messages.contains { $0.content == "Blocked" })
|
#expect(!viewModel.messages.contains { $0.content == "Blocked" })
|
||||||
}
|
}
|
||||||
|
|
||||||
@Test @MainActor
|
// NOTE: Tampered-signature rejection is enforced once, off the main
|
||||||
func handleNostrEvent_rejectsInvalidSignature() async throws {
|
// actor, at the relay boundary (events only reach the inbound pipeline
|
||||||
let (viewModel, _) = makeTestableViewModel()
|
// after verification) — see NostrRelayManagerTests
|
||||||
let geohash = "u4pruydq"
|
// `test_receiveEvent_invalidSignatureDoesNotPoisonDuplicateCache` and
|
||||||
let identity = try NostrIdentity.generate()
|
// `test_receiveGiftWrap_tamperedSignatureIsDroppedAndDoesNotPoisonDedup`.
|
||||||
|
|
||||||
viewModel.switchLocationChannel(to: .location(GeohashChannel(level: .city, geohash: geohash)))
|
|
||||||
|
|
||||||
let event = NostrEvent(
|
|
||||||
pubkey: identity.publicKeyHex,
|
|
||||||
createdAt: Date(),
|
|
||||||
kind: .ephemeralEvent,
|
|
||||||
tags: [["g", geohash]],
|
|
||||||
content: "Valid"
|
|
||||||
)
|
|
||||||
var signed = try event.sign(with: identity.schnorrSigningKey())
|
|
||||||
signed.id = "deadbeef"
|
|
||||||
|
|
||||||
viewModel.handleNostrEvent(signed)
|
|
||||||
|
|
||||||
try? await Task.sleep(nanoseconds: 100_000_000)
|
|
||||||
viewModel.publicMessagePipeline.flushIfNeeded()
|
|
||||||
|
|
||||||
#expect(!viewModel.messages.contains { $0.content == "Tampered" })
|
|
||||||
}
|
|
||||||
|
|
||||||
@Test @MainActor
|
@Test @MainActor
|
||||||
func subscribeGiftWrap_rejectsOversizedEmbeddedPacket() async throws {
|
func subscribeGiftWrap_rejectsOversizedEmbeddedPacket() async throws {
|
||||||
@@ -565,9 +545,14 @@ struct ChatViewModelNostrExtensionTests {
|
|||||||
|
|
||||||
viewModel.handleGiftWrap(giftWrap, id: recipient)
|
viewModel.handleGiftWrap(giftWrap, id: recipient)
|
||||||
|
|
||||||
try? await Task.sleep(nanoseconds: 50_000_000)
|
// Gift-wrap decryption runs off the main actor; wait for the ack
|
||||||
|
// (sent even for blocked senders) to know processing finished.
|
||||||
|
let didAck = await TestHelpers.waitUntil(
|
||||||
|
{ viewModel.sentGeoDeliveryAcks.contains(messageID) },
|
||||||
|
timeout: 5.0
|
||||||
|
)
|
||||||
|
#expect(didAck)
|
||||||
#expect(viewModel.privateChats[convKey] == nil)
|
#expect(viewModel.privateChats[convKey] == nil)
|
||||||
#expect(viewModel.sentGeoDeliveryAcks.contains(messageID))
|
|
||||||
}
|
}
|
||||||
|
|
||||||
@Test @MainActor
|
@Test @MainActor
|
||||||
|
|||||||
@@ -326,26 +326,11 @@ struct ChatViewModelPresenceHandlingTests {
|
|||||||
#expect(viewModel.geohashParticipantCount(for: activeGeohash) >= 1)
|
#expect(viewModel.geohashParticipantCount(for: activeGeohash) >= 1)
|
||||||
}
|
}
|
||||||
|
|
||||||
@Test func subscribeNostrEvent_samplingInvalidSignatureDoesNotPoisonDedup() async throws {
|
// NOTE: Tampered-signature rejection (and the forged-copy dedup-poisoning
|
||||||
let (viewModel, _) = makeTestableViewModel()
|
// invariant) is enforced once, off the main actor, at the relay boundary —
|
||||||
let sampleGeohash = "u4pru"
|
// the sampling path only ever sees verified events. See
|
||||||
let identity = try NostrIdentity.generate()
|
// NostrRelayManagerTests
|
||||||
let event = NostrEvent(
|
// `test_receiveEvent_invalidSignatureDoesNotPoisonDuplicateCache`.
|
||||||
pubkey: identity.publicKeyHex,
|
|
||||||
createdAt: Date(),
|
|
||||||
kind: .geohashPresence,
|
|
||||||
tags: [["g", sampleGeohash]],
|
|
||||||
content: ""
|
|
||||||
)
|
|
||||||
let signed = try event.sign(with: identity.schnorrSigningKey())
|
|
||||||
var invalid = signed
|
|
||||||
invalid.sig = String(repeating: "0", count: 128)
|
|
||||||
|
|
||||||
viewModel.subscribeNostrEvent(invalid, gh: sampleGeohash)
|
|
||||||
viewModel.subscribeNostrEvent(signed, gh: sampleGeohash)
|
|
||||||
|
|
||||||
#expect(viewModel.geohashParticipantCount(for: sampleGeohash) == 1)
|
|
||||||
}
|
|
||||||
|
|
||||||
// MARK: - Test Helper
|
// MARK: - Test Helper
|
||||||
|
|
||||||
|
|||||||
@@ -77,8 +77,10 @@ final class PerformanceBaselineTests: XCTestCase {
|
|||||||
// MARK: - 1a. Nostr inbound event handling (fresh events)
|
// MARK: - 1a. Nostr inbound event handling (fresh events)
|
||||||
|
|
||||||
/// `NostrInboundPipeline.handleNostrEvent` for never-seen geo events
|
/// `NostrInboundPipeline.handleNostrEvent` for never-seen geo events
|
||||||
/// (kind 20000): signature verification, dedup record, presence/nickname
|
/// (kind 20000): dedup record, presence/nickname bookkeeping, and
|
||||||
/// bookkeeping, and public-message ingest scheduling.
|
/// public-message ingest scheduling. Schnorr signature verification is
|
||||||
|
/// NOT part of this path anymore — it runs exactly once, off the main
|
||||||
|
/// actor, in `NostrRelayManager` before delivery.
|
||||||
func testNostrInboundEventHandling_freshEvents() throws {
|
func testNostrInboundEventHandling_freshEvents() throws {
|
||||||
let events = try Self.makeSignedGeohashEvents(count: 500)
|
let events = try Self.makeSignedGeohashEvents(count: 500)
|
||||||
// A fresh context per measure pass so every event takes the
|
// A fresh context per measure pass so every event takes the
|
||||||
@@ -106,8 +108,9 @@ final class PerformanceBaselineTests: XCTestCase {
|
|||||||
|
|
||||||
/// The dedup-hit path: identical events replayed. Duplicates dominate
|
/// The dedup-hit path: identical events replayed. Duplicates dominate
|
||||||
/// real relay traffic (the same event arrives from several relays), so
|
/// real relay traffic (the same event arrives from several relays), so
|
||||||
/// this path runs hundreds of times a minute in busy geohashes. Note it
|
/// this path runs hundreds of times a minute in busy geohashes. It is a
|
||||||
/// still pays full Schnorr signature verification before the dedup check.
|
/// pure dedup lookup: no crypto (verification happens upstream in
|
||||||
|
/// `NostrRelayManager`, and only for the first-seen copy).
|
||||||
func testNostrInboundEventHandling_duplicateEvents() throws {
|
func testNostrInboundEventHandling_duplicateEvents() throws {
|
||||||
let events = try Self.makeSignedGeohashEvents(count: 500)
|
let events = try Self.makeSignedGeohashEvents(count: 500)
|
||||||
let context = PerfNostrContext()
|
let context = PerfNostrContext()
|
||||||
|
|||||||
@@ -724,6 +724,77 @@ final class NostrRelayManagerTests: XCTestCase {
|
|||||||
XCTAssertEqual(context.manager.debugDuplicateInboundEventDropCount(forSubscriptionID: "events"), 0)
|
XCTAssertEqual(context.manager.debugDuplicateInboundEventDropCount(forSubscriptionID: "events"), 0)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// The relay boundary is the single signature-verification point for the
|
||||||
|
/// whole inbound path (downstream pipelines no longer re-verify), so a
|
||||||
|
/// tampered gift wrap (kind 1059, the DM/mailbox path) must be dropped
|
||||||
|
/// here — and must not poison the dedup cache against the genuine copy.
|
||||||
|
func test_receiveGiftWrap_tamperedSignatureIsDroppedAndDoesNotPoisonDedup() async throws {
|
||||||
|
let firstRelayURL = "wss://giftwrap-one.example"
|
||||||
|
let secondRelayURL = "wss://giftwrap-two.example"
|
||||||
|
let context = makeContext(permission: .denied)
|
||||||
|
let sender = try NostrIdentity.generate()
|
||||||
|
let recipient = try NostrIdentity.generate()
|
||||||
|
let giftWrap = try NostrProtocol.createPrivateMessage(
|
||||||
|
content: "psst",
|
||||||
|
recipientPubkey: recipient.publicKeyHex,
|
||||||
|
senderIdentity: sender
|
||||||
|
)
|
||||||
|
let tampered = invalidSignatureCopy(of: giftWrap)
|
||||||
|
var receivedIDs: [String] = []
|
||||||
|
|
||||||
|
context.manager.subscribe(
|
||||||
|
filter: makeFilter(),
|
||||||
|
id: "gift-wraps",
|
||||||
|
relayUrls: [firstRelayURL, secondRelayURL]
|
||||||
|
) { event in
|
||||||
|
receivedIDs.append(event.id)
|
||||||
|
}
|
||||||
|
let subscriptionsSent = await waitUntil {
|
||||||
|
context.sessionFactory.latestConnection(for: firstRelayURL)?.sentStrings.count == 1 &&
|
||||||
|
context.sessionFactory.latestConnection(for: secondRelayURL)?.sentStrings.count == 1
|
||||||
|
}
|
||||||
|
XCTAssertTrue(subscriptionsSent)
|
||||||
|
|
||||||
|
try context.sessionFactory.latestConnection(for: firstRelayURL)?.emitEventMessage(subscriptionID: "gift-wraps", event: tampered)
|
||||||
|
try context.sessionFactory.latestConnection(for: secondRelayURL)?.emitEventMessage(subscriptionID: "gift-wraps", event: giftWrap)
|
||||||
|
|
||||||
|
let countedOnBothRelays = await waitUntil {
|
||||||
|
context.manager.relays.first(where: { $0.url == firstRelayURL })?.messagesReceived == 1 &&
|
||||||
|
context.manager.relays.first(where: { $0.url == secondRelayURL })?.messagesReceived == 1
|
||||||
|
}
|
||||||
|
XCTAssertTrue(countedOnBothRelays)
|
||||||
|
XCTAssertEqual(receivedIDs, [giftWrap.id])
|
||||||
|
XCTAssertEqual(context.manager.debugDuplicateInboundEventDropCount, 0)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Signature verification moved off-main into a single serial consumer;
|
||||||
|
/// several frames buffered on one socket must still be delivered to the
|
||||||
|
/// handler in arrival order.
|
||||||
|
func test_receiveEvent_deliversBackToBackEventsInArrivalOrder() async throws {
|
||||||
|
let relayURL = "wss://ordered.example"
|
||||||
|
let context = makeContext(permission: .denied)
|
||||||
|
let events = try (0..<12).map { try makeSignedEvent(content: "ordered-\($0)") }
|
||||||
|
var receivedIDs: [String] = []
|
||||||
|
|
||||||
|
context.manager.subscribe(filter: makeFilter(), id: "ordered", relayUrls: [relayURL]) { event in
|
||||||
|
receivedIDs.append(event.id)
|
||||||
|
}
|
||||||
|
let subscriptionSent = await waitUntil {
|
||||||
|
context.sessionFactory.latestConnection(for: relayURL)?.sentStrings.count == 1
|
||||||
|
}
|
||||||
|
XCTAssertTrue(subscriptionSent)
|
||||||
|
|
||||||
|
for event in events {
|
||||||
|
try context.sessionFactory.latestConnection(for: relayURL)?.emitEventMessage(subscriptionID: "ordered", event: event)
|
||||||
|
}
|
||||||
|
|
||||||
|
let allDelivered = await waitUntil(timeout: 5.0) {
|
||||||
|
receivedIDs.count == events.count
|
||||||
|
}
|
||||||
|
XCTAssertTrue(allDelivered)
|
||||||
|
XCTAssertEqual(receivedIDs, events.map(\.id))
|
||||||
|
}
|
||||||
|
|
||||||
func test_receiveEvent_withoutHandlerStillTracksReceivedCount() async throws {
|
func test_receiveEvent_withoutHandlerStillTracksReceivedCount() async throws {
|
||||||
let relayURL = "wss://missing-handler.example"
|
let relayURL = "wss://missing-handler.example"
|
||||||
let context = makeContext(permission: .denied)
|
let context = makeContext(permission: .denied)
|
||||||
@@ -1695,7 +1766,11 @@ private final class MockRelayConnection: NostrRelayConnectionProtocol {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func receive(completionHandler: @escaping (Result<URLSessionWebSocketTask.Message, Error>) -> Void) {
|
func receive(completionHandler: @escaping (Result<URLSessionWebSocketTask.Message, Error>) -> Void) {
|
||||||
receiveHandler = completionHandler
|
if !pendingResults.isEmpty {
|
||||||
|
completionHandler(pendingResults.removeFirst())
|
||||||
|
} else {
|
||||||
|
receiveHandler = completionHandler
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func sendPing(pongReceiveHandler: @escaping (Error?) -> Void) {
|
func sendPing(pongReceiveHandler: @escaping (Error?) -> Void) {
|
||||||
@@ -1728,15 +1803,24 @@ private final class MockRelayConnection: NostrRelayConnectionProtocol {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func emitRawString(_ string: String) throws {
|
func emitRawString(_ string: String) throws {
|
||||||
let handler = receiveHandler
|
deliver(.success(.string(string)))
|
||||||
receiveHandler = nil
|
|
||||||
handler?(.success(.string(string)))
|
|
||||||
}
|
}
|
||||||
|
|
||||||
private func emit(jsonObject: Any) throws {
|
private func emit(jsonObject: Any) throws {
|
||||||
let data = try JSONSerialization.data(withJSONObject: jsonObject)
|
let data = try JSONSerialization.data(withJSONObject: jsonObject)
|
||||||
let handler = receiveHandler
|
deliver(.success(.data(data)))
|
||||||
receiveHandler = nil
|
}
|
||||||
handler?(.success(.data(data)))
|
|
||||||
|
// Frames emitted before the manager re-arms `receive` are queued so
|
||||||
|
// back-to-back emissions model a socket with several buffered frames.
|
||||||
|
private var pendingResults: [Result<URLSessionWebSocketTask.Message, Error>] = []
|
||||||
|
|
||||||
|
private func deliver(_ result: Result<URLSessionWebSocketTask.Message, Error>) {
|
||||||
|
if let handler = receiveHandler {
|
||||||
|
receiveHandler = nil
|
||||||
|
handler(result)
|
||||||
|
} else {
|
||||||
|
pendingResults.append(result)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user