55d061d · skrepka · git.iko.soy

git.iko.soy/skrepka

Strengthen federation resilience: recv rate limits, drain blocks, server masking, concurrency cap, capacity signals, gossip dedup

55d061diko6 months ago

server.knot+84−22

compare with master

unifiedsplit

@@ -77,9 +77,13 @@ onlineGossipTtl : Int<S>

77onlineGossipTtl = 3600 -- 1 hour; TTL announced when a key authenticates

78offlineGossipMinTtl : Int<S>

79offlineGossipMinTtl = 60 -- minimum TTL for offline gossip events

80+maxFederationConcurrency = 10 -- max concurrent outbound federation requests per cycle

80maxForwardRetriesPerServer = 100

81maxForwardRetriesPerServer = 100

81maxGossipHops = 5 -- max hops a gossip event can propagate across servers

82maxGossipHops = 5 -- max hops a gossip event can propagate across servers

82gossipOriginHops = 2 -- hops set when this server originates a gossip event

83gossipOriginHops = 2 -- hops set when this server originates a gossip event

84+gossipDedupWindow : Int<Ms>

85+gossipDedupWindow = 60000 -- suppress re-broadcast of identical gossip events within this window

86+maxRecentGossip = 1000

83-- Exponential backoff for failed federation forwards

87-- Exponential backoff for failed federation forwards

84retryBackoffBase : Int<Ms>

88retryBackoffBase : Int<Ms>

85retryBackoffBase = 30000 -- 30 seconds

89retryBackoffBase = 30000 -- 30 seconds

@@ -218,6 +222,8 @@ data ForwardResult = ForwardDuplicate {} | ForwardRejected {} | ForwardStored {}

218

222

219type IpRateEntry = {ip: Text, windowStart: Timestamp, count: Int}

223type IpRateEntry = {ip: Text, windowStart: Timestamp, count: Int}

220type LookupRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}

224type LookupRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}

225+-- Tracks recently re-broadcast gossip events to suppress amplification from duplicate cycles

226+type RecentGossipEntry = {pubkey: PubkeyHex, server: Text, timestamp: Timestamp, seenAt: Timestamp}

221type PeerStatus = {server: Text, failures: Int, lastFailedAt: Timestamp, firstFailedAt: Timestamp, backoffMs: Int, retriable: Bool}

227type PeerStatus = {server: Text, failures: Int, lastFailedAt: Timestamp, firstFailedAt: Timestamp, backoffMs: Int, retriable: Bool}

222

228

223-- ============================================

229-- ============================================

@@ -308,6 +314,8 @@ type PeerStatus = {server: Text, failures: Int, lastFailedAt: Timestamp, firstFa

308

314

309*recvRates : [RecvRateEntry]

315*recvRates : [RecvRateEntry]

310

316

317+*recentGossip : [RecentGossipEntry]

318+

311-- ============================================

319-- ============================================

312-- Routes

320-- Routes

313-- ============================================

321-- ============================================

@@ -367,7 +375,7 @@ route MaintenanceApi where

367 = Drain

375 = Drain

368

376

369route HealthApi where

377route HealthApi where

370− GET /health -> {ok: Bool, pruneHealthy: Bool, heartbeatHealthy: Bool} = Health

378+ GET /health -> {ok: Bool, pruneHealthy: Bool, heartbeatHealthy: Bool, capacityHealthy: Bool} = Health

371

379

372-- Returns counts of in-memory collections (requires federation or maintenance secret)

380-- Returns counts of in-memory collections (requires federation or maintenance secret)

373route StatsApi where

381route StatsApi where

@@ -669,9 +677,12 @@ evictOldestAckedIds = \items n ->

669-- Constant-time string comparison to prevent timing side-channels.

677-- Constant-time string comparison to prevent timing side-channels.

670-- Always iterates maxTokenLen characters regardless of input length;

678-- Always iterates maxTokenLen characters regardless of input length;

671-- rejects empty strings to avoid vacuous equality.

679-- rejects empty strings to avoid vacuous equality.

680+-- Loop terminates strictly on counter (n starts at max of both lengths and maxTokenLen,

681+-- so both strings are exhausted by the time n reaches 0). Avoiding s1/s2 emptiness

682+-- checks in the branch prevents a potential timing side-channel from branch prediction.

672safeCompareLoop : Text -> Text -> Int -> Int -> Int

683safeCompareLoop : Text -> Text -> Int -> Int -> Int

673safeCompareLoop = \s1 s2 n result ->

684safeCompareLoop = \s1 s2 n result ->

674− if n <= 0 && s1 == "" && s2 == ""

685+ if n <= 0

675 then result

686 then result

676 else safeCompareLoop (drop 1 s1) (drop 1 s2) (n - 1) (result + (if take 1 s1 == take 1 s2 then 0 else 1))

687 else safeCompareLoop (drop 1 s1) (drop 1 s2) (n - 1) (result + (if take 1 s1 == take 1 s2 then 0 else 1))

677

688

@@ -816,15 +827,23 @@ validGossipEvents = \events t ->

816 GossipOffline {} -> {online: acc.online, offline: union acc.offline (validate e)}

827 GossipOffline {} -> {online: acc.online, offline: union acc.offline (validate e)}

817 _ -> acc) {online: [], offline: []} events

