@@ -1189,14 +1189,21 @@ uniqueBucketsFor = \items getKey ->
1189 let b = pubkeyBucket (getKey x) in
1189 let b = pubkeyBucket (getKey x) in
1190 if any acc (\i -> i == b) then acc else union acc [b]) [] items
1190 if any acc (\i -> i == b) then acc else union acc [b]) [] items
1192−-- Prune a single message bucket and return kept count + per-recipient counts.
1192+-- Prune a single message bucket and atomically adjust shared counters.
1193−-- Since messages are sharded by recipient pubkey, each bucket's recipient counts
1193+-- Each bucket prune decrements totalMessageCount and recipientMsgCounts by exactly
1194−-- are disjoint from other buckets — simple union merges them.
1194+-- what was removed, preserving concurrent modifications from poll-ack or send.
1195−pruneMessageBucketCounting = \b messageCutoff -> atomic do
1195+pruneMessageBucket = \b messageCutoff -> atomic do
1196 bucket <- readMessageBucket b
1196 bucket <- readMessageBucket b
1197 let kept = do m <- bucket; where m.receivedAt > messageCutoff; yield m
1197 let kept = do m <- bucket; where m.receivedAt > messageCutoff; yield m
1198+ let removedCount = count bucket - count kept
1198 writeMessageBucket b kept
1199 writeMessageBucket b kept
1199− yield {count: count kept, recipients: recomputeRecipientCounts kept}
1200+ if removedCount > 0 then do
1201+ tc <- *totalMessageCount
1202+ set *totalMessageCount = max 0 (tc - removedCount)
1203+ rc <- *recipientMsgCounts
1204+ let removed = do m <- bucket; where m.receivedAt <= messageCutoff; yield m
1205+ set *recipientMsgCounts = batchDecrementRecipientCounts rc (countByRecipient removed)
1201-- ============================================
1208-- ============================================
1202-- Recipient Count Helpers
1209-- Recipient Count Helpers
@@ -1223,9 +1230,6 @@ countByRecipient = \items ->
1223 then do c <- acc; yield (if c.pubkey == x.toKey then {pubkey: c.pubkey, count: c.count + 1} else c)
1230 then do c <- acc; yield (if c.pubkey == x.toKey then {pubkey: c.pubkey, count: c.count + 1} else c)
1224 else union acc [{pubkey: x.toKey, count: 1}]) [] items
1231 else union acc [{pubkey: x.toKey, count: 1}]) [] items
1226−recomputeRecipientCounts : [Message] -> [RecipientCount]
1227−recomputeRecipientCounts = countByRecipient
1229-- Batch-decrement recipient counts: subtract per-key removal counts in a single pass
1233-- Batch-decrement recipient counts: subtract per-key removal counts in a single pass
1230-- instead of calling adjustRecipientCount in a loop.
1234-- instead of calling adjustRecipientCount in a loop.
1231batchDecrementRecipientCounts : [RecipientCount] -> [RecipientCount] -> [RecipientCount]
1235batchDecrementRecipientCounts : [RecipientCount] -> [RecipientCount] -> [RecipientCount]
@@ -1330,6 +1334,12 @@ safeCompare = \a b ->
1330 if take 1 a == "" || take 1 b == "" then False {}
1334 if take 1 a == "" || take 1 b == "" then False {}
1331 else safeCompareLoop a b maxTokenLen 0 0 == 0
1335 else safeCompareLoop a b maxTokenLen 0 0 == 0
1337+-- Constant-time IP binding check: returns True if IPs match or either is empty (no binding).
1338+-- Uses safeCompareLoop to prevent timing side-channels on IP comparison.
1339+matchIp : Text -> Text -> Bool
1340+matchIp = \clientIp sessionIp ->
1341+ clientIp == "" || sessionIp == "" || safeCompareLoop clientIp sessionIp maxServerNameLen 0 0 == 0
1333-- Generic sliding-window rate limiter. Returns whether the request is rate-limited
1343-- Generic sliding-window rate limiter. Returns whether the request is rate-limited
1334-- and the updated entries list. Callers provide match/create/update functions for their entry type.
1344-- and the updated entries list. Callers provide match/create/update functions for their entry type.
1335checkRateWindow = \entries matchKey makeNew updateCount maxRate windowMs maxEntries t ->
1345checkRateWindow = \entries matchKey makeNew updateCount maxRate windowMs maxEntries t ->
@@ -1509,31 +1519,26 @@ pruneExpired = \t -> do
1509 transformAllSessionBuckets (\bucket -> do x <- bucket; where x.expiresAt > t; yield x)
1519 transformAllSessionBuckets (\bucket -> do x <- bucket; where x.expiresAt > t; yield x)
1510 sessionCount <- countAllSessions
1520 sessionCount <- countAllSessions
1511 set *totalSessionCount = sessionCount
1521 set *totalSessionCount = sessionCount
1512− -- Prune messages across all 16 buckets. Each bucket prune is atomic (short lock).
1522+ -- Prune messages across all 16 buckets. Each bucket prune is atomic and adjusts
1513− -- Recipient counts are rebuilt incrementally per-bucket — since messages are sharded
1523+ -- shared counters (totalMessageCount, recipientMsgCounts) by delta, preserving
1514− -- by recipient pubkey, each bucket's counts are disjoint, so simple union merges them.
1524+ -- concurrent modifications from poll-ack or send-message operations.
1515 let messageCutoff = t - unackedMessageMaxAge
1525 let messageCutoff = t - unackedMessageMaxAge
1516− r0 <- pruneMessageBucketCounting 0 messageCutoff
1526+ pruneMessageBucket 0 messageCutoff
1517− r1 <- pruneMessageBucketCounting 1 messageCutoff
1527+ pruneMessageBucket 1 messageCutoff
1518− r2 <- pruneMessageBucketCounting 2 messageCutoff
1528+ pruneMessageBucket 2 messageCutoff
1519− r3 <- pruneMessageBucketCounting 3 messageCutoff
1529+ pruneMessageBucket 3 messageCutoff
1520− r4 <- pruneMessageBucketCounting 4 messageCutoff
1530+ pruneMessageBucket 4 messageCutoff
1521− r5 <- pruneMessageBucketCounting 5 messageCutoff
1531+ pruneMessageBucket 5 messageCutoff
1522− r6 <- pruneMessageBucketCounting 6 messageCutoff
1532+ pruneMessageBucket 6 messageCutoff
1523− r7 <- pruneMessageBucketCounting 7 messageCutoff
1533+ pruneMessageBucket 7 messageCutoff
1524− r8 <- pruneMessageBucketCounting 8 messageCutoff
1534+ pruneMessageBucket 8 messageCutoff
1525− r9 <- pruneMessageBucketCounting 9 messageCutoff
1535+ pruneMessageBucket 9 messageCutoff
1526− r10 <- pruneMessageBucketCounting 10 messageCutoff
1536+ pruneMessageBucket 10 messageCutoff
1527− r11 <- pruneMessageBucketCounting 11 messageCutoff
1537+ pruneMessageBucket 11 messageCutoff
1528− r12 <- pruneMessageBucketCounting 12 messageCutoff
1538+ pruneMessageBucket 12 messageCutoff
1529− r13 <- pruneMessageBucketCounting 13 messageCutoff
1539+ pruneMessageBucket 13 messageCutoff
1530− r14 <- pruneMessageBucketCounting 14 messageCutoff
1540+ pruneMessageBucket 14 messageCutoff
1531− r15 <- pruneMessageBucketCounting 15 messageCutoff
1541+ pruneMessageBucket 15 messageCutoff
1532− let prunedTotal = r0.count + r1.count + r2.count + r3.count + r4.count + r5.count + r6.count + r7.count + r8.count + r9.count + r10.count + r11.count + r12.count + r13.count + r14.count + r15.count
1533− let prunedRecipients = union r0.recipients (union r1.recipients (union r2.recipients (union r3.recipients (union r4.recipients (union r5.recipients (union r6.recipients (union r7.recipients (union r8.recipients (union r9.recipients (union r10.recipients (union r11.recipients (union r12.recipients (union r13.recipients (union r14.recipients r15.recipients))))))))))))))
1535− set *totalMessageCount = prunedTotal
1536− set *recipientMsgCounts = prunedRecipients
1537 -- Prune federation state
1542 -- Prune federation state
1538 atomic do
1543 atomic do
1539 let retainCutoff = t - messageMaxAge
1544 let retainCutoff = t - messageMaxAge
@@ -1768,7 +1773,7 @@ withSessionAuth = \authorization t clientIp onUnauth onAuth -> do
1768 case authenticateSession sessions token t of
1773 case authenticateSession sessions token t of
1769 Unauthorized {} -> onUnauth
1774 Unauthorized {} -> onUnauth
1770 Authorized {pubkey, ip} ->
1775 Authorized {pubkey, ip} ->
1771− if clientIp != "" && ip != "" && clientIp != ip
1776+ if not (matchIp clientIp ip)
1772 then onUnauth
1777 then onUnauth
1773 else onAuth pubkey
1778 else onAuth pubkey
@@ -2219,7 +2224,7 @@ pollOnce = \token cursor clientIp -> do
2219 case authenticateSession sessions token t of
2224 case authenticateSession sessions token t of
2220 Unauthorized {} -> yield (Nothing {})
2225 Unauthorized {} -> yield (Nothing {})
2221 Authorized {pubkey, ip} ->
2226 Authorized {pubkey, ip} ->
2222− if clientIp != "" && ip != "" && clientIp != ip
2227+ if not (matchIp clientIp ip)
2223 then yield (Nothing {})
2228 then yield (Nothing {})
2225 -- Refresh session expiry (sliding window keeps active sessions alive)
2230 -- Refresh session expiry (sliding window keeps active sessions alive)
@@ -2831,6 +2836,9 @@ handleLogout = \authorization clientIp respond -> do
2831 t <- now
2836 t <- now
2832 globalRateLimited <- checkGlobalRate t
2837 globalRateLimited <- checkGlobalRate t
2833 if globalRateLimited then yield (respond {ok: False {}, authorized: False {}})
2838 if globalRateLimited then yield (respond {ok: False {}, authorized: False {}})
2840+ authLimited <- checkAuthGlobalRate t
2841+ if authLimited then yield (respond {ok: False {}, authorized: False {}})
2835 case extractToken authorization of
2843 case extractToken authorization of
2836 Nothing {} -> yield (respond {ok: False {}, authorized: False {}})
2844 Nothing {} -> yield (respond {ok: False {}, authorized: False {}})
@@ -2933,6 +2941,13 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
2933 withFederationAuth authorization t respond fromServer requestTimestamp (not (isLocalServer originServer) && not (isLocalServer fromServer)) (do
2941 withFederationAuth authorization t respond fromServer requestTimestamp (not (isLocalServer originServer) && not (isLocalServer fromServer)) (do
2934 when (fromServer != originServer)
2942 when (fromServer != originServer)
2935 (println ("forward: relay from " ++ fromServer ++ " (origin: " ++ originServer ++ ") for " ++ toKey))
2943 (println ("forward: relay from " ++ fromServer ++ " (origin: " ++ originServer ++ ") for " ++ toKey))
2944+ -- Relay origin validation: when per-peer secrets are configured, only accept
2945+ -- relays claiming an origin that is also a configured peer
2946+ if fromServer != originServer && count federationPeerSecrets > 0 && none federationPeerSecrets (\p -> normalizeServerName p.server == originServer)
2948+ println ("forward: rejecting relay with unknown origin " ++ originServer ++ " from " ++ fromServer)
2949+ yield (respond {ok: False {}, error: "unknown_origin"})
2936 -- Replay protection: reject forwards with timestamps older than messageMaxAge.
2951 -- Replay protection: reject forwards with timestamps older than messageMaxAge.
2937 -- Legitimate queued messages have timestamps within this window. Replayed captures
2952 -- Legitimate queued messages have timestamps within this window. Replayed captures
2938 -- become stale once dedup state (messages + ackedIds + deliveredIds) is pruned (~30 days).
2953 -- become stale once dedup state (messages + ackedIds + deliveredIds) is pruned (~30 days).
@@ -3408,6 +3423,8 @@ main = do
3408 && maxSessions > 0 && maxChallenges > 0 && backgroundPruneInterval > 0
3423 && maxSessions > 0 && maxChallenges > 0 && backgroundPruneInterval > 0
3409 && pollTimeoutMs > 0 && challengeExpiry > 0 && sessionExpiry > 0
3424 && pollTimeoutMs > 0 && challengeExpiry > 0 && sessionExpiry > 0
3410 && maxPresenceTtl > 0 && maxForwardsEntries > 0 && maxAckedIds > 0
3425 && maxPresenceTtl > 0 && maxForwardsEntries > 0 && maxAckedIds > 0
3426+ && maxForwardRetries > 0 && maxGossipHops > 0 && maxPollsPerKey > 0
3427+ && gossipOriginHops > 0
3411 -- Relationship invariants: these misconfigurations cause subtle bugs
3428 -- Relationship invariants: these misconfigurations cause subtle bugs
3412 -- (e.g., messages pruned before their stated max age, backoff that never increases)
3429 -- (e.g., messages pruned before their stated max age, backoff that never increases)
3413 let configRelationsOk = unackedMessageMaxAge <= messageMaxAge
3430 let configRelationsOk = unackedMessageMaxAge <= messageMaxAge
@@ -3424,11 +3441,12 @@ main = do
3424 && gossipDedupWindow < presenceRefreshInterval
3441 && gossipDedupWindow < presenceRefreshInterval
3425 && federationFetchTimeoutMs <= backgroundPruneInterval
3442 && federationFetchTimeoutMs <= backgroundPruneInterval
3426 && maxAckOverflow <= maxAckedIds
3443 && maxAckOverflow <= maxAckedIds
3444+ && gossipOriginHops <= maxGossipHops
3427 let peerSecretsOk = none federationPeerSecrets (\p -> isBadServerName (normalizeServerName p.server) || p.secret == "")
3445 let peerSecretsOk = none federationPeerSecrets (\p -> isBadServerName (normalizeServerName p.server) || p.secret == "")
3428 if not configOk
3446 if not configOk
3429 then println "FATAL: invalid configuration — capacity limits and intervals must be > 0, server not started"
3447 then println "FATAL: invalid configuration — capacity limits and intervals must be > 0, server not started"
3430 else if not configRelationsOk
3448 else if not configRelationsOk
3431− then println "FATAL: invalid configuration — relationship invariants violated (check: unackedMessageMaxAge <= messageMaxAge, challengeExpiry < sessionExpiry, retryBackoffBase <= retryBackoffMax, backgroundPruneInterval < healthPruneMaxAge, pollTimeoutMs < sessionExpiry, gossipDedupWindow < presenceRefreshInterval, federationFetchTimeoutMs <= backgroundPruneInterval, maxAckOverflow <= maxAckedIds, per-key limits <= global limits), server not started"
3449+ then println "FATAL: invalid configuration — relationship invariants violated (check: unackedMessageMaxAge <= messageMaxAge, challengeExpiry < sessionExpiry, retryBackoffBase <= retryBackoffMax, backgroundPruneInterval < healthPruneMaxAge, pollTimeoutMs < sessionExpiry, gossipDedupWindow < presenceRefreshInterval, federationFetchTimeoutMs <= backgroundPruneInterval, maxAckOverflow <= maxAckedIds, gossipOriginHops <= maxGossipHops, per-key limits <= global limits), server not started"
3432 else if not peerSecretsOk
3450 else if not peerSecretsOk
3433 then println "FATAL: invalid federationPeerSecrets — server names must pass validation and secrets must be non-empty"
3451 then println "FATAL: invalid federationPeerSecrets — server names must pass validation and secrets must be non-empty"