839f0fb · skrepka · git.iko.soy

git.iko.soy/skrepka

Enhance server security, reliability, performance: challenge IP binding, timing leak fix, atomic block split, federation error logging, optimized message counting

839f0fbiko6 months ago

server.knot+167−133

compare with master

unifiedsplit

@@ -682,10 +682,10 @@ nextSeqTs = \t seqCounter -> if t > seqCounter then t else seqCounter + 1

682-- Deduplicate message IDs, preserving order, capped at maxMessageIds

683dedupMessageIds : [MessageIdHex] -> [MessageIdHex]

684dedupMessageIds = \messageIds ->

685+ let capped = takeRelation maxMessageIds messageIds in

685 let result = fold (\acc mid ->

686 let result = fold (\acc mid ->

686− if acc.n >= maxMessageIds then acc

687− else if any acc.items (\a -> a == mid) then acc

687+ if any acc.items (\a -> a == mid) then acc

688− else {n: acc.n + 1, items: union acc.items [mid]}) {n: 0, items: []} messageIds

688+ else {n: acc.n + 1, items: union acc.items [mid]}) {n: 0, items: []} capped

689 in result.items

690

691-- Remove the N oldest acked entries (by ackedAt timestamp)

@@ -707,12 +707,8 @@ safeCompareLoop = \s1 s2 n result ->

707

708safeCompare : Text -> Text -> Bool

709safeCompare = \a b ->

710− let lenA = count (chars a) in

710+ if a == "" || b == "" then False {}

711− let lenB = count (chars b) in

712− safeCompareLoop a b (max maxTokenLen (max lenA lenB)) 0

711+ else safeCompareLoop a b maxTokenLen 0 == 0

713− + (if lenA != lenB then 1 else 0)

714− + (if a == "" then 1 else 0)

715− + (if b == "" then 1 else 0) == 0

716

712

717-- Generic sliding-window rate limiter. Returns whether the request is rate-limited

713-- Generic sliding-window rate limiter. Returns whether the request is rate-limited

718-- and the updated entries list. Callers provide match/create/update functions for their entry type.

714-- and the updated entries list. Callers provide match/create/update functions for their entry type.

@@ -825,6 +821,14 @@ countMessagesFor : [Message] -> PubkeyHex -> Timestamp -> Int

825countMessagesFor = \messages toKey t ->

821countMessagesFor = \messages toKey t ->

826 count (do m <- messages; where m.toKey == toKey && m.receivedAt >= t - messageMaxAge; yield m)

822 count (do m <- messages; where m.toKey == toKey && m.receivedAt >= t - messageMaxAge; yield m)

827

823

824+-- Count total messages and messages for a specific key in a single pass (avoids two O(n) scans)

825+messageCounts : [Message] -> PubkeyHex -> Timestamp -> {total: Int, forKey: Int}

826+messageCounts = \messages toKey t ->

827+ fold (\acc m ->

828+ {total: acc.total + 1,

829+ forKey: if m.toKey == toKey && m.receivedAt >= t - messageMaxAge then acc.forKey + 1 else acc.forKey}

830+ ) {total: 0, forKey: 0} messages

831+

828hasMessage : MessageIdHex -> PubkeyHex -> [{messageId: MessageIdHex, toKey: PubkeyHex | a}] -> Bool

832hasMessage : MessageIdHex -> PubkeyHex -> [{messageId: MessageIdHex, toKey: PubkeyHex | a}] -> Bool

829hasMessage = \mid toKey items -> any items (\x -> x.messageId == mid && x.toKey == toKey)

833hasMessage = \mid toKey items -> any items (\x -> x.messageId == mid && x.toKey == toKey)

830

834

@@ -867,54 +871,69 @@ sendToEachN = \servers action remaining ->

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

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

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

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

869pruneExpired = \t -> do

873pruneExpired = \t -> do

870− challenges <- *challenges

874+ -- Prune presence (must be atomic for before/after comparison)

871− set *challenges = do x <- challenges; where x.expiresAt > t; yield x

875+ presenceResult <- atomic do

872− sessions <- *sessions

873− set *sessions = do x <- sessions; where x.expiresAt > t; yield x

874− presenceBefore <- *presence

876+ presenceBefore <- *presence

875− let peerServersBefore = collectPeerServers presenceBefore (\p -> p.server)

877+ let peerServersBefore = collectPeerServers presenceBefore (\p -> p.server)

876− let presenceAfter = do x <- presenceBefore; where x.expiresAt > t; yield x

878+ let presenceAfter = do x <- presenceBefore; where x.expiresAt > t; yield x

877− set *presence = presenceAfter

879+ set *presence = presenceAfter

880+ yield {expiredLocal: expiredLocalPresence presenceBefore presenceAfter t, peerServers: peerServersBefore}

881+ -- Prune auth state

882+ atomic do

883+ challenges <- *challenges

884+ set *challenges = do x <- challenges; where x.expiresAt > t; yield x

885+ sessions <- *sessions

886+ set *sessions = do x <- sessions; where x.expiresAt > t; yield x

887+ -- Prune messages (largest collection, isolated for minimal lock time)

888+ atomic do

878− let messageCutoff = t - unackedMessageMaxAge

889+ let messageCutoff = t - unackedMessageMaxAge

879− let retainCutoff = t - messageMaxAge

880− messages <- *messages

890+ messages <- *messages

881− set *messages = do m <- messages; where m.receivedAt > messageCutoff; yield m

891+ set *messages = do m <- messages; where m.receivedAt > messageCutoff; yield m

892+ -- Prune federation state

893+ atomic do

894+ let retainCutoff = t - messageMaxAge

