6c298d9 · skrepka · git.iko.soy

git.iko.soy/skrepka

Improve relay security and reliability: origin validation, per-key rate limits, poll concurrency cap, ack overflow buffer

6c298d9iko6 months ago

server.knot+120−51

compare with master

unifiedsplit

@@ -31,11 +31,13 @@ maxTokenLen = 256

31maxChallenges = 10000

32maxSessions = 10000

33maxAckedIds = 10000

34+maxAckOverflow = 200

34maxChallengesPerKey = 5

35maxChallengesPerKey = 5

35maxSessionsPerKey = 10

36maxSessionsPerKey = 10

36challengeExpiry : Int<Ms>

37challengeExpiry : Int<Ms>

37challengeExpiry = 60000 -- 1 minute

38challengeExpiry = 60000 -- 1 minute

38−challengeRateLimit = 50

39+maxChallengeRatePerKey = 5

40+maxChallengeRateEntries = 1000

39challengeRateWindowMs : Int<Ms>

41challengeRateWindowMs : Int<Ms>

40challengeRateWindowMs = 1000

42challengeRateWindowMs = 1000

41maxSendRatePerKey = 20

43maxSendRatePerKey = 20

@@ -61,6 +63,8 @@ backgroundPruneInterval : Int<Ms>

61backgroundPruneInterval = 30000 -- 30 seconds

63backgroundPruneInterval = 30000 -- 30 seconds

62pollTimeoutMs : Int<Ms>

64pollTimeoutMs : Int<Ms>

63pollTimeoutMs = 25000 -- long-poll heartbeat interval

65pollTimeoutMs = 25000 -- long-poll heartbeat interval

66+maxPollsPerKey = 3

67+maxActivePolls = 5000

64presenceRefreshInterval : Int<Ms>

68presenceRefreshInterval : Int<Ms>

65presenceRefreshInterval = 300000 -- 5 minutes

69presenceRefreshInterval = 300000 -- 5 minutes

66presenceRefreshThreshold : Int<Ms>

70presenceRefreshThreshold : Int<Ms>

@@ -130,6 +134,8 @@ type ForwardEntry = {messageId: MessageIdHex, toServer: ServerName, createdAt: T

130type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: ServerName}

134type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: ServerName}

131type ForwardFailure = {server: Text, failedAt: Timestamp, failures: Int}

135type ForwardFailure = {server: Text, failedAt: Timestamp, failures: Int}

132type SendRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}

136type SendRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}

137+type ChallengeRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}

138+type ActivePoll = {pubkey: PubkeyHex, startedAt: Timestamp}

133data AuthResult = Authorized {pubkey: PubkeyHex} | Unauthorized {}

139data AuthResult = Authorized {pubkey: PubkeyHex} | Unauthorized {}

134data SendResult = SendFailed {status: MessageStatus} | SendDone {delivery: DeliveryStatus, remoteServers: [ServerName]} | SendDuplicate {}

140data SendResult = SendFailed {status: MessageStatus} | SendDone {delivery: DeliveryStatus, remoteServers: [ServerName]} | SendDuplicate {}

135data ForwardResult = ForwardDuplicate {} | ForwardRejected {} | ForwardStored {}

141data ForwardResult = ForwardDuplicate {} | ForwardRejected {} | ForwardStored {}

@@ -160,8 +166,7 @@ data ForwardResult = ForwardDuplicate {} | ForwardRejected {} | ForwardStored {}

160

166

161*lastPresenceRefresh : Timestamp

167*lastPresenceRefresh : Timestamp

162

168

163−*challengeRateTs : Timestamp

169+*challengeRates : [ChallengeRateEntry]

164−*challengeRateCount : Int

165

170

166*pollWakeup : Timestamp

171*pollWakeup : Timestamp

167

172

@@ -182,6 +187,10 @@ data ForwardResult = ForwardDuplicate {} | ForwardRejected {} | ForwardStored {}

182*messageSignal14 : Timestamp

187*messageSignal14 : Timestamp

183*messageSignal15 : Timestamp

188*messageSignal15 : Timestamp

184

189

190+*activePolls : [ActivePoll]

191+

192+*ackOverflow : [AckedEntry]

193+

185-- ============================================

194-- ============================================

186-- Routes

195-- Routes

187-- ============================================

196-- ============================================

@@ -212,7 +221,7 @@ route FederationApi where

212 POST {events: [GossipEvent], hopsRemaining: Int, fromServer: ServerName} /gossip headers {authorization: Text}