828 _ -> acc) {online: [], offline: []} events

818

829

819−-- Execute an action for each server in parallel (fork per server)

830+-- Execute an action for each server, forking up to maxFederationConcurrency

831+-- and processing the remainder sequentially to bound resource usage.

820sendToEach : [ServerName] -> (ServerName -> (a {})) -> a {}

832sendToEach : [ServerName] -> (ServerName -> (a {})) -> a {}

821−sendToEach = \servers action ->

833+sendToEach = \servers action -> sendToEachN servers action maxFederationConcurrency

834+

835+sendToEachN = \servers action remaining ->

822 case headOf servers of

836 case headOf servers of

823 Nothing {} -> yield {}

837 Nothing {} -> yield {}

824 Just {value: server} -> do

838 Just {value: server} -> do

825 let rest = do s <- servers; where s != server; yield s

839 let rest = do s <- servers; where s != server; yield s

840+ if remaining > 0

841+ then do

826− fork (action server)

842+ fork (action server)

827− sendToEach rest action

843+ sendToEachN rest action (remaining - 1)

844+ else do

845+ action server

846+ sendToEachN rest action 0

828

847

829-- Remove all expired entries from every collection. Returns expired local presence

848-- Remove all expired entries from every collection. Returns expired local presence

830-- entries (for broadcasting offline gossip) and the set of peer servers known before pruning.

849-- entries (for broadcasting offline gossip) and the set of peer servers known before pruning.

@@ -861,6 +880,8 @@ pruneExpired = \t -> do

861 set *lookupRates = do r <- lookupRates; where t - r.windowStart < lookupRateWindowMs; yield r

880 set *lookupRates = do r <- lookupRates; where t - r.windowStart < lookupRateWindowMs; yield r

862 recvRates <- *recvRates

881 recvRates <- *recvRates

863 set *recvRates = do r <- recvRates; where t - r.windowStart < recvRateWindowMs; yield r

882 set *recvRates = do r <- recvRates; where t - r.windowStart < recvRateWindowMs; yield r

883+ recentGossip <- *recentGossip

884+ set *recentGossip = do r <- recentGossip; where t - r.seenAt < gossipDedupWindow; yield r

864 ackOverflow <- *ackOverflow

885 ackOverflow <- *ackOverflow

865 set *ackOverflow = do a <- ackOverflow; where a.ackedAt > retainCutoff; yield a

886 set *ackOverflow = do a <- ackOverflow; where a.ackedAt > retainCutoff; yield a

866 deliveredIds <- *deliveredIds