882− forwards <- *forwards

895+ forwards <- *forwards

883− set *forwards = do f <- forwards; where f.createdAt > retainCutoff; yield f

896+ set *forwards = do f <- forwards; where f.createdAt > retainCutoff; yield f

884− ackedIds <- *ackedIds

897+ ackedIds <- *ackedIds

885− set *ackedIds = do a <- ackedIds; where a.ackedAt > retainCutoff; yield a

898+ set *ackedIds = do a <- ackedIds; where a.ackedAt > retainCutoff; yield a

886− sendRates <- *sendRates

887− set *sendRates = do r <- sendRates; where t - r.windowStart < sendRateWindowMs; yield r

888− forwardFailures <- *forwardFailures

899+ forwardFailures <- *forwardFailures

889− set *forwardFailures = do f <- forwardFailures; where t - f.failedAt < maxServerDowntime; yield f

900+ set *forwardFailures = do f <- forwardFailures; where t - f.failedAt < maxServerDowntime; yield f

901+ ackOverflow <- *ackOverflow

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

903+ deliveredIds <- *deliveredIds

904+ set *deliveredIds = do d <- deliveredIds; where d.deliveredAt > retainCutoff; yield d

905+ -- Prune rate limiters and ephemeral state

906+ atomic do

907+ sendRates <- *sendRates

908+ set *sendRates = do r <- sendRates; where t - r.windowStart < sendRateWindowMs; yield r

890− challengeRates <- *challengeRates

909+ challengeRates <- *challengeRates

891− set *challengeRates = do r <- challengeRates; where t - r.windowStart < challengeRateWindowMs; yield r

910+ set *challengeRates = do r <- challengeRates; where t - r.windowStart < challengeRateWindowMs; yield r

892− federationRates <- *federationRates

911+ federationRates <- *federationRates

893− set *federationRates = do r <- federationRates; where t - r.windowStart < federationRateWindowMs; yield r

912+ set *federationRates = do r <- federationRates; where t - r.windowStart < federationRateWindowMs; yield r

894− activePolls <- *activePolls

913+ activePolls <- *activePolls

895− set *activePolls = do p <- activePolls; where t - p.startedAt < pollTimeoutMs * 3; yield p

914+ set *activePolls = do p <- activePolls; where t - p.startedAt < pollTimeoutMs * 3; yield p

896− ipRates <- *ipRates

915+ ipRates <- *ipRates

897− set *ipRates = do r <- ipRates; where t - r.windowStart < ipRateWindowMs; yield r

916+ set *ipRates = do r <- ipRates; where t - r.windowStart < ipRateWindowMs; yield r

898− lookupRates <- *lookupRates

917+ lookupRates <- *lookupRates

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

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

900− recvRates <- *recvRates

919+ recvRates <- *recvRates

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

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

902− recentGossip <- *recentGossip

921+ recentGossip <- *recentGossip

903− set *recentGossip = do r <- recentGossip; where t - r.seenAt < gossipDedupWindow; yield r

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

904− ackOverflow <- *ackOverflow

923+ yield presenceResult

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

906− deliveredIds <- *deliveredIds

907− set *deliveredIds = do d <- deliveredIds; where d.deliveredAt > retainCutoff; yield d

908− yield {expiredLocal: expiredLocalPresence presenceBefore presenceAfter t, peerServers: peerServersBefore}

909

924

910-- Opportunistic pruning: runs at most once per second, triggered by request handlers

925-- Opportunistic pruning: runs at most once per second, triggered by request handlers

911maybePrune = \t -> do

926maybePrune = \t -> do

927+ shouldRun <- atomic do

912− lastPruned <- *lastPrunedAt

928+ lastPruned <- *lastPrunedAt

913− if t - lastPruned < 1000

929+ if t - lastPruned < 1000

930+ then yield (False {})

931+ else do

932+ set *lastPrunedAt = t

933+ yield (True {})

934+ if not shouldRun

914 then yield {expiredLocal: [], peerServers: []}

935 then yield {expiredLocal: [], peerServers: []}

915− else do

916− set *lastPrunedAt = t

917− pruneExpired t

936+ else pruneExpired t

918

937

919splitAtN = \items n ->

938splitAtN = \items n ->

920 fold (\acc e ->

939 fold (\acc e ->

@@ -936,7 +955,7 @@ sendBatchedGossipToServer = \server allEvents hops ->

936 authorization: federationAuth,

955 authorization: federationAuth,

937 events: batch, hopsRemaining: hops, fromServer: serverHostN})

956 events: batch, hopsRemaining: hops, fromServer: serverHostN})

938 case result of

957 case result of

939− Err _ -> println ("federation: gossip to " ++ server ++ " failed")

958+ Err e -> println ("federation: gossip to " ++ server ++ " failed: " ++ show e)

940 Ok _ -> yield {})

959 Ok _ -> yield {})

941

960

942sendBatchedGossip : [ServerName] -> [GossipEvent] -> Int -> IO {network} {}

961sendBatchedGossip : [ServerName] -> [GossipEvent] -> Int -> IO {network} {}

@@ -969,7 +988,7 @@ mergeGossipPresence = \presence deduped trustedOffline fromServer t ->

969 withinCapacity: capResult.withinCapacity}

988 withinCapacity: capResult.withinCapacity}

970

989

971maybePruneAndBroadcast = \t -> do

990maybePruneAndBroadcast = \t -> do

972− pruneResult <- atomic do maybePrune t

991+ pruneResult <- maybePrune t

973 forkBroadcastOffline t pruneResult

992 forkBroadcastOffline t pruneResult

974

993

975checkFederationRate = \server t -> atomic do

