mirror of
https://github.com/permissionlesstech/bitchat.git
synced 2026-07-27 01:05:20 +00:00
perf: verify Nostr inbound signatures once, off the main actor (#1352)
Moves inbound Nostr Schnorr verification and NIP-17 gift-wrap decryption off the main actor into per-relay bounded AsyncStream pipelines (parallel across relays, ordered per relay), dedups before crypto, and removes the redundant second re-verify in NostrInboundPipeline/GeoPresenceTracker. Adds a panic-wipe generation counter so in-flight decrypts are discarded after a wipe. Rebased onto main; #1451's 256 KiB pre-parse byte cap and tag limits are preserved and still enforced before any crypto/allocation.
This commit is contained in:
@@ -48,7 +48,14 @@ private struct URLSessionAdapter: NostrRelaySessionProtocol {
|
||||
let base: URLSession
|
||||
|
||||
func webSocketTask(with url: URL) -> NostrRelayConnectionProtocol {
|
||||
URLSessionWebSocketTaskAdapter(base: base.webSocketTask(with: url))
|
||||
let task = base.webSocketTask(with: url)
|
||||
// Byte bound per inbound frame; without it the per-relay buffer cap
|
||||
// (nostrInboundPerRelayBufferCap) bounds FRAMES but not BYTES, and a
|
||||
// hostile relay could pile up cap × 1 MiB (URLSession default) per
|
||||
// connection. See TransportConfig.nostrInboundMaxFrameBytes for the
|
||||
// sizing rationale. Oversized frames fail the receive with an error.
|
||||
task.maximumMessageSize = TransportConfig.nostrInboundMaxFrameBytes
|
||||
return URLSessionWebSocketTaskAdapter(base: task)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -106,7 +113,10 @@ private extension NostrRelayManagerDependencies {
|
||||
@MainActor
|
||||
final class NostrRelayManager: ObservableObject {
|
||||
static let shared = NostrRelayManager()
|
||||
// Track gift-wraps (kind 1059) we initiated so we can log OK acks at info
|
||||
// Track gift-wraps (kind 1059) we initiated so we can log OK acks at info.
|
||||
// Entries are removed only on OK acks (or panic wipe); relays that never
|
||||
// ack leave entries behind for the process lifetime. Observability-only
|
||||
// state, bounded in practice by outbound DM volume.
|
||||
private(set) static var pendingGiftWrapIDs = Set<String>()
|
||||
static func registerPendingGiftWrap(id: String) {
|
||||
pendingGiftWrapIDs.insert(id)
|
||||
@@ -239,7 +249,33 @@ final class NostrRelayManager: ObservableObject {
|
||||
|
||||
// Bump generation to invalidate scheduled reconnects when we reset/disconnect
|
||||
private var connectionGeneration: Int = 0
|
||||
|
||||
|
||||
// Per-relay 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.
|
||||
//
|
||||
// Each relay connection owns its OWN AsyncStream + consumer task, so N
|
||||
// relays verify in parallel while every relay's frames stay in arrival
|
||||
// order (a single subscription's events for a relay all arrive on that
|
||||
// relay's socket, so per-relay ordering preserves per-subscription
|
||||
// ordering). A burst of EVENT frames from one busy/malicious relay only
|
||||
// blocks that relay's own verification backlog — DMs, OKs, EOSEs, and
|
||||
// events from every other relay keep flowing on their own pipelines.
|
||||
//
|
||||
// Each stream is bounded (`.bufferingNewest`) so a relay flooding faster
|
||||
// than its verification drains sheds its own oldest frames instead of
|
||||
// growing memory without bound; it can never starve other relays.
|
||||
//
|
||||
// Continuations live in a lock-guarded, `Sendable` router (see
|
||||
// `InboundFrameRouter` at file scope) so the raw socket receive callback
|
||||
// (which is NOT main-actor isolated) can route a frame to the right relay
|
||||
// stream without a per-frame main hop, while the main actor owns pipeline
|
||||
// creation/teardown. The expensive work (Schnorr verify) is what runs
|
||||
// off-main; the yield stays cheap.
|
||||
private let inboundRouter = InboundFrameRouter()
|
||||
|
||||
init() {
|
||||
self.dependencies = .live()
|
||||
hasMutualFavorites = dependencies.hasMutualFavorites()
|
||||
@@ -293,7 +329,70 @@ final class NostrRelayManager: ObservableObject {
|
||||
}
|
||||
.store(in: &cancellables)
|
||||
}
|
||||
|
||||
|
||||
deinit {
|
||||
inboundRouter.finishAll()
|
||||
}
|
||||
|
||||
/// Ensure a serial off-main consumer pipeline exists for a relay. Called on
|
||||
/// the main actor when a socket is (re)armed for receiving. Idempotent.
|
||||
///
|
||||
/// Ordering within the relay 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 ensureRelayInboundPipeline(for relayUrl: String) {
|
||||
let started = inboundRouter.startPipeline(for: relayUrl) { [weak self] stream in
|
||||
Task.detached(priority: .userInitiated) {
|
||||
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: relayUrl
|
||||
) else {
|
||||
continue
|
||||
}
|
||||
guard event.isValidSignature() else {
|
||||
SecureLogger.warning(
|
||||
"⚠️ Dropped invalid Nostr event id=\(event.id.prefix(16))… sub=\(subId) relay=\(relayUrl)",
|
||||
category: .session
|
||||
)
|
||||
continue
|
||||
}
|
||||
await self.deliverVerifiedInboundEvent(subscriptionID: subId, event: event, from: relayUrl)
|
||||
case .eose, .ok, .notice:
|
||||
await self.handleParsedMessage(parsed, from: relayUrl)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if started {
|
||||
SecureLogger.debug("🧵 Started inbound verify pipeline for \(relayUrl)", category: .session)
|
||||
}
|
||||
}
|
||||
|
||||
/// Tear down a relay's inbound pipeline (socket gone or state wiped). The
|
||||
/// consumer drains any already-buffered frames before finishing, so
|
||||
/// in-flight verified events are still delivered.
|
||||
private func teardownRelayInboundPipeline(for relayUrl: String) {
|
||||
inboundRouter.finishPipeline(for: relayUrl)
|
||||
}
|
||||
|
||||
private func teardownAllRelayInboundPipelines() {
|
||||
inboundRouter.finishAll()
|
||||
}
|
||||
|
||||
/// Connect to all configured relays
|
||||
func connect() {
|
||||
// Global network policy gate
|
||||
@@ -308,6 +407,8 @@ final class NostrRelayManager: ObservableObject {
|
||||
task.cancel(with: .goingAway, reason: nil)
|
||||
}
|
||||
connections.removeAll()
|
||||
// Sockets are gone; drop every relay's inbound verify pipeline.
|
||||
teardownAllRelayInboundPipelines()
|
||||
markRelaySocketsClosed(resetState: false)
|
||||
// Sockets are gone, so per-relay subscription state is cleared — but
|
||||
// durable intent (subscriptionRequestState, messageHandlers, parked
|
||||
@@ -340,6 +441,7 @@ final class NostrRelayManager: ObservableObject {
|
||||
task.cancel(with: .goingAway, reason: nil)
|
||||
}
|
||||
connections.removeAll()
|
||||
teardownAllRelayInboundPipelines()
|
||||
markRelaySocketsClosed(resetState: true)
|
||||
subscriptions.removeAll()
|
||||
pendingSubscriptions.removeAll()
|
||||
@@ -681,6 +783,7 @@ final class NostrRelayManager: ObservableObject {
|
||||
connection.cancel(with: .goingAway, reason: nil)
|
||||
}
|
||||
connections.removeValue(forKey: url)
|
||||
teardownRelayInboundPipeline(for: url)
|
||||
subscriptions.removeValue(forKey: url)
|
||||
pendingSubscriptions.removeValue(forKey: url)
|
||||
}
|
||||
@@ -1020,7 +1123,11 @@ final class NostrRelayManager: ObservableObject {
|
||||
|
||||
connections[urlString] = task
|
||||
task.resume()
|
||||
|
||||
|
||||
// Bring up this relay's own serial verify pipeline before arming the
|
||||
// socket, so inbound frames have somewhere to land.
|
||||
ensureRelayInboundPipeline(for: urlString)
|
||||
|
||||
// Start receiving messages
|
||||
receiveMessage(from: task, relayUrl: urlString)
|
||||
|
||||
@@ -1095,15 +1202,14 @@ final class NostrRelayManager: ObservableObject {
|
||||
|
||||
switch result {
|
||||
case .success(let message):
|
||||
// Parse off-main to reduce UI jank, then hop back for state updates
|
||||
Task.detached(priority: .utility) {
|
||||
guard let parsed = ParsedInbound(message) else { return }
|
||||
await MainActor.run {
|
||||
guard self.connections[relayUrl] === task else { return }
|
||||
self.handleParsedMessage(parsed, from: relayUrl)
|
||||
}
|
||||
}
|
||||
|
||||
// Hand the raw frame to this relay's serial inbound pipeline:
|
||||
// parsing and signature verification run off-main, in arrival
|
||||
// order, independently of every other relay's pipeline. Routing
|
||||
// through the lock-guarded router keeps this off the main actor
|
||||
// (no per-frame main hop).
|
||||
self.inboundRouter.yield(InboundFrame(message: message), to: relayUrl)
|
||||
|
||||
|
||||
// Continue receiving
|
||||
Task { @MainActor in
|
||||
guard self.connections[relayUrl] === task else { return }
|
||||
@@ -1122,35 +1228,55 @@ final class NostrRelayManager: ObservableObject {
|
||||
// Note: declared at file scope below to avoid MainActor isolation inside this class
|
||||
// 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) {
|
||||
switch parsed {
|
||||
case .event(let subId, let event):
|
||||
if let index = self.relays.firstIndex(where: { $0.url == relayUrl }) {
|
||||
self.relays[index].messagesReceived += 1
|
||||
}
|
||||
guard event.isValidSignature() else {
|
||||
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 .event:
|
||||
// Events flow through the serial inbound pipeline (precheck →
|
||||
// off-main signature verification → deliverVerifiedInboundEvent)
|
||||
// and never reach this fallback.
|
||||
assertionFailure("inbound EVENT bypassed the verified pipeline")
|
||||
case .eose(let subId):
|
||||
if var tracker = eoseTrackers[subId] {
|
||||
// An EOSE proves the relay received the REQ even if the local
|
||||
@@ -1285,6 +1411,7 @@ final class NostrRelayManager: ObservableObject {
|
||||
) {
|
||||
if let connection, connections[relayUrl] !== connection { return }
|
||||
connections.removeValue(forKey: relayUrl)
|
||||
teardownRelayInboundPipeline(for: relayUrl)
|
||||
subscriptions.removeValue(forKey: relayUrl)
|
||||
let awaitingConfirmation = confirmedSends.compactMap { eventID, state in
|
||||
state.awaitingRelays.contains(relayUrl) ? eventID : nil
|
||||
@@ -1379,8 +1506,9 @@ final class NostrRelayManager: ObservableObject {
|
||||
if let connection = connections[normalizedRelayUrl] {
|
||||
connection.cancel(with: .goingAway, reason: nil)
|
||||
connections.removeValue(forKey: normalizedRelayUrl)
|
||||
teardownRelayInboundPipeline(for: normalizedRelayUrl)
|
||||
}
|
||||
|
||||
|
||||
// Attempt immediate reconnection
|
||||
connectToRelay(normalizedRelayUrl)
|
||||
}
|
||||
@@ -1473,6 +1601,77 @@ final class NostrRelayManager: ObservableObject {
|
||||
|
||||
// MARK: - Off-main inbound parsing helpers (file scope, non-isolated)
|
||||
|
||||
/// A single raw socket frame awaiting off-main parse + Schnorr verification.
|
||||
private struct InboundFrame: Sendable {
|
||||
let message: URLSessionWebSocketTask.Message
|
||||
}
|
||||
|
||||
/// Lock-guarded registry of per-relay inbound streams.
|
||||
///
|
||||
/// The raw WebSocket receive callback is not main-actor isolated, so it needs a
|
||||
/// `Sendable` path to route a frame to the correct relay's stream without a
|
||||
/// per-frame hop onto the main actor. Pipeline lifecycle (start/finish) is
|
||||
/// driven from the main actor; frame delivery (`yield`) can come from any
|
||||
/// thread. All access is serialized by a single lock — contention is negligible
|
||||
/// because the guarded critical section is only a dictionary lookup + yield.
|
||||
private final class InboundFrameRouter: @unchecked Sendable {
|
||||
private let lock = NSLock()
|
||||
private var continuations: [String: AsyncStream<InboundFrame>.Continuation] = [:]
|
||||
private var tasks: [String: Task<Void, Never>] = [:]
|
||||
|
||||
/// Start a relay's stream + consumer if one does not already exist.
|
||||
/// Returns true when a new pipeline was created. The bounded
|
||||
/// `.bufferingNewest` policy makes a single relay shed its OWN oldest
|
||||
/// frames under a flood, never other relays' frames. Buffered memory per
|
||||
/// relay is bounded (not eliminated) at the frame cap times the per-frame
|
||||
/// byte cap (`maximumMessageSize`) — see TransportConfig.
|
||||
func startPipeline(
|
||||
for relayUrl: String,
|
||||
makeConsumer: (AsyncStream<InboundFrame>) -> Task<Void, Never>
|
||||
) -> Bool {
|
||||
lock.lock()
|
||||
defer { lock.unlock() }
|
||||
if continuations[relayUrl] != nil { return false }
|
||||
let (stream, continuation) = AsyncStream<InboundFrame>.makeStream(
|
||||
bufferingPolicy: .bufferingNewest(TransportConfig.nostrInboundPerRelayBufferCap)
|
||||
)
|
||||
continuations[relayUrl] = continuation
|
||||
tasks[relayUrl] = makeConsumer(stream)
|
||||
return true
|
||||
}
|
||||
|
||||
/// Route a frame to a relay's stream. No-op if the relay has no live
|
||||
/// pipeline (socket already torn down) — the frame is simply dropped, which
|
||||
/// is safe for best-effort Nostr inbound.
|
||||
func yield(_ frame: InboundFrame, to relayUrl: String) {
|
||||
lock.lock()
|
||||
let continuation = continuations[relayUrl]
|
||||
lock.unlock()
|
||||
continuation?.yield(frame)
|
||||
}
|
||||
|
||||
/// Finish a relay's stream. The consumer drains any already-buffered frames
|
||||
/// before exiting, so in-flight verified events are still delivered.
|
||||
func finishPipeline(for relayUrl: String) {
|
||||
lock.lock()
|
||||
let continuation = continuations.removeValue(forKey: relayUrl)
|
||||
tasks.removeValue(forKey: relayUrl)
|
||||
lock.unlock()
|
||||
continuation?.finish()
|
||||
}
|
||||
|
||||
func finishAll() {
|
||||
lock.lock()
|
||||
let allContinuations = continuations
|
||||
continuations.removeAll()
|
||||
tasks.removeAll()
|
||||
lock.unlock()
|
||||
for continuation in allContinuations.values {
|
||||
continuation.finish()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private enum ParsedInbound {
|
||||
case event(subId: String, event: NostrEvent)
|
||||
case ok(eventId: String, success: Bool, reason: String)
|
||||
|
||||
@@ -87,6 +87,26 @@ enum TransportConfig {
|
||||
static let nostrMaxEventTags: Int = 64
|
||||
static let nostrMaxEventTagValues: Int = 16
|
||||
static let nostrMaxEventTagValueBytes: Int = 1024
|
||||
// Bounded per-relay inbound frame buffer. Each relay connection owns its
|
||||
// own serial verify pipeline; if a relay floods faster than its Schnorr
|
||||
// verification drains, the oldest buffered frames for THAT relay are
|
||||
// dropped (bufferingNewest) so one relay cannot stall other relays.
|
||||
// Nostr inbound is already best-effort (relays are redundant and events
|
||||
// replay), so dropping a flooding relay's backlog is safe. Together with
|
||||
// nostrInboundMaxFrameBytes this caps buffered inbound bytes at
|
||||
// cap × maxFrameBytes (128 MiB) per hostile relay — bounded, not zero.
|
||||
static let nostrInboundPerRelayBufferCap: Int = 256
|
||||
// Hard per-frame byte bound, applied as URLSessionWebSocketTask
|
||||
// .maximumMessageSize (oversized frames fail the receive instead of
|
||||
// buffering). BitChat's legitimate Nostr traffic is small: geohash chat /
|
||||
// presence events (kind 20000/20001), kind-1 notes, and NIP-17
|
||||
// gift-wrapped DMs carrying text payloads or receipts are all a few KiB,
|
||||
// and most public relays reject events beyond ~64–256 KiB anyway. 512 KiB
|
||||
// leaves an order-of-magnitude margin over anything we produce or expect
|
||||
// while halving the URLSession default (1 MiB), so a hostile relay's
|
||||
// worst-case buffered pile-up per connection is
|
||||
// nostrInboundPerRelayBufferCap × 512 KiB = 128 MiB instead of 256 MiB.
|
||||
static let nostrInboundMaxFrameBytes: Int = 512 * 1024
|
||||
|
||||
// Conversation store diagnostics (field observability)
|
||||
// Sample interval for the periodic store-audit "OK" heartbeat line
|
||||
|
||||
@@ -1361,6 +1361,11 @@ final class ChatViewModel: ObservableObject, BitchatDelegate, TransportEventDele
|
||||
// Geohash DM handlers can capture pre-wipe Nostr identities, so a plain
|
||||
// disconnect is not enough here.
|
||||
NostrRelayManager.shared.resetForPanicWipe()
|
||||
// Clearing relay handlers stops NEW events, but a detached gift-wrap
|
||||
// decrypt spawned just before the wipe still holds a pre-wipe key and
|
||||
// ciphertext; bump the pipeline's wipe generation so its result is
|
||||
// dropped at the main-actor delivery hop instead of landing here.
|
||||
nostrCoordinator.inbound.invalidateInFlightDecrypts()
|
||||
nostrRelayManager = nil
|
||||
|
||||
// Clear Nostr identity associations
|
||||
|
||||
@@ -103,7 +103,8 @@ final class GeoPresenceTracker {
|
||||
else {
|
||||
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 }
|
||||
|
||||
let existingCount = context.geoParticipantCount(for: gh)
|
||||
|
||||
@@ -78,20 +78,38 @@ extension ChatViewModel: NostrInboundPipelineContext {
|
||||
}
|
||||
}
|
||||
|
||||
/// The inbound Nostr hot path: raw relay events in, chat messages / Noise
|
||||
/// payloads out. Pure transformation plus dedup — no relay lifecycle.
|
||||
/// The inbound Nostr hot path: verified relay events in, chat messages /
|
||||
/// Noise payloads out. Pure transformation plus dedup — no relay lifecycle.
|
||||
///
|
||||
/// Ordering is deliberate and performance-critical: cheap rejects (kind,
|
||||
/// dedup lookup) run BEFORE Schnorr signature verification because duplicates
|
||||
/// dominate real relay traffic; events are recorded only AFTER verification so
|
||||
/// a forged-signature copy can never poison the dedup set; gift-wrap
|
||||
/// verification for the account mailbox runs off-main with an atomic
|
||||
/// main-actor check-and-record.
|
||||
/// Every event arriving here already had its Schnorr signature verified
|
||||
/// exactly once, off the main actor, by `NostrRelayManager`'s serial inbound
|
||||
/// pipeline (which records events into its own dedup cache only AFTER
|
||||
/// verification, so forged copies can't suppress genuine events). This
|
||||
/// pipeline therefore never re-verifies; it keeps its own event-ID dedup
|
||||
/// (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 {
|
||||
private weak var context: (any NostrInboundPipelineContext)?
|
||||
private let presence: GeoPresenceTracker
|
||||
private var geoEventLogCount = 0
|
||||
|
||||
/// Monotonic panic-wipe generation for this pipeline. A panic wipe clears
|
||||
/// relay handlers so no NEW events flow, but a detached decrypt task
|
||||
/// spawned just BEFORE the wipe — which strongly captures a pre-wipe Nostr
|
||||
/// private key and ciphertext — survives it. Spawn sites capture this
|
||||
/// value; the task compares it at its main-actor hops and drops its result
|
||||
/// (no delivery; the captured identity and plaintext die with the task)
|
||||
/// if `invalidateInFlightDecrypts()` bumped it in between.
|
||||
@MainActor private(set) var wipeGeneration: UInt64 = 0
|
||||
|
||||
/// Called from `ChatViewModel.panicClearAllData()` so plaintext decrypted
|
||||
/// with pre-wipe keys can never land in post-wipe state.
|
||||
@MainActor
|
||||
func invalidateInFlightDecrypts() {
|
||||
wipeGeneration &+= 1
|
||||
}
|
||||
|
||||
init(context: any NostrInboundPipelineContext, presence: GeoPresenceTracker) {
|
||||
self.context = context
|
||||
self.presence = presence
|
||||
@@ -100,17 +118,15 @@ final class NostrInboundPipeline {
|
||||
@MainActor
|
||||
func subscribeNostrEvent(_ event: NostrEvent) {
|
||||
guard let context else { return }
|
||||
// Cheap rejects (kind, dedup lookup) before Schnorr verification —
|
||||
// duplicates dominate real traffic and must not pay for crypto.
|
||||
// Only verified events are recorded, so a forged-signature copy can
|
||||
// never poison the dedup set and suppress the genuine event.
|
||||
// Cheap rejects (kind, dedup lookup) — duplicates dominate real
|
||||
// traffic. The signature was already verified (exactly once, off the
|
||||
// main actor) by NostrRelayManager before delivery.
|
||||
guard (event.kind == NostrProtocol.EventKind.ephemeralEvent.rawValue
|
||||
|| event.kind == NostrProtocol.EventKind.geohashPresence.rawValue),
|
||||
!context.hasProcessedNostrEvent(event.id)
|
||||
else {
|
||||
return
|
||||
}
|
||||
guard event.isValidSignature() else { return }
|
||||
|
||||
context.recordProcessedNostrEvent(event.id)
|
||||
|
||||
@@ -180,15 +196,14 @@ final class NostrInboundPipeline {
|
||||
@MainActor
|
||||
func handleNostrEvent(_ event: NostrEvent) {
|
||||
guard let context else { return }
|
||||
// Cheap rejects (kind, dedup lookup) before Schnorr verification —
|
||||
// duplicates dominate real traffic and must not pay for crypto.
|
||||
// Cheap rejects (kind, dedup lookup) — the signature was already
|
||||
// verified (exactly once, off the main actor) by NostrRelayManager.
|
||||
guard (event.kind == NostrProtocol.EventKind.ephemeralEvent.rawValue
|
||||
|| event.kind == NostrProtocol.EventKind.geohashPresence.rawValue)
|
||||
else {
|
||||
return
|
||||
}
|
||||
if context.hasProcessedNostrEvent(event.id) { return }
|
||||
guard event.isValidSignature() else { return }
|
||||
context.recordProcessedNostrEvent(event.id)
|
||||
|
||||
let powBits = NostrPoW.validatedDifficulty(idHex: event.id, tags: event.tags)
|
||||
@@ -273,112 +288,130 @@ final class NostrInboundPipeline {
|
||||
@MainActor
|
||||
func subscribeGiftWrap(_ giftWrap: NostrEvent, id: NostrIdentity) {
|
||||
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 giftWrap.isValidSignature() else { return }
|
||||
context.recordProcessedNostrEvent(giftWrap.id)
|
||||
|
||||
guard let (content, senderPubkey, rumorTs) = try? NostrProtocol.decryptPrivateMessage(
|
||||
giftWrap: giftWrap,
|
||||
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)
|
||||
// Group state travels only over mesh Noise sessions in v1; anything
|
||||
// claiming to be group traffic over Nostr is ignored.
|
||||
// Live voice is mesh-only: latency and relay cost make it
|
||||
// meaningless over Nostr.
|
||||
case .verifyChallenge, .verifyResponse, .groupInvite, .groupKeyUpdate, .vouch, .voiceFrame:
|
||||
break
|
||||
// Capture the wipe generation at spawn, alongside the per-geohash
|
||||
// identity (private key) the detached task strongly captures. A panic
|
||||
// wipe between spawn and delivery bumps the generation, and the task
|
||||
// drops its result instead of delivering plaintext post-wipe.
|
||||
let wipeGeneration = self.wipeGeneration
|
||||
Task.detached(priority: .userInitiated) { [weak self] in
|
||||
await self?.processGeohashGiftWrap(giftWrap, id: id, verbose: false, wipeGeneration: wipeGeneration)
|
||||
}
|
||||
}
|
||||
|
||||
@MainActor
|
||||
func handleGiftWrap(_ giftWrap: NostrEvent, id: NostrIdentity) {
|
||||
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) {
|
||||
return
|
||||
}
|
||||
guard giftWrap.isValidSignature() else { return }
|
||||
context.recordProcessedNostrEvent(giftWrap.id)
|
||||
|
||||
// Spawn-time wipe-generation capture; see subscribeGiftWrap.
|
||||
let wipeGeneration = self.wipeGeneration
|
||||
Task.detached(priority: .userInitiated) { [weak self] in
|
||||
await self?.processGeohashGiftWrap(giftWrap, id: id, verbose: true, wipeGeneration: wipeGeneration)
|
||||
}
|
||||
}
|
||||
|
||||
/// 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.
|
||||
///
|
||||
/// `wipeGeneration` is this pipeline's generation captured at spawn (the
|
||||
/// moment the pre-wipe `id` was captured); a mismatch at either main-actor
|
||||
/// hop means a panic wipe happened in between, so the task bails without
|
||||
/// decrypting (first hop) or without delivering the plaintext (second
|
||||
/// hop) — the captured identity and any decrypted material are simply
|
||||
/// dropped with the task.
|
||||
private func processGeohashGiftWrap(
|
||||
_ giftWrap: NostrEvent,
|
||||
id: NostrIdentity,
|
||||
verbose: Bool,
|
||||
wipeGeneration: UInt64
|
||||
) 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 {
|
||||
guard self.wipeGeneration == wipeGeneration else { return true }
|
||||
if context.hasProcessedNostrEvent(giftWrap.id) { return true }
|
||||
context.recordProcessedNostrEvent(giftWrap.id)
|
||||
return false
|
||||
}
|
||||
if alreadyProcessed { return }
|
||||
|
||||
guard let (content, senderPubkey, rumorTs) = try? NostrProtocol.decryptPrivateMessage(
|
||||
giftWrap: giftWrap,
|
||||
recipientIdentity: id
|
||||
) 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
|
||||
}
|
||||
|
||||
SecureLogger.debug(
|
||||
"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
|
||||
if verbose {
|
||||
SecureLogger.debug(
|
||||
"GeoDM: decrypted gift-wrap id=\(giftWrap.id.prefix(16))... from=\(senderPubkey.prefix(8))...",
|
||||
category: .session
|
||||
)
|
||||
case .delivered:
|
||||
context.handleDelivered(payload, senderPubkey: senderPubkey, convKey: convKey)
|
||||
case .readReceipt:
|
||||
context.handleReadReceipt(payload, senderPubkey: senderPubkey, convKey: convKey)
|
||||
// Group state travels only over mesh Noise sessions in v1; anything
|
||||
// claiming to be group traffic over Nostr is ignored.
|
||||
// Live voice is mesh-only: latency and relay cost make it
|
||||
// meaningless over Nostr.
|
||||
case .verifyChallenge, .verifyResponse, .groupInvite, .groupKeyUpdate, .vouch, .voiceFrame:
|
||||
break
|
||||
}
|
||||
|
||||
await MainActor.run {
|
||||
// A panic wipe during the off-main decrypt must not let the
|
||||
// pre-wipe plaintext reach post-wipe state; drop it here, atomic
|
||||
// with the wipe on the main actor.
|
||||
guard self.wipeGeneration == wipeGeneration else { return }
|
||||
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:
|
||||
context.handleReadReceipt(payload, senderPubkey: senderPubkey, convKey: convKey)
|
||||
// Group state travels only over mesh Noise sessions in v1; anything
|
||||
// claiming to be group traffic over Nostr is ignored.
|
||||
// Live voice is mesh-only: latency and relay cost make it
|
||||
// meaningless over Nostr.
|
||||
case .verifyChallenge, .verifyResponse, .groupInvite, .groupKeyUpdate, .vouch, .voiceFrame:
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@MainActor
|
||||
func handleNostrMessage(_ giftWrap: NostrEvent) {
|
||||
guard let context else { return }
|
||||
// Cheap dedup pre-check only; Schnorr verification runs off-main in
|
||||
// processNostrMessage, which then does the authoritative
|
||||
// check-and-record. Recording stays after verification so a
|
||||
// forged-signature copy can never poison the dedup set and suppress
|
||||
// the genuine event.
|
||||
// Cheap dedup pre-check only; processNostrMessage does the
|
||||
// authoritative check-and-record before the off-main NIP-17 unwrap.
|
||||
// The outer signature was already verified (exactly once, off the
|
||||
// main actor) by NostrRelayManager, and only verified events are
|
||||
// recorded, so a forged-signature copy can never poison the dedup
|
||||
// set and suppress the genuine event.
|
||||
if context.hasProcessedNostrEvent(giftWrap.id) { return }
|
||||
|
||||
Task.detached(priority: .userInitiated) { [weak self] in
|
||||
@@ -387,7 +420,6 @@ final class NostrInboundPipeline {
|
||||
}
|
||||
|
||||
func processNostrMessage(_ giftWrap: NostrEvent) async {
|
||||
guard giftWrap.isValidSignature() else { return }
|
||||
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.
|
||||
@@ -397,8 +429,13 @@ final class NostrInboundPipeline {
|
||||
return false
|
||||
}
|
||||
if alreadyProcessed { return }
|
||||
let currentIdentity: NostrIdentity? = await MainActor.run {
|
||||
context.currentNostrIdentity()
|
||||
// Fetch the identity and the wipe generation in ONE main-actor hop:
|
||||
// the generation then vouches for exactly this identity. A wipe after
|
||||
// this point bumps the generation and the delivery hop below drops
|
||||
// the decrypted result (same guard as processGeohashGiftWrap; this
|
||||
// account-mailbox path had the identical hazard).
|
||||
let (currentIdentity, wipeGeneration): (NostrIdentity?, UInt64) = await MainActor.run {
|
||||
(context.currentNostrIdentity(), self.wipeGeneration)
|
||||
}
|
||||
guard let currentIdentity else { return }
|
||||
|
||||
@@ -430,6 +467,9 @@ final class NostrInboundPipeline {
|
||||
let payload = NoisePayload.decode(packet.payload) {
|
||||
let messageTimestamp = Date(timeIntervalSince1970: TimeInterval(rumorTimestamp))
|
||||
await MainActor.run {
|
||||
// Drop pre-wipe plaintext if a panic wipe landed
|
||||
// during the off-main decrypt (see above).
|
||||
guard self.wipeGeneration == wipeGeneration else { return }
|
||||
context.registerNostrKeyMapping(senderPubkey, for: targetPeerID)
|
||||
|
||||
switch payload.type {
|
||||
|
||||
Reference in New Issue
Block a user