@@ -16,6 +16,8 @@ maxBlobLen = 65536
1616maxSignatureLen = 256
1717maxMessageIds = 100
1818maxGossipEvents = 100
19+maxPollEvents = 50
20+maxPresenceEntries = 10000
1921clockSkewTolerance = 5000 -- 5 seconds tolerance for clock skew between federated servers
2022
2123-- ============================================
@@ -40,6 +42,8 @@ type GossipEvent = {eventType: Text, pubkey: Text, server: Text, timestamp: Int,
4042
4143*seqCounter : Int
4244
45+*forwards : [{messageId: Text, toServer: Text, createdAt: Int}]
46+
4347-- ============================================
4448-- Routes
4549-- ============================================
@@ -110,10 +114,21 @@ extractToken = \authHeader ->
110114 else ""
111115
112116verifyFederationAuth = \authorization ->
113− extractToken authorization == federationSecret && not (federationSecret == "")
117+ not (federationSecret == "") && safeCompare (extractToken authorization) federationSecret
114118
115119clampTtl = \ttl -> if ttl < 1 then 1 else if ttl > maxPresenceTtl then maxPresenceTtl else ttl
116120
121+safeCompareLoop = \s1 s2 result ->
122+ if s1 == "" && s2 == ""
123+ then result
124+ else if s1 == ""
125+ then safeCompareLoop "" (drop 1 s2) (result + 1)
126+ else if s2 == ""
127+ then safeCompareLoop (drop 1 s1) "" (result + 1)
128+ else safeCompareLoop (drop 1 s1) (drop 1 s2) (result + (if take 1 s1 == take 1 s2 then 0 else 1))
129+
130+safeCompare = \a b -> safeCompareLoop a b 0 == 0
131+
117132sendToEach = \servers action ->
118133 if count servers == 0
119134 then yield {}
@@ -123,7 +138,9 @@ sendToEach = \servers action ->
123138 if server == ""
124139 then yield {}
125140 else do
126− action server
141+ catch (action server) (\err -> do
142+ println ("federation: error reaching " ++ server ++ ": " ++ show err)
143+ yield {})
127144 sendToEach rest action
128145
129146pruneExpired = \t -> do
@@ -152,6 +169,11 @@ pruneExpired = \t -> do
152169 m <- messages
153170 where m.receivedAt >= t - messageMaxAge
154171 yield m
172+ forwards <- *forwards
173+ set *forwards = do
174+ f <- forwards
175+ where f.createdAt >= t - messageMaxAge
176+ yield f
155177 yield {expiredLocal: (do
156178 p <- presenceBefore
157179 where p.expiresAt <= t
@@ -193,12 +215,21 @@ serve = \req -> case req of
193215 pruneResult <- atomic do
194216 pr <- pruneExpired t
195217 challenges <- *challenges
196− set *challenges = union challenges
218+ let existingCount = count (do c <- challenges; where c.pubkey == pubkey; yield c)
219+ let oldestChallengeExpiry = fold (\acc c2 -> if c2.pubkey == pubkey && (acc == -1 || c2.expiresAt < acc) then c2.expiresAt else acc) -1 challenges
220+ let cappedChallenges = if existingCount >= 5 && oldestChallengeExpiry >= 0
221+ then fold (\acc c ->
222+ if not acc.removed && c.pubkey == pubkey && c.expiresAt == oldestChallengeExpiry
223+ then {removed: True {}, items: acc.items}
224+ else {removed: acc.removed, items: union acc.items [c]}) {removed: False {}, items: []} challenges
225+ else {removed: False {}, items: challenges}
226+ set *challenges = union
227+ cappedChallenges.items
197228 [{pubkey: pubkey, challenge: challenge, expiresAt: t + 60000}]
198229 yield pr
230+ respond {challenge: challenge}
199231 broadcastOfflineGossip t pruneResult.peerServers pruneResult.expiredLocal
200− else yield {}
201− respond {challenge: (if valid then challenge else "")}
232+ else respond {challenge: ""}
202233
203234 VerifyAuth {pubkey, challenge, signature, respond} -> do
204235 t <- now
@@ -233,24 +264,25 @@ serve = \req -> case req of
233264 where s.pubkey == pubkey
234265 where s.expiresAt > t
235266 yield s
236− let oldestExpiry = fold (\acc s -> if s.expiresAt < acc then s.expiresAt else acc) expiry existingForKey
267+ let oldestExpiry = fold (\acc s -> if acc == -1 || s.expiresAt < acc then s.expiresAt else acc) -1 existingForKey
237268 let cappedFold = fold (\acc s ->
238269 if not acc.removed && s.pubkey == pubkey && s.expiresAt == oldestExpiry
239270 then {removed: True {}, sessions: acc.sessions}
240271 else {removed: acc.removed, sessions: union acc.sessions [s]}) {removed: False {}, sessions: []} sessions
241− let sessionsCapped = if count existingForKey >= 10
272+ let sessionsCapped = if count existingForKey >= 10 && oldestExpiry >= 0
242273 then cappedFold.sessions
243274 else sessions
244275 set *sessions = if sigValid
245276 then union sessionsCapped [{token: token, pubkey: pubkey, expiresAt: expiry}]
246277 else sessions
247− let updatedPresence = if sigValid
248− then union
249− (do
278+ let presenceWithout = do
250279 p <- presence
251280 where not (p.pubkey == pubkey && p.server == serverHost)
252− yield p)
253− [{pubkey: pubkey, server: serverHost, expiresAt: expiry, timestamp: t}]
281+ yield p
282+ let updatedPresence = if sigValid
283+ then if count presenceWithout < maxPresenceEntries
284+ then union presenceWithout [{pubkey: pubkey, server: serverHost, expiresAt: expiry, timestamp: t}]
285+ else presenceWithout
254286 else presence
255287 set *presence = updatedPresence
256288 let peerServers = if sigValid && not (federationSecret == "")
@@ -260,6 +292,9 @@ serve = \req -> case req of
260292 else union acc [p.server]) [] updatedPresence
261293 else []
262294 yield {sigValid: sigValid, peerServers: peerServers, expiredLocal: pr.expiredLocal, offlinePeers: pr.peerServers}
295+ if result.sigValid
296+ then respond {token: token, expiresAt: expiry}
297+ else respond {token: "", expiresAt: 0}
263298 if result.sigValid && not (federationSecret == "")
264299 then sendToEach result.peerServers (\server ->
265300 fetch (federationScheme ++ server) (RecvGossip {
@@ -270,9 +305,6 @@ serve = \req -> case req of
270305 p <- result.expiredLocal
271306 where not (p.pubkey == pubkey)
272307 yield p)
273− if result.sigValid
274− then respond {token: token, expiresAt: expiry}
275− else respond {token: "", expiresAt: 0}
276308
277309 Poll {cursor, authorization, respond} -> do
278310 t <- now
@@ -283,26 +315,36 @@ serve = \req -> case req of
283315 messages <- *messages
284316 seqCounter <- *seqCounter
285317 let session = lookupSession sessions token t
286− let evts = case session of
318+ let clampedCursor = if cursor < 0 then 0 else if cursor > seqCounter then seqCounter else cursor
319+ let matchingMsgs = case session of
287320 Nothing {} -> []
288321 Just {value: s} -> do
289322 m <- messages
290323 where m.toKey == s.pubkey
291324 where m.receivedAt >= t - messageMaxAge
292− where m.receivedAt > cursor
293− yield {id: m.messageId, eventType: "message",
294− payload: toJson {toKey: m.toKey, fromKey: m.fromKey,
295− encryptedBlob: m.encryptedBlob,
296− timestamp: m.timestamp}}
325+ where m.receivedAt > clampedCursor
326+ yield m
327+ let limitedMsgs = fold (\acc m ->
328+ if count acc >= maxPollEvents then acc
329+ else union acc [m]) [] matchingMsgs
330+ let evts = do
331+ m <- limitedMsgs
332+ yield {id: m.messageId, eventType: "message",
333+ payload: toJson {toKey: m.toKey, fromKey: m.fromKey,
334+ encryptedBlob: m.encryptedBlob,
335+ timestamp: m.timestamp}}
336+ let lastReceivedAt = fold (\acc m -> if m.receivedAt > acc then m.receivedAt else acc) clampedCursor limitedMsgs
297337 let authorized = case session of
298338 Nothing {} -> False {}
299339 _ -> True {}
300340 let newCursor = case session of
301341 Nothing {} -> -1
302− _ -> if seqCounter > cursor then seqCounter else cursor
342+ _ -> if count limitedMsgs >= maxPollEvents
343+ then lastReceivedAt
344+ else if seqCounter > clampedCursor then seqCounter else clampedCursor
303345 yield {events: evts, cursor: newCursor, authorized: authorized, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers}
304− broadcastOfflineGossip t result.peerServers result.expiredLocal
305346 respond {events: result.events, cursor: result.cursor, authorized: result.authorized}
347+ broadcastOfflineGossip t result.peerServers result.expiredLocal
306348
307349 SendMessage {to, encryptedBlob, timestamp, authorization, respond} -> do
308350 t <- now
@@ -353,18 +395,21 @@ serve = \req -> case req of
353395 else if localOnline then "delivered"
354396 else if count remoteServers > 0 then "federated"
355397 else "queued"
398+ forwards <- *forwards
399+ set *forwards = if shouldInsert
400+ then union forwards (do s <- remoteServers; yield {messageId: msgId, toServer: s, createdAt: t})
401+ else forwards
356402 yield {status: status, remoteServers: remoteServers, fromKey: fromKey, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers}
357− finalStatus <- if result.status == "federated" && not (federationSecret == "")
358− then do
359− sendToEach result.remoteServers (\server ->
360− fetch (federationScheme ++ server) (ForwardMessage {
361− toKey: to, fromKey: result.fromKey, encryptedBlob: encryptedBlob,
362− timestamp: clampedTs, originServer: serverHost, messageId: msgId,
363− authorization: "Bearer " ++ federationSecret}))
364− yield "federated"
365− else yield (if result.status == "federated" then "queued" else result.status)
366− broadcastOfflineGossip t result.peerServers result.expiredLocal
403+ let finalStatus = if result.status == "federated" && federationSecret == "" then "queued" else result.status
367404 respond {status: finalStatus, messageId: (if result.status == "unauthorized" then "" else if result.status == "invalid" then "" else msgId)}
405+ if result.status == "federated" && not (federationSecret == "")
406+ then sendToEach result.remoteServers (\server ->
407+ fetch (federationScheme ++ server) (ForwardMessage {
408+ toKey: to, fromKey: result.fromKey, encryptedBlob: encryptedBlob,
409+ timestamp: clampedTs, originServer: serverHost, messageId: msgId,
410+ authorization: "Bearer " ++ federationSecret}))
411+ else yield {}
412+ broadcastOfflineGossip t result.peerServers result.expiredLocal
368413
369414 AckMessages {messageIds, authorization, respond} -> do
370415 t <- now
@@ -403,6 +448,7 @@ serve = \req -> case req of
403448 let originServers = fold (\acc m ->
404449 if count (do a <- acc; where a == m.originServer; yield a) > 0 then acc
405450 else union acc [m.originServer]) [] fedMessages
451+ respond {ok: result.ok}
406452 if result.ok && count originServers > 0 && not (federationSecret == "")
407453 then sendToEach originServers (\server ->
408454 fetch (federationScheme ++ server) (FederationAck {
@@ -411,7 +457,6 @@ serve = \req -> case req of
411457 originServer: server, fromServer: serverHost}))
412458 else yield {}
413459 broadcastOfflineGossip t result.peerServers result.expiredLocal
414− respond {ok: result.ok}
415460
416461 LookupPresence {pubkey, authorization, respond} -> do
417462 t <- now
@@ -440,8 +485,8 @@ serve = \req -> case req of
440485 Just {value: p} -> p.server
441486 Nothing {} -> (fold (\acc p -> if p.expiresAt > acc.best then {server: p.server, best: p.expiresAt} else acc) {server: "", best: 0} valid).server
442487 yield {foundServer: foundServer, authorized: authorized, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers}
443− broadcastOfflineGossip t result.peerServers result.expiredLocal
444488 respond {online: not (result.foundServer == ""), server: result.foundServer, authorized: result.authorized}
489+ broadcastOfflineGossip t result.peerServers result.expiredLocal
445490
446491 RecvGossip {events, hopsRemaining, fromServer, authorization, respond} -> do
447492 t <- now
@@ -457,6 +502,7 @@ serve = \req -> case req of
457502 where not (tooLong e.pubkey maxPubkeyLen)
458503 where not (tooLong e.server maxPubkeyLen)
459504 where isValidHex e.pubkey
505+ where cappedHops < 2 || e.server == fromServer
460506 where not (e.server == serverHost)
461507 where e.timestamp <= t + clockSkewTolerance
462508 where e.timestamp + (clampTtl e.ttl) * 1000 > t
@@ -470,6 +516,7 @@ serve = \req -> case req of
470516 where not (tooLong e.pubkey maxPubkeyLen)
471517 where not (tooLong e.server maxPubkeyLen)
472518 where isValidHex e.pubkey
519+ where cappedHops < 2 || e.server == fromServer
473520 where not (e.server == serverHost)
474521 where e.timestamp <= t + clockSkewTolerance
475522 where t - e.timestamp < (clampTtl e.ttl) * 1000
@@ -560,18 +607,67 @@ serve = \req -> case req of
560607 where o.timestamp >= p.timestamp
561608 yield o) > 0
562609 yield p)
563− set *presence = if validRequest
564− then afterOffline
610+ let withinCapacity = count afterOffline <= maxPresenceEntries
611+ let updatesOnlyMerged = if withinCapacity then afterOffline
612+ else fold (\acc e ->
613+ if count (do
614+ p <- acc
615+ where p.pubkey == e.pubkey
616+ where p.server == e.server
617+ where p.expiresAt >= e.expiresAt
618+ yield p) > 0
619+ then acc
620+ else if count (do p <- presence; where p.pubkey == e.pubkey; where p.server == e.server; yield p) > 0
621+ then union
622+ (do
623+ p <- acc
624+ where not (p.pubkey == e.pubkey && p.server == e.server)
625+ yield p)
626+ [e]
627+ else acc) presence deduped
628+ let updatesOnlyAfterOffline = if withinCapacity then afterOffline
629+ else diff updatesOnlyMerged (do
630+ p <- updatesOnlyMerged
631+ k <- offlineOnlyKeys
632+ where p.pubkey == k.pubkey
633+ where p.server == k.server
634+ where count (do
635+ o <- offlineEntries
636+ where o.pubkey == k.pubkey
637+ where o.server == k.server
638+ where o.timestamp >= p.timestamp
639+ yield o) > 0
640+ yield p)
641+ let appliedOnlineUpdates = if withinCapacity then appliedOnline
642+ else do
643+ e <- deduped
644+ where count (do p <- presence; where p.pubkey == e.pubkey; where p.server == e.server; yield p) > 0
645+ where count (do
646+ p <- presence
647+ where p.pubkey == e.pubkey
648+ where p.server == e.server
649+ where p.expiresAt >= e.expiresAt
650+ yield p) == 0
651+ yield {pubkey: e.pubkey, server: e.server}
652+ let finalPresence = if validRequest
653+ then if withinCapacity then afterOffline else updatesOnlyAfterOffline
565654 else presence
566− yield {appliedKeys: (union appliedOnline appliedOffline), expiredLocal: pr.expiredLocal, peerServers: pr.peerServers}
655+ set *presence = finalPresence
656+ let gossipPeerServers = fold (\acc p ->
657+ if p.server == serverHost then acc
658+ else if count (do a <- acc; where a == p.server; yield a) > 0 then acc
659+ else union acc [p.server]) [] finalPresence
660+ yield {appliedKeys: (if withinCapacity then union appliedOnline appliedOffline else union appliedOnlineUpdates appliedOffline), expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, gossipPeerServers: gossipPeerServers}
567661 let validEvents = do
568662 e <- events
569663 where validRequest
664+ where e.eventType == "online" || e.eventType == "offline"
570665 where not (e.server == "")
571666 where not (e.pubkey == "")
572667 where not (tooLong e.pubkey maxPubkeyLen)
573668 where not (tooLong e.server maxPubkeyLen)
574669 where isValidHex e.pubkey
670+ where cappedHops < 2 || e.server == fromServer
575671 where not (e.server == serverHost)
576672 where e.timestamp <= t + clockSkewTolerance
577673 where e.timestamp + (clampTtl e.ttl) * 1000 > t
@@ -586,19 +682,14 @@ serve = \req -> case req of
586682 e <- validEvents
587683 where count (do a <- gossipResult.appliedKeys; where a.pubkey == e.pubkey; where a.server == e.server; yield a) > 0
588684 yield e
685+ respond {ok: validRequest}
589686 if validRequest && cappedHops > 1 && not (federationSecret == "") && count effectiveEvents > 0
590687 then do
591− peerServers <- atomic do
592− presence <- *presence
593− yield (fold (\acc p ->
594− if p.server == serverHost then acc
595− else if count (do a <- acc; where a == p.server; yield a) > 0 then acc
596− else union acc [p.server]) [] presence)
597688 let senderServers = fold (\acc e ->
598689 if count (do a <- acc; where a == e.server; yield a) > 0 then acc
599690 else union acc [e.server]) (if fromServer == "" then [] else [fromServer]) effectiveEvents
600691 let targets = do
601− s <- peerServers
692+ s <- gossipResult.gossipPeerServers
602693 where count (do ss <- senderServers; where ss == s; yield ss) == 0
603694 yield s
604695 sendToEach targets (\server ->
@@ -607,7 +698,6 @@ serve = \req -> case req of
607698 events: effectiveEvents, hopsRemaining: cappedHops - 1, fromServer: serverHost}))
608699 else yield {}
609700 broadcastOfflineGossip t gossipResult.peerServers gossipResult.expiredLocal
610− respond {ok: validRequest}
611701
612702 ForwardMessage {toKey, fromKey, encryptedBlob, timestamp, originServer, messageId, authorization, respond} -> do
613703 t <- now
@@ -615,10 +705,16 @@ serve = \req -> case req of
615705 let validOrigin = not (originServer == "") && not (originServer == serverHost) && not (tooLong originServer maxPubkeyLen)
616706 let validFields = not (toKey == "") && not (fromKey == "") && not (encryptedBlob == "") && not (messageId == "") && not (tooLong toKey maxPubkeyLen) && not (tooLong fromKey maxPubkeyLen) && not (tooLong encryptedBlob maxBlobLen) && not (tooLong messageId maxPubkeyLen) && isValidHex toKey && isValidHex fromKey && isValidHex encryptedBlob && isValidHex messageId
617707 let clampedTs = if timestamp > t then t else if timestamp < t - messageMaxAge then t else timestamp
618− pruneResult <- atomic do
708+ result <- atomic do
619709 pr <- pruneExpired t
710+ presence <- *presence
620711 messages <- *messages
621712 seqCounter <- *seqCounter
713+ let knownPeer = count (do
714+ p <- presence
715+ where p.server == originServer
716+ where p.expiresAt > t
717+ yield p) > 0
622718 let seqTs = if t > seqCounter then t else seqCounter + 1
623719 let exists = count (do
624720 m <- messages
@@ -626,16 +722,16 @@ serve = \req -> case req of
626722 where m.toKey == toKey
627723 where m.receivedAt >= t - messageMaxAge
628724 yield m) > 0
629− let shouldInsert = authorized && validOrigin && validFields && not exists
725+ let shouldInsert = authorized && validOrigin && validFields && knownPeer && not exists
630726 set *seqCounter = if shouldInsert then seqTs else seqCounter
631727 set *messages = if shouldInsert
632728 then union messages
633729 [{messageId: messageId, toKey: toKey, fromKey: fromKey,
634730 encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs}]
635731 else messages
636− yield pr
637− broadcastOfflineGossip t pruneResult.peerServers pruneResult.expiredLocal
638− respond {ok: authorized && validOrigin && validFields}
732+ yield {peerServers: pr.peerServers, expiredLocal: pr.expiredLocal, knownPeer: knownPeer}
733+ respond {ok: authorized && validOrigin && validFields && result.knownPeer}
734+ broadcastOfflineGossip t result.peerServers result.expiredLocal
639735
640736 FederationAck {messageIds, originServer, fromServer, authorization, respond} -> do
641737 t <- now
@@ -647,17 +743,29 @@ serve = \req -> case req of
647743 else if not (isValidHex mid) then acc
648744 else if count (do a <- acc; where a == mid; yield a) > 0 then acc
649745 else union acc [mid]) [] messageIds
650− let validAck = authorized && originServer == serverHost && not (fromServer == "") && not (fromServer == serverHost) && not (tooLong fromServer maxPubkeyLen)
746+ let validAckFields = authorized && originServer == serverHost && not (fromServer == "") && not (fromServer == serverHost) && not (tooLong fromServer maxPubkeyLen)
651747 pruneResult <- atomic do
652748 pr <- pruneExpired t
653749 messages <- *messages
654750 presence <- *presence
751+ forwards <- *forwards
752+ let knownPeer = count (do
753+ p <- presence
754+ where p.server == fromServer
755+ where p.expiresAt > t
756+ yield p) > 0
757+ let validAck = validAckFields && knownPeer
655758 set *messages = if validAck
656759 then diff messages (do
657760 m <- messages
658761 mid <- dedupedIds
659762 where m.messageId == mid
660763 where m.originServer == ""
764+ where count (do
765+ f <- forwards
766+ where f.messageId == mid
767+ where f.toServer == fromServer
768+ yield f) > 0
661769 where count (do
662770 p <- presence
663771 where p.pubkey == m.toKey
@@ -666,9 +774,9 @@ serve = \req -> case req of
666774 yield p) == 0
667775 yield m)
668776 else messages
669− yield pr
777+ yield {peerServers: pr.peerServers, expiredLocal: pr.expiredLocal, validAck: validAck}
778+ respond {ok: pruneResult.validAck}
670779 broadcastOfflineGossip t pruneResult.peerServers pruneResult.expiredLocal
671− respond {ok: validAck}
672780
673781-- ============================================
674782-- Entry Point