diff --git a/logos_delivery/waku/common/rate_limit/setting.nim b/logos_delivery/waku/common/rate_limit/setting.nim index e3669ebd6..8684a385b 100644 --- a/logos_delivery/waku/common/rate_limit/setting.nim +++ b/logos_delivery/waku/common/rate_limit/setting.nim @@ -11,6 +11,7 @@ type RateLimitedProtocol* = enum LIGHTPUSH PEEREXCHG FILTER + STORESYNC type ProtocolRateLimitSettings* = Table[RateLimitedProtocol, RateLimitSetting] @@ -24,10 +25,19 @@ let UnlimitedRateLimit*: RateLimitSetting = (0, 0.seconds) # all subscribed peers let FilterDefaultPerPeerRateLimit*: RateLimitSetting = (30, 1.minutes) +# Store sync sessions are heavier than single queries: each one runs a full +# range diff and may trigger message transfer. Honest peers initiate at most +# one session per sync interval (a minute or more), so this leaves generous +# headroom while bounding what a hostile peer can demand for free. +let StoreSyncDefaultRateLimit*: RateLimitSetting = (30, 5.minutes) + # For being used under GC-safe condition must use threadvar var DefaultProtocolRateLimit* {.threadvar.}: ProtocolRateLimitSettings -DefaultProtocolRateLimit = - {GLOBAL: UnlimitedRateLimit, FILTER: FilterDefaultPerPeerRateLimit}.toTable() +DefaultProtocolRateLimit = { + GLOBAL: UnlimitedRateLimit, + FILTER: FilterDefaultPerPeerRateLimit, + STORESYNC: StoreSyncDefaultRateLimit, +}.toTable() proc isUnlimited*(t: RateLimitSetting): bool {.inline.} = return t.volume <= 0 or t.period <= 0.seconds @@ -54,6 +64,8 @@ proc translate(sProtocol: string): RateLimitedProtocol {.raises: [ValueError].} return PEEREXCHG of "filter": return FILTER + of "storesync": + return STORESYNC else: raise newException(ValueError, "Unknown protocol definition: " & sProtocol) @@ -83,7 +95,7 @@ proc parse*( ## group4: Unit of period - only h:hour, m:minute, s:second, ms:millisecond allowed ## whitespaces are allowed lazily const parseRegex = - """^\s*((store|storev3|lightpush|px|filter)\s*:)?\s*(\d+)\s*\/\s*(\d+)\s*(s|h|m|ms)\s*$""" + """^\s*((storesync|store|storev3|lightpush|px|filter)\s*:)?\s*(\d+)\s*\/\s*(\d+)\s*(s|h|m|ms)\s*$""" const regexParseSize = re2(parseRegex) for settingStr in settings: let aSetting = settingStr.toLower() @@ -116,6 +128,7 @@ proc parse*( # due it is taken for protocols not defined in the list - thus those will not apply accidentally wrong settings. discard settingsTable.hasKeyOrPut(GLOBAL, UnlimitedRateLimit) discard settingsTable.hasKeyOrPut(FILTER, FilterDefaultPerPeerRateLimit) + discard settingsTable.hasKeyOrPut(STORESYNC, StoreSyncDefaultRateLimit) return ok(settingsTable) diff --git a/logos_delivery/waku/node/waku_node.nim b/logos_delivery/waku/node/waku_node.nim index 3b10d5b7b..21de0f360 100644 --- a/logos_delivery/waku/node/waku_node.nim +++ b/logos_delivery/waku/node/waku_node.nim @@ -376,7 +376,7 @@ proc mountStoreSync*( return err("store sync requires a mounted archive") let idsChannel = newAsyncQueue[(SyncID, PubsubTopic, ContentTopic)](0) - let wantsChannel = newAsyncQueue[(PeerId)](0) + let wantsChannel = newAsyncQueue[PeerId](0) let needsChannel = newAsyncQueue[(PeerId, WakuMessageHash)](0) let pubsubTopics = shards.mapIt($RelayShard(clusterId: cluster, shardId: it)) @@ -384,9 +384,17 @@ proc mountStoreSync*( node.storeSyncRange = storeSyncRange.seconds let recon = ?await SyncReconciliation.new( - pubsubTopics, contentTopics, node.peerManager, node.wakuArchive, - storeSyncRange.seconds, storeSyncInterval.seconds, storeSyncRelayJitter.seconds, - idsChannel, wantsChannel, needsChannel, + pubsubTopics, + contentTopics, + node.peerManager, + node.wakuArchive, + storeSyncRange.seconds, + storeSyncInterval.seconds, + storeSyncRelayJitter.seconds, + idsChannel, + wantsChannel, + needsChannel, + rateLimitSetting = Opt.some(node.rateLimitSettings.getSetting(STORESYNC)), ) node.wakuStoreReconciliation = recon diff --git a/logos_delivery/waku/waku_store_sync/protocols_metrics.nim b/logos_delivery/waku/waku_store_sync/protocols_metrics.nim index f76db8b47..ae4ca8bc5 100644 --- a/logos_delivery/waku/waku_store_sync/protocols_metrics.nim +++ b/logos_delivery/waku/waku_store_sync/protocols_metrics.nim @@ -20,6 +20,9 @@ declarePublicCounter total_bytes_exchanged, declarePublicCounter total_transfer_messages_rejected, "number of received transfer messages dropped by validation" +declarePublicCounter total_transfer_unsolicited_disconnects, + "number of transfer connections dropped because the peer had no recent reconciliation session" + declarePublicCounter total_transfer_messages_unverified, "number of received transfer messages accepted without proof verification (archives do not persist RLN proofs)" diff --git a/logos_delivery/waku/waku_store_sync/reconciliation.nim b/logos_delivery/waku/waku_store_sync/reconciliation.nim index d6731f355..3fad2a614 100644 --- a/logos_delivery/waku/waku_store_sync/reconciliation.nim +++ b/logos_delivery/waku/waku_store_sync/reconciliation.nim @@ -16,6 +16,7 @@ import ../common/nimchronos, ../common/protobuf, ../common/paging, + ../common/rate_limit/request_limiter, ../waku_enr, ../waku_core/codecs, ../waku_core/time, @@ -37,6 +38,14 @@ logScope: const DefaultStorageCap = 50_000 +# Reconciliation payloads grow with the number of differences: the range +# fan-out plus item-set elements come to roughly 40 B per difference, so a +# round reconciling a 300k-message diff is on the order of 12 MiB. 64 MiB +# clears the largest legitimate rounds the RLN-capped window can produce +# while still bounding what one length prefix can make us allocate +# (readLp(int.high) would accept an arbitrary claim). +const MaxSyncPayloadSize = 64 * 1024 * 1024 + type SyncReconciliation* = ref object of LPProtocol pubsubTopics: HashSet[PubsubTopic] # Empty set means accept all. See spec. contentTopics: HashSet[ContentTopic] # Empty set means accept all. See spec. @@ -50,12 +59,19 @@ type SyncReconciliation* = ref object of LPProtocol # Receive IDs from transfer protocol for storage idsRx: AsyncQueue[(SyncID, PubsubTopic, ContentTopic)] - # Send Hashes to transfer protocol for reception - localWantsTx: AsyncQueue[(PeerId)] + # Signal to the transfer protocol which peers have a sync session, + # opening a bounded window in which their transfers are accepted + localWantsTx: AsyncQueue[PeerId] # Send Hashes to transfer protocol for transmission remoteNeedsTx: AsyncQueue[(PeerId, WakuMessageHash)] + # DoS protection: bound how often peers can start sync sessions + requestRateLimiter: RequestRateLimiter + + # Peers with a sync session in flight; one session per peer at a time + activeSessions: HashSet[PeerId] + # params syncInterval: timer.Duration # Time between each synchronization attempt syncRange: timer.Duration # Amount of time in the past to sync @@ -185,12 +201,14 @@ proc processRequest( roundTrips = 0 diffs = 0 - # Signal to transfer protocol that this reconciliation is starting - await self.localWantsTx.addLast(conn.peerId) + # Open the transfer window for this peer: messages found missing during + # this session will be pushed by the peer over the transfer protocol, + # possibly well after the session itself ends. + self.localWantsTx.addLastNoWait(conn.peerId) while true: let readRes = catch: - await conn.readLp(int.high) + await conn.readLp(MaxSyncPayloadSize) let buffer: seq[byte] = readRes.valueOr: await conn.close() @@ -264,9 +282,6 @@ proc processRequest( continue - # Signal to transfer protocol that this reconciliation is done - await self.localWantsTx.addLast(conn.peerId) - reconciliation_roundtrips.observe(roundTrips) reconciliation_differences.observe(diffs) @@ -330,6 +345,12 @@ proc storeSynchronization*( self.peerManager.selectPeer(WakuReconciliationCodec).valueOr: return err("no suitable peer found for sync") + if self.activeSessions.containsOrIncl(peer.peerId): + return err("sync session already in progress with peer " & $peer.peerId) + + defer: + self.activeSessions.excl(peer.peerId) + let connOpt = await self.peerManager.dialPeer(peer, WakuReconciliationCodec) let conn: Connection = connOpt.valueOr: @@ -409,6 +430,7 @@ proc new*( idsRx: AsyncQueue[(SyncID, PubsubTopic, ContentTopic)], localWantsTx: AsyncQueue[PeerId], remoteNeedsTx: AsyncQueue[(PeerId, WakuMessageHash)], + rateLimitSetting: Opt[RateLimitSetting] = Opt.none(RateLimitSetting), ): Future[Result[T, string]] {.async.} = let res = await initFillStorage(syncRange, wakuArchive) let storage = @@ -429,18 +451,34 @@ proc new*( idsRx: idsRx, localWantsTx: localWantsTx, remoteNeedsTx: remoteNeedsTx, + requestRateLimiter: newRequestRateLimiter(rateLimitSetting), ) proc handler(conn: Connection, proto: string) {.async: (raises: [CancelledError]).} = - try: - (await sync.processRequest(conn)).isOkOr: - error "request processing error", error = error - except CatchableError: - error "exception in reconciliation handler", error = getCurrentExceptionMsg() + sync.requestRateLimiter.checkUsageLimit(WakuReconciliationCodec, conn): + if sync.activeSessions.containsOrIncl(conn.peerId): + info "rejecting sync request: session already in progress", remote = conn.peerId + await conn.close() + return + + defer: + sync.activeSessions.excl(conn.peerId) + + try: + (await sync.processRequest(conn)).isOkOr: + error "request processing error", error = error + except CatchableError: + error "exception in reconciliation handler", error = getCurrentExceptionMsg() + do: + info "sync request rejected due rate limit exceeded", + remote = conn.peerId, limit = $sync.requestRateLimiter.setting + await conn.close() sync.handler = handler sync.codec = WakuReconciliationCodec + setServiceLimitMetric(WakuReconciliationCodec, rateLimitSetting) + info "Store Reconciliation protocol initialized", sync_range = syncRange, sync_interval = syncInterval, relay_jitter = relayJitter diff --git a/logos_delivery/waku/waku_store_sync/transfer.nim b/logos_delivery/waku/waku_store_sync/transfer.nim index caa7b92a5..8b1f3fd34 100644 --- a/logos_delivery/waku/waku_store_sync/transfer.nim +++ b/logos_delivery/waku/waku_store_sync/transfer.nim @@ -1,7 +1,7 @@ {.push raises: [].} import - std/[sets, tables], + std/tables, results, chronicles, chronos, @@ -31,6 +31,12 @@ import logScope: topics = "waku transfer" +# How long after a reconciliation session a peer's transfer messages are +# still accepted. Pushes can lag the session (large backlogs are sent one +# message at a time) and periodic sync refreshes the window every interval, +# so this only needs to cover one catch-up burst. +const TransferAllowancePeriod = 10.minutes + type TransferValidator* = proc(msg: WakuMessage): Future[bool] {.gcsafe, raises: [].} ## Returns false when a received transfer message must be dropped ## (e.g. failed RLN proof verification). @@ -43,10 +49,14 @@ type SyncTransfer* = ref object of LPProtocol # Send IDs to reconciliation protocol for storage idsTx: AsyncQueue[(SyncID, PubsubTopic, ContentTopic)] - # Receive Hashes from reconciliation protocol for reception + # Receive session signals from the reconciliation protocol localWantsRx: AsyncQueue[PeerId] localWantsRxFut: Future[void] - inSessions: HashSet[PeerId] + + # DoS protection: peers with a recent reconciliation session and the + # deadline until which their transfer messages are accepted; anything + # from a peer outside its window is unsolicited. + allowedPeers: Table[PeerId, Moment] # Receive Hashes from reconciliation protocol for transmission remoteNeedsRx: AsyncQueue[(PeerId, WakuMessageHash)] @@ -84,14 +94,13 @@ proc openConnection( return ok(conn) proc wantsReceiverLoop(self: SyncTransfer) {.async.} = - ## Waits for peer ids of nodes - ## we are reconciliating with + ## Waits for peer ids of nodes we are reconciliating with; + ## each session (re)opens that peer's transfer window. while true: # infinite loop let peerId = await self.localWantsRx.popFirst() - if self.inSessions.containsOrIncl(peerId): - self.inSessions.excl(peerId) + self.allowedPeers[peerId] = Moment.now() + TransferAllowancePeriod return @@ -149,10 +158,15 @@ proc needsReceiverLoop(self: SyncTransfer) {.async.} = proc initProtocolHandler(self: SyncTransfer) = proc handler(conn: Connection, proto: string) {.async: (raises: [CancelledError]).} = while true: - ## removed DOS prototection until we can design something better - #[ if not self.inSessions.contains(conn.peerId): - error "unwanted peer, disconnecting", remote = conn.peerId - break ]# + ## DoS protection: only accept messages from peers we reconciled + ## with recently. The session signal is queued before any diff is + ## computed, so it always precedes the peer's first push. + let deadline = self.allowedPeers.getOrDefault(conn.peerId) + if deadline == default(Moment) or Moment.now() > deadline: + self.allowedPeers.del(conn.peerId) + total_transfer_unsolicited_disconnects.inc() + warn "unsolicited transfer, disconnecting", remote_peer_id = conn.peerId + break let readRes = catch: await conn.readLp(int64(DefaultMaxWakuMessageSize)) diff --git a/tests/common/test_ratelimit_setting.nim b/tests/common/test_ratelimit_setting.nim index 0add4b76f..1efff9127 100644 --- a/tests/common/test_ratelimit_setting.nim +++ b/tests/common/test_ratelimit_setting.nim @@ -47,23 +47,46 @@ suite "RateLimitSetting": let res4 = ProtocolRateLimitSettings.parse(@[test4]) let res5 = ProtocolRateLimitSettings.parse(@[test5]) + let expSync = StoreSyncDefaultRateLimit + check: res1.isOk() - res1.get() == {GLOBAL: exp1, FILTER: FilterDefaultPerPeerRateLimit}.toTable() + res1.get() == + {GLOBAL: exp1, FILTER: FilterDefaultPerPeerRateLimit, STORESYNC: expSync}.toTable() res2.isOk() res2.get() == - {GLOBAL: expU, FILTER: FilterDefaultPerPeerRateLimit, STOREV3: exp2}.toTable() + { + GLOBAL: expU, + FILTER: FilterDefaultPerPeerRateLimit, + STORESYNC: expSync, + STOREV3: exp2, + }.toTable() res2b.isOk() res2b.get() == - {GLOBAL: expU, FILTER: FilterDefaultPerPeerRateLimit, STOREV3: exp2b}.toTable() + { + GLOBAL: expU, + FILTER: FilterDefaultPerPeerRateLimit, + STORESYNC: expSync, + STOREV3: exp2b, + }.toTable() res3.isOk() res3.get() == - {GLOBAL: expU, FILTER: FilterDefaultPerPeerRateLimit, LIGHTPUSH: exp3}.toTable() + { + GLOBAL: expU, + FILTER: FilterDefaultPerPeerRateLimit, + STORESYNC: expSync, + LIGHTPUSH: exp3, + }.toTable() res4.isOk() res4.get() == - {GLOBAL: expU, FILTER: FilterDefaultPerPeerRateLimit, PEEREXCHG: exp4}.toTable() + { + GLOBAL: expU, + FILTER: FilterDefaultPerPeerRateLimit, + STORESYNC: expSync, + PEEREXCHG: exp4, + }.toTable() res5.isOk() - res5.get() == {GLOBAL: expU, FILTER: exp5}.toTable() + res5.get() == {GLOBAL: expU, FILTER: exp5, STORESYNC: expSync}.toTable() test "Parse rate limit setting - err": let test1 = "10/2d" @@ -95,6 +118,7 @@ suite "RateLimitSetting": let exp1 = { GLOBAL: (10, 2.minutes), FILTER: FilterDefaultPerPeerRateLimit, + STORESYNC: StoreSyncDefaultRateLimit, LIGHTPUSH: (2, 2.milliseconds), STOREV3: (3, 3.seconds), }.toTable() @@ -115,6 +139,7 @@ suite "RateLimitSetting": STOREV3: (3, 3.seconds), FILTER: (4, 42.milliseconds), PEEREXCHG: (10, 10.hours), + STORESYNC: StoreSyncDefaultRateLimit, }.toTable() let res2 = ProtocolRateLimitSettings.parse(test2) @@ -125,7 +150,10 @@ suite "RateLimitSetting": let test3 = @["store:3/3s", "storev3:4/42ms", "storev3:5/5s", "storev3:6/6s"] let exp3 = { - GLOBAL: expU, FILTER: FilterDefaultPerPeerRateLimit, STOREV3: (6, 6.seconds) + GLOBAL: expU, + FILTER: FilterDefaultPerPeerRateLimit, + STORESYNC: StoreSyncDefaultRateLimit, + STOREV3: (6, 6.seconds), }.toTable() let res3 = ProtocolRateLimitSettings.parse(test3) @@ -136,7 +164,11 @@ suite "RateLimitSetting": res3.get().getSetting(LIGHTPUSH) == expU let test4 = newSeq[string](0) - let exp4 = {GLOBAL: expU, FILTER: FilterDefaultPerPeerRateLimit}.toTable() + let exp4 = { + GLOBAL: expU, + FILTER: FilterDefaultPerPeerRateLimit, + STORESYNC: StoreSyncDefaultRateLimit, + }.toTable() let res4 = ProtocolRateLimitSettings.parse(test4) diff --git a/tests/waku_store_sync/sync_utils.nim b/tests/waku_store_sync/sync_utils.nim index 21d75dd97..1291ad23c 100644 --- a/tests/waku_store_sync/sync_utils.nim +++ b/tests/waku_store_sync/sync_utils.nim @@ -1,4 +1,4 @@ -import std/random, chronos, chronicles +import std/random, results, chronos, chronicles import logos_delivery/waku/[ @@ -7,6 +7,7 @@ import waku_store_sync/common, waku_store_sync/reconciliation, waku_store_sync/transfer, + common/rate_limit/setting, ], ../testlib/wakucore @@ -28,6 +29,7 @@ proc newTestWakuRecon*( idsRx: AsyncQueue[(SyncID, PubsubTopic, ContentTopic)], wantsTx: AsyncQueue[PeerId], needsTx: AsyncQueue[(PeerId, WakuMessageHash)], + rateLimitSetting: Opt[RateLimitSetting] = Opt.none(RateLimitSetting), ): Future[SyncReconciliation] {.async.} = let peerManager = PeerManager.new(switch) @@ -41,6 +43,7 @@ proc newTestWakuRecon*( idsRx = idsRx, localWantsTx = wantsTx, remoteNeedsTx = needsTx, + rateLimitSetting = rateLimitSetting, ) let proto = res.get() diff --git a/tests/waku_store_sync/test_protocol.nim b/tests/waku_store_sync/test_protocol.nim index 50d8b4bf5..2b0f6d3c8 100644 --- a/tests/waku_store_sync/test_protocol.nim +++ b/tests/waku_store_sync/test_protocol.nim @@ -21,6 +21,7 @@ import waku_archive/archive, waku_archive/driver, waku_archive/common, + common/rate_limit/setting, ], ../testlib/[wakucore, testasync], ../waku_archive/archive_utils, @@ -846,6 +847,56 @@ suite "Waku Sync: reconciliation": check needsQ.len == 0 + asyncTest "one sync session per peer at a time": + server = await newTestWakuRecon( + serverSwitch, @[], @[], DefaultSyncRange, idsChannel, localWants, remoteNeeds + ) + client = await newTestWakuRecon( + clientSwitch, @[], @[], DefaultSyncRange, idsChannel, localWants, remoteNeeds + ) + + # start two sessions towards the same peer concurrently: the session + # guard must reject the second while the first is still in flight + let fut1 = client.storeSynchronization(Opt.some(serverPeerInfo)) + let fut2 = client.storeSynchronization(Opt.some(serverPeerInfo)) + + let res1 = await fut1 + let res2 = await fut2 + + check: + res1.isOk() + res2.isErr() + + # once the first session is done, a new one is accepted again + let res3 = await client.storeSynchronization(Opt.some(serverPeerInfo)) + check res3.isOk() + + asyncTest "sync requests beyond the rate limit are rejected": + let limit: RateLimitSetting = (1, 1.minutes) + + server = await newTestWakuRecon( + serverSwitch, + @[], + @[], + DefaultSyncRange, + idsChannel, + localWants, + remoteNeeds, + rateLimitSetting = Opt.some(limit), + ) + client = await newTestWakuRecon( + clientSwitch, @[], @[], DefaultSyncRange, idsChannel, localWants, remoteNeeds + ) + + # the first session fits the budget and is served + let res1 = await client.storeSynchronization(Opt.some(serverPeerInfo)) + assert res1.isOk(), $res1.error + + # an immediate second session exceeds the server's budget: the server + # closes the connection without processing and the sync fails + let res2 = await client.storeSynchronization(Opt.some(serverPeerInfo)) + check res2.isErr() + suite "Waku Sync: transfer": var serverSwitch {.threadvar.}: Switch @@ -938,7 +989,7 @@ suite "Waku Sync: transfer": serverDriver = serverDriver.put(DefaultPubsubTopic, msgs) - # add server info to client want channel + # signal an active sync session with the server, opening its transfer window let want = serverPeerInfo.peerId await clientLocalWants.put(want) @@ -961,8 +1012,7 @@ suite "Waku Sync: transfer": check: response.messages.len > 0 - ## Disabled until we impl. DOS protection again - #[ asyncTest "Check the exact missing messages are received": + asyncTest "Check the exact missing messages are received": let timeSlice = calculateTimeRange() let timeWindow = int64(timeSlice.b) - int64(timeSlice.a) let (part, _) = divmod(timeWindow, 3) @@ -992,9 +1042,32 @@ suite "Waku Sync: transfer": let sid1 = await clientIds.get() let sid2 = await clientIds.get() - let received = [sid1.hash, sid2.hash].toHashSet() - let expected = [hB, hC].toHashSet + let received = [sid1[0].hash, sid2[0].hash].toHashSet() + let expected = [hB, hC].toHashSet() check received == expected - check clientIds.len == 0 ]# + check clientIds.len == 0 + + asyncTest "unsolicited transfer messages are dropped": + let msg = fakeWakuMessage() + let hash = computeMessageHash(DefaultPubsubTopic, msg) + + serverDriver = serverDriver.put(DefaultPubsubTopic, @[msg]) + + # no budget granted on the client side: reconciliation never + # determined we need anything from the server + let need = (clientPeerInfo.peerId, hash) + await serverRemoteNeeds.put(need) + + # give time for the (rejected) transfer attempt to happen + await sleepAsync(500.milliseconds) + + var query = ArchiveQuery() + query.includeData = true + query.hashes = @[hash] + + let res = await clientArchive.findMessages(query) + assert res.isOk(), $res.error + + check res.get().messages.len == 0 diff --git a/tools/confutils/cli_args.nim b/tools/confutils/cli_args.nim index 7181d0ff7..ee7f513a2 100644 --- a/tools/confutils/cli_args.nim +++ b/tools/confutils/cli_args.nim @@ -741,7 +741,7 @@ hence would have reachability issues.""", desc: "Rate limit settings for different protocols." & "Format: protocol:volume/period" & - " Where 'protocol' can be one of: if not defined it means a global setting" & + " Where 'protocol' can be one of: if not defined it means a global setting" & " 'volume' and period must be an integer value. " & " 'unit' must be one of - hours, minutes, seconds, milliseconds respectively. " & "Argument may be repeated.",