@@ -66,6 +66,11 @@ maxRecvRatePerKey = 10 -- max messages received by a key per window (across all
6666recvRateWindowMs : Int<Ms>
6767recvRateWindowMs = 1000
6868maxRecvRateEntries = 1000
69+-- Per-sender-per-recipient rate limiting (prevents one sender from filling a recipient's mailbox)
70+maxPairRatePerPair = 30 -- max messages per sender-recipient pair per window
71+pairRateWindowMs : Int<Ms>
72+pairRateWindowMs = 60000 -- 1 minute
73+maxPairRateEntries = 5000
6974-- Federation rate limiting (per-server)
7075maxFederationRatePerServer = 50 -- max federation requests per server per window
7176federationRateWindowMs : Int<Ms>
@@ -89,6 +94,7 @@ offlineGossipMinTtl = 60 -- minimum TTL for offline gossip events
8994maxFederationConcurrency = 10 -- max concurrent outbound federation requests per cycle
9095maxForwardRetriesPerServer = 100
9196maxForwardRetries = 10 -- max retries per individual forward entry before dropping
97+maxAckRetries = 10 -- max retries per ack entry before dropping to deliveredIds
9298maxGossipHops = 5 -- max hops a gossip event can propagate across servers
9399gossipOriginHops = 2 -- hops set when this server originates a gossip event
94100gossipDedupWindow : Int<Ms>
@@ -238,7 +244,7 @@ type PresenceEntry = {pubkey: PubkeyHex, server: AnyServerName, expiresAt: Times
238244-- Queued federation forward: message waiting to be delivered to a remote server
239245type ForwardEntry = {messageId: MessageIdHex, toServer: ServerName, createdAt: Timestamp, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: Text, retries: Int}
240246-- Tracks message IDs that were acked by a polling client, pending federation ack back to origin
241−type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: ServerName}
247+type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: ServerName, retries: Int}
242248-- Terminal state: message delivered and ack confirmed (used for dedup after ack overflow)
243249type DeliveredEntry = {messageId: MessageIdHex, toKey: PubkeyHex, deliveredAt: Timestamp}
244250-- Tracks consecutive failures per federation peer for exponential backoff
@@ -246,6 +252,7 @@ type ForwardFailure = {server: Text, failedAt: Timestamp, failures: Int, firstFa
246252-- Sliding-window rate limit state per pubkey
247253type SendRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}
248254type RecvRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}
255+type PairRateEntry = {sender: PubkeyHex, recipient: PubkeyHex, windowStart: Timestamp, count: Int}
249256type ChallengeRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}
250257-- Sliding-window rate limit state per federation peer server
251258type FederationRateEntry = {server: Text, windowStart: Timestamp, count: Int}
@@ -411,6 +418,8 @@ type RecentFederationRequest = {server: Text, timestamp: Timestamp, seenAt: Time
411418
412419*recvRates : [RecvRateEntry]
413420
421+*pairRates : [PairRateEntry]
422+
414423*recentGossip : [RecentGossipEntry]
415424
416425*lastFederationSuccessAt : Timestamp
@@ -452,7 +461,8 @@ type RecentFederationRequest = {server: Text, timestamp: Timestamp, seenAt: Time
452461*recentFederationRequests : [RecentFederationRequest]
453462
454463-- Pending verification tokens for federation callback auth (single-use, short-lived)
455−*pendingVerifications : [{token: Text, createdAt: Timestamp}]
464+-- scope binds the token to a specific request type (gossip/forward/ack) to prevent cross-use
465+*pendingVerifications : [{token: Text, createdAt: Timestamp, scope: Text}]
456466
457467-- ============================================
458468-- Routes
@@ -517,7 +527,7 @@ route FederationApi where
517527-- Federation callback verification: the receiving server calls back to the sender
518528-- to confirm a request token is genuine. No shared secret needed.
519529route FederationVerifyApi where
520− POST {token: Text} /federation/verify -> {ok: Bool} = VerifyToken
530+ POST {token: Text, scope: Text} /federation/verify -> {ok: Bool} = VerifyToken
521531
522532-- Triggers pruning of expired state and retry of pending federation ops
523533route MaintenanceApi where
@@ -823,13 +833,15 @@ verifyMaintenanceAuth = verifyAuthWith (\token -> verifySecret token maintenance
823833
824834-- Generate a single-use token for outbound federation auth.
825835-- The receiving server calls back to /federation/verify to confirm we issued it.
826−generateFederationAuth : IO {network} Text
827−generateFederationAuth = do
836+-- scope binds the token to a request type ("gossip", "forward", "ack") so a token
837+-- issued for one operation cannot authorize a different one.
838+generateFederationAuth : Text -> IO {network} Text
839+generateFederationAuth = \scope -> do
828840 token <- genToken
829841 t <- now
830842 atomic do
831843 pending <- *pendingVerifications
832− let updated = union pending [{token: token, createdAt: t}]
844+ let updated = union pending [{token: token, createdAt: t, scope: scope}]
833845 set *pendingVerifications = if count updated > maxPendingVerifications
834846 then evictOneOldest updated (\p -> p.createdAt)
835847 else updated
@@ -837,12 +849,13 @@ generateFederationAuth = do
837849
838850-- Verify an inbound federation token by callback to the sender's /federation/verify endpoint.
839851-- Single-use: the sender consumes the token on verification.
840−verifyFederationCallback : Text -> Text -> IO {network} Bool
841−verifyFederationCallback = \fromServer authorization ->
852+-- scope must match the scope the token was generated with; empty scope accepts any (backward compat).
853+verifyFederationCallback : Text -> Text -> Text -> IO {network} Bool
854+verifyFederationCallback = \fromServer authorization scope ->
842855 case extractToken authorization of
843856 Nothing {} -> yield (False {})
844857 Just {value: token} -> do
845− result <- fetch (federationUrl fromServer) (VerifyToken {token: token})
858+ result <- fetch (federationUrl fromServer) (VerifyToken {token: token, scope: scope})
846859 case result of
847860 Err e -> do
848861 println ("federation: verify callback to " ++ fromServer ++ " failed: " ++ show e)
@@ -1578,6 +1591,8 @@ pruneExpired = \t -> do
15781591 set *challengeRates = do r <- challengeRates; where t - r.windowStart < challengeRateWindowMs; yield r
15791592 recvRates <- *recvRates
15801593 set *recvRates = do r <- recvRates; where t - r.windowStart < recvRateWindowMs; yield r
1594+ pairRates <- *pairRates
1595+ set *pairRates = do r <- pairRates; where t - r.windowStart < pairRateWindowMs; yield r
15811596 -- Reset auth-tier global rate window
15821597 authWS <- *authGlobalRateWindowStart
15831598 when (t - authWS >= authGlobalRateWindowMs) (do
@@ -1630,7 +1645,7 @@ sendBatchedGossipToServer : ServerName -> [GossipEvent] -> Int -> IO {network} {
16301645sendBatchedGossipToServer = \server allEvents hops ->
16311646 forEachBatch allEvents maxGossipEvents (\batch -> do
16321647 t <- now
1633− auth <- generateFederationAuth
1648+ auth <- generateFederationAuth "gossip"
16341649 result <- fetchFederation server (RecvGossip {
16351650 authorization: auth,
16361651 events: batch, hopsRemaining: hops, fromServer: serverHostN,
@@ -1736,8 +1751,8 @@ checkPreAuth = \clientIp t -> do
17361751 if ipLimited then yield "ip_rate_limited"
17371752 else yield ""
17381753
1739−withFederationAuth = \authorization t respond fromServer requestTimestamp precondition action -> do
1740− verified <- verifyFederationCallback fromServer authorization
1754+withFederationAuth = \authorization t respond fromServer scope requestTimestamp precondition action -> do
1755+ verified <- verifyFederationCallback fromServer authorization scope
17411756 if not verified
17421757 then yield (respond {ok: False {}, error: "unauthorized"})
17431758 -- Replay protection: reject requests with timestamps too far from local clock.
@@ -1892,7 +1907,7 @@ retryForwardsForServer = \entries remaining ->
18921907 Just {value: f} -> do
18931908 let rest = do x <- entries; where x.messageId != f.messageId; yield x
18941909 t <- now
1895− auth <- generateFederationAuth
1910+ auth <- generateFederationAuth "forward"
18961911 result <- fetchFederation f.toServer (ForwardMessage {
18971912 toKey: f.toKey, encryptedBlob: f.encryptedBlob,
18981913 timestamp: f.timestamp, originServer: f.originServer, messageId: f.messageId,
@@ -1956,7 +1971,7 @@ forwardQueuedMessages = \targets t -> do
19561971
19571972fetchFederationAck = \origin messageIds -> do
19581973 t <- now
1959− auth <- generateFederationAuth
1974+ auth <- generateFederationAuth "ack"
19601975 fetchFederation origin (FederationAck {
19611976 authorization: auth,
19621977 messageIds: messageIds,
@@ -1971,6 +1986,23 @@ sendBatchedAcksToServer = \serverAcks server ->
19711986 Err e -> do
19721987 println ("federation: ack to " ++ server ++ " failed: " ++ show e)
19731988 recordFailure server
1989+ -- Increment retry counters; drop entries exceeding maxAckRetries to deliveredIds
1990+ dropped <- atomic do
1991+ ackedIds <- *ackedIds
1992+ let isTarget = \a -> a.originServer == server && hasMessage a.messageId a.toKey batch
1993+ let toDrop = do a <- ackedIds; where isTarget a && a.retries + 1 >= maxAckRetries; yield a
1994+ let dropCount = count toDrop
1995+ set *ackedIds = do a <- ackedIds
1996+ where not (isTarget a && a.retries + 1 >= maxAckRetries)
1997+ yield (if isTarget a then {messageId: a.messageId, toKey: a.toKey, ackedAt: a.ackedAt, originServer: a.originServer, retries: a.retries + 1} else a)
1998+ when (dropCount > 0) (do
1999+ deliveredIds <- *deliveredIds
2000+ let newDelivered = do d <- toDrop; yield {messageId: d.messageId, toKey: d.toKey, deliveredAt: d.ackedAt}
2001+ let combined = union deliveredIds newDelivered
2002+ set *deliveredIds = if count combined > maxDeliveredIds then (evictSmallest combined (count combined - maxDeliveredIds) (\d -> d.deliveredAt)).kept else combined)
2003+ yield dropCount
2004+ when (dropped > 0)
2005+ (println ("federation: dropped " ++ show dropped ++ " ack entries to " ++ server ++ " after " ++ show maxAckRetries ++ " retries"))
19742006 Ok _ -> do
19752007 clearFailure server
19762008 atomic do
@@ -2006,7 +2038,7 @@ retryEvictedAcks = \evictedEntries pendingAcks ->
20062038 println ("federation: evicted ack to " ++ server ++ " failed: " ++ show e)
20072039 overflow <- atomic do
20082040 ackedIds <- *ackedIds
2009− let reinsert = acksForServer evictedToRetry server
2041+ let reinsert = do a <- acksForServer evictedToRetry server; yield {messageId: a.messageId, toKey: a.toKey, ackedAt: a.ackedAt, originServer: a.originServer, retries: a.retries + 1}
20102042 let combined = union ackedIds reinsert
20112043 let excess = count combined - maxAckedIds
20122044 if excess <= 0
@@ -2055,6 +2087,23 @@ retryOverflowAcks = \overflow -> do
20552087 Err e -> do
20562088 println ("federation: overflow ack to " ++ server ++ " failed: " ++ show e)
20572089 recordFailure server
2090+ -- Increment retry counters in overflow; drop entries exceeding maxAckRetries to deliveredIds
2091+ dropped <- atomic do
2092+ ackOverflow <- *ackOverflow
2093+ let isTarget = \a -> a.originServer == server && hasMessage a.messageId a.toKey batch
2094+ let toDrop = do a <- ackOverflow; where isTarget a && a.retries + 1 >= maxAckRetries; yield a
2095+ let dropCount = count toDrop
2096+ set *ackOverflow = do a <- ackOverflow
2097+ where not (isTarget a && a.retries + 1 >= maxAckRetries)
2098+ yield (if isTarget a then {messageId: a.messageId, toKey: a.toKey, ackedAt: a.ackedAt, originServer: a.originServer, retries: a.retries + 1} else a)
2099+ when (dropCount > 0) (do
2100+ deliveredIds <- *deliveredIds
2101+ let newDelivered = do d <- toDrop; yield {messageId: d.messageId, toKey: d.toKey, deliveredAt: d.ackedAt}
2102+ let combined = union deliveredIds newDelivered
2103+ set *deliveredIds = if count combined > maxDeliveredIds then (evictSmallest combined (count combined - maxDeliveredIds) (\d -> d.deliveredAt)).kept else combined)
2104+ yield dropCount
2105+ when (dropped > 0)
2106+ (println ("federation: dropped " ++ show dropped ++ " overflow ack entries to " ++ server ++ " after " ++ show maxAckRetries ++ " retries"))
20582107 Ok _ -> do
20592108 clearFailure server
20602109 atomic do
@@ -2344,7 +2393,7 @@ pollableMessages = \messages pubkey cursor t ->
23442393
23452394-- Build ack entries for messages that came from a remote origin server (need federation ack)
23462395federatedAckEntries = \ackedMessages t ->
2347− do m <- ackedMessages; where m.originServer != ""; yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer}
2396+ do m <- ackedMessages; where m.originServer != ""; yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer, retries: 0}
23482397
23492398-- Remove forwards for specific message IDs destined to a given server
23502399forwardsExcluding = \forwards server messageIds ->
@@ -2660,6 +2709,15 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp
26602709 then yield {send: SendFailed {status: MsRateLimited {}}, senderRateLimited: True {}, globalCapReached: False {}}
26612710 else do
26622711 set *sendRates = rateResult.entries
2712+ pairRates <- *pairRates
2713+ let pairResult = checkRateWindow pairRates (\r -> r.sender == senderKey && r.recipient == to)
2714+ (\ts -> {sender: senderKey, recipient: to, windowStart: ts, count: 1})
2715+ (\r -> {sender: r.sender, recipient: r.recipient, windowStart: r.windowStart, count: r.count + 1})
2716+ maxPairRatePerPair pairRateWindowMs maxPairRateEntries t
2717+ set *pairRates = pairResult.entries
2718+ if pairResult.limited
2719+ then yield {send: SendFailed {status: MsRateLimited {}}, senderRateLimited: False {}, globalCapReached: False {}}
2720+ else do
26632721 recvRates <- *recvRates
26642722 let recvResult = checkRateWindow recvRates (\r -> r.pubkey == to)
26652723 (\ts -> {pubkey: to, windowStart: ts, count: 1})
@@ -2717,7 +2775,7 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp
27172775 yield (any forwards (\f -> f.messageId == effectiveMsgId && f.toServer == server))
27182776 when stillPending (do
27192777 fwdT <- now
2720− fwdAuth <- generateFederationAuth
2778+ fwdAuth <- generateFederationAuth "forward"
27212779 fwdResult <- fetchFederation server (ForwardMessage {
27222780 toKey: to, encryptedBlob: encryptedBlob,
27232781 timestamp: clampedTs, originServer: serverHostN, messageId: effectiveMsgId,
@@ -2907,11 +2965,12 @@ handleLogout = \authorization clientIp respond -> do
29072965-- POST /federation/verify: Confirm a token was issued by this server.
29082966-- Called by the receiving server during federation callback verification.
29092967-- Tokens are single-use and expire after verificationTokenExpiry.
2910−handleVerifyToken = \token respond -> do
2968+handleVerifyToken = \token scope respond -> do
29112969 t <- now
29122970 found <- atomic do
29132971 pending <- *pendingVerifications
2914− let match = any pending (\p -> safeCompare p.token token && t - p.createdAt < verificationTokenExpiry)
2972+ -- Empty scope accepts any (backward compat with peers that don't send scope yet)
2973+ let match = any pending (\p -> safeCompare p.token token && (scope == "" || p.scope == scope) && t - p.createdAt < verificationTokenExpiry)
29152974 when match (set *pendingVerifications = do p <- pending; where not (safeCompare p.token token); yield p)
29162975 yield match
29172976 yield (respond {ok: found})
@@ -2923,7 +2982,7 @@ handleVerifyToken = \token respond -> do
29232982handleRecvGossip = \events hopsRemaining fromServer requestTimestamp authorization respond -> do
29242983 t <- now
29252984 let cappedHops = clamp 0 maxGossipHops hopsRemaining
2926− withFederationAuth authorization t respond fromServer requestTimestamp (cappedHops >= 1 && not (isLocalServer fromServer) && count events <= maxGossipEvents) (do
2985+ withFederationAuth authorization t respond fromServer "gossip" requestTimestamp (cappedHops >= 1 && not (isLocalServer fromServer) && count events <= maxGossipEvents) (do
29272986 let validated = validGossipEvents events t
29282987 let onlineEntries = validated.online
29292988 let trustedOffline = trustedOfflineEvents validated.offline fromServer
@@ -2970,7 +3029,7 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp authorizati
29703029-- twice) and forwardCandidates (won't re-forward to the same server).
29713030handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp authorization respond -> do
29723031 t <- now
2973− withFederationAuth authorization t respond fromServer requestTimestamp (not (isLocalServer originServer) && not (isLocalServer fromServer)) (do
3032+ withFederationAuth authorization t respond fromServer "forward" requestTimestamp (not (isLocalServer originServer) && not (isLocalServer fromServer)) (do
29743033 when (fromServer != originServer)
29753034 (println ("forward: relay from " ++ fromServer ++ " (origin: " ++ originServer ++ ") for " ++ toKey))
29763035 -- Replay protection: reject forwards with timestamps older than messageMaxAge.
@@ -3034,7 +3093,7 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
30343093-- recipient has no local presence and no other pending forwards for that message.
30353094handleFederationAck = \messageIds originServer fromServer requestTimestamp authorization respond -> do
30363095 t <- now
3037− withFederationAuth authorization t respond fromServer requestTimestamp (isLocalServer originServer && not (isLocalServer fromServer)) (do
3096+ withFederationAuth authorization t respond fromServer "ack" requestTimestamp (isLocalServer originServer && not (isLocalServer fromServer)) (do
30383097 let dedupedIds = dedupMessageIds messageIds
30393098 -- Clear forward entries for the acking server. Extract removed forwards' toKey
30403099 -- so we can reclaim origin messages that are no longer needed.
@@ -3333,7 +3392,7 @@ serve = \req -> case req of
33333392 RecvGossip {events, hopsRemaining, fromServer, requestTimestamp, authorization, respond} -> handleRecvGossip events hopsRemaining fromServer requestTimestamp authorization respond
33343393 ForwardMessage {toKey, encryptedBlob, timestamp, originServer, messageId, fromServer, requestTimestamp, authorization, respond} -> handleForwardMessage toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp authorization respond
33353394 FederationAck {messageIds, originServer, fromServer, requestTimestamp, authorization, respond} -> handleFederationAck messageIds originServer fromServer requestTimestamp authorization respond
3336− VerifyToken {token, respond} -> handleVerifyToken token respond
3395+ VerifyToken {token, scope, respond} -> handleVerifyToken token scope respond
33373396 Prune {authorization, respond} -> handlePrune authorization respond
33383397 Drain {authorization, respond} -> handleDrain authorization respond
33393398 Undrain {authorization, respond} -> handleUndrain authorization respond
@@ -3452,6 +3511,7 @@ main = do
34523511 && maxForwardRetries > 0 && maxGossipHops > 0 && maxPollsPerKey > 0
34533512 && gossipOriginHops > 0
34543513 && verificationTokenExpiry > 0
3514+ && maxAckRetries > 0 && maxPairRatePerPair > 0 && pairRateWindowMs > 0
34553515 -- Relationship invariants: these misconfigurations cause subtle bugs
34563516 -- (e.g., messages pruned before their stated max age, backoff that never increases)
34573517 let configRelationsOk = unackedMessageMaxAge <= messageMaxAge
@@ -3491,6 +3551,8 @@ main = do
34913551 println (" message max age: " ++ show messageMaxAge ++ "ms")
34923552 println (" global rate limit: " ++ show maxGlobalRequestRate ++ "/" ++ show globalRateWindowMs ++ "ms")
34933553 println (" auth rate limit: " ++ show maxAuthGlobalRate ++ "/" ++ show authGlobalRateWindowMs ++ "ms")
3554+ println (" pair rate limit: " ++ show maxPairRatePerPair ++ "/" ++ show pairRateWindowMs ++ "ms per sender-recipient")
3555+ println (" ack retry limit: " ++ show maxAckRetries ++ " per entry")
34943556 when (federationCircuitBreakerThreshold > 0)
34953557 (println (" circuit breaker: " ++ show federationCircuitBreakerThreshold ++ " failing peers"))
34963558 when (autoSnapshotInterval > 0)