mirror of
https://github.com/logos-messaging/logos-delivery.git
synced 2026-07-24 21:43:19 +00:00
feat: DoS guards for store sync - rate limit, session guards, bounded reads
The sync server had no protection (the old transfer guard was 'removed DOS prototection until we can design something better'): any peer could open unlimited reconciliation sessions, claim an int.high-sized payload with one length prefix, and push unsolicited transfer messages for free. - reconciliation: a RequestRateLimiter now fronts the server handler (new 'storesync' bucket in --rate-limit, default 30 sessions/5 min, the cheap check running before any processing); one active session per peer at a time, enforced on both the serving and initiating side; payload reads bounded to 64 MiB (large legitimate diffs are ~40 B per difference, so the largest honest rounds stay well under) - transfer: messages are only accepted from peers with a reconciliation session in the last 10 minutes; each session (re)opens that peer's window, unsolicited pushes are dropped, counted and disconnected. A stricter accept-only-granted-hashes model was tried and rejected: for an empty or far-behind receiver the sender computes the diff, so the receiver cannot enumerate the hashes it is owed - re-enables the transfer test disabled 'until we impl. DOS protection again'; adds session-guard, rate-limit and unsolicited-drop regression tests Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
darshankabariya
parent
1e8a710f69
commit
e2a34e365d
@ -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)
|
||||
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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)"
|
||||
|
||||
|
||||
@ -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
|
||||
|
||||
|
||||
@ -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))
|
||||
|
||||
@ -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)
|
||||
|
||||
|
||||
@ -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()
|
||||
|
||||
@ -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
|
||||
|
||||
@ -741,7 +741,7 @@ hence would have reachability issues.""",
|
||||
desc:
|
||||
"Rate limit settings for different protocols." &
|
||||
"Format: protocol:volume/period<unit>" &
|
||||
" Where 'protocol' can be one of: <store|storev3|lightpush|px|filter> if not defined it means a global setting" &
|
||||
" Where 'protocol' can be one of: <store|storev3|storesync|lightpush|px|filter> if not defined it means a global setting" &
|
||||
" 'volume' and period must be an integer value. " &
|
||||
" 'unit' must be one of <h|m|s|ms> - hours, minutes, seconds, milliseconds respectively. " &
|
||||
"Argument may be repeated.",
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user