221 POST {events: [GossipEvent], hopsRemaining: Int, fromServer: ServerName} /gossip headers {authorization: Text}

213 -> {ok: Bool}

222 -> {ok: Bool}

214 = RecvGossip

223 = RecvGossip

215− POST {toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: ServerName, messageId: MessageIdHex} /forward headers {authorization: Text}

224+ POST {toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: ServerName, messageId: MessageIdHex, fromServer: ServerName} /forward headers {authorization: Text}

216 -> {ok: Bool}

225 -> {ok: Bool}

217 = ForwardMessage

226 = ForwardMessage

218 POST {messageIds: [MessageIdHex], originServer: ServerName, fromServer: ServerName} /ack headers {authorization: Text}

227 POST {messageIds: [MessageIdHex], originServer: ServerName, fromServer: ServerName} /ack headers {authorization: Text}

@@ -536,6 +545,14 @@ capGlobal = \items maxCount getKey ->

536capPerKeyAndGlobal = \items matchKey maxPerKey getExpiry t maxGlobal ->

545capPerKeyAndGlobal = \items matchKey maxPerKey getExpiry t maxGlobal ->

537 capGlobal (capPerKey items matchKey maxPerKey getExpiry t) maxGlobal getExpiry

546 capGlobal (capPerKey items matchKey maxPerKey getExpiry t) maxGlobal getExpiry

538

547

548+removeOneForKey = \items pubkey ->

549+ let result = fold (\acc x ->

550+ if not acc.removed && x.pubkey == pubkey

551+ then {removed: True {}, items: acc.items}

552+ else {removed: acc.removed, items: union acc.items [x]}

553+ ) {removed: False {}, items: []} items

554+ in result.items

555+

539collectPeerServers : [a] -> (a -> Text) -> [Text]

556collectPeerServers : [a] -> (a -> Text) -> [Text]

540collectPeerServers = \items getServer ->

557collectPeerServers = \items getServer ->

541 fold (\acc item ->

558 fold (\acc item ->

@@ -635,6 +652,12 @@ pruneExpired = \t -> do

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

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

636 forwardFailures <- *forwardFailures

653 forwardFailures <- *forwardFailures

637 set *forwardFailures = do f <- forwardFailures; where t - f.failedAt < retryBackoffMax * 2; yield f

654 set *forwardFailures = do f <- forwardFailures; where t - f.failedAt < retryBackoffMax * 2; yield f

655+ challengeRates <- *challengeRates

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

657+ activePolls <- *activePolls

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

659+ ackOverflow <- *ackOverflow

660+ set *ackOverflow = do a <- ackOverflow; where a.ackedAt > messageCutoff; yield a

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

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

639

662

640maybePrune = \t -> do

663maybePrune = \t -> do

@@ -760,7 +783,7 @@ retryForwardsForServer = \entries remaining ->

760 result <- fetch (federationUrl f.toServer) (ForwardMessage {

783 result <- fetch (federationUrl f.toServer) (ForwardMessage {

761 toKey: f.toKey, encryptedBlob: f.encryptedBlob,

784 toKey: f.toKey, encryptedBlob: f.encryptedBlob,

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

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

763− authorization: federationAuth})

786+ fromServer: serverHostN, authorization: federationAuth})

764 case result of

787 case result of

765 Err _ -> do

788 Err _ -> do

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

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

@@ -796,11 +819,11 @@ forwardQueuedMessages = \targets t -> do

796 sendToEach servers (\server ->

819 sendToEach servers (\server ->

797 retryForwardsForServer (forwardsForServer newForwards server) maxForwardRetriesPerServer)

820 retryForwardsForServer (forwardsForServer newForwards server) maxForwardRetriesPerServer)

798

821

799−fetchFederationAck = \server messageIds ->

822+fetchFederationAck = \origin messageIds ->

800− fetch (federationUrl server) (FederationAck {

823+ fetch (federationUrl origin) (FederationAck {

801 authorization: federationAuth,

824 authorization: federationAuth,

802 messageIds: messageIds,

825 messageIds: messageIds,

803− originServer: server, fromServer: serverHostN})

826+ originServer: origin, fromServer: serverHostN})

804

827

805sendBatchedAcksToServer : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: ServerName, | a}] -> ServerName -> IO {network} {}

828sendBatchedAcksToServer : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: ServerName, | a}] -> ServerName -> IO {network} {}

806sendBatchedAcksToServer = \serverAcks server ->

