git.iko.soy/skrepka
on 8815b21 help sign in
skrepka/server.knot
1175 lines 52 kB history raw
1-- Skrepka Relay Server
2-- Single-server implementation of the Skrepka protocol (v0.1)
3-- Binary data (keys, signatures, blobs) uses hex encoding (bytesToHex/bytesFromHex)
5-- ============================================
6-- Configuration
7-- ============================================
9serverHost = "localhost:8080"
10federationSecret = "" -- must be set to enable federation
11maintenanceSecret = "" -- set to enable /maintenance/prune independently of federation
12federationScheme = "https://" -- scheme for federation HTTP calls
13messageMaxAge = 2592000000 -- 30 days in milliseconds
14maxPresenceTtl = 3600 -- max presence TTL from gossip in seconds
15maxPubkeyLen = 256
16maxBlobLen = 65536
17maxSignatureLen = 256
18maxMessageIds = 100
19maxGossipEvents = 100
20maxPollEvents = 50
21maxPresenceEntries = 10000
22maxMessagesPerRecipient = 500
23clockSkewTolerance = 5000 -- 5 seconds tolerance for clock skew between federated servers
24maxForwardsEntries = 10000
25maxServerNameLen = 128
26maxMessageIdLen = 128
27maxTokenLen = 256
28maxChallenges = 10000
29maxSessions = 10000
30maxAckedIds = 10000
31maxChallengesPerKey = 5
32maxSessionsPerKey = 10
33challengeExpiry = 60000 -- 1 minute in milliseconds
34sessionExpiry = 3600000 -- 1 hour in milliseconds
35onlineGossipTtl = 3600 -- 1 hour in seconds
37-- ============================================
38-- Types
39-- ============================================
41type PollEvent = {id: Text, eventType: Text, payload: Text}
43type GossipEvent = {eventType: Text, pubkey: Text, server: Text, timestamp: Int, ttl: Int}
45-- ============================================
46-- Persisted State
47-- ============================================
49*challenges : [{pubkey: Text, challenge: Text, expiresAt: Int}]
51*sessions : [{token: Text, pubkey: Text, expiresAt: Int}]
53*messages : [{messageId: Text, toKey: Text, fromKey: Text, encryptedBlob: Text, timestamp: Int, originServer: Text, receivedAt: Int}]
55*presence : [{pubkey: Text, server: Text, expiresAt: Int, timestamp: Int}]
57*seqCounter : Int
59*forwards : [{messageId: Text, toServer: Text, createdAt: Int, toKey: Text, fromKey: Text, encryptedBlob: Text, timestamp: Int}]
61*ackedIds : [{messageId: Text, toKey: Text, ackedAt: Int, originServer: Text}]
63*lastPrunedAt : Int
65-- ============================================
66-- Routes
67-- ============================================
69route AuthApi where
70 /auth
71 POST {pubkey: Text} /challenge
72 -> {challenge: Text}
73 = ReqChallenge
74 POST {pubkey: Text, challenge: Text, signature: Text} /verify
75 -> {token: Text, expiresAt: Int}
76 = VerifyAuth
78route ClientApi where
79 GET {cursor: Int, timeout: Int} /poll headers {authorization: Text}
80 -> {events: [PollEvent], cursor: Int, authorized: Bool}
81 = Poll
82 /messages
83 POST {to: Text, encryptedBlob: Text, timestamp: Int} / headers {authorization: Text}
84 -> {status: Text, messageId: Text}
85 = SendMessage
86 POST {messageIds: [Text]} /ack headers {authorization: Text}
87 -> {ok: Bool}
88 = AckMessages
89 GET /lookup/{pubkey: Text} headers {authorization: Text}
90 -> {online: Bool, server: Text, authorized: Bool}
91 = LookupPresence
93route FederationApi where
94 /federation
95 POST {events: [GossipEvent], hopsRemaining: Int, fromServer: Text} /gossip headers {authorization: Text}
96 -> {ok: Bool}
97 = RecvGossip
98 POST {toKey: Text, fromKey: Text, encryptedBlob: Text, timestamp: Int, originServer: Text, messageId: Text} /forward headers {authorization: Text}
99 -> {ok: Bool}
100 = ForwardMessage
101 POST {messageIds: [Text], originServer: Text, fromServer: Text} /ack headers {authorization: Text}
102 -> {ok: Bool}
103 = FederationAck
105route MaintenanceApi where
106 POST {} /maintenance/prune headers {authorization: Text}
107 -> {ok: Bool}
108 = Prune
110route Api = AuthApi | ClientApi | FederationApi | MaintenanceApi
112-- ============================================
113-- Helpers
114-- ============================================
116hexDigit = \n ->
117 let n = if n < 0 then 0 else if n > 15 then 15 else n
118 if n == 0 then "0" else if n == 1 then "1" else if n == 2 then "2" else if n == 3 then "3"
119 else if n == 4 then "4" else if n == 5 then "5" else if n == 6 then "6" else if n == 7 then "7"
120 else if n == 8 then "8" else if n == 9 then "9" else if n == 10 then "a" else if n == 11 then "b"
121 else if n == 12 then "c" else if n == 13 then "d" else if n == 14 then "e" else "f"
123byteToHex = \n -> hexDigit (n / 16) ++ hexDigit (n - (n / 16) * 16)
125intTo3Hex = \n ->
126 let b0 = n / 65536
127 let r0 = n - b0 * 65536
128 let b1 = r0 / 256
129 let b2 = r0 - b1 * 256
130 byteToHex b0 ++ byteToHex b1 ++ byteToHex b2
132genToken = do
133 a <- randomInt 16777216
134 b <- randomInt 16777216
135 c <- randomInt 16777216
136 d <- randomInt 16777216
137 e <- randomInt 16777216
138 f <- randomInt 16777216
139 g <- randomInt 16777216
140 h <- randomInt 16777216
141 yield (intTo3Hex a ++ intTo3Hex b ++ intTo3Hex c ++ intTo3Hex d ++ intTo3Hex e ++ intTo3Hex f ++ intTo3Hex g ++ intTo3Hex h)
143tooLong = \s maxLen -> not (drop maxLen s == "")
145isHexChar = \c ->
146 c == "0" || c == "1" || c == "2" || c == "3" || c == "4" || c == "5" || c == "6" || c == "7" || c == "8" || c == "9" ||
147 c == "a" || c == "b" || c == "c" || c == "d" || c == "e" || c == "f" ||
148 c == "A" || c == "B" || c == "C" || c == "D" || c == "E" || c == "F"
150isValidHexPairs = \s ->
151 if s == "" then True {}
152 else if drop 1 s == "" then False {}
153 else if isHexChar (take 1 s) && isHexChar (take 1 (drop 1 s)) then isValidHexPairs (drop 2 s)
154 else False {}
156isValidHex = \s -> not (s == "") && isValidHexPairs s
158normalizeHex = \s -> bytesToHex (bytesFromHex s)
160containsBadUrlChar = \s ->
161 if s == "" then False {}
162 else if take 1 s == "/" || take 1 s == "?" || take 1 s == "#" || take 1 s == " " || take 1 s == "@" || take 1 s == "%" || take 1 s == "\\"
163 || take 1 (bytesToHex (textToBytes (take 1 s))) == "0" || take 1 (bytesToHex (textToBytes (take 1 s))) == "1" || bytesToHex (textToBytes (take 1 s)) == "7f"
164 || not (drop 2 (bytesToHex (textToBytes (take 1 s))) == "")
165 then True {}
166 else containsBadUrlChar (drop 1 s)
168isDigitChar = \c -> c == "0" || c == "1" || c == "2" || c == "3" || c == "4" || c == "5" || c == "6" || c == "7" || c == "8" || c == "9"
170isCgnatRange = \s ->
171 if not (take 4 s == "100.") then False {}
172 else if (take 1 (drop 4 s) == "7" || take 1 (drop 4 s) == "8" || take 1 (drop 4 s) == "9")
173 && isDigitChar (take 1 (drop 5 s)) && take 1 (drop 6 s) == "."
174 && isAllDigitsOrDotsUntilColonOrEnd (drop 7 s)
175 then True {}
176 else if take 1 (drop 4 s) == "6"
177 && (take 1 (drop 5 s) == "4" || take 1 (drop 5 s) == "5" || take 1 (drop 5 s) == "6" || take 1 (drop 5 s) == "7" || take 1 (drop 5 s) == "8" || take 1 (drop 5 s) == "9")
178 && take 1 (drop 6 s) == "."
179 && isAllDigitsOrDotsUntilColonOrEnd (drop 7 s)
180 then True {}
181 else if take 1 (drop 4 s) == "1" && (take 1 (drop 5 s) == "0" || take 1 (drop 5 s) == "1")
182 && isDigitChar (take 1 (drop 6 s)) && take 1 (drop 7 s) == "."
183 && isAllDigitsOrDotsUntilColonOrEnd (drop 8 s)
184 then True {}
185 else if take 1 (drop 4 s) == "1" && take 1 (drop 5 s) == "2"
186 && (take 1 (drop 6 s) == "0" || take 1 (drop 6 s) == "1" || take 1 (drop 6 s) == "2" || take 1 (drop 6 s) == "3"
187 || take 1 (drop 6 s) == "4" || take 1 (drop 6 s) == "5" || take 1 (drop 6 s) == "6" || take 1 (drop 6 s) == "7")
188 && take 1 (drop 7 s) == "."
189 && isAllDigitsOrDotsUntilColonOrEnd (drop 8 s)
190 then True {}
191 else False {}
193containsColon = \s ->
194 if s == "" then False {}
195 else if take 1 s == ":" then True {}
196 else containsColon (drop 1 s)
198containsMultipleColons = \s ->
199 if s == "" then False {}
200 else if take 1 s == ":" then containsColon (drop 1 s)
201 else containsMultipleColons (drop 1 s)
203hasOctalOctet = \s ->
204 if take 1 s == "0" && isDigitChar (take 1 (drop 1 s)) && isAllDigitsOrDotsUntilColonOrEnd (drop 2 s) then True {}
205 else hasOctalOctetAfterDot s
207hasOctalOctetAfterDot = \s ->
208 if s == "" then False {}
209 else if take 1 s == "." && take 1 (drop 1 s) == "0" && isDigitChar (take 1 (drop 2 s)) && isAllDigitsOrDotsUntilColonOrEnd (drop 3 s) then True {}
210 else hasOctalOctetAfterDot (drop 1 s)
212hasHexPrefix = \s ->
213 if take 2 s == "0x" || take 2 s == "0X" then True {}
214 else hasHexPrefixAfterDot s
216hasHexPrefixAfterDot = \s ->
217 if s == "" then False {}
218 else if take 1 s == "." && (take 2 (drop 1 s) == "0x" || take 2 (drop 1 s) == "0X") then True {}
219 else hasHexPrefixAfterDot (drop 1 s)
221isAllDigitsOrDotsUntilColonOrEnd = \s ->
222 if s == "" then True {}
223 else if take 1 s == ":" then True {}
224 else if isDigitChar (take 1 s) || take 1 s == "." then isAllDigitsOrDotsUntilColonOrEnd (drop 1 s)
225 else False {}
227isAllDigitsUntilColonOrEnd = \s ->
228 if s == "" then True {}
229 else if take 1 s == ":" then True {}
230 else if isDigitChar (take 1 s) then isAllDigitsUntilColonOrEnd (drop 1 s)
231 else False {}
233toLowerChar = \c ->
234 if c == "A" then "a" else if c == "B" then "b" else if c == "C" then "c"
235 else if c == "D" then "d" else if c == "E" then "e" else if c == "F" then "f"
236 else if c == "G" then "g" else if c == "H" then "h" else if c == "I" then "i"
237 else if c == "J" then "j" else if c == "K" then "k" else if c == "L" then "l"
238 else if c == "M" then "m" else if c == "N" then "n" else if c == "O" then "o"
239 else if c == "P" then "p" else if c == "Q" then "q" else if c == "R" then "r"
240 else if c == "S" then "s" else if c == "T" then "t" else if c == "U" then "u"
241 else if c == "V" then "v" else if c == "W" then "w" else if c == "X" then "x"
242 else if c == "Y" then "y" else if c == "Z" then "z"
243 else c
245toLower = \s ->
246 if s == "" then ""
247 else toLowerChar (take 1 s) ++ toLower (drop 1 s)
249isTrailingDots = \s ->
250 if s == "" then True {}
251 else if take 1 s == ":" then True {}
252 else if take 1 s == "." then isTrailingDots (drop 1 s)
253 else False {}
255stripTrailingDot = \s ->
256 if s == "" then ""
257 else if take 1 s == "." && isTrailingDots (drop 1 s) then stripTrailingDot (drop 1 s)
258 else take 1 s ++ stripTrailingDot (drop 1 s)
260normalizeServerName = \s -> toLower (stripTrailingDot s)
262serverHostN = normalizeServerName serverHost
264containsLocalhostSuffix = \s ->
265 if s == "" then False {}
266 else if take 10 s == ".localhost" && (drop 10 s == "" || take 1 (drop 10 s) == ":") then True {}
267 else containsLocalhostSuffix (drop 1 s)
269isBadServerName = \s ->
270 let sn = toLower (stripTrailingDot s)
271 sn == ""
272 || take 1 sn == "."
273 || take 1 sn == ":"
274 || containsBadUrlChar sn
275 || sn == "localhost" || take 10 sn == "localhost:"
276 || sn == "localhost.localdomain" || take 22 sn == "localhost.localdomain:"
277 || containsLocalhostSuffix sn
278 || take 4 sn == "127." || take 3 sn == "10."
279 || take 8 sn == "192.168." || take 7 sn == "169.254."
280 || take 2 sn == "0." || take 1 sn == "["
281 || take 7 sn == "172.16." || take 7 sn == "172.17." || take 7 sn == "172.18." || take 7 sn == "172.19."
282 || take 7 sn == "172.20." || take 7 sn == "172.21." || take 7 sn == "172.22." || take 7 sn == "172.23."
283 || take 7 sn == "172.24." || take 7 sn == "172.25." || take 7 sn == "172.26." || take 7 sn == "172.27."
284 || take 7 sn == "172.28." || take 7 sn == "172.29." || take 7 sn == "172.30." || take 7 sn == "172.31."
285 || take 7 sn == "198.18." || take 7 sn == "198.19."
286 || take 8 sn == "192.0.0."
287 || take 8 sn == "192.0.2."
288 || take 10 sn == "192.88.99."
289 || take 12 sn == "198.51.100."
290 || take 11 sn == "203.0.113."
291 || take 2 sn == "::"
292 || isCgnatRange sn
293 || containsMultipleColons sn
294 || hasOctalOctet sn
295 || hasHexPrefix sn
296 || (isDigitChar (take 1 sn) && isAllDigitsUntilColonOrEnd sn)
297 || take 4 sn == "224." || take 4 sn == "225." || take 4 sn == "226." || take 4 sn == "227."
298 || take 4 sn == "228." || take 4 sn == "229." || take 4 sn == "230." || take 4 sn == "231."
299 || take 4 sn == "232." || take 4 sn == "233." || take 4 sn == "234." || take 4 sn == "235."
300 || take 4 sn == "236." || take 4 sn == "237." || take 4 sn == "238." || take 4 sn == "239."
301 || take 4 sn == "240." || take 4 sn == "241." || take 4 sn == "242." || take 4 sn == "243."
302 || take 4 sn == "244." || take 4 sn == "245." || take 4 sn == "246." || take 4 sn == "247."
303 || take 4 sn == "248." || take 4 sn == "249." || take 4 sn == "250." || take 4 sn == "251."
304 || take 4 sn == "252." || take 4 sn == "253." || take 4 sn == "254." || take 4 sn == "255."
306lookupSession = \sessions token t ->
307 single (do
308 s <- sessions
309 where s.token == token
310 where s.expiresAt > t
311 yield s)
313extractToken = \authHeader ->
314 if (take 1 authHeader == "B" || take 1 authHeader == "b")
315 && (take 1 (drop 1 authHeader) == "e" || take 1 (drop 1 authHeader) == "E")
316 && (take 1 (drop 2 authHeader) == "a" || take 1 (drop 2 authHeader) == "A")
317 && (take 1 (drop 3 authHeader) == "r" || take 1 (drop 3 authHeader) == "R")
318 && (take 1 (drop 4 authHeader) == "e" || take 1 (drop 4 authHeader) == "E")
319 && (take 1 (drop 5 authHeader) == "r" || take 1 (drop 5 authHeader) == "R")
320 && take 1 (drop 6 authHeader) == " "
321 && not (tooLong (drop 7 authHeader) maxTokenLen)
322 then drop 7 authHeader
323 else ""
325verifyFederationAuth = \authorization ->
326 not (federationSecret == "") && safeCompare (extractToken authorization) federationSecret
328verifyMaintenanceAuth = \authorization ->
329 let token = extractToken authorization
330 let fedOk = not (federationSecret == "") && safeCompare token federationSecret
331 let maintOk = not (maintenanceSecret == "") && safeCompare token maintenanceSecret
332 fedOk || maintOk
334clampTtl = \ttl -> if ttl < 1 then 1 else if ttl > maxPresenceTtl then maxPresenceTtl else ttl
336evictOldestAckedIds = \items n ->
337 if n <= 0 then items
338 else
339 let oldest = fold (\acc a -> if acc.ackedAt == -1 || a.ackedAt < acc.ackedAt then a else acc) {messageId: "", toKey: "", ackedAt: -1, originServer: ""} items
340 in if oldest.ackedAt == -1 then items
341 else evictOldestAckedIds (fold (\acc a ->
342 if not acc.removed && a == oldest
343 then {removed: True {}, items: acc.items}
344 else {removed: acc.removed, items: union acc.items [a]}) {removed: False {}, items: []} items).items (n - 1)
346safeCompareLoop = \s1 s2 result ->
347 if s2 == ""
348 then result + (if s1 == "" then 0 else 1)
349 else safeCompareLoop (drop 1 s1) (drop 1 s2) (result + (if take 1 s1 == take 1 s2 then 0 else 1))
351safeCompare = \a b ->
352 safeCompareLoop a b 0 + (if a == "" then 1 else 0) + (if b == "" then 1 else 0) == 0
354sendToEach = \servers action ->
355 if count servers == 0
356 then yield {}
357 else do
358 let server = fold (\acc s -> if acc == "" then s else acc) "" servers
359 let rest = do s <- servers; where not (s == server); yield s
360 if server == ""
361 then yield {}
362 else do
363 catch (action server) (\err ->
364 println ("federation: error reaching " ++ server ++ ": " ++ show err))
365 sendToEach rest action
367pruneExpired = \t -> do
368 challenges <- *challenges
369 set *challenges = do
370 c <- challenges
371 where c.expiresAt > t
372 yield c
373 sessions <- *sessions
374 set *sessions = do
375 s <- sessions
376 where s.expiresAt > t
377 yield s
378 presenceBefore <- *presence
379 let peerServersBefore = fold (\acc p ->
380 if p.server == serverHostN then acc
381 else if count (do a <- acc; where a == p.server; yield a) > 0 then acc
382 else union acc [p.server]) [] presenceBefore
383 let presenceAfter = do
384 p <- presenceBefore
385 where p.expiresAt > t
386 yield p
387 set *presence = presenceAfter
388 messages <- *messages
389 set *messages = do
390 m <- messages
391 where m.receivedAt >= t - messageMaxAge
392 yield m
393 forwards <- *forwards
394 set *forwards = do
395 f <- forwards
396 where f.createdAt >= t - messageMaxAge
397 yield f
398 ackedIds <- *ackedIds
399 set *ackedIds = do
400 a <- ackedIds
401 where a.ackedAt >= t - messageMaxAge
402 yield a
403 yield {expiredLocal: (do
404 p <- presenceBefore
405 where p.expiresAt <= t
406 where p.server == serverHostN
407 where count (do
408 q <- presenceAfter
409 where q.pubkey == p.pubkey
410 where q.server == serverHostN
411 yield q) == 0
412 yield p), peerServers: peerServersBefore}
414maybePrune = \t -> do
415 lastPruned <- *lastPrunedAt
416 if t - lastPruned < 1000
417 then yield {expiredLocal: [], peerServers: []}
418 else do
419 set *lastPrunedAt = t
420 pruneExpired t
422sendBatchedGossip = \peerServers allEvents ->
423 if count allEvents == 0
424 then yield {}
425 else do
426 let batch = fold (\acc e ->
427 if count acc >= maxGossipEvents then acc
428 else union acc [e]) [] allEvents
429 let rest = (fold (\acc e ->
430 if acc.taken < maxGossipEvents
431 then {taken: acc.taken + 1, items: acc.items}
432 else {taken: acc.taken, items: union acc.items [e]}) {taken: 0, items: []} allEvents).items
433 sendToEach peerServers (\server ->
434 fetch (federationScheme ++ server) (RecvGossip {
435 authorization: "Bearer " ++ federationSecret,
436 events: batch, hopsRemaining: 2, fromServer: serverHostN}))
437 sendBatchedGossip peerServers rest
439broadcastOfflineGossip = \t peerServers expiredLocal -> do
440 let expiredEntries = fold (\acc p ->
441 if count (do a <- acc; where a.pubkey == p.pubkey; yield a) > 0 then acc
442 else union acc [{pubkey: p.pubkey, ttl: clampTtl ((p.expiresAt - p.timestamp) / 1000)}]) [] expiredLocal
443 if not (federationSecret == "") && count expiredEntries > 0 && count peerServers > 0
444 then do
445 let gossipEvents = do
446 e <- expiredEntries
447 yield {eventType: "offline", pubkey: e.pubkey, server: serverHostN, timestamp: t, ttl: e.ttl}
448 sendBatchedGossip peerServers gossipEvents
449 else yield {}
451retryPendingForwards = \pending ->
452 if count pending == 0
453 then yield {}
454 else do
455 let f = fold (\acc x -> if acc.messageId == "" then x else acc) {messageId: "", toServer: "", toKey: "", fromKey: "", encryptedBlob: "", timestamp: 0} pending
456 let rest = do x <- pending; where not (x.messageId == f.messageId && x.toServer == f.toServer); yield x
457 if f.messageId == ""
458 then yield {}
459 else do
460 catch (do
461 fetch (federationScheme ++ f.toServer) (ForwardMessage {
462 toKey: f.toKey, fromKey: f.fromKey, encryptedBlob: f.encryptedBlob,
463 timestamp: f.timestamp, originServer: serverHostN, messageId: f.messageId,
464 authorization: "Bearer " ++ federationSecret})
465 yield {}) (\err ->
466 println ("prune: retry forward to " ++ f.toServer ++ " failed: " ++ show err))
467 retryPendingForwards rest
469-- ============================================
470-- Request Handler
471-- ============================================
473serve = \req -> case req of
475 ReqChallenge {pubkey, respond} -> do
476 let valid = not (pubkey == "") && not (tooLong pubkey maxPubkeyLen) && isValidHex pubkey
477 let pubkey = if isValidHex pubkey then normalizeHex pubkey else pubkey
478 challenge <- genToken
479 t <- now
480 if valid
481 then do
482 pruneResult <- atomic do
483 pr <- maybePrune t
484 challenges <- *challenges
485 let existingCount = count (do c <- challenges; where c.pubkey == pubkey; yield c)
486 let oldestChallengeExpiry = fold (\acc c2 -> if c2.pubkey == pubkey && (acc == -1 || c2.expiresAt < acc) then c2.expiresAt else acc) -1 challenges
487 let cappedChallenges = if existingCount >= maxChallengesPerKey && oldestChallengeExpiry >= 0
488 then fold (\acc c ->
489 if not acc.removed && c.pubkey == pubkey && c.expiresAt == oldestChallengeExpiry
490 then {removed: True {}, items: acc.items}
491 else {removed: acc.removed, items: union acc.items [c]}) {removed: False {}, items: []} challenges
492 else {removed: False {}, items: challenges}
493 let globalOldestExpiry = if count cappedChallenges.items >= maxChallenges
494 then fold (\acc c -> if acc == -1 || c.expiresAt < acc then c.expiresAt else acc) -1 cappedChallenges.items
495 else -1
496 let globalCapped = if globalOldestExpiry >= 0
497 then (fold (\acc c ->
498 if not acc.removed && c.expiresAt == globalOldestExpiry
499 then {removed: True {}, items: acc.items}
500 else {removed: acc.removed, items: union acc.items [c]}) {removed: False {}, items: []} cappedChallenges.items).items
501 else cappedChallenges.items
502 set *challenges = union
503 globalCapped
504 [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry}]
505 yield pr
506 respond {challenge: challenge}
507 broadcastOfflineGossip t pruneResult.peerServers pruneResult.expiredLocal
508 else respond {challenge: ""}
510 VerifyAuth {pubkey, challenge, signature, respond} -> do
511 let pubkey = if isValidHex pubkey then normalizeHex pubkey else pubkey
512 t <- now
513 token <- genToken
514 let expiry = t + sessionExpiry
515 result <- atomic do
516 pr <- maybePrune t
517 challenges <- *challenges
518 sessions <- *sessions
519 presence <- *presence
520 let matched = single (do
521 c <- challenges
522 where c.pubkey == pubkey
523 where c.challenge == challenge
524 where c.expiresAt > t
525 yield c)
526 let sigValid = case matched of
527 Nothing {} -> False {}
528 _ -> if tooLong signature maxSignatureLen
529 then False {}
530 else if not (isValidHex pubkey) || not (isValidHex signature)
531 then False {}
532 else verify (bytesFromHex pubkey) (textToBytes challenge) (bytesFromHex signature)
533 set *challenges = case matched of
534 Just {value: _} -> do
535 c <- challenges
536 where not (c.pubkey == pubkey && c.challenge == challenge)
537 yield c
538 Nothing {} -> challenges
539 let existingForKey = do
540 s <- sessions
541 where s.pubkey == pubkey
542 where s.expiresAt > t
543 yield s
544 let oldestExpiry = fold (\acc s -> if acc == -1 || s.expiresAt < acc then s.expiresAt else acc) -1 existingForKey
545 let cappedFold = fold (\acc s ->
546 if not acc.removed && s.pubkey == pubkey && s.expiresAt == oldestExpiry
547 then {removed: True {}, sessions: acc.sessions}
548 else {removed: acc.removed, sessions: union acc.sessions [s]}) {removed: False {}, sessions: []} sessions
549 let sessionsCapped = if count existingForKey >= maxSessionsPerKey && oldestExpiry >= 0
550 then cappedFold.sessions
551 else sessions
552 let globalOldestSessionExpiry = if count sessionsCapped >= maxSessions
553 then fold (\acc s -> if acc == -1 || s.expiresAt < acc then s.expiresAt else acc) -1 sessionsCapped
554 else -1
555 let globalSessionsCapped = if globalOldestSessionExpiry >= 0
556 then (fold (\acc s ->
557 if not acc.removed && s.expiresAt == globalOldestSessionExpiry
558 then {removed: True {}, items: acc.items}
559 else {removed: acc.removed, items: union acc.items [s]}) {removed: False {}, items: []} sessionsCapped).items
560 else sessionsCapped
561 set *sessions = if sigValid
562 then union globalSessionsCapped [{token: token, pubkey: pubkey, expiresAt: expiry}]
563 else sessions
564 let presenceWithout = do
565 p <- presence
566 where not (p.pubkey == pubkey && p.server == serverHostN)
567 yield p
568 let oldestPresenceExp = fold (\acc p -> if acc == -1 || p.expiresAt < acc then p.expiresAt else acc) -1 presenceWithout
569 let presenceEvicted = if count presenceWithout >= maxPresenceEntries && oldestPresenceExp >= 0
570 then (fold (\acc p ->
571 if not acc.removed && p.expiresAt == oldestPresenceExp
572 then {removed: True {}, items: acc.items}
573 else {removed: acc.removed, items: union acc.items [p]}) {removed: False {}, items: []} presenceWithout).items
574 else presenceWithout
575 let updatedPresence = if sigValid
576 then union presenceEvicted [{pubkey: pubkey, server: serverHostN, expiresAt: expiry, timestamp: t}]
577 else presence
578 set *presence = updatedPresence
579 let peerServers = if sigValid && not (federationSecret == "")
580 then fold (\acc p ->
581 if p.server == serverHostN then acc
582 else if count (do a <- acc; where a == p.server; yield a) > 0 then acc
583 else union acc [p.server]) [] updatedPresence
584 else []
585 yield {sigValid: sigValid, peerServers: peerServers, expiredLocal: pr.expiredLocal, offlinePeers: pr.peerServers}
586 if result.sigValid
587 then respond {token: token, expiresAt: expiry}
588 else respond {token: "", expiresAt: 0}
589 if result.sigValid && not (federationSecret == "")
590 then sendToEach result.peerServers (\server ->
591 fetch (federationScheme ++ server) (RecvGossip {
592 authorization: "Bearer " ++ federationSecret,
593 events: [{eventType: "online", pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl}], hopsRemaining: 2, fromServer: serverHostN}))
594 else yield {}
595 broadcastOfflineGossip t result.offlinePeers (do
596 p <- result.expiredLocal
597 where not (p.pubkey == pubkey)
598 yield p)
600 Poll {cursor, timeout, authorization, respond} -> do
601 t <- now
602 let clampedTimeout = if timeout < 0 then 0 else if timeout > 60000 then 60000 else timeout
603 let deadline = t + clampedTimeout
604 let token = extractToken authorization
605 result <- atomic do
606 tNow <- now
607 pr <- maybePrune tNow
608 sessions <- *sessions
609 messages <- *messages
610 seqCounter <- *seqCounter
611 let session = lookupSession sessions token tNow
612 let clampedCursor = if cursor < 0 then 0 else if cursor > seqCounter then seqCounter else cursor
613 let matchingMsgs = case session of
614 Nothing {} -> []
615 Just {value: s} -> do
616 m <- messages
617 where m.toKey == s.pubkey
618 where m.receivedAt >= tNow - messageMaxAge
619 where m.receivedAt > clampedCursor
620 yield m
621 let limitedMsgs = fold (\acc m ->
622 if count acc >= maxPollEvents then acc
623 else union acc [m]) [] matchingMsgs
624 let authorized = case session of
625 Nothing {} -> False {}
626 _ -> True {}
627 if authorized && count limitedMsgs == 0 && tNow < deadline
628 then retry
629 else do
630 let evts = do
631 m <- limitedMsgs
632 yield {id: m.messageId, eventType: "message",
633 payload: toJson {toKey: m.toKey, fromKey: m.fromKey,
634 encryptedBlob: m.encryptedBlob,
635 timestamp: m.timestamp}}
636 let lastReceivedAt = fold (\acc m -> if m.receivedAt > acc then m.receivedAt else acc) clampedCursor limitedMsgs
637 let newCursor = case session of
638 Nothing {} -> -1
639 _ -> if count limitedMsgs >= maxPollEvents
640 then lastReceivedAt
641 else if seqCounter > clampedCursor then seqCounter else clampedCursor
642 yield {events: evts, cursor: newCursor, authorized: authorized, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers}
643 respond {events: result.events, cursor: result.cursor, authorized: result.authorized}
644 tAfterPoll <- now
645 broadcastOfflineGossip tAfterPoll result.peerServers result.expiredLocal
647 SendMessage {to, encryptedBlob, timestamp, authorization, respond} -> do
648 let validInput = not (to == "") && not (encryptedBlob == "") && not (tooLong to maxPubkeyLen) && not (tooLong encryptedBlob maxBlobLen) && isValidHex to && isValidHex encryptedBlob
649 let to = if isValidHex to then normalizeHex to else to
650 let encryptedBlob = if isValidHex encryptedBlob then normalizeHex encryptedBlob else encryptedBlob
651 t <- now
652 msgId <- genToken
653 let clampedTs = if timestamp > t then t else if timestamp < t - messageMaxAge then t - messageMaxAge else timestamp
654 result <- atomic do
655 pr <- maybePrune t
656 sessions <- *sessions
657 messages <- *messages
658 presence <- *presence
659 let session = lookupSession sessions (extractToken authorization) t
660 seqCounter <- *seqCounter
661 let seqTs = if t > seqCounter then t else seqCounter + 1
662 let recipientMsgCount = count (do
663 m <- messages
664 where m.toKey == to
665 where m.receivedAt >= t - messageMaxAge
666 yield m)
667 let shouldInsert = case session of
668 Nothing {} -> False {}
669 Just {value: _} -> validInput && recipientMsgCount < maxMessagesPerRecipient
670 set *seqCounter = if shouldInsert then seqTs else seqCounter
671 set *messages = case session of
672 Nothing {} -> messages
673 Just {value: s} -> if validInput && recipientMsgCount < maxMessagesPerRecipient
674 then union messages
675 [{messageId: msgId, toKey: to, fromKey: s.pubkey,
676 encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}]
677 else messages
678 let localOnline = case session of
679 Nothing {} -> False {}
680 Just {value: _} -> count (do
681 p <- presence
682 where p.pubkey == to
683 where p.server == serverHostN
684 where p.expiresAt > t
685 yield p) > 0
686 let remoteServers = case session of
687 Nothing {} -> []
688 Just {value: _} -> do
689 p <- presence
690 where p.pubkey == to
691 where not (p.server == serverHostN)
692 where p.expiresAt > t
693 yield p.server
694 let fromKey = case session of
695 Nothing {} -> ""
696 Just {value: s} -> s.pubkey
697 let status = case session of
698 Nothing {} -> "unauthorized"
699 _ -> if not validInput then "invalid"
700 else if localOnline then "delivered"
701 else if count remoteServers > 0 then "federated"
702 else "queued"
703 forwards <- *forwards
704 let forwardsAtCapacity = count forwards >= maxForwardsEntries
705 set *forwards = if shouldInsert && not forwardsAtCapacity && count remoteServers > 0
706 then union forwards (do s <- remoteServers; yield {messageId: msgId, toServer: s, createdAt: t, toKey: to, fromKey: fromKey, encryptedBlob: encryptedBlob, timestamp: clampedTs})
707 else forwards
708 yield {status: status, remoteServers: remoteServers, fromKey: fromKey, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, forwardsAtCapacity: forwardsAtCapacity}
709 let finalStatus = if result.status == "federated" && (federationSecret == "" || result.forwardsAtCapacity) then "queued" else result.status
710 respond {status: finalStatus, messageId: (if result.status == "unauthorized" then "" else if result.status == "invalid" then "" else msgId)}
711 if count result.remoteServers > 0 && not (result.status == "unauthorized") && not (result.status == "invalid") && not (federationSecret == "")
712 then sendToEach result.remoteServers (\server ->
713 fetch (federationScheme ++ server) (ForwardMessage {
714 toKey: to, fromKey: result.fromKey, encryptedBlob: encryptedBlob,
715 timestamp: clampedTs, originServer: serverHostN, messageId: msgId,
716 authorization: "Bearer " ++ federationSecret}))
717 else yield {}
718 if result.forwardsAtCapacity && count result.remoteServers > 0 && not (result.status == "unauthorized") && not (result.status == "invalid")
719 then println ("sendMessage: forwards at capacity (" ++ show maxForwardsEntries ++ "), skipping forward tracking")
720 else yield {}
721 broadcastOfflineGossip t result.peerServers result.expiredLocal
723 AckMessages {messageIds, authorization, respond} -> do
724 t <- now
725 let dedupedIds = fold (\acc mid ->
726 if mid == "" then acc
727 else if count acc >= maxMessageIds then acc
728 else if tooLong mid maxMessageIdLen then acc
729 else if not (isValidHex mid) then acc
730 else if count (do a <- acc; where a == normalizeHex mid; yield a) > 0 then acc
731 else union acc [normalizeHex mid]) [] messageIds
732 result <- atomic do
733 pr <- maybePrune t
734 sessions <- *sessions
735 messages <- *messages
736 let session = lookupSession sessions (extractToken authorization) t
737 let ackedMessages = case session of
738 Nothing {} -> []
739 Just {value: s} -> do
740 m <- messages
741 mid <- dedupedIds
742 where m.messageId == mid
743 where m.toKey == s.pubkey
744 yield m
745 set *messages = case session of
746 Nothing {} -> messages
747 Just {value: _} -> diff messages ackedMessages
748 let ackedFederated = do
749 m <- ackedMessages
750 where not (m.originServer == "")
751 where not (m.originServer == serverHostN)
752 yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer}
753 ackedIds <- *ackedIds
754 let combinedAckedIds = union ackedIds ackedFederated
755 let excessCount = count combinedAckedIds - maxAckedIds
756 set *ackedIds = if excessCount <= 0
757 then combinedAckedIds
758 else evictOldestAckedIds combinedAckedIds excessCount
759 let ok = case session of
760 Nothing {} -> False {}
761 _ -> True {}
762 yield {ok: ok, ackedMessages: ackedMessages, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers}
763 let fedMessages = do
764 m <- result.ackedMessages
765 where not (m.originServer == "")
766 where not (m.originServer == serverHostN)
767 yield m
768 let originServers = fold (\acc m ->
769 if count (do a <- acc; where a == m.originServer; yield a) > 0 then acc
770 else union acc [m.originServer]) [] fedMessages
771 respond {ok: result.ok}
772 if result.ok && count originServers > 0 && not (federationSecret == "")
773 then sendToEach originServers (\server ->
774 fetch (federationScheme ++ server) (FederationAck {
775 authorization: "Bearer " ++ federationSecret,
776 messageIds: (do m <- fedMessages; where m.originServer == server; yield m.messageId),
777 originServer: server, fromServer: serverHostN}))
778 else yield {}
779 broadcastOfflineGossip t result.peerServers result.expiredLocal
781 LookupPresence {pubkey, authorization, respond} -> do
782 let validPubkey = not (pubkey == "") && not (tooLong pubkey maxPubkeyLen) && isValidHex pubkey
783 let pubkey = if isValidHex pubkey then normalizeHex pubkey else pubkey
784 t <- now
785 result <- atomic do
786 pr <- maybePrune t
787 sessions <- *sessions
788 presence <- *presence
789 let session = lookupSession sessions (extractToken authorization) t
790 let authorized = case session of
791 Nothing {} -> False {}
792 _ -> True {}
793 let valid = do
794 p <- presence
795 where p.pubkey == pubkey
796 where p.expiresAt > t
797 yield p
798 let foundServer = case session of
799 Nothing {} -> ""
800 Just {value: _} ->
801 if not validPubkey || count valid == 0
802 then ""
803 else case single (do
804 p <- valid
805 where p.server == serverHostN
806 yield p) of
807 Just {value: p} -> p.server
808 Nothing {} -> (fold (\acc p -> if p.expiresAt > acc.best then {server: p.server, best: p.expiresAt} else acc) {server: "", best: 0} valid).server
809 yield {foundServer: foundServer, authorized: authorized, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers}
810 respond {online: not (result.foundServer == ""), server: result.foundServer, authorized: result.authorized}
811 broadcastOfflineGossip t result.peerServers result.expiredLocal
813 RecvGossip {events, hopsRemaining, fromServer, authorization, respond} -> do
814 t <- now
815 let authorized = verifyFederationAuth authorization
816 let cappedHops = if hopsRemaining > 5 then 5 else hopsRemaining
817 let fromServer = normalizeServerName fromServer
818 let validRequest = authorized && cappedHops >= 1 && not (fromServer == "") && not (fromServer == serverHostN) && not (tooLong fromServer maxServerNameLen) && not (isBadServerName fromServer) && count events <= maxGossipEvents
819 let onlineEntries = do
820 e <- events
821 where validRequest
822 where e.eventType == "online"
823 where not (e.server == "")
824 where not (e.pubkey == "")
825 where not (tooLong e.pubkey maxPubkeyLen)
826 where not (tooLong e.server maxServerNameLen)
827 where not (isBadServerName e.server)
828 where isValidHex e.pubkey
829 where not (normalizeServerName e.server == serverHostN)
830 where e.timestamp <= t + clockSkewTolerance
831 where e.timestamp + (clampTtl e.ttl) * 1000 > t
832 yield {pubkey: normalizeHex e.pubkey, server: normalizeServerName e.server, expiresAt: e.timestamp + (clampTtl e.ttl) * 1000, timestamp: e.timestamp}
833 let offlineEntries = do
834 e <- events
835 where validRequest
836 where e.eventType == "offline"
837 where not (e.server == "")
838 where not (e.pubkey == "")
839 where not (tooLong e.pubkey maxPubkeyLen)
840 where not (tooLong e.server maxServerNameLen)
841 where not (isBadServerName e.server)
842 where isValidHex e.pubkey
843 where not (normalizeServerName e.server == serverHostN)
844 where e.timestamp <= t + clockSkewTolerance
845 where t - e.timestamp < (clampTtl e.ttl) * 1000
846 yield {pubkey: normalizeHex e.pubkey, server: normalizeServerName e.server, timestamp: e.timestamp}
847 let filteredOnline = do
848 e <- onlineEntries
849 where count (do
850 o <- offlineEntries
851 where o.pubkey == e.pubkey
852 where o.server == e.server
853 where o.timestamp > e.timestamp
854 yield o) == 0
855 yield {pubkey: e.pubkey, server: e.server, expiresAt: e.expiresAt, timestamp: e.timestamp}
856 let affectedKeys = union
857 (do
858 e <- onlineEntries
859 yield {pubkey: e.pubkey, server: e.server})
860 (do
861 e <- offlineEntries
862 yield {pubkey: e.pubkey, server: e.server})
863 let deduped = fold (\acc e ->
864 if count (do
865 a <- acc
866 where a.pubkey == e.pubkey
867 where a.server == e.server
868 where a.expiresAt >= e.expiresAt
869 yield a) > 0
870 then acc
871 else union
872 (do
873 a <- acc
874 where not (a.pubkey == e.pubkey && a.server == e.server)
875 yield a)
876 [e]) [] filteredOnline
877 gossipResult <- atomic do
878 pr <- maybePrune t
879 presence <- *presence
880 let fromServerTrusted = count (do p <- presence; where p.server == fromServer; yield p) > 0
881 let trustedDeduped = do
882 e <- deduped
883 where e.server == fromServer || fromServerTrusted || count (do p <- presence; where p.server == e.server; yield p) > 0
884 yield e
885 let trustedOffline = do
886 e <- offlineEntries
887 where e.server == fromServer || fromServerTrusted || count (do p <- presence; where p.server == e.server; yield p) > 0
888 yield e
889 let appliedOnline = do
890 e <- trustedDeduped
891 where count (do
892 p <- presence
893 where p.pubkey == e.pubkey
894 where p.server == e.server
895 where p.expiresAt >= e.expiresAt
896 yield p) == 0
897 yield {pubkey: e.pubkey, server: e.server}
898 let merged = fold (\acc e ->
899 if count (do
900 p <- acc
901 where p.pubkey == e.pubkey
902 where p.server == e.server
903 where p.expiresAt >= e.expiresAt
904 yield p) > 0
905 then acc
906 else union
907 (do
908 p <- acc
909 where not (p.pubkey == e.pubkey && p.server == e.server)
910 yield p)
911 [e]) presence trustedDeduped
912 let offlineOnlyKeys = do
913 k <- affectedKeys
914 where count (do d <- trustedDeduped; where d.pubkey == k.pubkey; where d.server == k.server; yield d) == 0
915 yield k
916 let appliedOffline = do
917 k <- offlineOnlyKeys
918 where count (do
919 p <- presence
920 where p.pubkey == k.pubkey
921 where p.server == k.server
922 where count (do
923 o <- trustedOffline
924 where o.pubkey == k.pubkey
925 where o.server == k.server
926 where o.timestamp >= p.timestamp
927 yield o) > 0
928 yield p) > 0
929 yield {pubkey: k.pubkey, server: k.server}
930 let afterOffline = diff merged (do
931 p <- merged
932 k <- offlineOnlyKeys
933 where p.pubkey == k.pubkey
934 where p.server == k.server
935 where count (do
936 o <- trustedOffline
937 where o.pubkey == k.pubkey
938 where o.server == k.server
939 where o.timestamp >= p.timestamp
940 yield o) > 0
941 yield p)
942 let withinCapacity = count afterOffline <= maxPresenceEntries
943 let updatesOnlyMerged = if withinCapacity then afterOffline
944 else fold (\acc e ->
945 if count (do
946 p <- acc
947 where p.pubkey == e.pubkey
948 where p.server == e.server
949 where p.expiresAt >= e.expiresAt
950 yield p) > 0
951 then acc
952 else if count (do p <- presence; where p.pubkey == e.pubkey; where p.server == e.server; yield p) > 0
953 then union
954 (do
955 p <- acc
956 where not (p.pubkey == e.pubkey && p.server == e.server)
957 yield p)
958 [e]
959 else acc) presence trustedDeduped
960 let updatesOnlyAfterOffline = if withinCapacity then afterOffline
961 else diff updatesOnlyMerged (do
962 p <- updatesOnlyMerged
963 k <- offlineOnlyKeys
964 where p.pubkey == k.pubkey
965 where p.server == k.server
966 where count (do
967 o <- trustedOffline
968 where o.pubkey == k.pubkey
969 where o.server == k.server
970 where o.timestamp >= p.timestamp
971 yield o) > 0
972 yield p)
973 let reclaimCapacity = if withinCapacity then 0 else maxPresenceEntries - count updatesOnlyAfterOffline
974 let newEntries = if withinCapacity then []
975 else do
976 e <- trustedDeduped
977 where count (do p <- presence; where p.pubkey == e.pubkey; where p.server == e.server; yield p) == 0
978 where count (do p <- updatesOnlyAfterOffline; where p.pubkey == e.pubkey; where p.server == e.server; yield p) == 0
979 yield e
980 let reclaimedEntries = if withinCapacity then []
981 else fold (\acc e ->
982 if count acc >= reclaimCapacity then acc
983 else union acc [e]) [] newEntries
984 let afterReclaim = if withinCapacity then afterOffline
985 else union updatesOnlyAfterOffline reclaimedEntries
986 let appliedOnlineUpdates = if withinCapacity then appliedOnline
987 else union
988 (do
989 e <- trustedDeduped
990 where count (do p <- presence; where p.pubkey == e.pubkey; where p.server == e.server; yield p) > 0
991 where count (do
992 p <- presence
993 where p.pubkey == e.pubkey
994 where p.server == e.server
995 where p.expiresAt >= e.expiresAt
996 yield p) == 0
997 yield {pubkey: e.pubkey, server: e.server})
998 (do
999 e <- reclaimedEntries
1000 yield {pubkey: e.pubkey, server: e.server})
1001 let finalPresence = if validRequest
1002 then if withinCapacity then afterOffline else afterReclaim
1003 else presence
1004 set *presence = finalPresence
1005 let gossipPeerServers = fold (\acc p ->
1006 if p.server == serverHostN then acc
1007 else if count (do a <- acc; where a == p.server; yield a) > 0 then acc
1008 else union acc [p.server]) [] finalPresence
1009 yield {appliedKeys: (if withinCapacity then union appliedOnline appliedOffline else union appliedOnlineUpdates appliedOffline), expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, gossipPeerServers: gossipPeerServers, withinCapacity: withinCapacity}
1010 let validEvents = do
1011 e <- events
1012 where validRequest
1013 where e.eventType == "online" || e.eventType == "offline"
1014 where not (e.server == "")
1015 where not (e.pubkey == "")
1016 where not (tooLong e.pubkey maxPubkeyLen)
1017 where not (tooLong e.server maxServerNameLen)
1018 where not (isBadServerName e.server)
1019 where isValidHex e.pubkey
1020 where not (normalizeServerName e.server == serverHostN)
1021 where e.timestamp <= t + clockSkewTolerance
1022 where e.timestamp + (clampTtl e.ttl) * 1000 > t
1023 where not (e.eventType == "online" && count (do
1024 o <- offlineEntries
1025 where o.pubkey == normalizeHex e.pubkey
1026 where o.server == normalizeServerName e.server
1027 where o.timestamp > e.timestamp
1028 yield o) > 0)
1029 where not (e.eventType == "offline" && count (do
1030 o <- onlineEntries
1031 where o.pubkey == normalizeHex e.pubkey
1032 where o.server == normalizeServerName e.server
1033 where o.timestamp > e.timestamp
1034 yield o) > 0)
1035 yield e
1036 let effectiveEvents = do
1037 e <- validEvents
1038 where count (do a <- gossipResult.appliedKeys; where a.pubkey == normalizeHex e.pubkey; where a.server == normalizeServerName e.server; yield a) > 0
1039 yield {eventType: e.eventType, pubkey: normalizeHex e.pubkey, server: normalizeServerName e.server, timestamp: e.timestamp, ttl: clampTtl e.ttl}
1040 respond {ok: validRequest}
1041 if not gossipResult.withinCapacity
1042 then println ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer)
1043 else yield {}
1044 if validRequest && cappedHops > 1 && not (federationSecret == "") && count effectiveEvents > 0
1045 then do
1046 let targets = do
1047 s <- gossipResult.gossipPeerServers
1048 where not (s == fromServer)
1049 yield s
1050 sendToEach targets (\server ->
1051 fetch (federationScheme ++ server) (RecvGossip {
1052 authorization: "Bearer " ++ federationSecret,
1053 events: effectiveEvents, hopsRemaining: cappedHops - 1, fromServer: serverHostN}))
1054 else yield {}
1055 broadcastOfflineGossip t gossipResult.peerServers gossipResult.expiredLocal
1057 ForwardMessage {toKey, fromKey, encryptedBlob, timestamp, originServer, messageId, authorization, respond} -> do
1058 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 maxMessageIdLen) && isValidHex toKey && isValidHex fromKey && isValidHex encryptedBlob && isValidHex messageId
1059 let toKey = if isValidHex toKey then normalizeHex toKey else toKey
1060 let fromKey = if isValidHex fromKey then normalizeHex fromKey else fromKey
1061 let encryptedBlob = if isValidHex encryptedBlob then normalizeHex encryptedBlob else encryptedBlob
1062 let messageId = if isValidHex messageId then normalizeHex messageId else messageId
1063 let originServer = normalizeServerName originServer
1064 t <- now
1065 let authorized = verifyFederationAuth authorization
1066 let validOrigin = not (originServer == "") && not (originServer == serverHostN) && not (tooLong originServer maxServerNameLen) && not (isBadServerName originServer)
1067 let clampedTs = if timestamp > t then t else if timestamp < t - messageMaxAge then t - messageMaxAge else timestamp
1068 result <- atomic do
1069 pr <- maybePrune t
1070 messages <- *messages
1071 seqCounter <- *seqCounter
1072 let seqTs = if t > seqCounter then t else seqCounter + 1
1073 let recipientMsgCount = count (do
1074 m <- messages
1075 where m.toKey == toKey
1076 where m.receivedAt >= t - messageMaxAge
1077 yield m)
1078 let existsInMessages = count (do
1079 m <- messages
1080 where m.messageId == messageId
1081 where m.toKey == toKey
1082 where m.receivedAt >= t - messageMaxAge
1083 yield m) > 0
1084 ackedIds <- *ackedIds
1085 let existsInAcked = count (do
1086 a <- ackedIds
1087 where a.messageId == messageId
1088 where a.toKey == toKey
1089 yield a) > 0
1090 let exists = existsInMessages || existsInAcked
1091 let shouldInsert = authorized && validOrigin && validFields && not exists && recipientMsgCount < maxMessagesPerRecipient
1092 set *seqCounter = if shouldInsert then seqTs else seqCounter
1093 set *messages = if shouldInsert
1094 then union messages
1095 [{messageId: messageId, toKey: toKey, fromKey: fromKey,
1096 encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs}]
1097 else messages
1098 yield {peerServers: pr.peerServers, expiredLocal: pr.expiredLocal, inserted: shouldInsert, alreadyExists: exists}
1099 respond {ok: result.inserted || (authorized && validOrigin && validFields && result.alreadyExists)}
1100 broadcastOfflineGossip t result.peerServers result.expiredLocal
1102 FederationAck {messageIds, originServer, fromServer, authorization, respond} -> do
1103 let fromServer = normalizeServerName fromServer
1104 let originServer = normalizeServerName originServer
1105 t <- now
1106 let authorized = verifyFederationAuth authorization
1107 let dedupedIds = fold (\acc mid ->
1108 if mid == "" then acc
1109 else if count acc >= maxMessageIds then acc
1110 else if tooLong mid maxMessageIdLen then acc
1111 else if not (isValidHex mid) then acc
1112 else if count (do a <- acc; where a == normalizeHex mid; yield a) > 0 then acc
1113 else union acc [normalizeHex mid]) [] messageIds
1114 let validAckFields = authorized && originServer == serverHostN && not (fromServer == "") && not (fromServer == serverHostN) && not (tooLong fromServer maxServerNameLen) && not (isBadServerName fromServer)
1115 pruneResult <- atomic do
1116 pr <- maybePrune t
1117 forwards <- *forwards
1118 let validAck = validAckFields
1119 let forwardsToRemove = do
1120 f <- forwards
1121 mid <- dedupedIds
1122 where f.messageId == mid
1123 where f.toServer == fromServer
1124 yield f
1125 set *forwards = if validAck
1126 then diff forwards forwardsToRemove
1127 else forwards
1128 yield {peerServers: pr.peerServers, expiredLocal: pr.expiredLocal, validAck: validAck}
1129 respond {ok: pruneResult.validAck}
1130 broadcastOfflineGossip t pruneResult.peerServers pruneResult.expiredLocal
1132 Prune {authorization, respond} -> do
1133 t <- now
1134 let authorized = verifyMaintenanceAuth authorization
1135 if authorized
1136 then do
1137 result <- atomic do
1138 set *lastPrunedAt = t
1139 pr <- pruneExpired t
1140 forwards <- *forwards
1141 let pendingForwards = do
1142 f <- forwards
1143 yield {toServer: f.toServer, toKey: f.toKey, fromKey: f.fromKey, encryptedBlob: f.encryptedBlob, timestamp: f.timestamp, messageId: f.messageId}
1144 ackedIds <- *ackedIds
1145 let pendingAcks = do
1146 a <- ackedIds
1147 where not (a.originServer == "")
1148 where not (a.originServer == serverHostN)
1149 yield a
1150 yield {expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, pendingForwards: pendingForwards, pendingAcks: pendingAcks}
1151 respond {ok: True {}}
1152 broadcastOfflineGossip t result.peerServers result.expiredLocal
1153 if not (federationSecret == "") && count result.pendingForwards > 0
1154 then retryPendingForwards result.pendingForwards
1155 else yield {}
1156 if not (federationSecret == "") && count result.pendingAcks > 0
1157 then do
1158 let ackOriginServers = fold (\acc a ->
1159 if count (do s <- acc; where s == a.originServer; yield s) > 0 then acc
1160 else union acc [a.originServer]) [] result.pendingAcks
1161 sendToEach ackOriginServers (\server ->
1162 fetch (federationScheme ++ server) (FederationAck {
1163 authorization: "Bearer " ++ federationSecret,
1164 messageIds: (do a <- result.pendingAcks; where a.originServer == server; yield a.messageId),
1165 originServer: server, fromServer: serverHostN}))
1166 else yield {}
1167 else respond {ok: False {}}
1169-- ============================================
1170-- Entry Point
1171-- ============================================
1173main = do
1174 println "Skrepka relay server starting on port 8080..."
1175 listen 8080 serve