994checkFederationRate = \server t -> atomic do

@@ -1042,7 +1061,9 @@ withFederationAuth = \authorization t respond fromServer precondition action ->

1042 println ("federation: rate limited (" ++ show maxFederationRatePerServer ++ "/s per server), rejecting for " ++ fromServer)

1061 println ("federation: rate limited (" ++ show maxFederationRatePerServer ++ "/s per server), rejecting for " ++ fromServer)

1043 maybePruneAndBroadcast t

1062 maybePruneAndBroadcast t

1044 yield (respond {ok: False {}, error: "rate_limited"})

1063 yield (respond {ok: False {}, error: "rate_limited"})

1064+ else do

1065+ clearFailure fromServer

1045− else action

1066+ action

1046

1067

1047withSessionAuth = \authorization t onUnauth onAuth -> do

1068withSessionAuth = \authorization t onUnauth onAuth -> do

1048 sessions <- *sessions

1069 sessions <- *sessions

@@ -1098,8 +1119,8 @@ retryForwardsForServer = \entries remaining ->

1098 timestamp: f.timestamp, originServer: serverHostN, messageId: f.messageId,

1119 timestamp: f.timestamp, originServer: serverHostN, messageId: f.messageId,

1099 fromServer: serverHostN, authorization: federationAuth})

1120 fromServer: serverHostN, authorization: federationAuth})

1100 case result of

1121 case result of

1101− Err _ -> do

1122+ Err e -> do

1102− println ("federation: forward to " ++ f.toServer ++ " failed for " ++ f.messageId)

1123+ println ("federation: forward to " ++ f.toServer ++ " failed for " ++ f.messageId ++ ": " ++ show e)

1103 recordFailure f.toServer

1124 recordFailure f.toServer

1104 Ok _ -> do

1125 Ok _ -> do

1105 removeForward f.messageId f.toServer

1126 removeForward f.messageId f.toServer

@@ -1147,8 +1168,8 @@ sendBatchedAcksToServer = \serverAcks server ->