887 deliveredIds <- *deliveredIds

@@ -1601,6 +1622,9 @@ verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t ->

1601-- Acks for federated messages are sent to origin servers in the background.

1622-- Acks for federated messages are sent to origin servers in the background.

1602handlePoll = \cursor authorization respond -> do

1623handlePoll = \cursor authorization respond -> do

1603 t <- now

1624 t <- now

1625+ draining <- *draining

1626+ if draining then yield (respond {events: [], cursor: 0, authorized: False {}, rateLimited: False {}})

1627+ else do

1604 globalRateLimited <- checkGlobalRate t

1628 globalRateLimited <- checkGlobalRate t

1605 if globalRateLimited then yield (respond {events: [], cursor: 0, authorized: False {}, rateLimited: True {}})

1629 if globalRateLimited then yield (respond {events: [], cursor: 0, authorized: False {}, rateLimited: True {}})

1606 else do

1630 else do

@@ -1619,6 +1643,9 @@ handlePoll = \cursor authorization respond -> do

1619-- Rate-limited per sender key. If recipient is known on a remote server, forwards via federation.

1643-- Rate-limited per sender key. If recipient is known on a remote server, forwards via federation.

1620handleSendMessage = \to encryptedBlob timestamp messageId authorization respond -> do

1644handleSendMessage = \to encryptedBlob timestamp messageId authorization respond -> do

1621 t <- now

1645 t <- now

1646+ draining <- *draining

1647+ if draining then yield (respond {status: MsRejected {}, messageId: ""})

1648+ else do

1622 globalRateLimited <- checkGlobalRate t

1649 globalRateLimited <- checkGlobalRate t

1623 if globalRateLimited then yield (respond {status: MsRateLimited {}, messageId: ""})

1650 if globalRateLimited then yield (respond {status: MsRateLimited {}, messageId: ""})

1624 else do

1651 else do

@@ -1706,6 +1733,9 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization respond

1706-- Returns local server if key is online locally, otherwise the remote with latest expiry.

1733-- Returns local server if key is online locally, otherwise the remote with latest expiry.

1707handleLookupPresence = \pubkey authorization respond -> do

1734handleLookupPresence = \pubkey authorization respond -> do

1708 t <- now

1735 t <- now

1736+ draining <- *draining

1737+ if draining then yield (respond {server: "", authorized: False {}})

1738+ else do

1709 globalRateLimited <- checkGlobalRate t

1739 globalRateLimited <- checkGlobalRate t

1710 if globalRateLimited then yield (respond {server: "", authorized: False {}})

1740 if globalRateLimited then yield (respond {server: "", authorized: False {}})

1711 else do

1741 else do

@@ -1725,7 +1755,10 @@ handleLookupPresence = \pubkey authorization respond -> do

1725 else do

1755 else do

1726 presence <- *presence

1756 presence <- *presence

1727 let foundServer = bestServerForKey (presenceForKey presence pubkey t)

1757 let foundServer = bestServerForKey (presenceForKey presence pubkey t)

1758+ -- Mask remote server names: clients only need to know local/remote/unknown,

1759+ -- not which specific peer server a key is homed on (prevents metadata leakage)

1760+ let maskedServer = if foundServer == "" then "" else if isLocalServer foundServer then foundServer else "remote"

1728− yield {pr: pr, server: foundServer, authorized: True {}})

1761+ yield {pr: pr, server: maskedServer, authorized: True {}})

1729 forkBroadcastOffline t result.pr

1762 forkBroadcastOffline t result.pr

1730 yield (respond {server: result.server, authorized: result.authorized})

1763 yield (respond {server: result.server, authorized: result.authorized})

1731

1764

@@ -1752,10 +1785,19 @@ handleRecvGossip = \events hopsRemaining fromServer authorization respond -> do

1752 let effectiveEvents = union (effectiveGossipEvents filteredOnline gossipResult.gm.appliedKeys (GossipOnline {})) (effectiveGossipEvents filteredOffline gossipResult.gm.appliedKeys (GossipOffline {}))

