mirror of
https://github.com/permissionlesstech/bitchat.git
synced 2026-07-27 09:05:19 +00:00
Gateway mode: opt-in mesh↔Nostr uplink for geohash channels (#1384)
* Add capability bits to announce TLV Announces now carry an optional capabilities TLV (0x05): a little-endian bitfield with named bits for upcoming features (prekeys, wifiBulk, gateway, groups, board, vouch, meshDiagnostics). Old clients skip the unknown TLV; peers without it decode as nil so features can distinguish "legacy peer" from "advertises nothing". PeerCapabilities lives in BitFoundation with a minimal-length encoding that preserves unknown bits for forward compatibility. Peer capabilities are stored in the BLE peer registry on verified announce and exposed via BLEService.peerCapabilities(_:). The local advertisement set is empty until each feature ships its bit. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Gateway mode: opt-in mesh↔Nostr uplink for geohash channels An opt-in "internet gateway" toggle lets one connected phone bridge the local geohash channel for mesh-only peers: signed kind-20000 events ride a new nostrCarrier (0x28) packet — directed to the gateway for uplink, broadcast with TTL for downlink — with Schnorr verification at every hop, CourierStore-style quotas, and explicit loop-prevention rules. - BitFoundation: MessageType.nostrCarrier = 0x28 - NostrCarrierPacket: 2-byte-length TLV codec (direction, geohash, signed event JSON), 16 KiB cap, tolerant decoder - GatewayService: closure-injected policy layer — verify gates (sig, kind, #g tag, age, size), uplink quotas (10/min/depositor rate limit, offline queue of 20 total / 5 per depositor, drop-oldest, flush on reconnect), downlink budget (30/min, bounded drop-oldest backlog), bounded loop-prevention ID sets - BLEService: runtime capability bits (advertise .gateway only while the toggle is on, re-announce on change), signed directed uplink sends, carrier ingress with depositor signature verification - Mesh-only senders uplink automatically from sendGeohash when no relay is connected and a reachable peer advertises .gateway; once-per- channel "sent via mesh gateway" notice - UI: gateway toggle beside the Tor toggle, globe header indicator, VoiceOver labels, xcstrings entries Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Gateway: harden downlink freshness, uplink verify ordering, and drain Fixes the confirmed downlink/uplink defects from the PR #1384 review + Codex findings: - Downlink age + #g gate (Codex P2 / review #1): rebroadcastRelayEvent now drops events outside the same freshness window receivers enforce and whose #g tag mismatches the carrier geohash, BEFORE spending any budget — so a 1h/200-event channel-resubscribe backfill no longer burns the 30/min BLE budget on events every receiver drops. - Rate-limit + dedup before Schnorr (review #2): handleUplinkDeposit now runs cheap structural checks + carried-ID dedup + rate-token consume before isValidSignature(), so a replay flood is bounded by cheap work instead of unbounded main-actor verifies. - Quota-dropped deposits not rendered (review #3): enqueueUplink reports acceptance and injectInbound only fires for events actually published/queued, ending the local-timeline divergence. - Drain timer + mark-after-send (Codex P2 / review #4): a burst beyond budget now arms a timer to drain when the window frees; rebroadcast IDs are marked only after an event is actually sent, so overflow- dropped events stay retryable. - Symmetric publish path (review #5): the gateway publish closure now refuses when no geo relay is known, matching the local send path instead of publishing dead traffic to default relays. - Loop-rule doc (review #7): softened to reflect that rule 3 is a call-site convention with unit-tested backstops; added tests for the publishedEventIDs backstop, downlink freshness/mismatch, drain timer, and quota-drop non-injection. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Gateway: stop self-echo of uplinked events onto the mesh Every event a gateway uplinks to the relays comes back through its own geohash subscription. `rebroadcastRelayEvent` deduped against `meshBroadcastEventIDs`, `rebroadcastEventIDs`, and `pendingDownlinks`, but not `publishedEventIDs` — so an event this gateway just published was downlink-rebroadcast onto the same mesh it originated from, doubling BLE airtime per uplinked message and able to starve the 30/min downlink budget on a busy channel (device-confirmed, filed on #1384). Fix: also skip the downlink rebroadcast when the event id is in `publishedEventIDs`. That set is already the bounded (drop-oldest, capacity maxTrackedEventIDs) loop-rule-2 uplink cache, populated only by `publish()`, so genuine inbound-from-internet events (never published here) still rebroadcast normally. Reconciles cleanly with the existing loop-prevention sets — no new state. Adds a GatewayServiceTests case asserting an uplinked event that echoes back via the subscription is not rebroadcast, while a genuine inbound event still is. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> --------- Co-authored-by: jack <jackjackbits@users.noreply.github.com> Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
jack
co-authored by
jack
Claude Fable 5
jack
Claude Fable 5
parent
2360140760
commit
d4f0c49787
@@ -12,7 +12,7 @@ enum BLEOutboundPacketPolicy {
|
||||
switch MessageType(rawValue: packetType) {
|
||||
case .noiseEncrypted, .noiseHandshake:
|
||||
return true
|
||||
case .none, .announce, .message, .leave, .requestSync, .fragment, .fileTransfer, .courierEnvelope, .boardPost:
|
||||
case .none, .announce, .message, .leave, .requestSync, .fragment, .fileTransfer, .courierEnvelope, .boardPost, .nostrCarrier:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
@@ -112,6 +112,11 @@ struct BLEPeerRegistry {
|
||||
peers[peerID.toShort()]?.capabilities ?? []
|
||||
}
|
||||
|
||||
/// Peers whose last verified announce advertised the given capability.
|
||||
func peers(advertising capability: PeerCapabilities) -> [PeerID] {
|
||||
peers.values.filter { $0.capabilities.contains(capability) }.map(\.peerID)
|
||||
}
|
||||
|
||||
func displayNicknames(selfNickname: String) -> [PeerID: String] {
|
||||
let connected = peers.filter { $0.value.isConnected }
|
||||
let tuples = connected.map { ($0.key, $0.value.nickname, true) }
|
||||
|
||||
@@ -51,8 +51,11 @@ struct BLEReceivePipeline {
|
||||
// Courier envelopes are directed opaque ciphertext like DMs; a
|
||||
// remote handover toward a relayed announce rides this same
|
||||
// deterministic relay treatment instead of the broadcast clamp.
|
||||
// Directed nostrCarrier uplinks (mesh-only peer -> gateway) need
|
||||
// the same multi-hop treatment to reach a non-adjacent gateway.
|
||||
isDirectedEncrypted: (packet.type == MessageType.noiseEncrypted.rawValue
|
||||
|| packet.type == MessageType.courierEnvelope.rawValue) && packet.recipientID != nil,
|
||||
|| packet.type == MessageType.courierEnvelope.rawValue
|
||||
|| packet.type == MessageType.nostrCarrier.rawValue) && packet.recipientID != nil,
|
||||
isFragment: packet.type == MessageType.fragment.rawValue,
|
||||
isDirectedFragment: packet.type == MessageType.fragment.rawValue && packet.recipientID != nil,
|
||||
isHandshake: packet.type == MessageType.noiseHandshake.rawValue,
|
||||
|
||||
@@ -62,6 +62,14 @@ final class BLEService: NSObject {
|
||||
// Local-only store-and-forward counters; nil in unit tests.
|
||||
var sfMetrics: StoreAndForwardMetrics?
|
||||
|
||||
// Gateway mode: sink for received nostrCarrier packets (set by app
|
||||
// wiring, called on the main actor after transport-level checks) and the
|
||||
// runtime-toggled capability bits ORed into `PeerCapabilities.localSupported`
|
||||
// for every announce. `directedToUs` distinguishes an uplink deposit
|
||||
// addressed to this device from a downlink broadcast.
|
||||
var onNostrCarrierPacket: (@MainActor (_ payload: Data, _ from: PeerID, _ directedToUs: Bool) -> Void)?
|
||||
private var runtimeCapabilities: PeerCapabilities = [] // collectionsQueue
|
||||
|
||||
#if DEBUG
|
||||
// Test-only tap on the outbound pipeline so multi-node tests can ferry
|
||||
// packets between in-process service instances.
|
||||
@@ -638,6 +646,32 @@ final class BLEService: NSObject {
|
||||
collectionsQueue.sync { peerRegistry.capabilities(for: peerID) }
|
||||
}
|
||||
|
||||
/// Enables or disables a runtime-advertised capability bit (e.g. the
|
||||
/// internet-gateway toggle) and re-announces so peers learn promptly.
|
||||
/// Build-time bits stay in `PeerCapabilities.localSupported`.
|
||||
func setLocalCapability(_ capability: PeerCapabilities, enabled: Bool) {
|
||||
let changed: Bool = collectionsQueue.sync(flags: .barrier) {
|
||||
let before = runtimeCapabilities
|
||||
if enabled {
|
||||
runtimeCapabilities.insert(capability)
|
||||
} else {
|
||||
runtimeCapabilities.remove(capability)
|
||||
}
|
||||
return runtimeCapabilities != before
|
||||
}
|
||||
guard changed else { return }
|
||||
sendAnnounce(forceSend: true)
|
||||
}
|
||||
|
||||
/// Reachable peers currently advertising the `.gateway` capability.
|
||||
func reachableGatewayPeers() -> [PeerID] {
|
||||
let now = Date()
|
||||
return collectionsQueue.sync {
|
||||
peerRegistry.peers(advertising: .gateway)
|
||||
.filter { peerRegistry.isReachable($0, now: now) }
|
||||
}
|
||||
}
|
||||
|
||||
func getPeerNicknames() -> [PeerID: String] {
|
||||
return collectionsQueue.sync {
|
||||
peerRegistry.displayNicknames(selfNickname: myNickname)
|
||||
@@ -1272,16 +1306,16 @@ final class BLEService: NSObject {
|
||||
let noisePub = noiseService.getStaticPublicKeyData() // For noise handshakes and peer identification
|
||||
let signingPub = noiseService.getSigningPublicKeyData() // For signature verification
|
||||
|
||||
let connectedPeerIDs: [Data] = collectionsQueue.sync {
|
||||
peerRegistry.connectedRoutingData
|
||||
let (connectedPeerIDs, advertisedCapabilities): ([Data], PeerCapabilities) = collectionsQueue.sync {
|
||||
(peerRegistry.connectedRoutingData, PeerCapabilities.localSupported.union(runtimeCapabilities))
|
||||
}
|
||||
|
||||
|
||||
let announcement = AnnouncementPacket(
|
||||
nickname: myNickname,
|
||||
noisePublicKey: noisePub,
|
||||
signingPublicKey: signingPub,
|
||||
directNeighbors: connectedPeerIDs,
|
||||
capabilities: PeerCapabilities.localSupported
|
||||
capabilities: advertisedCapabilities
|
||||
)
|
||||
|
||||
guard let payload = announcement.encode() else {
|
||||
@@ -2762,6 +2796,81 @@ extension BLEService {
|
||||
}
|
||||
}
|
||||
|
||||
// MARK: Gateway carrier (nostrCarrier)
|
||||
|
||||
/// Sign and send an encoded `toGateway` carrier payload directed at a
|
||||
/// gateway peer. The packet is signed so the gateway can key its uplink
|
||||
/// quotas to an authenticated depositor; the carried Nostr event has its
|
||||
/// own Schnorr signature for content authenticity. Returns false when
|
||||
/// the gateway is not reachable or signing fails.
|
||||
func sendNostrCarrier(_ payload: Data, to gatewayPeer: PeerID) -> Bool {
|
||||
guard isPeerReachable(gatewayPeer) else { return false }
|
||||
let packet = BitchatPacket(
|
||||
type: MessageType.nostrCarrier.rawValue,
|
||||
senderID: myPeerIDData,
|
||||
recipientID: Data(hexString: gatewayPeer.id),
|
||||
timestamp: UInt64(Date().timeIntervalSince1970 * 1000),
|
||||
payload: payload,
|
||||
signature: nil,
|
||||
ttl: messageTTL
|
||||
)
|
||||
guard let signed = noiseService.signPacket(packet) else { return false }
|
||||
messageQueue.async { [weak self] in
|
||||
// broadcastPacket applies a known route when one exists and
|
||||
// otherwise floods the directed packet like a DM, so a gateway
|
||||
// that is reachable but multi-hop still gets the deposit.
|
||||
self?.broadcastPacket(signed)
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
/// Broadcast an encoded `fromGateway` carrier payload on the mesh with
|
||||
/// the default TTL. Unsigned at the packet layer — receivers verify the
|
||||
/// carried event's own Schnorr signature.
|
||||
func broadcastNostrCarrier(_ payload: Data) {
|
||||
let packet = BitchatPacket(
|
||||
type: MessageType.nostrCarrier.rawValue,
|
||||
senderID: myPeerIDData,
|
||||
recipientID: nil,
|
||||
timestamp: UInt64(Date().timeIntervalSince1970 * 1000),
|
||||
payload: payload,
|
||||
signature: nil,
|
||||
ttl: messageTTL
|
||||
)
|
||||
messageQueue.async { [weak self] in
|
||||
self?.broadcastPacket(packet)
|
||||
}
|
||||
}
|
||||
|
||||
/// Transport-level handling for a received nostrCarrier packet; policy
|
||||
/// (verification of the carried event, quotas, loop prevention) lives in
|
||||
/// `GatewayService` behind `onNostrCarrierPacket`.
|
||||
private func handleNostrCarrier(_ packet: BitchatPacket, from peerID: PeerID) {
|
||||
let senderID = PeerID(hexData: packet.senderID)
|
||||
let directedToUs: Bool
|
||||
if let recipientID = packet.recipientID {
|
||||
// Carriers addressed elsewhere ride the generic relay path untouched.
|
||||
guard recipientID == myPeerIDData else { return }
|
||||
// Uplink deposit: quotas are keyed by the depositor, so the
|
||||
// packet signature must verify against the sender's announced
|
||||
// signing key. Unlike courier deposits the depositor may be
|
||||
// multi-hop away, so ingress-link identity is not required.
|
||||
let signingKey = collectionsQueue.sync { peerRegistry.info(for: senderID)?.signingPublicKey }
|
||||
guard let signingKey,
|
||||
noiseService.verifyPacketSignature(packet, publicKey: signingKey) else {
|
||||
SecureLogger.debug("🌐 nostrCarrier uplink from \(senderID.id.prefix(8))… rejected (missing/invalid packet signature)", category: .security)
|
||||
return
|
||||
}
|
||||
directedToUs = true
|
||||
} else {
|
||||
directedToUs = false
|
||||
}
|
||||
let payload = packet.payload
|
||||
notifyUI { [weak self] in
|
||||
self?.onNostrCarrierPacket?(payload, senderID, directedToUs)
|
||||
}
|
||||
}
|
||||
|
||||
// MARK: Link capability snapshots (thread-safe via bleQueue)
|
||||
|
||||
private func readLinkState<T>(_ body: (BLELinkStateStore) -> T) -> T {
|
||||
@@ -3269,6 +3378,8 @@ extension BLEService {
|
||||
case .boardPost:
|
||||
// Invalid or deleted posts must not spread; skip the relay step.
|
||||
guard handleBoardPost(packet, from: senderID) else { return }
|
||||
case .nostrCarrier:
|
||||
handleNostrCarrier(packet, from: peerID)
|
||||
|
||||
case .leave:
|
||||
handleLeave(packet, from: senderID)
|
||||
|
||||
@@ -0,0 +1,492 @@
|
||||
//
|
||||
// GatewayService.swift
|
||||
// bitchat
|
||||
//
|
||||
// This is free and unencumbered software released into the public domain.
|
||||
// For more information, see <https://unlicense.org>
|
||||
//
|
||||
|
||||
import BitFoundation
|
||||
import BitLogger
|
||||
import Combine
|
||||
import Foundation
|
||||
|
||||
/// Policy engine for gateway mode: an opt-in "share my internet with the
|
||||
/// mesh" bridge. While the toggle is on, this device advertises the
|
||||
/// `.gateway` capability bit, publishes signed geohash events deposited by
|
||||
/// mesh-only peers to Nostr relays (uplink), and rebroadcasts inbound relay
|
||||
/// events onto the mesh (downlink) so mesh-only peers can take part in the
|
||||
/// local geohash channel. Mesh-only peers need no toggle: their uplink
|
||||
/// engages automatically when relays are unreachable and a gateway peer
|
||||
/// exists.
|
||||
///
|
||||
/// Threat model:
|
||||
/// - Keys never leave the originating device. Mesh-only senders sign events
|
||||
/// locally with their per-geohash ephemeral identity; the gateway carries
|
||||
/// only the finished, signed event.
|
||||
/// - The gateway cannot forge or alter events: every carried event is
|
||||
/// Schnorr-verified here before it is published or rebroadcast, and again
|
||||
/// independently by relays and receivers.
|
||||
/// - Carried contents are public geohash chat, already plaintext on Nostr,
|
||||
/// so the mesh carrier adds no confidentiality loss.
|
||||
///
|
||||
/// Loop-prevention rules:
|
||||
/// 1. An event learned from a `fromGateway` mesh broadcast is never
|
||||
/// re-published to relays, never re-uplinked, and never rebroadcast
|
||||
/// (`meshBroadcastEventIDs`), so a second gateway on the same mesh cannot
|
||||
/// echo mesh-carried traffic back out. Mesh-level propagation of the
|
||||
/// original broadcast packet is the TTL relay's job, not ours.
|
||||
/// 2. An uplink deposit is published at most once (`publishedEventIDs`) and
|
||||
/// a relay event is rebroadcast at most once (`rebroadcastEventIDs`), so
|
||||
/// repeat deposits and relay echoes are absorbed. An event this gateway
|
||||
/// itself uplinked (`publishedEventIDs`) is additionally never
|
||||
/// downlink-rebroadcast: it originated on this mesh, so echoing it back
|
||||
/// when our own relay subscription redelivers it would double BLE airtime
|
||||
/// (the device-confirmed self-echo bug).
|
||||
/// 3. Uplink is only attempted for locally composed events at the send site
|
||||
/// (`GeohashSubscriptionManager.sendGeohash`); events received over the
|
||||
/// carrier never re-enter the uplink path. This is a call-site convention;
|
||||
/// the `meshBroadcastEventIDs`/`publishedEventIDs` backstops in
|
||||
/// `uplinkViaMesh` enforce it defensively and are unit-tested.
|
||||
/// Rules 1 and 2 are enforced here and unit-tested.
|
||||
/// Rebroadcast storms at the mesh layer are additionally bounded by the BLE
|
||||
/// `MessageDeduplicator` and packet TTL, and receivers dedup carried events
|
||||
/// against their own relay subscriptions via the Nostr event-ID cache in
|
||||
/// `NostrInboundPipeline`.
|
||||
///
|
||||
/// All dependencies are closure-injected (repo convention) so the policy
|
||||
/// layer is unit-testable without relays or radios.
|
||||
@MainActor
|
||||
final class GatewayService: ObservableObject {
|
||||
enum Limits {
|
||||
/// Uplink deposits held while relays are unreachable (CourierStore-style
|
||||
/// bounded mailbag: bounded total, bounded per depositor).
|
||||
static let maxQueuedUplinks = 20
|
||||
static let maxQueuedUplinksPerDepositor = 5
|
||||
/// Uplink deposits accepted per depositor per minute.
|
||||
static let uplinkEventsPerMinutePerDepositor = 10
|
||||
/// Downlink mesh rebroadcasts per minute — BLE airtime is precious.
|
||||
/// Beyond the budget events queue (bounded, drop-oldest) and drain on
|
||||
/// a scheduled timer once the window frees (also re-driven by the next
|
||||
/// inbound relay event); a quiet channel does not strand its backlog.
|
||||
static let downlinkEventsPerMinute = 30
|
||||
static let maxPendingDownlinks = 30
|
||||
/// Accepted clock skew for a carried ephemeral event; anything older
|
||||
/// is stale replay the relays would drop anyway.
|
||||
static let maxEventAgeSeconds: TimeInterval = 15 * 60
|
||||
/// Bounded loop-prevention ID caches (oldest evicted).
|
||||
static let maxTrackedEventIDs = 512
|
||||
}
|
||||
|
||||
struct QueuedUplink {
|
||||
let depositor: PeerID
|
||||
let geohash: String
|
||||
let event: NostrEvent
|
||||
let queuedAt: Date
|
||||
}
|
||||
|
||||
static let shared = GatewayService()
|
||||
|
||||
/// The user toggle. While true this device advertises `.gateway` and
|
||||
/// bridges mesh <-> Nostr for geohash channels.
|
||||
@Published private(set) var isEnabled: Bool
|
||||
|
||||
// MARK: Wiring (set once by the bootstrapper; fakes in tests)
|
||||
|
||||
/// Publishes a verified event to the geo relays for a geohash.
|
||||
var publishToRelays: (@MainActor (NostrEvent, String) -> Void)?
|
||||
/// Broadcasts an encoded `fromGateway` carrier payload on the mesh.
|
||||
var broadcastToMesh: (@MainActor (Data) -> Void)?
|
||||
/// Sends an encoded `toGateway` carrier payload directed to a gateway
|
||||
/// peer. Returns false when the transport could not accept it.
|
||||
var sendToGatewayPeer: (@MainActor (Data, PeerID) -> Bool)?
|
||||
/// Reachable mesh peers currently advertising the `.gateway` capability.
|
||||
var availableGatewayPeers: (@MainActor () -> [PeerID])?
|
||||
/// Whether any Nostr relay connection is currently working.
|
||||
var relaysConnected: (@MainActor () -> Bool)?
|
||||
/// The geohash channel the local user is viewing, if any.
|
||||
var currentGeohash: (@MainActor () -> String?)?
|
||||
/// Injects a verified carried event into the same inbound pipeline as
|
||||
/// relay-received events (blocking, rate limits, dedup, rendering).
|
||||
var injectInbound: (@MainActor (NostrEvent) -> Void)?
|
||||
/// Fired on toggle changes (advertise/withdraw the capability bit and
|
||||
/// force a re-announce).
|
||||
var onEnabledChanged: (@MainActor (Bool) -> Void)?
|
||||
/// Schedules a downlink-drain closure to run after a delay. Injected so
|
||||
/// the drain timer is deterministic in tests; nil arms a real `Task`.
|
||||
var scheduleDrainTimer: (@MainActor (TimeInterval, @escaping @MainActor () -> Void) -> Void)?
|
||||
|
||||
// MARK: State
|
||||
|
||||
/// Loop rule 1: event IDs seen in `fromGateway` mesh broadcasts.
|
||||
private var meshBroadcastEventIDs: BoundedIDSet
|
||||
/// Loop rule 2 (uplink): event IDs this gateway already published.
|
||||
private var publishedEventIDs: BoundedIDSet
|
||||
/// Loop rule 2 (downlink): event IDs this gateway already rebroadcast.
|
||||
private var rebroadcastEventIDs: BoundedIDSet
|
||||
|
||||
private(set) var queuedUplinks: [QueuedUplink] = []
|
||||
private var uplinkDepositTimes: [PeerID: [Date]] = [:]
|
||||
private var downlinkSendTimes: [Date] = []
|
||||
private var pendingDownlinks: [(event: NostrEvent, geohash: String)] = []
|
||||
/// True while a drain timer is armed, so a burst schedules at most one.
|
||||
private var downlinkDrainScheduled = false
|
||||
|
||||
private let defaults: UserDefaults
|
||||
private let now: () -> Date
|
||||
private static let enabledKey = "gateway.userEnabled"
|
||||
|
||||
init(defaults: UserDefaults = .standard, now: @escaping () -> Date = Date.init) {
|
||||
self.defaults = defaults
|
||||
self.now = now
|
||||
self.isEnabled = defaults.bool(forKey: Self.enabledKey)
|
||||
self.meshBroadcastEventIDs = BoundedIDSet(capacity: Limits.maxTrackedEventIDs)
|
||||
self.publishedEventIDs = BoundedIDSet(capacity: Limits.maxTrackedEventIDs)
|
||||
self.rebroadcastEventIDs = BoundedIDSet(capacity: Limits.maxTrackedEventIDs)
|
||||
}
|
||||
|
||||
// MARK: - Toggle
|
||||
|
||||
func setEnabled(_ enabled: Bool) {
|
||||
guard enabled != isEnabled else { return }
|
||||
isEnabled = enabled
|
||||
defaults.set(enabled, forKey: Self.enabledKey)
|
||||
if !enabled {
|
||||
queuedUplinks.removeAll()
|
||||
pendingDownlinks.removeAll()
|
||||
uplinkDepositTimes.removeAll()
|
||||
}
|
||||
SecureLogger.info("🌐 Gateway mode \(enabled ? "enabled" : "disabled")", category: .session)
|
||||
onEnabledChanged?(enabled)
|
||||
}
|
||||
|
||||
// MARK: - Mesh carrier ingress (both roles)
|
||||
|
||||
/// Entry point for received `nostrCarrier` packets. `directedToUs` is
|
||||
/// true for packets addressed to this device (uplink deposits); false
|
||||
/// for broadcasts (downlink rebroadcasts from a gateway).
|
||||
func handleMeshCarrier(_ payload: Data, from peerID: PeerID, directedToUs: Bool) {
|
||||
guard let carrier = NostrCarrierPacket.decode(payload) else {
|
||||
SecureLogger.debug("🌐 Gateway: dropping undecodable carrier from \(peerID.id.prefix(8))…", category: .session)
|
||||
return
|
||||
}
|
||||
switch carrier.direction {
|
||||
case .toGateway:
|
||||
// Uplink deposits are directed; a broadcast toGateway is malformed.
|
||||
guard directedToUs else { return }
|
||||
handleUplinkDeposit(carrier, from: peerID)
|
||||
case .fromGateway:
|
||||
// Downlink rides broadcast only; a directed fromGateway is malformed.
|
||||
guard !directedToUs else { return }
|
||||
handleDownlinkBroadcast(carrier)
|
||||
}
|
||||
}
|
||||
|
||||
// MARK: - Uplink (gateway role: mesh peer -> internet)
|
||||
|
||||
private func handleUplinkDeposit(_ carrier: NostrCarrierPacket, from depositor: PeerID) {
|
||||
guard isEnabled else { return }
|
||||
// Cheap structural checks first (parse, size, geohash, kind, #g tag,
|
||||
// age) — no crypto — so junk and stale replays are dropped before we
|
||||
// ever pay for a MainActor Schnorr verify.
|
||||
guard let event = structurallyValidEvent(from: carrier) else {
|
||||
SecureLogger.debug("🌐 Gateway: rejected uplink deposit from \(depositor.id.prefix(8))… (failed validation)", category: .security)
|
||||
return
|
||||
}
|
||||
// Dedup by the carried event ID BEFORE verification. Loop rule 1: a
|
||||
// fromGateway-learned event is mesh-carried and must never be
|
||||
// re-published. Loop rule 2: repeat deposits of an already handled
|
||||
// event are absorbed. A replay of one valid deposit is short-circuited
|
||||
// here without a per-packet signature verify.
|
||||
guard !meshBroadcastEventIDs.contains(event.id),
|
||||
!publishedEventIDs.contains(event.id),
|
||||
!queuedUplinks.contains(where: { $0.event.id == event.id }) else {
|
||||
return
|
||||
}
|
||||
// Consume the per-depositor rate token BEFORE the expensive verify so
|
||||
// a flood of distinct forged/junk deposits is bounded by cheap work,
|
||||
// not by main-actor Schnorr verifications.
|
||||
guard allowUplinkDeposit(from: depositor) else {
|
||||
SecureLogger.debug("🌐 Gateway: rate-limited uplink deposit from \(depositor.id.prefix(8))…", category: .session)
|
||||
return
|
||||
}
|
||||
// Only now pay for cryptographic verification; receivers verify again.
|
||||
guard event.isValidSignature() else {
|
||||
SecureLogger.debug("🌐 Gateway: rejected uplink deposit from \(depositor.id.prefix(8))… (bad signature)", category: .security)
|
||||
return
|
||||
}
|
||||
|
||||
let accepted: Bool
|
||||
if relaysConnected?() ?? false {
|
||||
publish(event, geohash: carrier.geohash)
|
||||
accepted = true
|
||||
} else {
|
||||
accepted = enqueueUplink(QueuedUplink(depositor: depositor, geohash: carrier.geohash, event: event, queuedAt: now()))
|
||||
}
|
||||
|
||||
// Only render on our own timeline what we actually accepted for
|
||||
// publish or queue: a quota-dropped deposit is never published and,
|
||||
// being directed, no other peer will ever see it, so showing it would
|
||||
// diverge our timeline permanently from what reached the channel.
|
||||
if accepted, currentGeohash?() == carrier.geohash {
|
||||
injectInbound?(event)
|
||||
}
|
||||
}
|
||||
|
||||
/// Publish everything queued while relays were unreachable. Called when
|
||||
/// relay connectivity comes back.
|
||||
func flushQueuedUplinks() {
|
||||
guard isEnabled, relaysConnected?() ?? false, !queuedUplinks.isEmpty else { return }
|
||||
let queued = queuedUplinks
|
||||
queuedUplinks.removeAll()
|
||||
for item in queued where !publishedEventIDs.contains(item.event.id) {
|
||||
publish(item.event, geohash: item.geohash)
|
||||
}
|
||||
}
|
||||
|
||||
private func publish(_ event: NostrEvent, geohash: String) {
|
||||
publishedEventIDs.insert(event.id)
|
||||
publishToRelays?(event, geohash)
|
||||
SecureLogger.info("🌐 Gateway: published carried event \(event.id.prefix(8))… to relays for #\(geohash)", category: .session)
|
||||
}
|
||||
|
||||
/// Returns true when the item was actually stored for later publish.
|
||||
@discardableResult
|
||||
private func enqueueUplink(_ item: QueuedUplink) -> Bool {
|
||||
let fromDepositor = queuedUplinks.filter { $0.depositor == item.depositor }.count
|
||||
guard fromDepositor < Limits.maxQueuedUplinksPerDepositor else {
|
||||
SecureLogger.debug("🌐 Gateway: uplink queue quota reached for \(item.depositor.id.prefix(8))…", category: .session)
|
||||
return false
|
||||
}
|
||||
if queuedUplinks.count >= Limits.maxQueuedUplinks {
|
||||
queuedUplinks.removeFirst(queuedUplinks.count - Limits.maxQueuedUplinks + 1)
|
||||
}
|
||||
queuedUplinks.append(item)
|
||||
return true
|
||||
}
|
||||
|
||||
private func allowUplinkDeposit(from depositor: PeerID) -> Bool {
|
||||
let cutoff = now().addingTimeInterval(-60)
|
||||
var times = uplinkDepositTimes[depositor, default: []]
|
||||
times.removeAll { $0 < cutoff }
|
||||
guard times.count < Limits.uplinkEventsPerMinutePerDepositor else {
|
||||
uplinkDepositTimes[depositor] = times
|
||||
return false
|
||||
}
|
||||
times.append(now())
|
||||
uplinkDepositTimes[depositor] = times
|
||||
// Bound the tracker itself against a churn of spoofed depositors.
|
||||
if uplinkDepositTimes.count > Limits.maxTrackedEventIDs {
|
||||
uplinkDepositTimes = uplinkDepositTimes.filter { !$0.value.isEmpty && $0.value.contains { $0 >= cutoff } }
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// MARK: - Downlink (gateway role: internet -> mesh)
|
||||
|
||||
/// Called for every event the gateway's own geohash-channel subscription
|
||||
/// delivers. Wraps it in a `fromGateway` carrier and broadcasts it on
|
||||
/// the mesh, within the airtime budget.
|
||||
func rebroadcastRelayEvent(_ event: NostrEvent, geohash: String) {
|
||||
guard isEnabled, broadcastToMesh != nil else { return }
|
||||
guard event.kind == NostrProtocol.EventKind.ephemeralEvent.rawValue else { return }
|
||||
// Freshness + geohash gate BEFORE spending any budget. A channel
|
||||
// (re)subscribe backfills up to an hour of history (limit 200), but
|
||||
// every receiver's `validatedEvent` drops anything older than the
|
||||
// same window — so rebroadcasting backfill would burn the whole
|
||||
// per-minute budget on events no mesh peer accepts. Also require the
|
||||
// event's own `#g` tag to match the carrier geohash.
|
||||
guard isFresh(event),
|
||||
event.tags.contains(where: { $0.count >= 2 && $0[0] == "g" && $0[1] == geohash }) else {
|
||||
return
|
||||
}
|
||||
// Loop rule 1: never rebroadcast mesh-carried events back onto the
|
||||
// mesh. Loop rule 2 (self-echo): never rebroadcast an event this
|
||||
// gateway itself uplinked (`publishedEventIDs`) — it originated on this
|
||||
// very mesh, so our own relay subscription echoing it back must not
|
||||
// double the BLE airtime by pushing it out again. Loop rule 2
|
||||
// (downlink): rebroadcast each genuine inbound relay event at most once
|
||||
// — but mark only AFTER it is actually sent (in `drainPendingDownlinks`),
|
||||
// so an event dropped by the queue overflow stays retryable on relay
|
||||
// redelivery. Guard against a redelivery re-queueing an event that is
|
||||
// still waiting to be sent.
|
||||
guard !meshBroadcastEventIDs.contains(event.id),
|
||||
!publishedEventIDs.contains(event.id),
|
||||
!rebroadcastEventIDs.contains(event.id),
|
||||
!pendingDownlinks.contains(where: { $0.event.id == event.id }) else {
|
||||
return
|
||||
}
|
||||
// Verify before spending BLE airtime; receivers verify again.
|
||||
guard event.isValidSignature() else { return }
|
||||
|
||||
pendingDownlinks.append((event, geohash))
|
||||
if pendingDownlinks.count > Limits.maxPendingDownlinks {
|
||||
// Bandwidth guard: drop-oldest — fresher chat is worth more. The
|
||||
// dropped event is not yet in `rebroadcastEventIDs`, so a later
|
||||
// relay redelivery can still carry it.
|
||||
pendingDownlinks.removeFirst(pendingDownlinks.count - Limits.maxPendingDownlinks)
|
||||
}
|
||||
drainPendingDownlinks()
|
||||
}
|
||||
|
||||
private func drainPendingDownlinks() {
|
||||
let cutoff = now().addingTimeInterval(-60)
|
||||
downlinkSendTimes.removeAll { $0 < cutoff }
|
||||
while !pendingDownlinks.isEmpty,
|
||||
downlinkSendTimes.count < Limits.downlinkEventsPerMinute {
|
||||
let (event, geohash) = pendingDownlinks.removeFirst()
|
||||
// A queued event may have aged past the window while it waited;
|
||||
// don't burn airtime on what receivers would now drop.
|
||||
guard isFresh(event) else { continue }
|
||||
guard let carrier = NostrCarrierPacket(direction: .fromGateway, geohash: geohash, event: event),
|
||||
let payload = carrier.encode() else { continue }
|
||||
broadcastToMesh?(payload)
|
||||
// Mark-after-send: only now is the relay event definitively
|
||||
// rebroadcast (loop rule 2).
|
||||
rebroadcastEventIDs.insert(event.id)
|
||||
downlinkSendTimes.append(now())
|
||||
}
|
||||
// Budget exhausted with events still queued: arm a timer to drain when
|
||||
// the window frees, instead of stranding them until the next inbound
|
||||
// relay event (which may never come on a channel that went quiet).
|
||||
scheduleDownlinkDrainIfNeeded()
|
||||
}
|
||||
|
||||
/// Arms a single timer to drain the backlog once the per-minute window
|
||||
/// frees. No-op when nothing is pending or a drain is already scheduled.
|
||||
private func scheduleDownlinkDrainIfNeeded() {
|
||||
guard !pendingDownlinks.isEmpty, !downlinkDrainScheduled else { return }
|
||||
// The window frees when the oldest recorded send ages out of 60s.
|
||||
let oldest = downlinkSendTimes.min() ?? now()
|
||||
let delay = max(0.05, 60 - now().timeIntervalSince(oldest))
|
||||
downlinkDrainScheduled = true
|
||||
let fire: @MainActor () -> Void = { [weak self] in
|
||||
guard let self else { return }
|
||||
self.downlinkDrainScheduled = false
|
||||
self.drainPendingDownlinks()
|
||||
}
|
||||
if let scheduleDrainTimer {
|
||||
scheduleDrainTimer(delay, fire)
|
||||
} else {
|
||||
Task { @MainActor in
|
||||
try? await Task.sleep(nanoseconds: UInt64(delay * 1_000_000_000))
|
||||
fire()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// MARK: - Downlink (receiver role: carried event arrives over mesh)
|
||||
|
||||
private func handleDownlinkBroadcast(_ carrier: NostrCarrierPacket) {
|
||||
guard let event = validatedEvent(from: carrier) else { return }
|
||||
// Mark only AFTER signature verification, so a forged copy carrying a
|
||||
// real event's ID cannot poison the never-republish set, and use the
|
||||
// marking as dedup: the same broadcast relayed along several mesh
|
||||
// paths injects once (the pipeline's Nostr event-ID cache additionally
|
||||
// dedups against our own relay subscription).
|
||||
guard meshBroadcastEventIDs.insert(event.id) else { return }
|
||||
// Only inject events for the channel we're viewing; the inbound
|
||||
// pipeline files public messages under the current geohash.
|
||||
guard currentGeohash?() == carrier.geohash else { return }
|
||||
injectInbound?(event)
|
||||
}
|
||||
|
||||
// MARK: - Uplink (sender role: mesh-only peer with no relays)
|
||||
|
||||
/// Hands a locally signed event to a mesh gateway peer when we have no
|
||||
/// working relay connection. Returns true when the event was sent.
|
||||
///
|
||||
/// v1 is deliberately fire-and-forget: no gateway ack. The event also
|
||||
/// stays in `NostrRelayManager`'s own pending queue, so if our internet
|
||||
/// comes back the relays dedup the duplicate publish by event ID.
|
||||
///
|
||||
/// Loop rule 3: call sites only pass freshly composed events (see
|
||||
/// `GeohashSubscriptionManager.sendGeohash`); received carrier events
|
||||
/// never reach this path, and the mesh-carried guard below backstops it.
|
||||
func uplinkViaMesh(event: NostrEvent, geohash: String) -> Bool {
|
||||
if relaysConnected?() ?? true { return false }
|
||||
guard !meshBroadcastEventIDs.contains(event.id),
|
||||
!publishedEventIDs.contains(event.id) else {
|
||||
return false
|
||||
}
|
||||
// A single gateway is enough — relays fan out from there, and BLE
|
||||
// airtime is precious.
|
||||
guard let gateway = availableGatewayPeers?().first else { return false }
|
||||
guard let carrier = NostrCarrierPacket(direction: .toGateway, geohash: geohash, event: event),
|
||||
let payload = carrier.encode() else {
|
||||
return false
|
||||
}
|
||||
guard sendToGatewayPeer?(payload, gateway) ?? false else { return false }
|
||||
SecureLogger.info("🌐 Gateway: uplinked event \(event.id.prefix(8))… for #\(geohash) via mesh gateway \(gateway.id.prefix(8))…", category: .session)
|
||||
return true
|
||||
}
|
||||
|
||||
// MARK: - Validation
|
||||
|
||||
/// Structural and cryptographic checks every carried event must pass
|
||||
/// before a gateway publishes it or a receiver displays it. Ordered
|
||||
/// cheap-first; Schnorr verification runs last.
|
||||
private func validatedEvent(from carrier: NostrCarrierPacket) -> NostrEvent? {
|
||||
guard let event = structurallyValidEvent(from: carrier),
|
||||
event.isValidSignature() else {
|
||||
return nil
|
||||
}
|
||||
return event
|
||||
}
|
||||
|
||||
/// The cheap half of `validatedEvent`: parse + size + geohash + kind +
|
||||
/// `#g` tag + freshness, with NO signature verification. Callers that can
|
||||
/// dedup or rate-limit on the carried ID run this first so the expensive
|
||||
/// Schnorr verify is reached only for events that survive the cheap gates.
|
||||
private func structurallyValidEvent(from carrier: NostrCarrierPacket) -> NostrEvent? {
|
||||
guard carrier.eventJSON.count <= NostrCarrierPacket.maxEventJSONBytes,
|
||||
Self.isValidGeohash(carrier.geohash),
|
||||
let event = carrier.event(),
|
||||
event.kind == NostrProtocol.EventKind.ephemeralEvent.rawValue,
|
||||
event.tags.contains(where: { $0.count >= 2 && $0[0] == "g" && $0[1] == carrier.geohash }),
|
||||
isFresh(event) else {
|
||||
return nil
|
||||
}
|
||||
return event
|
||||
}
|
||||
|
||||
/// True when `event.created_at` is within the accepted clock skew — the
|
||||
/// SAME freshness window receivers enforce, so a gateway never spends
|
||||
/// airtime on events every receiver would drop as stale.
|
||||
private func isFresh(_ event: NostrEvent) -> Bool {
|
||||
abs(now().timeIntervalSince1970 - TimeInterval(event.created_at)) <= Limits.maxEventAgeSeconds
|
||||
}
|
||||
|
||||
static func isValidGeohash(_ geohash: String) -> Bool {
|
||||
let allowed = Set("0123456789bcdefghjkmnpqrstuvwxyz")
|
||||
return (1...NostrCarrierPacket.maxGeohashLength).contains(geohash.count)
|
||||
&& geohash.allSatisfy { allowed.contains($0) }
|
||||
}
|
||||
}
|
||||
|
||||
/// Insertion-ordered string set with a fixed capacity; the oldest entry is
|
||||
/// evicted when full.
|
||||
private struct BoundedIDSet {
|
||||
private var members: Set<String> = []
|
||||
private var order: [String] = []
|
||||
let capacity: Int
|
||||
|
||||
init(capacity: Int) {
|
||||
self.capacity = capacity
|
||||
}
|
||||
|
||||
func contains(_ id: String) -> Bool {
|
||||
members.contains(id)
|
||||
}
|
||||
|
||||
/// Returns false when the ID was already present.
|
||||
@discardableResult
|
||||
mutating func insert(_ id: String) -> Bool {
|
||||
guard members.insert(id).inserted else { return false }
|
||||
order.append(id)
|
||||
if order.count > capacity {
|
||||
members.remove(order.removeFirst())
|
||||
}
|
||||
return true
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user