@@ -110,10 +110,11 @@ type PresenceEntry = {pubkey: PubkeyHex, server: AnyServerName, expiresAt: Times
110110type ForwardEntry = {messageId: MessageIdHex, toServer: ServerName, createdAt: Timestamp, toKey: PubkeyHex, fromKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp}
111111type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: ServerName}
112112data AuthResult = Authorized {pubkey: PubkeyHex} | Unauthorized {}
113−data SendResult = SendFailed {status: MessageStatus} | SendDone {delivery: DeliveryStatus, remoteServers: [ServerName], fromKey: PubkeyHex}
113+data SendResult = SendFailed {status: MessageStatus} | SendDone {delivery: DeliveryStatus, remoteServers: [ServerName], forwardServers: [ServerName], fromKey: PubkeyHex}
114114data VerifyResult = VerifyFailed {} | VerifyOk {peerServers: [ServerName]}
115115data AckResult = AckUnauthorized {} | AckDone {pendingAcks: [AckedEntry], evictedEntries: [AckedEntry]}
116−type PollResponse = {events: [PollEvent], cursor: Timestamp, authorized: Bool}
116+data PollResult = PollUnauthorized {} | PollOk {events: [PollEvent], cursor: Timestamp}
117+data ForwardResult = ForwardDuplicate {} | ForwardRejected {} | ForwardStored {}
117118
118119
119120-- ============================================
@@ -200,6 +201,10 @@ headOf = \items -> fold (\acc x -> case acc of
200201 Nothing {} -> Just {value: x}
201202 _ -> acc) Nothing {} items
202203
204+maxBy = \items getKey -> fold (\acc x -> case acc of
205+ Nothing {} -> Just {value: x}
206+ Just {value: v} -> if getKey x > getKey v then Just {value: x} else acc) Nothing {} items
207+
203208hexDigit = \n -> take 1 (drop (clamp 0 15 n) "0123456789abcdef")
204209
205210byteToHex = \n -> hexDigit (n / 16) ++ hexDigit (n - (n / 16) * 16)
@@ -220,7 +225,7 @@ genToken = genTokenN 8
220225
221226tooLong = \s maxLen -> drop maxLen s != ""
222227
223−isHexChar = \c -> contains c "0123456789abcdefABCDEF"
228+isHexChar = \c -> contains c "0123456789abcdef"
224229
225230isValidHexPairs = \s ->
226231 if s == "" then True {}
@@ -352,10 +357,6 @@ authenticateSession = \sessions token t ->
352357 Nothing {} -> Unauthorized {}
353358 Just {value: s} -> Authorized {pubkey: s.pubkey}
354359
355−authenticateRequest = \authorization t -> do
356− sessions <- *sessions
357− yield (authenticateSession sessions (extractToken authorization) t)
358−
359360extractToken = \authHeader ->
360361 let token = drop 7 authHeader in
361362 if toLower (take 7 authHeader) == "bearer " && not (tooLong token maxTokenLen)
@@ -621,10 +622,27 @@ effectiveGossipEvents = \entries appliedKeys eventType -> do
621622 where any appliedKeys (\a -> samePubkeyServer a e)
622623 yield {eventType: eventType, pubkey: e.pubkey, server: e.server, timestamp: e.timestamp, ttl: e.ttl}
623624
625+mergeGossipPresence = \presence deduped trustedOffline fromServer t ->
626+ let trustedDeduped = do e <- deduped; where e.server == fromServer || any presence (\p -> p.server == e.server && p.expiresAt > t); yield {expiresAt: e.expiresAt, pubkey: e.pubkey, server: e.server, timestamp: e.timestamp} in
627+ let appliedOnline = do e <- trustedDeduped; where none presence (\p -> samePubkeyServer p e && p.expiresAt >= e.expiresAt); yield (pubkeyServer e) in
628+ let merged = mergeByPubkeyServer presence trustedDeduped in
629+ let afterOffline = removeOffline merged trustedOffline t in
630+ let appliedOffline = do p <- diff merged afterOffline; yield (pubkeyServer p) in
631+ let capResult = capGossipPresence {afterOffline: afterOffline, appliedOnline: appliedOnline, trustedDeduped: trustedDeduped, presence: presence, trustedOffline: trustedOffline, t: t} in
632+ {finalPresence: capResult.finalPresence,
633+ appliedKeys: union capResult.appliedOnline appliedOffline,
634+ gossipPeerServers: collectPeerServers capResult.finalPresence (\p -> p.server),
635+ withinCapacity: capResult.withinCapacity}
636+
624637rejectFederation = \t respond -> do
625638 maybePruneAndBroadcast t
626639 yield (respond {ok: False {}})
627640
641+withFederationAuth = \authorization t respond extraCheck action ->
642+ if not (verifyFederationAuth authorization) || extraCheck
643+ then rejectFederation t respond
644+ else action
645+
628646maybePruneAndBroadcast = \t -> do
629647 pruneResult <- atomic do maybePrune t
630648 forkBroadcastOffline t pruneResult
@@ -658,13 +676,16 @@ retryPendingForwards = \pending -> do
658676 retryForwardsForServer (do x <- pending; where x.toServer == server; yield x) maxForwardRetriesPerServer)
659677 yield {}
660678
679+fetchFederationAck = \server messageIds ->
680+ fetch (federationUrl server) (FederationAck {
681+ authorization: federationAuth,
682+ messageIds: messageIds,
683+ originServer: server, fromServer: serverHostN})
684+
661685sendBatchedAcksToServer : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: ServerName, | a}] -> ServerName -> IO {network} {}
662686sendBatchedAcksToServer = \serverAcks server ->
663687 forEachBatch serverAcks maxMessageIds (\batch -> do
664− result <- fetch (federationUrl server) (FederationAck {
665− authorization: federationAuth,
666− messageIds: (do a <- batch; yield a.messageId),
667− originServer: server, fromServer: serverHostN})
688+ result <- fetchFederationAck server (do a <- batch; yield a.messageId)
668689 whenOk result (atomic do
669690 ackedIds <- *ackedIds
670691 set *ackedIds = do
@@ -686,10 +707,7 @@ retryEvictedAcks = \evictedEntries pendingAcks ->
686707 in do
687708 let evictedOriginServers = collectPeerServers evictedToRetry (\a -> a.originServer)
688709 sendToEach evictedOriginServers (\server -> do
689− fetchResult <- fetch (federationUrl server) (FederationAck {
690− authorization: federationAuth,
691− messageIds: (do a <- evictedToRetry; where a.originServer == server; yield a.messageId),
692− originServer: server, fromServer: serverHostN})
710+ fetchResult <- fetchFederationAck server (do a <- evictedToRetry; where a.originServer == server; yield a.messageId)
693711 case fetchResult of
694712 Err _ -> do
695713 dropped <- atomic do
@@ -725,7 +743,7 @@ pollOnce = \token cursor -> do
725743 let auth = authenticateSession sessions token t
726744 case auth of
727745 Unauthorized {} ->
728− yield {events: [], cursor: 0, authorized: False {}}
746+ yield (PollUnauthorized {})
729747 Authorized {pubkey} -> do
730748 seqCounter <- *seqCounter
731749 messages <- *messages
@@ -742,19 +760,20 @@ pollOnce = \token cursor -> do
742760 encryptedBlob: m.encryptedBlob,
743761 timestamp: m.timestamp}}
744762 let lastReceivedAt = fold (\acc m -> max acc m.receivedAt) clampedCursor limitedMsgs
745− yield {events: evts, cursor: lastReceivedAt, authorized: True {}}
763+ yield (PollOk {events: evts, cursor: lastReceivedAt})
746764
747−pollLoop : Text -> Int -> Int<Ms> -> IO {clock} PollResponse
765+pollLoop : Text -> Int -> Int<Ms> -> IO {clock} PollResult
748766pollLoop = \token cursor deadline -> do
749767 result <- pollOnce token cursor
750− if not result.authorized then yield result
751− else do
752− t <- now
753− if count result.events == 0 && t < deadline
754− then do
755− sleep (clamp 1 500 (deadline - t))
756− pollLoop token cursor deadline
757− else yield result
768+ case result of
769+ PollUnauthorized {} -> yield result
770+ PollOk {events} -> do
771+ t <- now
772+ if count events == 0 && t < deadline
773+ then do
774+ sleep (clamp 1 500 (deadline - t))
775+ pollLoop token cursor deadline
776+ else yield result
758777
759778-- ============================================
760779-- Request Handler
@@ -827,7 +846,9 @@ serve = \req -> case req of
827846 result <- pollLoop token cursor deadline
828847 tAfterPoll <- now
829848 maybePruneAndBroadcast tAfterPoll
830− yield (respond result)
849+ yield (respond (case result of
850+ PollUnauthorized {} -> {events: [], cursor: 0, authorized: False {}}
851+ PollOk {events: evts, cursor: c} -> {events: evts, cursor: c, authorized: True {}}))
831852
832853 SendMessage {to, encryptedBlob, timestamp, authorization, respond} -> do
833854 t <- now
@@ -835,7 +856,8 @@ serve = \req -> case req of
835856 let clampedTs = clampTimestamp timestamp t
836857 result <- atomic do
837858 pr <- maybePrune t
838− auth <- authenticateRequest authorization t
859+ sessions <- *sessions
860+ let auth = authenticateSession sessions (extractToken authorization) t
839861 case auth of
840862 Unauthorized {} ->
841863 yield {pr: pr, send: SendFailed {status: MsUnauthorized {}}}
@@ -863,9 +885,9 @@ serve = \req -> case req of
863885 set *forwards = if shouldForward
864886 then union forwards (do s <- forwardServers; yield {messageId: msgId, toServer: s, createdAt: t, toKey: to, fromKey: fromKey, encryptedBlob: encryptedBlob, timestamp: clampedTs})
865887 else forwards
866− yield {pr: pr, send: SendDone {delivery: delivery, remoteServers: remoteServers, fromKey: fromKey}}
888+ yield {pr: pr, send: SendDone {delivery: delivery, remoteServers: remoteServers, forwardServers: forwardServers, fromKey: fromKey}}
867889 case result.send of
868− SendDone {delivery, remoteServers, fromKey} ->
890+ SendDone {delivery, remoteServers, forwardServers, fromKey} ->
869891 fork (do
870892 when (count remoteServers > 0 && federationEnabled) (do
871893 sendToEach remoteServers (\server -> do
@@ -873,7 +895,8 @@ serve = \req -> case req of
873895 toKey: to, fromKey: fromKey, encryptedBlob: encryptedBlob,
874896 timestamp: clampedTs, originServer: serverHostN, messageId: msgId,
875897 authorization: federationAuth})
876− whenOk fwdResult (removeForward msgId server))
898+ when (any forwardServers (\s -> s == server))
899+ (whenOk fwdResult (removeForward msgId server)))
877900 case delivery of
878901 DsQueued {} -> println ("sendMessage: forwards at capacity (" ++ show maxForwardsEntries ++ "), skipping forward tracking")
879902 _ -> yield {})
@@ -888,7 +911,8 @@ serve = \req -> case req of
888911 let dedupedIds = dedupMessageIds messageIds
889912 result <- atomic do
890913 pr <- maybePrune t
891− auth <- authenticateRequest authorization t
914+ sessions <- *sessions
915+ let auth = authenticateSession sessions (extractToken authorization) t
892916 case auth of
893917 Unauthorized {} ->
894918 yield {pr: pr, ack: AckUnauthorized {}}
@@ -903,7 +927,7 @@ serve = \req -> case req of
903927 m <- ackedMessages
904928 case m.origin of
905929 FederatedOrigin {server} -> yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: server}
906− _ -> []
930+ LocalOrigin {} -> []
907931 ackedIds <- *ackedIds
908932 let combinedAckedIds = union ackedIds ackedFederated
909933 let evictedEntries = let excessCount = count combinedAckedIds - maxAckedIds in
@@ -924,7 +948,8 @@ serve = \req -> case req of
924948 t <- now
925949 result <- atomic do
926950 pr <- maybePrune t
927− auth <- authenticateRequest authorization t
951+ sessions <- *sessions
952+ let auth = authenticateSession sessions (extractToken authorization) t
928953 case auth of
929954 Unauthorized {} ->
930955 yield {pr: pr, server: "", authorized: False {}}
@@ -936,9 +961,7 @@ serve = \req -> case req of
936961 yield p
937962 let localOnline = any valid (\p -> isLocalServer p.server)
938963 let remoteValid = do p <- valid; where not (isLocalServer p.server); yield p
939− let bestRemote = fold (\acc p -> case acc of
940− Nothing {} -> Just {value: p}
941− Just {value: v} -> if p.expiresAt > v.expiresAt then Just {value: p} else acc) Nothing {} remoteValid
964+ let bestRemote = maxBy remoteValid (\p -> p.expiresAt)
942965 let foundServer = if localOnline then serverHostN
943966 else case bestRemote of Just {value: p} -> p.server; Nothing {} -> ""
944967 yield {pr: pr, server: foundServer, authorized: True {}}
@@ -947,91 +970,73 @@ serve = \req -> case req of
947970
948971 RecvGossip {events, hopsRemaining, fromServer, authorization, respond} -> do
949972 t <- now
950− let authorized = verifyFederationAuth authorization
951973 let cappedHops = clamp 0 maxGossipHops hopsRemaining
952− if not authorized || cappedHops < 1 || isLocalServer fromServer || count events > maxGossipEvents
953− then rejectFederation t respond
954− else do
955− let validated = validGossipEvents events t
956− let onlineEntries = validated.online
957− let trustedOffline = do e <- validated.offline; where e.server == fromServer; yield e
958− let filteredOnline = resolveByTimestamp onlineEntries trustedOffline
959− let filteredOffline = resolveByTimestamp trustedOffline onlineEntries
960− let deduped = mergeByPubkeyServer [] filteredOnline
961− gossipResult <- atomic do
962− pr <- maybePrune t
963− presence <- *presence
964− let trustedDeduped = do
965− e <- deduped
966− where e.server == fromServer || any presence (\p -> p.server == e.server && p.expiresAt > t)
967− yield {expiresAt: e.expiresAt, pubkey: e.pubkey, server: e.server, timestamp: e.timestamp}
968− let appliedOnline = do
969− e <- trustedDeduped
970− where none presence (\p -> samePubkeyServer p e && p.expiresAt >= e.expiresAt)
971− yield (pubkeyServer e)
972− let merged = mergeByPubkeyServer presence trustedDeduped
973− let afterOffline = removeOffline merged trustedOffline t
974− let appliedOffline = do p <- diff merged afterOffline; yield (pubkeyServer p)
975− let capResult = capGossipPresence {afterOffline: afterOffline, appliedOnline: appliedOnline, trustedDeduped: trustedDeduped, presence: presence, trustedOffline: trustedOffline, t: t}
976− set *presence = capResult.finalPresence
977− let gossipPeerServers = collectPeerServers capResult.finalPresence (\p -> p.server)
978− yield {appliedKeys: union capResult.appliedOnline appliedOffline, pr: pr, gossipPeerServers: gossipPeerServers, withinCapacity: capResult.withinCapacity}
979− let effectiveEvents = union (effectiveGossipEvents filteredOnline gossipResult.appliedKeys (GossipOnline {})) (effectiveGossipEvents filteredOffline gossipResult.appliedKeys (GossipOffline {}))
980− when (not gossipResult.withinCapacity)
981− (println ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer))
982− fork (do
983− when (cappedHops > 1 && federationEnabled && count effectiveEvents > 0) (
984− let targets = do
985− s <- gossipResult.gossipPeerServers
986− where s != fromServer
987− yield s
988− in sendBatchedGossip targets effectiveEvents (cappedHops - 1))
989− broadcastOfflineGossip t gossipResult.pr.peerServers gossipResult.pr.expiredLocal)
990− yield (respond {ok: True {}})
974+ withFederationAuth authorization t respond (cappedHops < 1 || isLocalServer fromServer || count events > maxGossipEvents) (do
975+ let validated = validGossipEvents events t
976+ let onlineEntries = validated.online
977+ let trustedOffline = do e <- validated.offline; where e.server == fromServer; yield e
978+ let filteredOnline = resolveByTimestamp onlineEntries trustedOffline
979+ let filteredOffline = resolveByTimestamp trustedOffline onlineEntries
980+ let deduped = mergeByPubkeyServer [] filteredOnline
981+ gossipResult <- atomic do
982+ pr <- maybePrune t
983+ presence <- *presence
984+ let gm = mergeGossipPresence presence deduped trustedOffline fromServer t
985+ set *presence = gm.finalPresence
986+ yield {gm: gm, pr: pr}
987+ let effectiveEvents = union (effectiveGossipEvents filteredOnline gossipResult.gm.appliedKeys (GossipOnline {})) (effectiveGossipEvents filteredOffline gossipResult.gm.appliedKeys (GossipOffline {}))
988+ when (not gossipResult.gm.withinCapacity)
989+ (println ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer))
990+ fork (do
991+ when (cappedHops > 1 && federationEnabled && count effectiveEvents > 0) (
992+ let targets = do
993+ s <- gossipResult.gm.gossipPeerServers
994+ where s != fromServer
995+ yield s
996+ in sendBatchedGossip targets effectiveEvents (cappedHops - 1))
997+ broadcastOfflineGossip t gossipResult.pr.peerServers gossipResult.pr.expiredLocal)
998+ yield (respond {ok: True {}}))
991999
9921000 ForwardMessage {toKey, fromKey, encryptedBlob, timestamp, originServer, messageId, authorization, respond} -> do
9931001 t <- now
994− let authorized = verifyFederationAuth authorization
995− if not authorized || isLocalServer originServer
996− then rejectFederation t respond
997− else do
998− let clampedTs = clampTimestamp timestamp t
999− result <- atomic do
1000− pr <- maybePrune t
1001− messages <- *messages
1002− ackedIds <- *ackedIds
1003− if hasMessage messageId toKey messages || hasMessage messageId toKey ackedIds
1004− then yield {pr: pr, ok: True {}}
1005− else if countMessagesFor messages toKey t >= maxMessagesPerRecipient
1006− then yield {pr: pr, ok: False {}}
1007− else do
1008− seqCounter <- *seqCounter
1009− let seqTs = nextSeqTs t seqCounter
1010− set *seqCounter = seqTs
1011− set *messages = union messages
1012− [{messageId: messageId, toKey: toKey, fromKey: fromKey,
1013− encryptedBlob: encryptedBlob, timestamp: clampedTs, origin: FederatedOrigin {server: originServer}, receivedAt: seqTs}]
1014− yield {pr: pr, ok: True {}}
1015− forkBroadcastOffline t result.pr
1016− yield (respond {ok: result.ok})
1002+ withFederationAuth authorization t respond (isLocalServer originServer) (do
1003+ let clampedTs = clampTimestamp timestamp t
1004+ result <- atomic do
1005+ pr <- maybePrune t
1006+ messages <- *messages
1007+ ackedIds <- *ackedIds
1008+ if hasMessage messageId toKey messages || hasMessage messageId toKey ackedIds
1009+ then yield {pr: pr, forward: ForwardDuplicate {}}
1010+ else if countMessagesFor messages toKey t >= maxMessagesPerRecipient
1011+ then yield {pr: pr, forward: ForwardRejected {}}
1012+ else do
1013+ seqCounter <- *seqCounter
1014+ let seqTs = nextSeqTs t seqCounter
1015+ set *seqCounter = seqTs
1016+ set *messages = union messages
1017+ [{messageId: messageId, toKey: toKey, fromKey: fromKey,
1018+ encryptedBlob: encryptedBlob, timestamp: clampedTs, origin: FederatedOrigin {server: originServer}, receivedAt: seqTs}]
1019+ yield {pr: pr, forward: ForwardStored {}}
1020+ forkBroadcastOffline t result.pr
1021+ yield (respond {ok: case result.forward of
1022+ ForwardRejected {} -> False {}
1023+ ForwardDuplicate {} -> True {}
1024+ ForwardStored {} -> True {}}))
10171025
10181026 FederationAck {messageIds, originServer, fromServer, authorization, respond} -> do
10191027 t <- now
1020− let authorized = verifyFederationAuth authorization
1021− if not authorized || not (isLocalServer originServer) || isLocalServer fromServer
1022− then rejectFederation t respond
1023− else do
1024− let dedupedIds = dedupMessageIds messageIds
1025− pruneResult <- atomic do
1026− pr <- maybePrune t
1027− forwards <- *forwards
1028− set *forwards = do
1029− f <- forwards
1030− where not (f.toServer == fromServer && any dedupedIds (\mid -> f.messageId == mid))
1031− yield f
1032− yield pr
1033− forkBroadcastOffline t pruneResult
1034− yield (respond {ok: True {}})
1028+ withFederationAuth authorization t respond (not (isLocalServer originServer) || isLocalServer fromServer) (do
1029+ let dedupedIds = dedupMessageIds messageIds
1030+ pruneResult <- atomic do
1031+ pr <- maybePrune t
1032+ forwards <- *forwards
1033+ set *forwards = do
1034+ f <- forwards
1035+ where not (f.toServer == fromServer && any dedupedIds (\mid -> f.messageId == mid))
1036+ yield f
1037+ yield pr
1038+ forkBroadcastOffline t pruneResult
1039+ yield (respond {ok: True {}}))
10351040
10361041 Prune {authorization, respond} -> do
10371042 t <- now
@@ -1047,7 +1052,7 @@ backgroundPrune = do
10471052 sleep backgroundPruneInterval
10481053 t <- now
10491054 pruneAndRetry t
1050− fork backgroundPrune
1055+ backgroundPrune
10511056
10521057main = do
10531058 println "Skrepka relay server starting on port 8080..."