1785 let effectiveEvents = union (effectiveGossipEvents filteredOnline gossipResult.gm.appliedKeys (GossipOnline {})) (effectiveGossipEvents filteredOffline gossipResult.gm.appliedKeys (GossipOffline {}))

1753 when (not gossipResult.gm.withinCapacity)

1786 when (not gossipResult.gm.withinCapacity)

1754 (println ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer))

1787 (println ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer))

1788+ -- Suppress re-broadcast of events we've already gossiped recently (prevents amplification)

1789+ freshEvents <- atomic do

1790+ recentGossip <- *recentGossip

1791+ let active = do r <- recentGossip; where t - r.seenAt < gossipDedupWindow; yield r

1792+ let fresh = do e <- effectiveEvents; where none active (\r -> r.pubkey == e.pubkey && r.server == e.server && r.timestamp == e.timestamp); yield e

1793+ let newRecent = do e <- fresh; yield {pubkey: e.pubkey, server: e.server, timestamp: e.timestamp, seenAt: t}

1794+ let updated = union active newRecent

1795+ set *recentGossip = if count updated > maxRecentGossip then do r <- updated; where t - r.seenAt < gossipDedupWindow; yield r else updated

1796+ yield fresh

1755 let newlyOnlineRemote = newlyOnlineRemoteKeys filteredOnline gossipResult.gm.appliedKeys

1797 let newlyOnlineRemote = newlyOnlineRemoteKeys filteredOnline gossipResult.gm.appliedKeys

1756 fork (do

1798 fork (do

1757− when (cappedHops > 1 && federationEnabled && count effectiveEvents > 0) (

1799+ when (cappedHops > 1 && federationEnabled && count freshEvents > 0) (

1758− sendBatchedGossip (gossipTargets gossipResult.gm.gossipPeerServers fromServer) effectiveEvents (cappedHops - 1))

1800+ sendBatchedGossip (gossipTargets gossipResult.gm.gossipPeerServers fromServer) freshEvents (cappedHops - 1))

1759 broadcastOfflineGossip t gossipResult.pr.peerServers gossipResult.pr.expiredLocal

1801 broadcastOfflineGossip t gossipResult.pr.peerServers gossipResult.pr.expiredLocal

1760 when (federationEnabled && count newlyOnlineRemote > 0)

1802 when (federationEnabled && count newlyOnlineRemote > 0)

1761 (forwardQueuedMessages newlyOnlineRemote t))

1803 (forwardQueuedMessages newlyOnlineRemote t))

@@ -1763,7 +1805,7 @@ handleRecvGossip = \events hopsRemaining fromServer authorization respond -> do

1763

1805

1764-- POST /federation/forward: Receive a message forwarded from a peer server.

1806-- POST /federation/forward: Receive a message forwarded from a peer server.

1765-- Only accepts messages where fromServer == originServer (no cascade forwarding).

1807-- Only accepts messages where fromServer == originServer (no cascade forwarding).

1766−-- Deduplicates and stores; rejects if recipient mailbox is full.

1808+-- Deduplicates and stores; rejects if recipient mailbox is full or recv rate exceeded.

1767handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fromServer authorization respond -> do

1809handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fromServer authorization respond -> do

1768 t <- now

1810 t <- now

