@@ -73,6 +73,7 @@ clockSkewTolerance : Int<Ms>
7373clockSkewTolerance = 5000 -- 5 seconds tolerance for clock skew between federated servers
7474maxForwardsEntries = 10000
7575maxForwardsPerKey = 50 -- max pending forwards per recipient pubkey (prevents one key from starving others)
76+maxForwardsPerSenderPerKey = 10 -- max pending forwards per sender-recipient pair (prevents one sender from monopolizing a recipient's queue)
7677maxForwardsPerServer = 2000 -- max pending forwards per destination server (prevents one unreachable peer from starving others)
7778maxServerNameLen = 128
7879maxMessageIdLen = 128
@@ -212,6 +213,12 @@ maxGossipEventsPerServer = 500 -- max total gossip events per server per window
212213gossipEventRateWindowMs : Int<Ms>
213214gossipEventRateWindowMs = 60000 -- 1 minute
214215maxGossipEventRateEntries = 100
216+-- Gossip batch backoff: when a single gossip request carries more than this many fresh events,
217+-- delay re-broadcast by a random interval up to gossipBatchBackoffMs. Prevents amplification
218+-- storms when peers recover from downtime and send bursts of presence updates.
219+gossipBatchBackoffThreshold = 50
220+gossipBatchBackoffMs : Int<Ms>
221+gossipBatchBackoffMs = 5000 -- max random delay before re-gossiping large batches
215222-- Global federation connection cap (bounds total concurrent outbound connections across all paths)
216223maxGlobalFederationConcurrency = 30
217224-- Peer blocklist: max blocked servers tracked at runtime (managed via /maintenance/block, /maintenance/unblock)
@@ -256,6 +263,11 @@ peerKeyTtlMs : Int<Ms>
256263peerKeyTtlMs = 86400000 -- 24 hours; peer keys change rarely (server restart or rotation)
257264peerKeyRefetchCooldownMs : Int<Ms>
258265peerKeyRefetchCooldownMs = 60000 -- 1 minute; minimum interval between re-fetch attempts per server
266+-- Require ClientIp header on auth requests. When True, challenge requests with
267+-- empty clientIp are rejected. Prevents unbound sessions if the reverse proxy is
268+-- temporarily misconfigured (no ClientIp header forwarding). Sessions created without
269+-- IP binding permanently accept requests from any IP.
270+requireClientIp = True {}
259271-- IP-based rate limiting (reverse proxy sets ClientIp header)
260272maxIpRatePerIp = 10 -- max auth requests per IP per window
261273ipRateWindowMs : Int<Ms>
@@ -827,6 +839,12 @@ maxPeerLatencyEntries = 100
827839*nonceBucket14 : [ForwardNonce]
828840*nonceBucket15 : [ForwardNonce]
829841
842+-- Nonce eviction replay protection: tracks the seenAt of the oldest retained nonce
843+-- after LRU eviction. Forwards with timestamps older than this cutoff cannot be
844+-- verified for dedup (their nonces may have been evicted) and are rejected.
845+-- 0 = no eviction has occurred (all nonces retained), accept all timestamps.
846+*nonceEvictionCutoff : Timestamp
847+
830848-- Federation verify per-IP rate limiting (tighter than general IP rate)
831849*verifyIpRates : [IpRateEntry]
832850
@@ -977,7 +995,7 @@ route AuthApi where
977995route ClientApi where
978996 -- Long-poll for new messages; cursor tracks last-seen receivedAt timestamp
979997 POST {cursor: Timestamp, generation: Int, watchKeys: [PubkeyHex]} /poll headers {authorization: Text, clientIp: Text}
980− -> {events: [PollEvent], cursor: Timestamp, authorized: Bool, rateLimited: Bool, draining: Bool, generation: Int, presenceUpdates: [{pubkey: PubkeyHex, online: Bool}], heartbeat: Bool, retryAfterMs: Int, cursorReset: Bool, hasMore: Bool, cursorStale: Bool, messageExpirations: [{messageId: MessageIdHex, toKey: PubkeyHex, expiredAt: Timestamp}], messageDeliveries: [{messageId: MessageIdHex, toKey: PubkeyHex, deliveredAt: Timestamp}], droppedDeliveryReceipts: Int, droppedExpirationNotifs: Int}
998+ -> {events: [PollEvent], cursor: Timestamp, authorized: Bool, rateLimited: Bool, draining: Bool, generation: Int, presenceUpdates: [{pubkey: PubkeyHex, online: Bool}], heartbeat: Bool, retryAfterMs: Int, cursorReset: Bool, hasMore: Bool, cursorStale: Bool, messageExpirations: [{messageId: MessageIdHex, toKey: PubkeyHex, expiredAt: Timestamp}], messageDeliveries: [{messageId: MessageIdHex, toKey: PubkeyHex, deliveredAt: Timestamp}], droppedDeliveryReceipts: Int, droppedExpirationNotifs: Int, loadPct: Int}
981999 = Poll
9821000 /messages
9831001 -- Send an E2E encrypted message to a recipient pubkey.
@@ -986,7 +1004,7 @@ route ClientApi where
9861004 -- priority: optional urgency level (0 = normal, higher = more urgent, clamped to [0, maxMessagePriority]).
9871005 -- High-priority messages are forwarded first during federation retries.
9881006 POST {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: OptionalMessageId, expiresInMs: Int, priority: Int} / headers {authorization: Text, clientIp: Text}
989− -> {status: MessageStatus, messageId: OptionalMessageId, retryAfterMs: Int, error: Text}
1007+ -> {status: MessageStatus, messageId: OptionalMessageId, retryAfterMs: Int, error: Text, loadPct: Int}
9901008 = SendMessage
9911009 -- Check delivery status of a sent message (caller must know recipient to prove ownership)
9921010 POST {to: PubkeyHex, messageId: MessageIdHex} /status headers {authorization: Text, clientIp: Text}
@@ -2743,6 +2761,14 @@ updateForwardNonce = \nb messageId toKey t -> do
27432761 then case minBy evictedNonces (\n -> n.seenAt) of Nothing {} -> 0; Just {value: n} -> t - n.seenAt
27442762 else 0
27452763 writeNonceBucket nb evictedNonces
2764+ -- Update nonce eviction cutoff: the seenAt of the oldest retained nonce.
2765+ -- Forwards with timestamps older than this cannot be verified for dedup.
2766+ when (lruEvicted > 0) (do
2767+ let oldestRetainedSeenAt = case minBy evictedNonces (\n -> n.seenAt) of Nothing {} -> t; Just {value: n} -> n.seenAt
2768+ cutoff <- *nonceEvictionCutoff
2769+ -- Only advance the cutoff (never regress): a bucket with older retained nonces
2770+ -- shouldn't lower the cutoff set by a bucket with newer evictions.
2771+ when (oldestRetainedSeenAt > cutoff) (set *nonceEvictionCutoff = oldestRetainedSeenAt))
27462772 yield {lruEvicted: lruEvicted, oldestRetainedAge: oldestRetainedAge}
27472773
27482774-- Log nonce LRU eviction if any occurred. Call outside atomic blocks.
@@ -4564,13 +4590,15 @@ forwardQueuedMessages = \targets t -> do
45644590 relevantMessages <- readSelectedBuckets neededBuckets
45654591 let candidates = forwardCandidates remoteTargets relevantMessages forwards t
45664592 totalFwdCount <- *totalForwardCount
4567− -- Per-key and per-server caps: filter candidates to respect both limits
4593+ -- Per-key, per-server, and per-sender-per-key caps: filter candidates to respect all limits.
4594+ -- Per-sender cap prevents a single sender from monopolizing a recipient's forward queue.
45684595 let perKeyCapped = fold (\acc c ->
45694596 let existingForKey = count (do f <- forwards; where f.toKey == c.toKey; yield f)
45704597 + count (do a <- acc; where a.toKey == c.toKey; yield a)
45714598 in let existingForServer = count (do f <- forwards; where f.toServer == c.toServer; yield f)
45724599 + count (do a <- acc; where a.toServer == c.toServer; yield a)
4573− in if existingForKey >= maxForwardsPerKey || existingForServer >= maxForwardsPerServer then acc else union acc [c]) [] candidates
4600+ in let existingForSenderKey = if c.senderKey != "" then count (do f <- forwards; where f.senderKey == c.senderKey && f.toKey == c.toKey; yield f) + count (do a <- acc; where a.senderKey == c.senderKey && a.toKey == c.toKey; yield a) else 0
4601+ in if existingForKey >= maxForwardsPerKey || existingForServer >= maxForwardsPerServer || (c.senderKey != "" && existingForSenderKey >= maxForwardsPerSenderPerKey) then acc else union acc [c]) [] candidates
45744602 let capped = takeRelation (maxForwardsEntries - totalFwdCount) perKeyCapped
45754603 -- Write back only the affected forward buckets (not all 16)
45764604 let combined = union forwards capped
@@ -5426,6 +5454,13 @@ remoteOnlyTargets = \targets localOnline ->
54265454-- Rate-limited per key; globally capped. Challenge expires in 1 minute.
54275455handleChallenge = \pubkey clientIp respond -> do
54285456 t <- now
5457+ -- Reject challenges without ClientIp when requireClientIp is enabled.
5458+ -- Prevents creating sessions with no IP binding (permanently accept from any IP).
5459+ if requireClientIp && clientIp == ""
5460+ then do
5461+ logEvent "warn" "missing_client_ip" ("endpoint=challenge key=" ++ pubkey)
5462+ yield (respond {challenge: "", ok: False {}, expiresAt: 0, retryAfterMs: 0, error: "client_ip_required"})
5463+ else do
54295464 preAuth <- checkPreAuth clientIp t
54305465 if preAuth != ""
54315466 then do
@@ -5613,10 +5648,10 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
56135648 -- Cursor staleness: cursor is older than message retention, so messages were missed.
56145649 -- Different from cursorReset (server restart): staleness means client was offline too long.
56155650 let cursorStale = effectiveCursor > 0 && t - effectiveCursor > unackedMessageMaxAge
5616− if dp >= 2 then yield (respond {events: [], cursor: 0, authorized: True {}, rateLimited: False {}, draining: True {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: 0, cursorReset: cursorWasReset, hasMore: False {}, cursorStale: False {}, messageExpirations: [], messageDeliveries: [], droppedDeliveryReceipts: 0, droppedExpirationNotifs: 0})
5651+ if dp >= 2 then yield (respond {events: [], cursor: 0, authorized: True {}, rateLimited: False {}, draining: True {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: 0, cursorReset: cursorWasReset, hasMore: False {}, cursorStale: False {}, messageExpirations: [], messageDeliveries: [], droppedDeliveryReceipts: 0, droppedExpirationNotifs: 0, loadPct: 0})
56175652 else do
56185653 globalRateLimited <- checkGlobalRate t
5619− if globalRateLimited then yield (respond {events: [], cursor: 0, authorized: False {}, rateLimited: True {}, draining: False {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: globalRateWindowMs, cursorReset: cursorWasReset, hasMore: False {}, cursorStale: False {}, messageExpirations: [], messageDeliveries: [], droppedDeliveryReceipts: 0, droppedExpirationNotifs: 0})
5654+ if globalRateLimited then yield (respond {events: [], cursor: 0, authorized: False {}, rateLimited: True {}, draining: False {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: globalRateWindowMs, cursorReset: cursorWasReset, hasMore: False {}, cursorStale: False {}, messageExpirations: [], messageDeliveries: [], droppedDeliveryReceipts: 0, droppedExpirationNotifs: 0, loadPct: 0})
56205655 else do
56215656 -- Capture presenceVersion before long-poll wait to skip unchanged lookups
56225657 startPresenceVersion <- *presenceVersion
@@ -5678,7 +5713,10 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
56785713 let deResult = drainDropCounter den result.pubkey
56795714 when (deResult.count > 0) (set *droppedExpirationNotifs = deResult.remaining)
56805715 yield {dr: drResult.count, de: deResult.count}
5681− yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized, rateLimited: result.pollSlotFull, draining: False {}, generation: serverGen, presenceUpdates: presenceUpdates, heartbeat: isHeartbeat, retryAfterMs: pollRetryMs, cursorReset: cursorWasReset, hasMore: result.hasMore, cursorStale: cursorStale, messageExpirations: expirations, messageDeliveries: deliveries, droppedDeliveryReceipts: drops.dr, droppedExpirationNotifs: drops.de})
5716+ -- Expose server load to clients: 0 = healthy (<=80% capacity), up to 90 = near-full.
5717+ -- Clients can voluntarily reduce poll frequency or batch sends when loadPct > 0.
5718+ loadPct <- do scale <- *cachedAdaptiveScale; yield (if scale >= 100 then 0 else 100 - scale)
5719+ yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized, rateLimited: result.pollSlotFull, draining: False {}, generation: serverGen, presenceUpdates: presenceUpdates, heartbeat: isHeartbeat, retryAfterMs: pollRetryMs, cursorReset: cursorWasReset, hasMore: result.hasMore, cursorStale: cursorStale, messageExpirations: expirations, messageDeliveries: deliveries, droppedDeliveryReceipts: drops.dr, droppedExpirationNotifs: drops.de, loadPct: loadPct})
56825720
56835721-- Core send logic: dedup, rate limiting, store message, create forwards.
56845722-- Called inside an atomic block by both handleSendMessage and sendBatchInner.
@@ -5754,7 +5792,12 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId expiresIn
57545792 -- Per-key cap: prevents one key from consuming disproportionate forward slots
57555793 let keyFwdCount = count (do f <- fwdBkt; where f.toKey == to; yield f)
57565794 let keyCap = maxForwardsPerKey - keyFwdCount
5757− let cappedFwds = takeRelation (min globalCap keyCap) newFwds
5795+ -- Per-sender-per-recipient cap: prevents a single sender from monopolizing
5796+ -- a recipient's forward queue, ensuring at least (maxForwardsPerKey /
5797+ -- maxForwardsPerSenderPerKey) distinct senders can queue forwards.
5798+ let senderKeyFwdCount = if senderKey != "" then count (do f <- fwdBkt; where f.senderKey == senderKey && f.toKey == to; yield f) else 0
5799+ let senderKeyCap = if senderKey != "" && senderKeyFwdCount >= maxForwardsPerSenderPerKey then 0 else min globalCap keyCap
5800+ let cappedFwds = takeRelation senderKeyCap newFwds
57585801 writeFwdBucket fbi (union fwdBkt cappedFwds)
57595802 set *totalForwardCount = totalFwdCount + count cappedFwds
57605803 -- Return whether any forwards were actually enqueued (vs dropped by capacity cap)
@@ -5891,14 +5934,14 @@ handleSendMessage = \to encryptedBlob timestamp messageId expiresInMs priority a
58915934 let clampedPriority = clamp 0 maxMessagePriority priority
58925935 t <- now
58935936 dp <- *drainPhase
5894− if dp >= 2 then yield (respond {status: MsDraining {}, messageId: "", retryAfterMs: 0, error: "draining"})
5937+ if dp >= 2 then yield (respond {status: MsDraining {}, messageId: "", retryAfterMs: 0, error: "draining", loadPct: 0})
58955938 else do
58965939 globalRateLimited <- checkGlobalRate t
5897− if globalRateLimited then yield (respond {status: MsRateLimited {}, messageId: "", retryAfterMs: 0, error: "rate_limited:" ++ show globalRateWindowMs ++ "ms"})
5940+ if globalRateLimited then yield (respond {status: MsRateLimited {}, messageId: "", retryAfterMs: 0, error: "rate_limited:" ++ show globalRateWindowMs ++ "ms", loadPct: 0})
58985941 else do
58995942 effectiveMsgId <- if messageId != "" then yield messageId else genToken
59005943 if t - timestamp > messageMaxAge
5901− then yield (respond {status: MsRejected {}, messageId: "", retryAfterMs: 0, error: "stale_timestamp"})
5944+ then yield (respond {status: MsRejected {}, messageId: "", retryAfterMs: 0, error: "stale_timestamp", loadPct: 0})
59025945 else do
59035946 when (timestamp > t + clockSkewTolerance)
59045947 (logWarn ("sendMessage: future timestamp detected (+" ++ show (timestamp - t) ++ "ms ahead), clamping to now"))
@@ -5914,12 +5957,13 @@ handleSendMessage = \to encryptedBlob timestamp messageId expiresInMs priority a
59145957 MsRejected {} -> when (result.retryAfterMs == 0) (logWarn ("sendMessage: rejected for " ++ effectiveMsgId ++ " (global capacity full)"))
59155958 _ -> yield {}
59165959 forkSendForwards effectiveMsgId to encryptedBlob clampedTs result.remoteServers
5960+ loadPct <- do scale <- *cachedAdaptiveScale; yield (if scale >= 100 then 0 else 100 - scale)
59175961 yield (case result.status of
5918− MsDuplicate {} -> respond {status: MsDuplicate {}, messageId: effectiveMsgId, retryAfterMs: 0, error: ""}
5919− MsDelivered {} -> respond {status: MsDelivered {}, messageId: effectiveMsgId, retryAfterMs: 0, error: ""}
5920− MsFederated {} -> respond {status: MsFederated {}, messageId: effectiveMsgId, retryAfterMs: 0, error: ""}
5921− MsQueued {} -> respond {status: MsQueued {}, messageId: effectiveMsgId, retryAfterMs: 0, error: ""}
5922− _ -> respond {status: result.status, messageId: "", retryAfterMs: result.retryAfterMs, error: "rejected"})
5962+ MsDuplicate {} -> respond {status: MsDuplicate {}, messageId: effectiveMsgId, retryAfterMs: 0, error: "", loadPct: loadPct}
5963+ MsDelivered {} -> respond {status: MsDelivered {}, messageId: effectiveMsgId, retryAfterMs: 0, error: "", loadPct: loadPct}
5964+ MsFederated {} -> respond {status: MsFederated {}, messageId: effectiveMsgId, retryAfterMs: 0, error: "", loadPct: loadPct}
5965+ MsQueued {} -> respond {status: MsQueued {}, messageId: effectiveMsgId, retryAfterMs: 0, error: "", loadPct: loadPct}
5966+ _ -> respond {status: result.status, messageId: "", retryAfterMs: result.retryAfterMs, error: "rejected", loadPct: loadPct})
59235967
59245968-- GET /lookup/:pubkey: Look up the best known server for a pubkey.
59255969-- Returns local server if key is online locally, otherwise the remote with latest expiry.
@@ -6267,6 +6311,14 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
62676311 let newlyOnlineRemote = newlyOnlineRemoteKeys filteredOnline gossipResult.gm.appliedKeys
62686312 fork (do
62696313 when (cappedHops > 1 && federationEnabled && count freshEvents > 0) (do
6314+ -- Gossip batch backoff: when a large batch of fresh events arrives (e.g., peer
6315+ -- recovering from downtime), delay re-broadcast by a random interval to prevent
6316+ -- amplification storms. Multiple peers recovering simultaneously would otherwise
6317+ -- generate O(peers^2 * events) gossip traffic.
6318+ when (count freshEvents > gossipBatchBackoffThreshold) (do
6319+ backoff <- randomInt gossipBatchBackoffMs
6320+ logEvent "info" "gossip_batch_backoff" ("events=" ++ show (count freshEvents) ++ " from=" ++ fromServer ++ " backoffMs=" ++ show backoff)
6321+ sleep backoff)
62706322 let targets = gossipTargets gossipResult.gm.gossipPeerServers fromServer
62716323 -- Split first-party (from the sender) vs third-party (relayed) events.
62726324 -- Third-party events get fewer hops to limit amplification from inflated hopsRemaining.
@@ -6330,7 +6382,7 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
63306382 presenceBkt <- readPresenceBucket bkt
63316383 let hasPresence = any presenceBkt (\p -> p.pubkey == toKey && p.expiresAt > t)
63326384 if not hasPresence
6333− then yield {forward: ForwardRejected {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: True {}}
6385+ then yield {forward: ForwardRejected {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: True {}, nonceCutoffRejected: False {}}
63346386 else do
63356387 bucket <- readMessageBucket bkt
63366388 ackedBkt <- readAckedBucket bkt
@@ -6342,28 +6394,35 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
63426394 -- messages (originServer="") whose nonces were recorded by sendMessageCore.
63436395 let isDup = checkMessageDup bucket messageId toKey || hasMessage messageId toKey ackedBkt || hasMessage messageId toKey delivBkt || any nonceBkt (\n -> n.messageId == messageId && n.toKey == toKey)
63446396 if isDup
6345− then yield {forward: ForwardDuplicate {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: False {}}
6397+ then yield {forward: ForwardDuplicate {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: False {}, nonceCutoffRejected: False {}}
63466398 else do
6399+ -- Nonce eviction replay protection: if LRU eviction has occurred, we cannot
6400+ -- verify dedup for messages whose nonces may have been evicted. Reject forwards
6401+ -- with timestamps older than the eviction cutoff as a safety measure.
6402+ nonceCutoff <- *nonceEvictionCutoff
6403+ if nonceCutoff > 0 && timestamp > 0 && timestamp < nonceCutoff
6404+ then yield {forward: ForwardRejected {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: False {}, nonceCutoffRejected: True {}}
6405+ else do
63476406 totalMsgCount <- *totalMessageCount
63486407 if totalMsgCount >= maxMessages
6349− then yield {forward: ForwardRejected {}, globalCapReached: True {}, recvRateLimited: False {}, noPresence: False {}}
6408+ then yield {forward: ForwardRejected {}, globalCapReached: True {}, recvRateLimited: False {}, noPresence: False {}, nonceCutoffRejected: False {}}
63506409 else do
63516410 recipientCounts <- readRcBucket bkt
63526411 let recipientCount = getRecipientCount toKey recipientCounts
63536412 if recipientCount >= maxMessagesPerRecipient
6354− then yield {forward: ForwardRejected {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: False {}}
6413+ then yield {forward: ForwardRejected {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: False {}, nonceCutoffRejected: False {}}
63556414 else do
63566415 recvRates <- *recvRates
63576416 recvRate <- effectiveRecvRate
63586417 let recvResult = checkPubkeyRate recvRates toKey recvRate recvRateWindowMs maxRecvRateEntries t
63596418 if recvResult.limited
6360− then yield {forward: ForwardRejected {}, globalCapReached: False {}, recvRateLimited: True {}, noPresence: False {}}
6419+ then yield {forward: ForwardRejected {}, globalCapReached: False {}, recvRateLimited: True {}, noPresence: False {}, nonceCutoffRejected: False {}}
63616420 else do
63626421 set *recvRates = recvResult.entries
63636422 federationPairRates <- *federationPairRates
63646423 let fedPairResult = checkServerRecipientRate federationPairRates fromServer toKey maxFederationPairRatePerPair federationPairRateWindowMs maxFederationPairRateEntries t
63656424 if fedPairResult.limited
6366− then yield {forward: ForwardRejected {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: False {}}
6425+ then yield {forward: ForwardRejected {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: False {}, nonceCutoffRejected: False {}}
63676426 else do
63686427 set *federationPairRates = fedPairResult.entries
63696428 seqCounter <- *seqCounter
@@ -6381,21 +6440,23 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
63816440 signalMessageForKey toKey seqTs
63826441 -- Record forward nonce for replay protection beyond message/ack dedup lifetime.
63836442 recordForwardNonce messageId toKey t
6384− yield {forward: ForwardStored {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: False {}}
6443+ yield {forward: ForwardStored {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: False {}, nonceCutoffRejected: False {}}
63856444 if result.noPresence then do
63866445 logWarn ("forward: rejected for " ++ toKey ++ " — no active presence on this server (from " ++ fromServer ++ ")")
63876446 yield (respond {ok: False {}, error: "no_presence"})
63886447 else do
63896448 case result.forward of
63906449 ForwardRejected {} ->
6391− if result.globalCapReached
6450+ if result.nonceCutoffRejected
6451+ then logEvent "warn" "forward_nonce_cutoff" ("messageId=" ++ messageId ++ " key=" ++ toKey ++ " from=" ++ fromServer ++ " — timestamp older than nonce eviction cutoff, possible replay")
6452+ else if result.globalCapReached
63926453 then logWarn ("forward: global message cap reached (" ++ show maxMessages ++ "), rejecting for " ++ toKey)
63936454 else if result.recvRateLimited
63946455 then logEvent "warn" "rate_limited_recv" ("key=" ++ toKey ++ " limit=" ++ show maxRecvRatePerKey)
63956456 else logWarn ("forward: rejected for " ++ toKey ++ " (mailbox full, max " ++ show maxMessagesPerRecipient ++ ")")
63966457 _ -> yield {}
63976458 yield (case result.forward of
6398− ForwardRejected {} -> respond {ok: False {}, error: "mailbox_full"}
6459+ ForwardRejected {} -> respond {ok: False {}, error: if result.nonceCutoffRejected then "nonce_cutoff" else "mailbox_full"}
63996460 _ -> respond {ok: True {}, error: ""}))
64006461
64016462-- Generate sender delivery receipts for messages confirmed delivered via federation ack.
@@ -6426,13 +6487,17 @@ fedDeliveryReceipts = \bucketIds forwards t -> case headOf bucketIds of
64266487 let combined = union sd deduped
64276488 let overCap = count combined > maxSenderDeliveries
64286489 if overCap then do
6429− let evicted = extractOldest combined (\d -> d.deliveredAt)
6430− set *senderDeliveries = evicted.items
6431− case evicted.candidate of
6432− Nothing {} -> yield {}
6433− Just {value: ev} -> do
6434− ddr <- *droppedDeliveryReceipts
6435− set *droppedDeliveryReceipts = incrementDropCounter ddr ev.senderKey 1 t
6490+ -- Batch eviction: evict 2% of capacity instead of 1 entry at a time.
6491+ -- Amortizes the O(n) scan cost across multiple inserts, reducing overhead
6492+ -- under sustained capacity pressure from O(n) per insert to O(n) per 2% fills.
6493+ let evictCount = max 1 (count combined - maxSenderDeliveries + maxSenderDeliveries / 50)
6494+ let eviction = evictSmallest combined evictCount (\d -> d.deliveredAt)
6495+ set *senderDeliveries = eviction.kept
6496+ -- Track per-sender drop counts for all evicted entries
6497+ let dropCounts = fold (\acc e -> incrementDropCounter acc e.senderKey 1 t) [] eviction.evicted
6498+ when (count dropCounts > 0) (do
6499+ ddr <- *droppedDeliveryReceipts
6500+ set *droppedDeliveryReceipts = fold (\acc dc -> incrementDropCounter acc dc.senderKey dc.count t) ddr dropCounts)
64366501 else set *senderDeliveries = combined)
64376502 fedDeliveryReceipts rest forwards t
64386503
@@ -6729,6 +6794,11 @@ forwardBucketGroupInner = \entries fromServer b t acc presenceBkt bucket ackedBk
67296794 if isDup
67306795 then forwardBucketGroupInner rest fromServer b t (union acc [{messageId: entry.messageId, ok: True {}}]) presenceBkt bucket ackedBkt delivBkt nonceBkt rc msgStored
67316796 else do
6797+ -- Nonce eviction replay protection (matches handleForwardMessage)
6798+ nonceCutoff <- *nonceEvictionCutoff
6799+ if nonceCutoff > 0 && entry.timestamp > 0 && entry.timestamp < nonceCutoff
6800+ then forwardBucketGroupInner rest fromServer b t (union acc [{messageId: entry.messageId, ok: False {}}]) presenceBkt bucket ackedBkt delivBkt nonceBkt rc msgStored
6801+ else do
67326802 totalMsgCount <- *totalMessageCount
67336803 let recipientCount = getRecipientCount entry.toKey rc
67346804 if totalMsgCount + msgStored >= maxMessages || recipientCount >= maxMessagesPerRecipient
@@ -7098,37 +7168,57 @@ computeAndCacheHealth = \t -> do
70987168 set *cachedHealthDegradedPeers = result.degradedPeers
70997169 yield result
71007170
7171+-- Return cached health state. Used for unauthenticated callers and rate-limited requests.
7172+-- Provides constant-time responses regardless of server state complexity.
7173+readCachedHealth = \respond -> do
7174+ cached <- atomic do
7175+ cOk <- *cachedHealthOk
7176+ cP <- *cachedHealthPrune
7177+ cH <- *cachedHealthHeartbeat
7178+ cC <- *cachedHealthCapacity
7179+ cF <- *cachedHealthFederation
7180+ cW <- *cachedHealthWatchdog
7181+ cN <- *cachedHealthNonce
7182+ cFQ <- *cachedHealthForwardQueue
7183+ cCW <- *cachedHealthCapacityWarning
7184+ cDP <- *cachedHealthDegradedPeers
7185+ dp <- *drainPhase
7186+ yield {ok: cOk && dp == 0, version: serverBuildVersion, pruneHealthy: cP, heartbeatHealthy: cH, capacityHealthy: cC, capacityWarning: cCW, federationHealthy: cF, watchdogHealthy: cW, nonceHealthy: cN, forwardQueueHealthy: cFQ, draining: dp > 0, degradedPeers: cDP}
7187+ yield (respond cached)
7188+
71017189handleHealth = \authorization respond -> do
71027190 t <- now
71037191 -- Health checks exempt from global rate limiting (monitoring traffic shouldn't starve real requests)
71047192 -- but have their own dedicated rate limiter to prevent DDoS on /health from saturating reads.
71057193 healthLimited <- checkHealthRate t
7106− if healthLimited then do
7107− -- Return cached last-known health (prevents DDoS on /health from masking real failures).
7108− -- Cache is kept fresh by computeAndCacheHealth called from both here and background prune.
7109− cached <- atomic do
7110− cOk <- *cachedHealthOk
7111− cP <- *cachedHealthPrune
7112− cH <- *cachedHealthHeartbeat
7113− cC <- *cachedHealthCapacity
7114− cF <- *cachedHealthFederation
7115− cW <- *cachedHealthWatchdog
7116− cN <- *cachedHealthNonce
7117− cFQ <- *cachedHealthForwardQueue
7118− cCW <- *cachedHealthCapacityWarning
7119− cDP <- *cachedHealthDegradedPeers
7120− dp <- *drainPhase
7121− yield {ok: cOk && dp == 0, version: serverBuildVersion, pruneHealthy: cP, heartbeatHealthy: cH, capacityHealthy: cC, capacityWarning: cCW, federationHealthy: cF, watchdogHealthy: cW, nonceHealthy: cN, forwardQueueHealthy: cFQ, draining: dp > 0, degradedPeers: cDP}
7122− yield (respond cached)
7194+ if healthLimited then readCachedHealth respond
71237195 else do
71247196 isAuthed <- verifyMaintenanceAuth authorization
7197+ -- Unauthenticated callers always get cached results (constant-time, no live computation).
7198+ -- This prevents timing side-channels: an attacker cannot infer server load by measuring
7199+ -- response time differences between cached (fast) and computed (slow) health checks.
7200+ -- Cache is kept fresh by computeAndCacheHealth called from background prune (~30s).
7201+ -- Only authenticated callers trigger live computation + cache refresh.
7202+ if not isAuthed
7203+ then do
7204+ cached <- atomic do
7205+ cOk <- *cachedHealthOk
7206+ dp <- *drainPhase
7207+ yield {ok: cOk && dp == 0, draining: dp > 0}
7208+ yield (respond {ok: cached.ok, version: serverBuildVersion, pruneHealthy: True {},
7209+ heartbeatHealthy: True {},
7210+ capacityHealthy: True {},
7211+ capacityWarning: False {},
7212+ federationHealthy: True {},
7213+ watchdogHealthy: True {},
7214+ nonceHealthy: True {},
7215+ forwardQueueHealthy: True {},
7216+ draining: cached.draining,
7217+ degradedPeers: 0})
7218+ else do
71257219 result <- computeAndCacheHealth t
71267220 let allOk = not result.isDraining && result.pruneOk && result.heartbeatOk && result.capacityOk && result.federationOk && result.watchdogOk && result.nonceOk && result.forwardQueueOk
7127− -- Unauthenticated: return real ok boolean so load balancers can route around unhealthy nodes,
7128− -- but hide subsystem breakdown to prevent targeted load-correlation attacks.
7129− -- Draining status is always visible (load balancers need it for routing decisions).
7130− if isAuthed
7131− then yield (respond {ok: allOk, version: serverBuildVersion, pruneHealthy: result.pruneOk,
7221+ yield (respond {ok: allOk, version: serverBuildVersion, pruneHealthy: result.pruneOk,
71327222 heartbeatHealthy: result.heartbeatOk,
71337223 capacityHealthy: result.capacityOk,
71347224 capacityWarning: result.capacityWarning,
@@ -7138,16 +7228,6 @@ handleHealth = \authorization respond -> do
71387228 forwardQueueHealthy: result.forwardQueueOk,
71397229 draining: result.isDraining,
71407230 degradedPeers: result.degradedPeers})
7141− else yield (respond {ok: allOk, version: serverBuildVersion, pruneHealthy: True {},
7142− heartbeatHealthy: True {},
7143− capacityHealthy: True {},
7144− capacityWarning: False {},
7145− federationHealthy: True {},
7146− watchdogHealthy: True {},
7147− nonceHealthy: True {},
7148− forwardQueueHealthy: True {},
7149− draining: result.isDraining,
7150− degradedPeers: 0})
71517231
71527232zeroStats = {ok: False {}, version: serverBuildVersion, capacity: {sessions: 0, sessionsMax: 0, sessionsPct: 0, messages: 0, messagesMax: 0, messagesPct: 0, totalMessageCount: 0, presence: 0, presenceMax: 0, presencePct: 0, forwards: 0, forwardsMax: 0, forwardsPct: 0, challenges: 0, challengesMax: 0, ackedIds: 0, ackedIdsMax: 0, ackOverflow: 0, ackOverflowMax: 0, deliveredIds: 0, deliveredIdsMax: 0, forwardNonces: 0, forwardNoncesMax: 0, activePolls: 0, activePollsMax: 0, recipientCounts: 0, messageBucketMax: 0, messageBucketMin: 0}, federation: {enabled: False {}, inFlight: 0, hungSlots: 0, avgLatencyMs: 0, maxLatencyMs: 0, callCount: 0, slowVerifies: 0, verifyFailures: 0, forwardFailures: 0, oldestForwardAge: 0, oldestAckAge: 0, recentGossip: 0, blockedPeers: 0, drainCompletedAt: 0, forwardServers: 0, maxForwardQueueDepth: 0, topFailingPeers: [], pendingCancels: 0, gossipSuccess: 0, gossipFail: 0}, rates: {send: 0, challenge: 0, federation: 0, federationPair: 0, global: 0, ip: 0, lookup: 0, recv: 0, pair: 0, gossipEvent: 0, verifyIp: 0, uniqueKeyIp: 0, pendingAckCleanups: 0, senderExpirations: 0, senderDeliveries: 0, rejectedGlobal: 0, rejectedIp: 0, rejectedAuth: 0, rejectedPerKey: 0, rejectedFederation: 0}, timing: {pruneCount: 0, pruneRunning: 0, heartbeatCount: 0, uptime: 0, lastPruneAt: 0, lastHeartbeatAt: 0, draining: False {}, drainPhase: 0, totalEntries: 0, activeRequests: 0, peakActiveRequests: 0, pollAvgMs: 0, pollMaxMs: 0, pollCount: 0, sendAvgMs: 0, sendMaxMs: 0, sendCount: 0}}
71537233
@@ -8727,6 +8807,7 @@ main = do
87278807 && clockSkewTolerance < challengeExpiry
87288808 && maxMessagesPerRecipient <= maxMessages
87298809 && maxForwardsPerKey <= maxForwardsEntries
8810+ && maxForwardsPerSenderPerKey <= maxForwardsPerKey
87308811 && maxForwardsPerServer <= maxForwardsEntries
87318812 && maxSessionsPerKey <= maxSessions
87328813 && maxChallengesPerKey <= maxChallenges
@@ -8744,7 +8825,7 @@ main = do
87448825 if not configOk
87458826 then logError "FATAL: invalid configuration — capacity limits and intervals must be > 0, server not started"
87468827 else if not configRelationsOk
8747− then logError "FATAL: invalid configuration — relationship invariants violated (check: unackedMessageMaxAge <= messageMaxAge, challengeExpiry < sessionExpiry, sessionExpiry <= maxSessionLifetime, retryBackoffBase <= retryBackoffMax, backgroundPruneInterval < healthPruneMaxAge, pollTimeoutMs < sessionExpiry, gossipDedupWindow < presenceRefreshInterval, federationFetchTimeoutMs <= backgroundPruneInterval, maxAckOverflow <= maxAckedIds, gossipOriginHops <= maxGossipHops, maxThirdPartyGossipHops <= maxGossipHops, verificationTokenExpiry <= federationFetchTimeoutMs, forwardNonceRetention >= messageMaxAge, maxGlobalFederationConcurrency >= maxFederationConcurrency, maxGossipEventsPerServer > maxGossipEvents, per-key/per-server limits <= global limits), server not started"
8828+ then logError "FATAL: invalid configuration — relationship invariants violated (check: unackedMessageMaxAge <= messageMaxAge, challengeExpiry < sessionExpiry, sessionExpiry <= maxSessionLifetime, retryBackoffBase <= retryBackoffMax, backgroundPruneInterval < healthPruneMaxAge, pollTimeoutMs < sessionExpiry, gossipDedupWindow < presenceRefreshInterval, federationFetchTimeoutMs <= backgroundPruneInterval, maxAckOverflow <= maxAckedIds, gossipOriginHops <= maxGossipHops, maxThirdPartyGossipHops <= maxGossipHops, verificationTokenExpiry <= federationFetchTimeoutMs, forwardNonceRetention >= messageMaxAge, maxGlobalFederationConcurrency >= maxFederationConcurrency, maxGossipEventsPerServer > maxGossipEvents, maxForwardsPerSenderPerKey <= maxForwardsPerKey, per-key/per-server limits <= global limits), server not started"
87488829 else do
87498830 t <- now
87508831 atomic do
@@ -8767,7 +8848,9 @@ main = do
87678848 (logWarn ("serverHost is " ++ serverHost ++ " — peers cannot reach this server for federation"))
87688849 when (federationEnabled && isBadServerName serverHostN)
87698850 (logError ("serverHost '" ++ serverHost ++ "' would be rejected by isBadServerName — federation peers cannot forward messages here. Set serverHost to a public DNS name."))
8770− logInfo "IP binding is active — sessions created with a ClientIp header will reject requests without one"
8851+ if requireClientIp
8852+ then logInfo "IP binding: required — challenge requests without ClientIp header are rejected"
8853+ else logInfo "IP binding: optional — sessions created with a ClientIp header will reject requests without one"
87718854 logWarn "SECURITY: Knot has no HTTP-layer request body size limit. Deploy behind a reverse proxy with a body limit (e.g., Caddy: request_body { max_size 256KB }) to prevent large-payload DoS."
87728855 when federationEnabled
87738856 (logWarn "SECURITY: DNS rebinding risk — federation URLs resolve DNS at fetch time, not at validation time. A peer hostname passing isBadServerName could rebind to a private IP between validation and fetch. Mitigate by blocking outbound connections to RFC1918/link-local/CGNAT ranges at the firewall level.")
@@ -8786,7 +8869,7 @@ main = do
87868869 logInfo (" max messages: " ++ show maxMessages ++ " (per recipient: " ++ show maxMessagesPerRecipient ++ ")")
87878870 logInfo (" max sessions: " ++ show maxSessions ++ " (per key: " ++ show maxSessionsPerKey ++ ")")
87888871 logInfo (" max presence: " ++ show maxPresenceEntries)
8789− logInfo (" max forwards: " ++ show maxForwardsEntries ++ " (per key: " ++ show maxForwardsPerKey ++ ", per server: " ++ show maxForwardsPerServer ++ ")")
8872+ logInfo (" max forwards: " ++ show maxForwardsEntries ++ " (per key: " ++ show maxForwardsPerKey ++ ", per sender-key: " ++ show maxForwardsPerSenderPerKey ++ ", per server: " ++ show maxForwardsPerServer ++ ")")
87908873 logInfo (" session expiry: " ++ show sessionExpiry ++ "ms (max lifetime: " ++ show maxSessionLifetime ++ "ms)")
87918874 logInfo (" message max age: " ++ show messageMaxAge ++ "ms")
87928875 logInfo (" max active requests: " ++ if maxActiveRequests > 0 then show maxActiveRequests else "unlimited")