@@ -74,7 +74,7 @@ type PollResult = PollUnauthorized {} | PollAuthorized {events: [PollEvent], cur
74type VerifyResult = VerifyFailed {} | VerifyOk {peerServers: [Text]}
74type VerifyResult = VerifyFailed {} | VerifyOk {peerServers: [Text]}
75type SendResult = SendUnauthorized {} | SendRejected {} | SendDone {status: MessageStatus, remoteServers: [Text], fromKey: PubkeyHex}
75type SendResult = SendUnauthorized {} | SendRejected {} | SendDone {status: MessageStatus, remoteServers: [Text], fromKey: PubkeyHex}
76type AckResult = AckUnauthorized {} | AckDone {pendingAcks: [AckedEntry], evictedEntries: [AckedEntry]}
76type AckResult = AckUnauthorized {} | AckDone {pendingAcks: [AckedEntry], evictedEntries: [AckedEntry]}
77−type LookupResult = LookupUnauthorized {} | LookupDone {foundServer: Text}
77+type LookupResult = LookupUnauthorized {} | LookupDone {foundServer: OptionalServerName}
80-- ============================================
80-- ============================================
@@ -151,6 +151,10 @@ route Api = AuthApi | ClientApi | FederationApi | MaintenanceApi
151any = \xs pred -> fold (\acc x -> if acc then acc else pred x) (False {}) xs
151any = \xs pred -> fold (\acc x -> if acc then acc else pred x) (False {}) xs
152none = \xs pred -> not (any xs pred)
152none = \xs pred -> not (any xs pred)
153when = \cond action -> if cond then action else yield {}
153when = \cond action -> if cond then action else yield {}
154+whenOk = \result action -> case result of
157+clamp = \lo hi x -> if x < lo then lo else if x > hi then hi else x
154headOf = \items -> fold (\acc x -> case acc of
158headOf = \items -> fold (\acc x -> case acc of
155 Nothing {} -> Just {value: x}
159 Nothing {} -> Just {value: x}
156 _ -> acc) Nothing {} items
160 _ -> acc) Nothing {} items
@@ -204,11 +208,6 @@ isAllowedServerChar = \c ->
204 || c == "m" || c == "n" || c == "o" || c == "p" || c == "q" || c == "r"
208 || c == "m" || c == "n" || c == "o" || c == "p" || c == "q" || c == "r"
205 || c == "s" || c == "t" || c == "u" || c == "v" || c == "w" || c == "x"
209 || c == "s" || c == "t" || c == "u" || c == "v" || c == "w" || c == "x"
206 || c == "y" || c == "z"
210 || c == "y" || c == "z"
207− || c == "A" || c == "B" || c == "C" || c == "D" || c == "E" || c == "F"
208− || c == "G" || c == "H" || c == "I" || c == "J" || c == "K" || c == "L"
209− || c == "M" || c == "N" || c == "O" || c == "P" || c == "Q" || c == "R"
210− || c == "S" || c == "T" || c == "U" || c == "V" || c == "W" || c == "X"
211− || c == "Y" || c == "Z"
213containsBadUrlChar = \s ->
212containsBadUrlChar = \s ->
214 if s == "" then False {}
213 if s == "" then False {}
@@ -373,9 +372,9 @@ authenticateSession = \sessions token t ->
373 Just {value: s} -> Authorized {pubkey: s.pubkey}
372 Just {value: s} -> Authorized {pubkey: s.pubkey}
375extractToken = \authHeader ->
374extractToken = \authHeader ->
375+ let token = drop 7 authHeader in
376− if toLower (take 7 authHeader) == "bearer "
376+ if toLower (take 7 authHeader) == "bearer " && not (tooLong token maxTokenLen)
377− && not (tooLong (drop 7 authHeader) maxTokenLen)
377+ then token
378− then drop 7 authHeader
381verifySecret = \token secret ->
380verifySecret = \token secret ->
@@ -388,9 +387,9 @@ verifyMaintenanceAuth = \authorization ->
388 let token = extractToken authorization in
387 let token = extractToken authorization in
389 verifySecret token federationSecret || verifySecret token maintenanceSecret
388 verifySecret token federationSecret || verifySecret token maintenanceSecret
391−clampTtl = \ttl -> if ttl < 1 then 1 else if ttl > maxPresenceTtl then maxPresenceTtl else ttl
390+clampTtl = clamp 1 maxPresenceTtl
393−clampTimestamp = \timestamp t -> if timestamp > t then t else if timestamp < t - messageMaxAge then t - messageMaxAge else timestamp
392+clampTimestamp = \timestamp t -> clamp (t - messageMaxAge) t timestamp
395nextSeqTs = \t seqCounter -> if t > seqCounter then t else seqCounter + 1
394nextSeqTs = \t seqCounter -> if t > seqCounter then t else seqCounter + 1
@@ -500,8 +499,7 @@ removeOffline = \presenceSet offlineOnlyKeys trustedOffline t ->
500countMessagesFor = \messages toKey t ->
499countMessagesFor = \messages toKey t ->
501 count (do
500 count (do
502 m <- messages
501 m <- messages
503− where m.toKey == toKey
504− where m.receivedAt >= t - messageMaxAge
502+ where m.toKey == toKey && m.receivedAt >= t - messageMaxAge
@@ -595,7 +593,8 @@ sendBatchedGossip = \peerServers allEvents ->
595broadcastOfflineGossip = \t peerServers expiredLocal -> do
593broadcastOfflineGossip = \t peerServers expiredLocal -> do
596 let expiredEntries = mergeByKey (\a b -> a.pubkey == b.pubkey) [] (do
594 let expiredEntries = mergeByKey (\a b -> a.pubkey == b.pubkey) [] (do
597 p <- expiredLocal
595 p <- expiredLocal
598− yield {pubkey: p.pubkey, ttl: (let rawTtl = (p.expiresAt - p.timestamp) / 1000 in if rawTtl < offlineGossipMinTtl then offlineGossipMinTtl else if rawTtl > maxPresenceTtl then maxPresenceTtl else rawTtl), expiresAt: p.expiresAt})
596+ let ttl = clamp offlineGossipMinTtl maxPresenceTtl ((p.expiresAt - p.timestamp) / 1000)
597+ yield {pubkey: p.pubkey, ttl: ttl, expiresAt: p.expiresAt})
599 when (federationEnabled && count expiredEntries > 0 && count peerServers > 0) (do
598 when (federationEnabled && count expiredEntries > 0 && count peerServers > 0) (do
600 let gossipEvents = do
599 let gossipEvents = do
601 e <- expiredEntries
600 e <- expiredEntries
@@ -605,6 +604,10 @@ broadcastOfflineGossip = \t peerServers expiredLocal -> do
605forkBroadcastOffline = \t result ->
604forkBroadcastOffline = \t result ->
606 fork (broadcastOfflineGossip t result.peerServers result.expiredLocal)
605 fork (broadcastOfflineGossip t result.peerServers result.expiredLocal)
607+maybePruneAndBroadcast = \t -> do
608+ pruneResult <- atomic do maybePrune t
609+ forkBroadcastOffline t pruneResult
608removeForward = \mid srv -> atomic do
611removeForward = \mid srv -> atomic do
609 forwards <- *forwards
612 forwards <- *forwards
610 set *forwards = do
613 set *forwards = do
@@ -624,9 +627,7 @@ retryForwardsForServer = \entries remaining ->
624 toKey: f.toKey, fromKey: f.fromKey, encryptedBlob: f.encryptedBlob,
627 toKey: f.toKey, fromKey: f.fromKey, encryptedBlob: f.encryptedBlob,
625 timestamp: f.timestamp, originServer: serverHostN, messageId: f.messageId,
628 timestamp: f.timestamp, originServer: serverHostN, messageId: f.messageId,
626 authorization: "Bearer " ++ federationSecret})
629 authorization: "Bearer " ++ federationSecret})
628− Ok _ -> removeForward f.messageId f.toServer
630+ whenOk result (removeForward f.messageId f.toServer)
630 retryForwardsForServer rest (remaining - 1)
631 retryForwardsForServer rest (remaining - 1)
632retryPendingForwards : [ForwardEntry] -> IO {network} {}
633retryPendingForwards : [ForwardEntry] -> IO {network} {}
@@ -643,15 +644,12 @@ sendBatchedAcksToServer = \serverAcks server ->
643 authorization: "Bearer " ++ federationSecret,
644 authorization: "Bearer " ++ federationSecret,
644 messageIds: (do a <- batch; yield a.messageId),
645 messageIds: (do a <- batch; yield a.messageId),
645 originServer: server, fromServer: serverHostN})
646 originServer: server, fromServer: serverHostN})
648− atomic do
647+ whenOk result (atomic do
649− ackedIds <- *ackedIds
648+ ackedIds <- *ackedIds
650− set *ackedIds = do
649+ set *ackedIds = do
651− a <- ackedIds
650+ a <- ackedIds
652− where not (a.originServer == server && any batch (\sa -> sa.messageId == a.messageId && sa.toKey == a.toKey))
651+ where not (a.originServer == server && any batch (\sa -> sa.messageId == a.messageId && sa.toKey == a.toKey))
653− yield a
652+ yield a))
656sendAcksAndClean : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: ServerName, | a}] -> [ServerName] -> IO {network} {}
654sendAcksAndClean : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: ServerName, | a}] -> [ServerName] -> IO {network} {}
657sendAcksAndClean = \pendingAcks servers ->
655sendAcksAndClean = \pendingAcks servers ->
@@ -685,12 +683,10 @@ pollOnce = \token cursor -> do
685 seqCounter <- *seqCounter
683 seqCounter <- *seqCounter
686 messages <- *messages
684 messages <- *messages
687 let cursorCeil = if seqCounter > t then seqCounter else t
685 let cursorCeil = if seqCounter > t then seqCounter else t
688− let clampedCursor = if cursor < 0 then 0 else if cursor > cursorCeil then cursorCeil else cursor
686+ let clampedCursor = clamp 0 cursorCeil cursor
689 let matchingMsgs = do
687 let matchingMsgs = do
690 m <- messages
688 m <- messages
691− where m.toKey == pubkey
692− where m.receivedAt >= t - messageMaxAge
689+ where m.toKey == pubkey && m.receivedAt >= t - messageMaxAge && m.receivedAt > clampedCursor
693− where m.receivedAt > clampedCursor
695 let limitedMsgs = boundedSmallest matchingMsgs maxPollEvents (\m -> m.receivedAt)
691 let limitedMsgs = boundedSmallest matchingMsgs maxPollEvents (\m -> m.receivedAt)
696 let evts = do
692 let evts = do
@@ -771,31 +767,32 @@ serve = \req -> case req of
771 then collectPeerServers updatedPresence (\p -> p.server)
767 then collectPeerServers updatedPresence (\p -> p.server)
773 yield {pr: pr, verify: VerifyOk {peerServers: peerServers}}
769 yield {pr: pr, verify: VerifyOk {peerServers: peerServers}}
774− fork (case result.verify of
771+ case result.verify of
775− VerifyOk {peerServers} -> do
772+ VerifyOk {peerServers} ->
776− when federationEnabled
773+ when federationEnabled
777− (sendToEach peerServers (\server -> do
774+ (sendToEach peerServers (\server -> do
778− _ <- fetch (federationScheme ++ server) (RecvGossip {
775+ _ <- fetch (federationScheme ++ server) (RecvGossip {
779− authorization: "Bearer " ++ federationSecret,
776+ authorization: "Bearer " ++ federationSecret,
780− events: [{eventType: "online", pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl}], hopsRemaining: 2, fromServer: serverHostN})
777+ events: [{eventType: "online", pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl}], hopsRemaining: 2, fromServer: serverHostN})
781− yield {}))
778+ yield {}))
780+ let expiredLocal = case result.verify of
782− broadcastOfflineGossip t result.pr.peerServers (do p <- result.pr.expiredLocal; where not (p.pubkey == pubkey); yield p)
781+ VerifyOk _ -> do p <- result.pr.expiredLocal; where not (p.pubkey == pubkey); yield p
783− VerifyFailed {} ->
782+ _ -> result.pr.expiredLocal
784− broadcastOfflineGossip t result.pr.peerServers result.pr.expiredLocal)
783+ broadcastOfflineGossip t result.pr.peerServers expiredLocal)
785 yield (case result.verify of
784 yield (case result.verify of
786 VerifyFailed _ -> respond {token: "", expiresAt: 0}
785 VerifyFailed _ -> respond {token: "", expiresAt: 0}
787 VerifyOk _ -> respond {token: token, expiresAt: expiry})
786 VerifyOk _ -> respond {token: token, expiresAt: expiry})
789 Poll {cursor, timeout, authorization, respond} -> do
788 Poll {cursor, timeout, authorization, respond} -> do
791− let clampedTimeout = if timeout < 0 then 0 else if timeout > 60000 then 60000 else timeout
790+ let clampedTimeout = clamp 0 60000 timeout
792 let deadline = t + clampedTimeout
791 let deadline = t + clampedTimeout
793 let token = extractToken authorization
792 let token = extractToken authorization
794 result <- pollLoop token cursor deadline
793 result <- pollLoop token cursor deadline
795 tAfterPoll <- now
794 tAfterPoll <- now
796− pruneResult <- atomic do
795+ maybePruneAndBroadcast tAfterPoll
797− maybePrune tAfterPoll
798− forkBroadcastOffline tAfterPoll pruneResult
799 yield (case result of
796 yield (case result of
800 PollUnauthorized {} -> respond {events: [], cursor: -1, authorized: False {}}
797 PollUnauthorized {} -> respond {events: [], cursor: -1, authorized: False {}}
801 PollAuthorized {events, cursor: c} -> respond {events: events, cursor: c, authorized: True {}})
798 PollAuthorized {events, cursor: c} -> respond {events: events, cursor: c, authorized: True {}})
@@ -825,7 +822,7 @@ serve = \req -> case req of
825 set *messages = union messages
822 set *messages = union messages
826 [{messageId: msgId, toKey: to, fromKey: fromKey,
823 [{messageId: msgId, toKey: to, fromKey: fromKey,
827 encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}]
824 encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}]
828− let toPresence = do p <- presence; where p.pubkey == to; where p.expiresAt > t; yield p
825+ let toPresence = do p <- presence; where p.pubkey == to && p.expiresAt > t; yield p
829 let localOnline = any toPresence (\p -> p.server == serverHostN)
826 let localOnline = any toPresence (\p -> p.server == serverHostN)
830 let remoteServers = do p <- toPresence; where not (p.server == serverHostN); yield p.server
827 let remoteServers = do p <- toPresence; where not (p.server == serverHostN); yield p.server
831 forwards <- *forwards
828 forwards <- *forwards
@@ -847,13 +844,10 @@ serve = \req -> case req of
847 toKey: to, fromKey: fromKey, encryptedBlob: encryptedBlob,
844 toKey: to, fromKey: fromKey, encryptedBlob: encryptedBlob,
848 timestamp: clampedTs, originServer: serverHostN, messageId: msgId,
845 timestamp: clampedTs, originServer: serverHostN, messageId: msgId,
849 authorization: "Bearer " ++ federationSecret})
846 authorization: "Bearer " ++ federationSecret})
851− Ok _ -> removeForward msgId server
847+ whenOk fwdResult (removeForward msgId server))
853 when (status == "queued")
848 when (status == "queued")
854 (println ("sendMessage: forwards at capacity (" ++ show maxForwardsEntries ++ "), skipping forward tracking")))
849 (println ("sendMessage: forwards at capacity (" ++ show maxForwardsEntries ++ "), skipping forward tracking")))
855− SendUnauthorized {} -> yield {}
856− SendRejected {} -> yield {}
850+ _ -> yield {}
857 broadcastOfflineGossip t result.pr.peerServers result.pr.expiredLocal)
851 broadcastOfflineGossip t result.pr.peerServers result.pr.expiredLocal)
858 yield (case result.send of
852 yield (case result.send of
859 SendUnauthorized _ -> respond {status: "unauthorized", messageId: ""}
853 SendUnauthorized _ -> respond {status: "unauthorized", messageId: ""}
@@ -923,7 +917,7 @@ serve = \req -> case req of
923 when (dropped > 0)
917 when (dropped > 0)
924 (println ("ackMessages: ackedIds at capacity (" ++ show maxAckedIds ++ "), " ++ show dropped ++ " ack entries permanently lost for " ++ server))
918 (println ("ackMessages: ackedIds at capacity (" ++ show maxAckedIds ++ "), " ++ show dropped ++ " ack entries permanently lost for " ++ server))
925 _ -> yield {}))
919 _ -> yield {}))
926− AckUnauthorized {} -> yield {}
920+ _ -> yield {}
927 broadcastOfflineGossip t result.pr.peerServers result.pr.expiredLocal)
921 broadcastOfflineGossip t result.pr.peerServers result.pr.expiredLocal)
928 yield (case result.ack of
922 yield (case result.ack of
929 AckUnauthorized _ -> respond {ok: False {}}
923 AckUnauthorized _ -> respond {ok: False {}}
@@ -943,12 +937,13 @@ serve = \req -> case req of
943 presence <- *presence
937 presence <- *presence
944 let valid = do
938 let valid = do
945 p <- presence
939 p <- presence
946− where p.pubkey == pubkey
940+ where p.pubkey == pubkey && p.expiresAt > t
947− where p.expiresAt > t
949− let foundServer = if any valid (\p -> p.server == serverHostN)
942+ let foundServer = (fold (\acc p ->
950− then serverHostN
943+ if acc.server == serverHostN then acc
944+ else if p.server == serverHostN then {server: serverHostN, best: p.expiresAt}
951− else (fold (\acc p -> if p.expiresAt > acc.best then {server: p.server, best: p.expiresAt} else acc) {server: "", best: 0} valid).server
945+ else if p.expiresAt > acc.best then {server: p.server, best: p.expiresAt}
946+ else acc) {server: "", best: 0} valid).server
952 yield {pr: pr, lookup: LookupDone {foundServer: foundServer}}
947 yield {pr: pr, lookup: LookupDone {foundServer: foundServer}}
953 forkBroadcastOffline t result.pr
948 forkBroadcastOffline t result.pr
954 yield (case result.lookup of
949 yield (case result.lookup of
@@ -962,8 +957,7 @@ serve = \req -> case req of
962 let fromServer = normalizeServerName fromServer
957 let fromServer = normalizeServerName fromServer
963 if not authorized || cappedHops < 1 || fromServer == serverHostN || count events > maxGossipEvents
958 if not authorized || cappedHops < 1 || fromServer == serverHostN || count events > maxGossipEvents
965− pruneResult <- atomic do maybePrune t
960+ maybePruneAndBroadcast t
966− forkBroadcastOffline t pruneResult
967 yield (respond {ok: False {}})
961 yield (respond {ok: False {}})
969 let allValidEntries = validGossipEvents events t
963 let allValidEntries = validGossipEvents events t