1769 withFederationAuth authorization t respond fromServer (not (isLocalServer originServer) && fromServer == originServer) (do

1811 withFederationAuth authorization t respond fromServer (not (isLocalServer originServer) && fromServer == originServer) (do

@@ -1778,15 +1820,24 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro

1778 else if countMessagesFor messages toKey t >= maxMessagesPerRecipient

1820 else if countMessagesFor messages toKey t >= maxMessagesPerRecipient

1779 then yield {pr: pr, forward: ForwardRejected {}}

1821 then yield {pr: pr, forward: ForwardRejected {}}

1780 else do

1822 else do

1823+ recvRates <- *recvRates

1824+ let recvResult = checkRateWindow recvRates (\r -> r.pubkey == toKey)

1825+ (\ts -> {pubkey: toKey, windowStart: ts, count: 1})

1826+ (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})

1827+ maxRecvRatePerKey recvRateWindowMs maxRecvRateEntries t

1828+ if recvResult.limited

1829+ then yield {pr: pr, forward: ForwardRejected {}}

1830+ else do

1831+ set *recvRates = recvResult.entries

1781− seqCounter <- *seqCounter

1832+ seqCounter <- *seqCounter

1782− let seqTs = nextSeqTs t seqCounter

1833+ let seqTs = nextSeqTs t seqCounter

1783− set *seqCounter = seqTs

1834+ set *seqCounter = seqTs

1784− let updatedMessages = union messages

1835+ let updatedMessages = union messages

1785− [{messageId: messageId, toKey: toKey,

1836+ [{messageId: messageId, toKey: toKey,

1786− encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs}]

1837+ encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs}]

1787− set *messages = updatedMessages

1838+ set *messages = updatedMessages

1788− signalMessageForKey toKey seqTs

1839+ signalMessageForKey toKey seqTs

1789− yield {pr: pr, forward: ForwardStored {}}

1840+ yield {pr: pr, forward: ForwardStored {}}

1790 case result.forward of

1841 case result.forward of

1791 ForwardRejected {} -> println ("forward: rejected for " ++ toKey ++ " (mailbox full, max " ++ show maxMessagesPerRecipient ++ ")")

1842 ForwardRejected {} -> println ("forward: rejected for " ++ toKey ++ " (mailbox full, max " ++ show maxMessagesPerRecipient ++ ")")

1792 _ -> yield {}

1843 _ -> yield {}

@@ -1864,12 +1915,23 @@ handleHealth = \respond -> do

1864 lastPrune <- *lastPrunedAt

1915 lastPrune <- *lastPrunedAt

1865 lastHb <- *lastHeartbeatAt

1916 lastHb <- *lastHeartbeatAt

1866 sa <- *startedAt

1917 sa <- *startedAt

1918+ presence <- *presence

1919+ forwards <- *forwards

1920+ sessions <- *sessions

1921+ activePolls <- *activePolls

1867 let warmup = t - sa < healthPruneMaxAge

1922 let warmup = t - sa < healthPruneMaxAge

1923+ -- Capacity pressure: flag when any capped collection exceeds 90% full

1924+ let capacityOk = count presence * 10 <= maxPresenceEntries * 9

1925+ && count forwards * 10 <= maxForwardsEntries * 9

1926+ && count sessions * 10 <= maxSessions * 9

1927+ && count activePolls * 10 <= maxActivePolls * 9

1868 yield {pruneOk: warmup || t - lastPrune < healthPruneMaxAge,

1928 yield {pruneOk: warmup || t - lastPrune < healthPruneMaxAge,

1869− heartbeatOk: warmup || t - lastHb < healthHeartbeatMaxAge}

1929+ heartbeatOk: warmup || t - lastHb < healthHeartbeatMaxAge,

1930+ capacityOk: capacityOk}

1870− yield (respond {ok: result.pruneOk && result.heartbeatOk,

1931+ yield (respond {ok: result.pruneOk && result.heartbeatOk && result.capacityOk,

1871 pruneHealthy: result.pruneOk,

1932 pruneHealthy: result.pruneOk,

1872− heartbeatHealthy: result.heartbeatOk})

1933+ heartbeatHealthy: result.heartbeatOk,

1934+ capacityHealthy: result.capacityOk})

1873

1935

1874handleStats = \authorization respond -> do

1936handleStats = \authorization respond -> do

1875 let authorized = verifyMaintenanceAuth authorization

1937 let authorized = verifyMaintenanceAuth authorization