829sendBatchedAcksToServer = \serverAcks server ->

@@ -827,19 +850,40 @@ retryEvictedAcks = \evictedEntries pendingAcks ->

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

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

828 case fetchResult of

851 case fetchResult of

829 Err _ -> do

852 Err _ -> do

830− dropped <- atomic do

853+ overflow <- atomic do

831 ackedIds <- *ackedIds

854 ackedIds <- *ackedIds

832 let reinsert = acksForServer evictedToRetry server

855 let reinsert = acksForServer evictedToRetry server

833 let combined = union ackedIds reinsert

856 let combined = union ackedIds reinsert

834 let excess = count combined - maxAckedIds

857 let excess = count combined - maxAckedIds

835− let afterEviction = if excess <= 0 then combined

858+ if excess <= 0

836− else evictOldestAckedIds combined excess

859+ then do set *ackedIds = combined; yield []

837− set *ackedIds = afterEviction

860+ else do

838− yield excess

861+ let evicted = takeRelation excess (sortBy (\a -> a.ackedAt) combined)

862+ set *ackedIds = diff combined evicted

863+ yield evicted

864+ when (count overflow > 0) (do

865+ dropped <- atomic do

866+ ackOverflow <- *ackOverflow

867+ let updated = union ackOverflow overflow

868+ let overflowExcess = count updated - maxAckOverflow

869+ if overflowExcess <= 0

870+ then do set *ackOverflow = updated; yield 0

871+ else do set *ackOverflow = evictOldestAckedIds updated overflowExcess; yield overflowExcess

839− when (dropped > 0)

872+ when (dropped > 0)

840− (println ("ackMessages: ackedIds at capacity (" ++ show maxAckedIds ++ "), " ++ show dropped ++ " ack entries permanently lost for " ++ server))

873+ (println ("ackMessages: overflow at capacity (" ++ show maxAckOverflow ++ "), " ++ show dropped ++ " ack entries permanently lost for " ++ server)))

841 Ok _ -> yield {})

874 Ok _ -> yield {})

842

875

876+retryOverflowAcks = \overflow ->

877+ let servers = collectPeerServers overflow (\a -> a.originServer) in

878+ sendToEach servers (\server ->

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

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

881+ case result of

882+ Err _ -> println ("federation: overflow ack to " ++ server ++ " failed")

883+ Ok _ -> atomic do

884+ ackOverflow <- *ackOverflow

885+ set *ackOverflow = do a <- ackOverflow; where not (a.originServer == server && hasMessage a.messageId a.toKey batch); yield a))

886+

843pruneAndRetry = \t -> do

887pruneAndRetry = \t -> do

844 result <- atomic do

888 result <- atomic do

845 set *lastPrunedAt = t

889 set *lastPrunedAt = t

@@ -847,17 +891,20 @@ pruneAndRetry = \t -> do

847 forwards <- *forwards

891 forwards <- *forwards

848 ackedIds2 <- *ackedIds

892 ackedIds2 <- *ackedIds

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

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

894+ ackOverflow <- *ackOverflow

895+ let pendingOverflow = do a <- ackOverflow; where not (isLocalServer a.originServer); yield a

850 lastRefresh <- *lastPresenceRefresh

896 lastRefresh <- *lastPresenceRefresh

851 let shouldRefresh = federationEnabled && t - lastRefresh >= presenceRefreshInterval

897 let shouldRefresh = federationEnabled && t - lastRefresh >= presenceRefreshInterval

852 presence <- *presence

898 presence <- *presence

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

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

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

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

855 when shouldRefresh (set *lastPresenceRefresh = t)

901 when shouldRefresh (set *lastPresenceRefresh = t)

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

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

857 broadcastOfflineGossip t result.peerServers result.expiredLocal

903 broadcastOfflineGossip t result.peerServers result.expiredLocal

858 when federationEnabled (do

904 when federationEnabled (do

859 retryPendingForwards result.pendingForwards

905 retryPendingForwards result.pendingForwards

860− sendAcksAndClean result.pendingAcks)

906+ sendAcksAndClean result.pendingAcks

907+ retryOverflowAcks result.pendingOverflow)