1147 forEachBatch serverAcks maxMessageIds (\batch -> do

1168 forEachBatch serverAcks maxMessageIds (\batch -> do

1148 result <- fetchFederationAck server (do a <- batch; yield a.messageId)

1169 result <- fetchFederationAck server (do a <- batch; yield a.messageId)

1149 case result of

1170 case result of

1150− Err _ -> do

1171+ Err e -> do

1151− println ("federation: ack to " ++ server ++ " failed")

1172+ println ("federation: ack to " ++ server ++ " failed: " ++ show e)

1152 recordFailure server

1173 recordFailure server

1153 Ok _ -> do

1174 Ok _ -> do

1154 clearFailure server

1175 clearFailure server

@@ -1177,7 +1198,8 @@ retryEvictedAcks = \evictedEntries pendingAcks ->

1177 sendToEach evictedOriginServers (\server -> do

1198 sendToEach evictedOriginServers (\server -> do

1178 fetchResult <- fetchFederationAck server (do a <- acksForServer evictedToRetry server; yield a.messageId)

1199 fetchResult <- fetchFederationAck server (do a <- acksForServer evictedToRetry server; yield a.messageId)

1179 case fetchResult of

1200 case fetchResult of

1180− Err _ -> do

1201+ Err e -> do

1202+ println ("federation: evicted ack to " ++ server ++ " failed: " ++ show e)

1181 overflow <- atomic do

1203 overflow <- atomic do

1182 ackedIds <- *ackedIds

1204 ackedIds <- *ackedIds

1183 let reinsert = acksForServer evictedToRetry server

1205 let reinsert = acksForServer evictedToRetry server

@@ -1220,8 +1242,8 @@ retryOverflowAcks = \overflow -> do

1220 forEachBatch (acksForServer overflow server) maxMessageIds (\batch -> do

1242 forEachBatch (acksForServer overflow server) maxMessageIds (\batch -> do

1221 result <- fetchFederationAck server (do a <- batch; yield a.messageId)

1243 result <- fetchFederationAck server (do a <- batch; yield a.messageId)

1222 case result of

1244 case result of

1223− Err _ -> do

1245+ Err e -> do

1224− println ("federation: overflow ack to " ++ server ++ " failed")

1246+ println ("federation: overflow ack to " ++ server ++ " failed: " ++ show e)

1225 recordFailure server

1247 recordFailure server

1226 Ok _ -> do

1248 Ok _ -> do

1227 clearFailure server

1249 clearFailure server

@@ -1233,9 +1255,9 @@ retryOverflowAcks = \overflow -> do

1233-- retry overflow acks, broadcast offline gossip, and refresh presence for local keys

1255-- retry overflow acks, broadcast offline gossip, and refresh presence for local keys

1234-- nearing expiry. Called by both the background task and the /maintenance/prune endpoint.

1256-- nearing expiry. Called by both the background task and the /maintenance/prune endpoint.

1235pruneAndRetry = \t -> do

1257pruneAndRetry = \t -> do

1258+ atomic do set *lastPrunedAt = t

1259+ pr <- pruneExpired t

1236 result <- atomic do

1260 result <- atomic do

1237− set *lastPrunedAt = t

1238− pr <- pruneExpired t

1239 forwards <- *forwards

1261 forwards <- *forwards

1240 ackedIds2 <- *ackedIds

1262 ackedIds2 <- *ackedIds

1241 let pendingAcks = do a <- ackedIds2; where not (isLocalServer a.originServer); yield a

1263 let pendingAcks = do a <- ackedIds2; where not (isLocalServer a.originServer); yield a

@@ -1247,8 +1269,8 @@ pruneAndRetry = \t -> do

1247 let localActive = if shouldRefresh then (do p <- presence; where isLocalServer p.server && p.expiresAt > t && p.expiresAt - t < presenceRefreshThreshold; yield p) else []

1269 let localActive = if shouldRefresh then (do p <- presence; where isLocalServer p.server && p.expiresAt > t && p.expiresAt - t < presenceRefreshThreshold; yield p) else []

1248 let refreshPeers = if shouldRefresh then collectPeerServers presence (\p -> p.server) else []

1270 let refreshPeers = if shouldRefresh then collectPeerServers presence (\p -> p.server) else []

1249 when shouldRefresh (set *lastPresenceRefresh = t)

1271 when shouldRefresh (set *lastPresenceRefresh = t)

1250− yield {expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, pendingForwards: forwards, pendingAcks: pendingAcks, pendingOverflow: pendingOverflow, localActive: localActive, refreshPeers: refreshPeers, shouldRefresh: shouldRefresh}

1272+ yield {pendingForwards: forwards, pendingAcks: pendingAcks, pendingOverflow: pendingOverflow, localActive: localActive, refreshPeers: refreshPeers, shouldRefresh: shouldRefresh}

1251− broadcastOfflineGossip t result.peerServers result.expiredLocal

1273+ broadcastOfflineGossip t pr.peerServers pr.expiredLocal

1252 when federationEnabled (do

1274 when federationEnabled (do

1253 retryPendingForwards result.pendingForwards

1275 retryPendingForwards result.pendingForwards

1254 sendAcksAndClean result.pendingAcks

1276 sendAcksAndClean result.pendingAcks

@@ -1474,6 +1496,10 @@ forwardCandidates = \targets messages forwards t ->

1474challengeMatch = \challenges pubkey challenge ip t ->

1496challengeMatch = \challenges pubkey challenge ip t ->

1475 any challenges (\c -> c.pubkey == pubkey && safeCompare c.challenge challenge && c.expiresAt > t && (ip == "" || c.ip == "" || c.ip == ip))

1497 any challenges (\c -> c.pubkey == pubkey && safeCompare c.challenge challenge && c.expiresAt > t && (ip == "" || c.ip == "" || c.ip == ip))

1476

1498

1499+-- Check if a challenge exists for this key, ignoring IP (for unconditional consumption)

1500+challengeExistsForKey = \challenges pubkey challenge t ->

1501+ any challenges (\c -> c.pubkey == pubkey && safeCompare c.challenge challenge && c.expiresAt > t)

1502+

1477challengesExcluding = \challenges pubkey challenge ->

1503challengesExcluding = \challenges pubkey challenge ->

1478 do c <- challenges; where c.pubkey != pubkey || c.challenge != challenge; yield c

1504 do c <- challenges; where c.pubkey != pubkey || c.challenge != challenge; yield c

1479

1505

@@ -1544,8 +1570,8 @@ handleChallenge = \pubkey clientIp respond -> do

1544 yield (respond {challenge: "", ok: False {}, expiresAt: 0})

1570 yield (respond {challenge: "", ok: False {}, expiresAt: 0})

1545 else do

1571 else do

1546 challenge <- genToken

1572 challenge <- genToken

1573+ pr <- maybePrune t

1547 result <- atomic do

1574 result <- atomic do

1548− pr <- maybePrune t

1549 challenges <- *challenges

1575 challenges <- *challenges

1550 let active = do c <- challenges; where c.expiresAt > t; yield c

1576 let active = do c <- challenges; where c.expiresAt > t; yield c

1551 let activeForKey = count (do c <- active; where c.pubkey == pubkey; yield c)

1577 let activeForKey = count (do c <- active; where c.pubkey == pubkey; yield c)

@@ -1555,7 +1581,7 @@ handleChallenge = \pubkey clientIp respond -> do

1555 union others (evictOneOldest forKey (\c -> c.expiresAt))

1581 union others (evictOneOldest forKey (\c -> c.expiresAt))

1556 else active

1582 else active

1557 if count cappedActive >= maxChallenges

1583 if count cappedActive >= maxChallenges

1558− then yield {pr: pr, accepted: False {}, rateLimited: False {}}

1584+ then yield {accepted: False {}, rateLimited: False {}}

1559 else do

1585 else do

1560 challengeRates <- *challengeRates

1586 challengeRates <- *challengeRates

1561 let rateResult = checkRateWindow challengeRates (\r -> r.pubkey == pubkey)

1587 let rateResult = checkRateWindow challengeRates (\r -> r.pubkey == pubkey)

@@ -1563,16 +1589,16 @@ handleChallenge = \pubkey clientIp respond -> do

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

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

1564 maxChallengeRatePerKey challengeRateWindowMs maxChallengeRateEntries t

1590 maxChallengeRatePerKey challengeRateWindowMs maxChallengeRateEntries t

1565 if rateResult.limited

1591 if rateResult.limited

1566− then yield {pr: pr, accepted: False {}, rateLimited: True {}}

1592+ then yield {accepted: False {}, rateLimited: True {}}

1567 else do

1593 else do

1568 set *challengeRates = rateResult.entries

1594 set *challengeRates = rateResult.entries

1569 set *challenges = union cappedActive [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry, ip: clientIp}]

1595 set *challenges = union cappedActive [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry, ip: clientIp}]

1570− yield {pr: pr, accepted: True {}, rateLimited: False {}}

1596+ yield {accepted: True {}, rateLimited: False {}}

1571 when result.rateLimited

1597 when result.rateLimited

1572 (println ("challenge: rate limited (" ++ show maxChallengeRatePerKey ++ "/s per key), rejecting for " ++ pubkey))

1598 (println ("challenge: rate limited (" ++ show maxChallengeRatePerKey ++ "/s per key), rejecting for " ++ pubkey))

1573 when (not result.accepted && not result.rateLimited)

1599 when (not result.accepted && not result.rateLimited)

1574 (println ("challenge: at capacity (" ++ show maxChallenges ++ "), rejecting for " ++ pubkey))

1600 (println ("challenge: at capacity (" ++ show maxChallenges ++ "), rejecting for " ++ pubkey))

1575− forkBroadcastOffline t result.pr

1601+ forkBroadcastOffline t pr

1576 yield (respond {challenge: if result.accepted then challenge else "", ok: result.accepted, expiresAt: if result.accepted then t + challengeExpiry else 0})

1602 yield (respond {challenge: if result.accepted then challenge else "", ok: result.accepted, expiresAt: if result.accepted then t + challengeExpiry else 0})

1577

1603

1578-- POST /auth/verify: Verify Ed25519 signature over challenge string.

1604-- POST /auth/verify: Verify Ed25519 signature over challenge string.

@@ -1602,15 +1628,16 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->

1602verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t -> do

1628verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t -> do

1603 let expiry = t + sessionExpiry

1629 let expiry = t + sessionExpiry

1604 let sigValid = verify (bytesFromHex pubkey) (textToBytes challenge) (bytesFromHex signature)

1630 let sigValid = verify (bytesFromHex pubkey) (textToBytes challenge) (bytesFromHex signature)

1631+ pr <- maybePrune t

1605 validation <- atomic do

1632 validation <- atomic do

1606− pr <- maybePrune t

1607 challenges <- *challenges

1633 challenges <- *challenges

1608 let hasMatch = challengeMatch challenges pubkey challenge clientIp t

1634 let hasMatch = challengeMatch challenges pubkey challenge clientIp t

1609− -- Consume challenge on any verify attempt (prevents signature brute-force)

1635+ -- Always consume challenge if it exists, even on IP mismatch (prevents brute-force from other IPs)

1636+ let exists = challengeExistsForKey challenges pubkey challenge t

1610− when hasMatch (set *challenges = challengesExcluding challenges pubkey challenge)

1637+ when exists (set *challenges = challengesExcluding challenges pubkey challenge)

1611 if not hasMatch || not sigValid

1638 if not hasMatch || not sigValid

1612− then yield {pr: pr, valid: False {}}

1639+ then yield {valid: False {}}

1613− else yield {pr: pr, valid: True {}}

1640+ else yield {valid: True {}}

1614 if validation.valid

1641 if validation.valid

1615 then do

1642 then do

1616 token <- genToken

1643 token <- genToken

@@ -1627,14 +1654,14 @@ verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t ->

1627 yield (if federationEnabled

1654 yield (if federationEnabled

1628 then collectPeerServers updatedPresence (\p -> p.server)

1655 then collectPeerServers updatedPresence (\p -> p.server)

1629 else [])

1656 else [])

1630− let expiredLocal = do p <- validation.pr.expiredLocal; where p.pubkey != pubkey; yield p

1657+ let expiredLocal = do p <- pr.expiredLocal; where p.pubkey != pubkey; yield p

1631 fork (do

1658 fork (do

1632 when federationEnabled

1659 when federationEnabled

1633 (sendBatchedGossip peerServers [{eventType: GossipOnline {}, pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl}] gossipOriginHops)

1660 (sendBatchedGossip peerServers [{eventType: GossipOnline {}, pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl}] gossipOriginHops)

1634− broadcastOfflineGossip t validation.pr.peerServers expiredLocal)

1661+ broadcastOfflineGossip t pr.peerServers expiredLocal)

1635 yield (respond {token: token, expiresAt: expiry})

1662 yield (respond {token: token, expiresAt: expiry})

1636 else do

1663 else do

1637− forkBroadcastOffline t validation.pr

1664+ forkBroadcastOffline t pr

1638 yield (respond {token: "", expiresAt: 0})

1665 yield (respond {token: "", expiresAt: 0})

1639

1666

1640-- POST /poll: Long-poll for new messages. Cursor-based: messages at or before cursor

1667-- POST /poll: Long-poll for new messages. Cursor-based: messages at or before cursor

@@ -1671,16 +1698,16 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization respond

1671 else do

1698 else do

1672 effectiveMsgId <- if messageId != "" then yield messageId else genToken

1699 effectiveMsgId <- if messageId != "" then yield messageId else genToken

1673 let clampedTs = clampTimestamp timestamp t

1700 let clampedTs = clampTimestamp timestamp t

1701+ pr <- maybePrune t

1674 result <- atomic do

1702 result <- atomic do

1675− pr <- maybePrune t

1676 withSessionAuth authorization t

1703 withSessionAuth authorization t

1677− (yield {pr: pr, send: SendFailed {status: MsUnauthorized {}}, senderRateLimited: False {}})

1704+ (yield {send: SendFailed {status: MsUnauthorized {}}, senderRateLimited: False {}, globalCapReached: False {}})

1678 (\senderKey -> do

1705 (\senderKey -> do

1679 messages <- *messages

1706 messages <- *messages

1680 ackedIds <- *ackedIds

1707 ackedIds <- *ackedIds

1681 deliveredIds <- *deliveredIds

1708 deliveredIds <- *deliveredIds

1682 if messageId != "" && (hasMessage messageId to messages || hasMessage messageId to ackedIds || hasMessage messageId to deliveredIds)

1709 if messageId != "" && (hasMessage messageId to messages || hasMessage messageId to ackedIds || hasMessage messageId to deliveredIds)

1683− then yield {pr: pr, send: SendDuplicate {}, senderRateLimited: False {}}

1710+ then yield {send: SendDuplicate {}, senderRateLimited: False {}, globalCapReached: False {}}

1684 else do

1711 else do

1685 sendRates <- *sendRates

1712 sendRates <- *sendRates

1686 let rateResult = checkRateWindow sendRates (\r -> r.pubkey == senderKey)

1713 let rateResult = checkRateWindow sendRates (\r -> r.pubkey == senderKey)

@@ -1688,7 +1715,7 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization respond

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

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

1689 maxSendRatePerKey sendRateWindowMs maxSendRateEntries t

1716 maxSendRatePerKey sendRateWindowMs maxSendRateEntries t

1690 if rateResult.limited

1717 if rateResult.limited

1691− then yield {pr: pr, send: SendFailed {status: MsRateLimited {}}, senderRateLimited: True {}}

1718+ then yield {send: SendFailed {status: MsRateLimited {}}, senderRateLimited: True {}, globalCapReached: False {}}

1692 else do

1719 else do

1693 set *sendRates = rateResult.entries

1720 set *sendRates = rateResult.entries

1694 recvRates <- *recvRates

1721 recvRates <- *recvRates

@@ -1697,13 +1724,14 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization respond

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

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

1698 maxRecvRatePerKey recvRateWindowMs maxRecvRateEntries t

1725 maxRecvRatePerKey recvRateWindowMs maxRecvRateEntries t

1699 if recvResult.limited

1726 if recvResult.limited

1700− then yield {pr: pr, send: SendFailed {status: MsRateLimited {}}, senderRateLimited: False {}}

1727+ then yield {send: SendFailed {status: MsRateLimited {}}, senderRateLimited: False {}, globalCapReached: False {}}

1701 else do

1728 else do

1702 set *recvRates = recvResult.entries

1729 set *recvRates = recvResult.entries

1703− if count messages >= maxMessages

1730+ let msgStats = messageCounts messages to t

1704− then yield {pr: pr, send: SendFailed {status: MsRejected {}}, senderRateLimited: False {}}

1731+ if msgStats.total >= maxMessages

1732+ then yield {send: SendFailed {status: MsRejected {}}, senderRateLimited: False {}, globalCapReached: True {}}

1705− else if countMessagesFor messages to t >= maxMessagesPerRecipient

1733+ else if msgStats.forKey >= maxMessagesPerRecipient

1706− then yield {pr: pr, send: SendFailed {status: MsRejected {}}, senderRateLimited: False {}}

1734+ then yield {send: SendFailed {status: MsRejected {}}, senderRateLimited: False {}, globalCapReached: False {}}

1707 else do

1735 else do

1708 presence <- *presence

1736 presence <- *presence

1709 seqCounter <- *seqCounter

1737 seqCounter <- *seqCounter

@@ -1725,9 +1753,11 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization respond

1725 let newFwds = do s <- remoteServers; where none forwards (\f -> f.messageId == effectiveMsgId && f.toServer == s); yield {messageId: effectiveMsgId, toServer: s, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs}

1753 let newFwds = do s <- remoteServers; where none forwards (\f -> f.messageId == effectiveMsgId && f.toServer == s); yield {messageId: effectiveMsgId, toServer: s, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs}

1726 let cappedFwds = takeRelation (maxForwardsEntries - count forwards) newFwds

1754 let cappedFwds = takeRelation (maxForwardsEntries - count forwards) newFwds

1727 set *forwards = union forwards cappedFwds)

1755 set *forwards = union forwards cappedFwds)

1728− yield {pr: pr, send: SendDone {delivery: delivery, remoteServers: remoteServers}, senderRateLimited: False {}})

1756+ yield {send: SendDone {delivery: delivery, remoteServers: remoteServers}, senderRateLimited: False {}, globalCapReached: False {}})

1729 when result.senderRateLimited

1757 when result.senderRateLimited

1730 (println ("sendMessage: rate limited (" ++ show maxSendRatePerKey ++ "/s per key)"))

1758 (println ("sendMessage: rate limited (" ++ show maxSendRatePerKey ++ "/s per key)"))

1759+ when result.globalCapReached

1760+ (println ("sendMessage: global message cap reached (" ++ show maxMessages ++ "), rejecting"))

1731 case result.send of

1761 case result.send of

1732 SendDone {delivery, remoteServers} ->

1762 SendDone {delivery, remoteServers} ->

1733 fork (do

1763 fork (do

@@ -1738,14 +1768,14 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization respond

1738 timestamp: clampedTs, originServer: serverHostN, messageId: effectiveMsgId,

1768 timestamp: clampedTs, originServer: serverHostN, messageId: effectiveMsgId,

1739 fromServer: serverHostN, authorization: federationAuth})

1769 fromServer: serverHostN, authorization: federationAuth})

1740 case fwdResult of

1770 case fwdResult of

1741− Err _ -> do

1771+ Err e -> do

1742− println ("federation: send forward to " ++ server ++ " failed for " ++ effectiveMsgId)

1772+ println ("federation: send forward to " ++ server ++ " failed for " ++ effectiveMsgId ++ ": " ++ show e)

1743 recordFailure server

1773 recordFailure server

1744 Ok _ -> do

1774 Ok _ -> do

1745 removeForward effectiveMsgId server

1775 removeForward effectiveMsgId server

1746 clearFailure server))

1776 clearFailure server))

1747− broadcastOfflineGossip t result.pr.peerServers result.pr.expiredLocal)

1777+ broadcastOfflineGossip t pr.peerServers pr.expiredLocal)

1748− _ -> forkBroadcastOffline t result.pr

1778+ _ -> forkBroadcastOffline t pr

1749 yield (case result.send of

1779 yield (case result.send of

1750 SendFailed {status} -> respond {status: status, messageId: ""}

1780 SendFailed {status} -> respond {status: status, messageId: ""}

1751 SendDuplicate {} -> respond {status: MsDuplicate {}, messageId: effectiveMsgId}

1781 SendDuplicate {} -> respond {status: MsDuplicate {}, messageId: effectiveMsgId}

@@ -1761,10 +1791,10 @@ handleLookupPresence = \pubkey authorization respond -> do

1761 globalRateLimited <- checkGlobalRate t

1791 globalRateLimited <- checkGlobalRate t

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

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

1763 else do

1793 else do

1794+ pr <- maybePrune t

1764 result <- atomic do

1795 result <- atomic do

1765− pr <- maybePrune t

1766 withSessionAuth authorization t

1796 withSessionAuth authorization t

1767− (yield {pr: pr, server: "", authorized: False {}})

1797+ (yield {server: "", authorized: False {}})

1768 (\senderKey -> do

1798 (\senderKey -> do

1769 lookupRates <- *lookupRates

1799 lookupRates <- *lookupRates

1770 let rateResult = checkRateWindow lookupRates (\r -> r.pubkey == senderKey)

1800 let rateResult = checkRateWindow lookupRates (\r -> r.pubkey == senderKey)

@@ -1773,15 +1803,15 @@ handleLookupPresence = \pubkey authorization respond -> do

1773 maxLookupRatePerKey lookupRateWindowMs maxLookupRateEntries t

1803 maxLookupRatePerKey lookupRateWindowMs maxLookupRateEntries t

1774 set *lookupRates = rateResult.entries

1804 set *lookupRates = rateResult.entries

1775 if rateResult.limited

1805 if rateResult.limited

1776− then yield {pr: pr, server: "", authorized: True {}}

1806+ then yield {server: "", authorized: True {}}

1777 else do

1807 else do

1778 presence <- *presence

1808 presence <- *presence

1779 let foundServer = bestServerForKey (presenceForKey presence pubkey t)

1809 let foundServer = bestServerForKey (presenceForKey presence pubkey t)

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

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

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

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

1782 let maskedServer = if foundServer == "" then "" else if isLocalServer foundServer then foundServer else "remote"

1812 let maskedServer = if foundServer == "" then "" else if isLocalServer foundServer then foundServer else "remote"

1783− yield {pr: pr, server: maskedServer, authorized: True {}})

1813+ yield {server: maskedServer, authorized: True {}})

1784− forkBroadcastOffline t result.pr

1814+ forkBroadcastOffline t pr

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

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

1786

1816

1787-- POST /federation/gossip: Receive presence gossip from a peer server.

1817-- POST /federation/gossip: Receive presence gossip from a peer server.

@@ -1798,12 +1828,12 @@ handleRecvGossip = \events hopsRemaining fromServer authorization respond -> do

1798 let filteredOnline = resolveByTimestamp onlineEntries trustedOffline

1828 let filteredOnline = resolveByTimestamp onlineEntries trustedOffline

1799 let filteredOffline = resolveByTimestamp trustedOffline onlineEntries

1829 let filteredOffline = resolveByTimestamp trustedOffline onlineEntries

1800 let deduped = mergeByPubkeyServer [] filteredOnline

1830 let deduped = mergeByPubkeyServer [] filteredOnline

1831+ pr <- maybePrune t

1801 gossipResult <- atomic do

1832 gossipResult <- atomic do

1802− pr <- maybePrune t

1803 presence <- *presence

1833 presence <- *presence

1804 let gm = mergeGossipPresence presence deduped trustedOffline fromServer t

1834 let gm = mergeGossipPresence presence deduped trustedOffline fromServer t

1805 set *presence = gm.finalPresence

1835 set *presence = gm.finalPresence

1806− yield {gm: gm, pr: pr}

1836+ yield {gm: gm}

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

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

1808 when (not gossipResult.gm.withinCapacity)

1838 when (not gossipResult.gm.withinCapacity)

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

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

@@ -1820,7 +1850,7 @@ handleRecvGossip = \events hopsRemaining fromServer authorization respond -> do

1820 fork (do

1850 fork (do

1821 when (cappedHops > 1 && federationEnabled && count freshEvents > 0) (

1851 when (cappedHops > 1 && federationEnabled && count freshEvents > 0) (

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

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

1823− broadcastOfflineGossip t gossipResult.pr.peerServers gossipResult.pr.expiredLocal

1853+ broadcastOfflineGossip t pr.peerServers pr.expiredLocal

1824 when (federationEnabled && count newlyOnlineRemote > 0)

1854 when (federationEnabled && count newlyOnlineRemote > 0)

1825 (forwardQueuedMessages newlyOnlineRemote t))

1855 (forwardQueuedMessages newlyOnlineRemote t))

1826 yield (respond {ok: True {}, error: ""}))

1856 yield (respond {ok: True {}, error: ""}))

@@ -1832,40 +1862,45 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro

1832 t <- now

1862 t <- now

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

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

1834 let clampedTs = clampTimestamp timestamp t

1864 let clampedTs = clampTimestamp timestamp t

1865+ pr <- maybePrune t

1835 result <- atomic do

1866 result <- atomic do

1836− pr <- maybePrune t

1837 messages <- *messages

1867 messages <- *messages

1838 ackedIds <- *ackedIds

1868 ackedIds <- *ackedIds

1839 deliveredIds <- *deliveredIds

1869 deliveredIds <- *deliveredIds

1840 if hasMessage messageId toKey messages || hasMessage messageId toKey ackedIds || hasMessage messageId toKey deliveredIds

1870 if hasMessage messageId toKey messages || hasMessage messageId toKey ackedIds || hasMessage messageId toKey deliveredIds

1841− then yield {pr: pr, forward: ForwardDuplicate {}}

1871+ then yield {forward: ForwardDuplicate {}, globalCapReached: False {}}

1842− else if count messages >= maxMessages

1872+ else do

1843− then yield {pr: pr, forward: ForwardRejected {}}

1873+ let msgStats = messageCounts messages toKey t

1874+ if msgStats.total >= maxMessages

1875+ then yield {forward: ForwardRejected {}, globalCapReached: True {}}

1844− else if countMessagesFor messages toKey t >= maxMessagesPerRecipient

1876+ else if msgStats.forKey >= maxMessagesPerRecipient

1845− then yield {pr: pr, forward: ForwardRejected {}}

1877+ then yield {forward: ForwardRejected {}, globalCapReached: False {}}

1846− else do

1878+ else do

1847− recvRates <- *recvRates

1879+ recvRates <- *recvRates

1848− let recvResult = checkRateWindow recvRates (\r -> r.pubkey == toKey)

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

1849− (\ts -> {pubkey: toKey, windowStart: ts, count: 1})

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

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

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

1851− maxRecvRatePerKey recvRateWindowMs maxRecvRateEntries t

1883+ maxRecvRatePerKey recvRateWindowMs maxRecvRateEntries t

1852− if recvResult.limited

1884+ if recvResult.limited

1853− then yield {pr: pr, forward: ForwardRejected {}}

1885+ then yield {forward: ForwardRejected {}, globalCapReached: False {}}

1854− else do

1886+ else do

1855− set *recvRates = recvResult.entries

1887+ set *recvRates = recvResult.entries

1856− seqCounter <- *seqCounter

1888+ seqCounter <- *seqCounter

1857− let seqTs = nextSeqTs t seqCounter

1889+ let seqTs = nextSeqTs t seqCounter

1858− set *seqCounter = seqTs

1890+ set *seqCounter = seqTs

1859− let updatedMessages = union messages

1891+ let updatedMessages = union messages

1860− [{messageId: messageId, toKey: toKey,

1892+ [{messageId: messageId, toKey: toKey,

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

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

1862− set *messages = updatedMessages

1894+ set *messages = updatedMessages

1863− signalMessageForKey toKey seqTs

1895+ signalMessageForKey toKey seqTs

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

1896+ yield {forward: ForwardStored {}, globalCapReached: False {}}

1865 case result.forward of

1897 case result.forward of

1898+ ForwardRejected {} ->

1899+ if result.globalCapReached

1900+ then println ("forward: global message cap reached (" ++ show maxMessages ++ "), rejecting for " ++ toKey)

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

1901+ else println ("forward: rejected for " ++ toKey ++ " (mailbox full, max " ++ show maxMessagesPerRecipient ++ ")")

1867 _ -> yield {}

1902 _ -> yield {}

1868− forkBroadcastOffline t result.pr

1903+ forkBroadcastOffline t pr

1869 yield (case result.forward of

1904 yield (case result.forward of

1870 ForwardRejected {} -> respond {ok: False {}, error: "mailbox_full"}

1905 ForwardRejected {} -> respond {ok: False {}, error: "mailbox_full"}

1871 _ -> respond {ok: True {}, error: ""}))

1906 _ -> respond {ok: True {}, error: ""}))

@@ -1877,8 +1912,8 @@ handleFederationAck = \messageIds originServer fromServer authorization respond

1877 t <- now

1912 t <- now

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

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

1879 let dedupedIds = dedupMessageIds messageIds

1914 let dedupedIds = dedupMessageIds messageIds

1880− pruneResult <- atomic do

1881− pr <- maybePrune t

1915+ pr <- maybePrune t

1916+ atomic do

1882 forwards <- *forwards

1917 forwards <- *forwards

1883 let ackedForwards = do f <- forwards; where f.toServer == fromServer && any dedupedIds (\mid -> f.messageId == mid); yield f

1918 let ackedForwards = do f <- forwards; where f.toServer == fromServer && any dedupedIds (\mid -> f.messageId == mid); yield f

1884 let updatedForwards = forwardsExcluding forwards fromServer dedupedIds

1919 let updatedForwards = forwardsExcluding forwards fromServer dedupedIds

@@ -1891,8 +1926,7 @@ handleFederationAck = \messageIds originServer fromServer authorization respond

1891 where none presence (\p -> p.pubkey == f.toKey && isLocalServer p.server && p.expiresAt > t)

1926 where none presence (\p -> p.pubkey == f.toKey && isLocalServer p.server && p.expiresAt > t)

1892 yield f

1927 yield f

1893 set *messages = do m <- messages; where none removable (\f -> m.messageId == f.messageId && m.toKey == f.toKey); yield m

1928 set *messages = do m <- messages; where none removable (\f -> m.messageId == f.messageId && m.toKey == f.toKey); yield m

1894− yield pr

1895− forkBroadcastOffline t pruneResult

1929+ forkBroadcastOffline t pr

1896 yield (respond {ok: True {}, error: ""}))

1930 yield (respond {ok: True {}, error: ""}))

1897

1931

1898-- POST /maintenance/drain: Broadcast offline gossip for all local keys (graceful shutdown).

1932-- POST /maintenance/drain: Broadcast offline gossip for all local keys (graceful shutdown).