861 when result.shouldRefresh (do

908 when result.shouldRefresh (do

862 let refreshEvents = do p <- result.localActive; yield {eventType: GossipOnline {}, pubkey: p.pubkey, server: serverHostN, timestamp: p.timestamp, ttl: clampTtl ((p.expiresAt - p.timestamp) / 1000)}

909 let refreshEvents = do p <- result.localActive; yield {eventType: GossipOnline {}, pubkey: p.pubkey, server: serverHostN, timestamp: p.timestamp, ttl: clampTtl ((p.expiresAt - p.timestamp) / 1000)}

863 when (count result.refreshPeers > 0 && count refreshEvents > 0)

910 when (count result.refreshPeers > 0 && count refreshEvents > 0)

@@ -959,31 +1006,51 @@ pollOnce = \token cursor -> do

959 case setup of

1006 case setup of

960 Nothing {} -> yield {authorized: False {}, events: [], cursor: 0, pendingAcks: [], evictedEntries: []}

1007 Nothing {} -> yield {authorized: False {}, events: [], cursor: 0, pendingAcks: [], evictedEntries: []}

961 Just {value: s} -> do

1008 Just {value: s} -> do

962− -- Phase 1: check for existing messages (no blocking)

1009+ -- Acquire poll slot (limit concurrent polls per key)

963− initialMessages <- *messages

1010+ acquired <- atomic do

964− let initial = pollableMessages initialMessages s.pubkey s.clampedCursor t

1011+ activePolls <- *activePolls

965− if count initial > 0

1012+ let pollsForKey = count (do p <- activePolls; where p.pubkey == s.pubkey; yield p)

966− then do

1013+ if pollsForKey >= maxPollsPerKey

967− result <- atomic do pollNewEvents s.pubkey s.clampedCursor t

1014+ then yield (False {})

1015+ else do

1016+ set *activePolls = if count activePolls >= maxActivePolls

1017+ then union (evictOneOldest activePolls (\p -> p.startedAt)) [{pubkey: s.pubkey, startedAt: t}]

1018+ else union activePolls [{pubkey: s.pubkey, startedAt: t}]

1019+ yield (True {})

1020+ if not acquired

968− yield {authorized: True {}, events: result.events, cursor: result.cursor, pendingAcks: s.pendingAcks, evictedEntries: s.evictedEntries}

1021+ then yield {authorized: True {}, events: [], cursor: cursor, pendingAcks: s.pendingAcks, evictedEntries: s.evictedEntries}

969 else do

1022 else do

1023+ pollResult <- do

1024+ -- Phase 1: check for existing messages (no blocking)

1025+ initialMessages <- *messages

1026+ let initial = pollableMessages initialMessages s.pubkey s.clampedCursor t

1027+ if count initial > 0

1028+ then do

1029+ result <- atomic do pollNewEvents s.pubkey s.clampedCursor t

1030+ yield {authorized: True {}, events: result.events, cursor: result.cursor}

1031+ else do

970− -- Phase 2: wait for signal or heartbeat (no *messages dependency)

1032+ -- Phase 2: wait for signal or heartbeat (no *messages dependency)

1033+ atomic do

1034+ wakeup <- *pollWakeup

1035+ signal <- readMessageSignal s.pubkey

1036+ if wakeup > s.startWakeup || signal > s.startSignal

1037+ then yield {}

1038+ else retry

1039+ -- Phase 3: re-authenticate and read messages after waking

1040+ t2 <- now

1041+ result <- atomic do

1042+ sessions2 <- *sessions

1043+ case authenticateSession sessions2 token t2 of

1044+ Unauthorized {} -> yield {authorized: False {}, events: [], cursor: s.clampedCursor}

1045+ Authorized {} -> do

1046+ r <- pollNewEvents s.pubkey s.clampedCursor t2

1047+ yield {authorized: True {}, events: r.events, cursor: r.cursor}

1048+ yield {authorized: result.authorized, events: result.events, cursor: result.cursor}

1049+ -- Release poll slot

971 atomic do

1050 atomic do

972− wakeup <- *pollWakeup

1051+ activePolls <- *activePolls

973− signal <- readMessageSignal s.pubkey

1052+ set *activePolls = removeOneForKey activePolls s.pubkey

974− if wakeup > s.startWakeup || signal > s.startSignal

975− then yield {}

976− else retry

977− -- Phase 3: re-authenticate and read messages after waking

978− t2 <- now

979− result <- atomic do

980− sessions2 <- *sessions

981− case authenticateSession sessions2 token t2 of

982− Unauthorized {} -> yield {authorized: False {}, events: [], cursor: s.clampedCursor}

983− Authorized {} -> do

984− r <- pollNewEvents s.pubkey s.clampedCursor t2

985− yield {authorized: True {}, events: r.events, cursor: r.cursor}

986− yield {authorized: result.authorized, events: result.events, cursor: result.cursor, pendingAcks: s.pendingAcks, evictedEntries: s.evictedEntries}

1053+ yield {authorized: pollResult.authorized, events: pollResult.events, cursor: pollResult.cursor, pendingAcks: s.pendingAcks, evictedEntries: s.evictedEntries}

987

1054

988-- ============================================

1055-- ============================================

989-- Derived Relations

1056-- Derived Relations

@@ -1093,19 +1160,21 @@ serve = \req -> case req of

1093 else if count active >= maxChallenges

1160 else if count active >= maxChallenges

1094 then yield {pr: pr, accepted: False {}, rateLimited: False {}}

1161 then yield {pr: pr, accepted: False {}, rateLimited: False {}}

1095 else do

1162 else do

1096− rateTs <- *challengeRateTs

1163+ challengeRates <- *challengeRates

1097− rateCount <- *challengeRateCount

1164+ let senderEntry = headOf (do r <- challengeRates; where r.pubkey == pubkey; yield r)

1098− let newWindow = t - rateTs >= challengeRateWindowMs

1165+ let isNewWindow = case senderEntry of Nothing {} -> True {}; Just {value: r} -> t - r.windowStart >= challengeRateWindowMs

1099− let currentCount = if newWindow then 0 else rateCount

1166+ let rateCount = case senderEntry of Nothing {} -> 0; Just {value: r} -> if isNewWindow then 0 else r.count

1100− if currentCount >= challengeRateLimit

1167+ if rateCount >= maxChallengeRatePerKey

1101 then yield {pr: pr, accepted: False {}, rateLimited: True {}}

1168 then yield {pr: pr, accepted: False {}, rateLimited: True {}}

1102 else do

1169 else do

1103− set *challengeRateTs = if newWindow then t else rateTs

1170+ let newRates = if isNewWindow

1104− set *challengeRateCount = currentCount + 1

1171+ then union (do r <- challengeRates; where r.pubkey != pubkey; yield r) [{pubkey: pubkey, windowStart: t, count: 1}]

1172+ else do r <- challengeRates; yield (if r.pubkey == pubkey then {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1} else r)

1173+ set *challengeRates = if count newRates > maxChallengeRateEntries then (do r <- newRates; where t - r.windowStart < challengeRateWindowMs; yield r) else newRates

1105 set *challenges = union active [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry}]

1174 set *challenges = union active [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry}]

1106 yield {pr: pr, accepted: True {}, rateLimited: False {}}

1175 yield {pr: pr, accepted: True {}, rateLimited: False {}}

1107 when result.rateLimited

1176 when result.rateLimited

1108− (println ("challenge: rate limited (" ++ show challengeRateLimit ++ "/s), rejecting for " ++ pubkey))

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

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

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

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

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

1111 forkBroadcastOffline t result.pr

1180 forkBroadcastOffline t result.pr

@@ -1211,7 +1280,7 @@ serve = \req -> case req of

1211 fwdResult <- fetch (federationUrl server) (ForwardMessage {

1280 fwdResult <- fetch (federationUrl server) (ForwardMessage {

1212 toKey: to, encryptedBlob: encryptedBlob,

1281 toKey: to, encryptedBlob: encryptedBlob,

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

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

1214− authorization: federationAuth})

1283+ fromServer: serverHostN, authorization: federationAuth})

1215 case fwdResult of

1284 case fwdResult of

1216 Err _ -> do

1285 Err _ -> do

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

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

@@ -1269,9 +1338,9 @@ serve = \req -> case req of

1269 (forwardQueuedMessages newlyOnlineRemote t))

1338 (forwardQueuedMessages newlyOnlineRemote t))

1270 yield (respond {ok: True {}}))

1339 yield (respond {ok: True {}}))

1271

1340

1272− ForwardMessage {toKey, encryptedBlob, timestamp, originServer, messageId, authorization, respond} -> do

1341+ ForwardMessage {toKey, encryptedBlob, timestamp, originServer, messageId, fromServer, authorization, respond} -> do

1273 t <- now

1342 t <- now

1274− withFederationAuth authorization t respond (not (isLocalServer originServer)) (do

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

1275 let clampedTs = clampTimestamp timestamp t

1344 let clampedTs = clampTimestamp timestamp t

1276 result <- atomic do

1345 result <- atomic do

1277 pr <- maybePrune t

1346 pr <- maybePrune t