git.iko.soy/skrepka
on adeccc3 help sign in
skrepka/server.knot
1262 lines 56 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 fedResult = safeCompare token (if federationSecret == "" then "x" else federationSecret)
331 let maintResult = safeCompare token (if maintenanceSecret == "" then "x" else maintenanceSecret)
332 let fedN = if fedResult then 1 else 0
333 let maintN = if maintResult then 1 else 0
334 let fedSecretSet = if federationSecret == "" then 0 else 1
335 let maintSecretSet = if maintenanceSecret == "" then 0 else 1
336 fedN * fedSecretSet + maintN * maintSecretSet > 0
338clampTtl = \ttl -> if ttl < 1 then 1 else if ttl > maxPresenceTtl then maxPresenceTtl else ttl
340evictOldestAckedIds = \items n ->
341 if n <= 0 then items
342 else
343 let toEvict = (fold (\acc a ->
344 if acc.size < n
345 then {size: acc.size + 1, heap: union acc.heap [a], heapMax: if a.ackedAt > acc.heapMax then a.ackedAt else acc.heapMax}
346 else if a.ackedAt < acc.heapMax
347 then
348 let removed = fold (\r x ->
349 if not r.done && x.ackedAt == acc.heapMax
350 then {done: True {}, items: r.items}
351 else {done: r.done, items: union r.items [x]}) {done: False {}, items: []} acc.heap
352 let newHeap = union removed.items [a]
353 let newMax = fold (\best x -> if x.ackedAt > best then x.ackedAt else best) -1 newHeap
354 {size: acc.size, heap: newHeap, heapMax: newMax}
355 else acc) {size: 0, heap: [], heapMax: -1} items).heap
356 in diff items toEvict
358safeCompareLoop = \s1 s2 n result ->
359 if n <= 0 && s1 == "" && s2 == ""
360 then result
361 else safeCompareLoop (drop 1 s1) (drop 1 s2) (n - 1) (result + (if take 1 s1 == take 1 s2 then 0 else 1))
363safeCompare = \a b ->
364 safeCompareLoop a b maxTokenLen 0 + (if a == "" then 1 else 0) + (if b == "" then 1 else 0) == 0
366sendToEach = \servers action ->
367 if count servers == 0
368 then yield {}
369 else do
370 let server = fold (\acc s -> if acc == "" then s else acc) "" servers
371 let rest = do s <- servers; where not (s == server); yield s
372 if server == ""
373 then yield {}
374 else do
375 catch (action server) (\err ->
376 println ("federation: error reaching " ++ server ++ ": " ++ show err))
377 sendToEach rest action
379pruneExpired = \t -> do
380 challenges <- *challenges
381 set *challenges = do
382 c <- challenges
383 where c.expiresAt > t
384 yield c
385 sessions <- *sessions
386 set *sessions = do
387 s <- sessions
388 where s.expiresAt > t
389 yield s
390 presenceBefore <- *presence
391 let peerServersBefore = fold (\acc p ->
392 if p.server == serverHostN then acc
393 else if count (do a <- acc; where a == p.server; yield a) > 0 then acc
394 else union acc [p.server]) [] presenceBefore
395 let presenceAfter = do
396 p <- presenceBefore
397 where p.expiresAt > t
398 yield p
399 set *presence = presenceAfter
400 messages <- *messages
401 set *messages = do
402 m <- messages
403 where m.receivedAt >= t - messageMaxAge
404 yield m
405 forwards <- *forwards
406 set *forwards = do
407 f <- forwards
408 where f.createdAt >= t - messageMaxAge
409 yield f
410 ackedIds <- *ackedIds
411 set *ackedIds = do
412 a <- ackedIds
413 where a.ackedAt >= t - messageMaxAge
414 yield a
415 yield {expiredLocal: (do
416 p <- presenceBefore
417 where p.expiresAt <= t
418 where p.server == serverHostN
419 where count (do
420 q <- presenceAfter
421 where q.pubkey == p.pubkey
422 where q.server == serverHostN
423 yield q) == 0
424 yield p), peerServers: peerServersBefore}
426maybePrune = \t -> do
427 lastPruned <- *lastPrunedAt
428 if t - lastPruned < 1000
429 then yield {expiredLocal: [], peerServers: []}
430 else do
431 set *lastPrunedAt = t
432 pruneExpired t
434sendBatchedGossip = \peerServers allEvents ->
435 if count allEvents == 0
436 then yield {}
437 else do
438 let batch = fold (\acc e ->
439 if count acc >= maxGossipEvents then acc
440 else union acc [e]) [] allEvents
441 let rest = (fold (\acc e ->
442 if acc.taken < maxGossipEvents
443 then {taken: acc.taken + 1, items: acc.items}
444 else {taken: acc.taken, items: union acc.items [e]}) {taken: 0, items: []} allEvents).items
445 sendToEach peerServers (\server ->
446 fetch (federationScheme ++ server) (RecvGossip {
447 authorization: "Bearer " ++ federationSecret,
448 events: batch, hopsRemaining: 2, fromServer: serverHostN}))
449 sendBatchedGossip peerServers rest
451broadcastOfflineGossip = \t peerServers expiredLocal -> do
452 let expiredEntries = fold (\acc p ->
453 if count (do a <- acc; where a.pubkey == p.pubkey; yield a) > 0 then acc
454 else union acc [{pubkey: p.pubkey, ttl: clampTtl ((p.expiresAt - p.timestamp) / 1000)}]) [] expiredLocal
455 if not (federationSecret == "") && count expiredEntries > 0 && count peerServers > 0
456 then do
457 let gossipEvents = do
458 e <- expiredEntries
459 yield {eventType: "offline", pubkey: e.pubkey, server: serverHostN, timestamp: t, ttl: e.ttl}
460 sendBatchedGossip peerServers gossipEvents
461 else yield {}
463retryPendingForwards = \pending ->
464 if count pending == 0
465 then yield {}
466 else do
467 let f = fold (\acc x -> if acc.messageId == "" then x else acc) {messageId: "", toServer: "", toKey: "", fromKey: "", encryptedBlob: "", timestamp: 0} pending
468 let rest = do x <- pending; where not (x.messageId == f.messageId && x.toServer == f.toServer); yield x
469 if f.messageId == ""
470 then yield {}
471 else do
472 catch (do
473 fetch (federationScheme ++ f.toServer) (ForwardMessage {
474 toKey: f.toKey, fromKey: f.fromKey, encryptedBlob: f.encryptedBlob,
475 timestamp: f.timestamp, originServer: serverHostN, messageId: f.messageId,
476 authorization: "Bearer " ++ federationSecret})
477 yield {}) (\err ->
478 println ("prune: retry forward to " ++ f.toServer ++ " failed: " ++ show err))
479 retryPendingForwards rest
481sendAcksAndClean = \pendingAcks servers ->
482 if count servers == 0
483 then yield {}
484 else do
485 let server = fold (\acc s -> if acc == "" then s else acc) "" servers
486 let rest = do s <- servers; where not (s == server); yield s
487 if server == ""
488 then yield {}
489 else do
490 let serverAcks = do a <- pendingAcks; where a.originServer == server; yield a
491 catch (do
492 fetch (federationScheme ++ server) (FederationAck {
493 authorization: "Bearer " ++ federationSecret,
494 messageIds: (do a <- serverAcks; yield a.messageId),
495 originServer: server, fromServer: serverHostN})
496 atomic do
497 ackedIds <- *ackedIds
498 set *ackedIds = diff ackedIds serverAcks) (\err ->
499 println ("federation: error sending ack to " ++ server ++ ": " ++ show err))
500 sendAcksAndClean pendingAcks rest
502pollOnce = \token cursor ->
503 atomic do
504 t <- now
505 sessions <- *sessions
506 messages <- *messages
507 seqCounter <- *seqCounter
508 let session = lookupSession sessions token t
509 let cursorCeil = if seqCounter > t then seqCounter else t
510 let clampedCursor = if cursor < 0 then 0 else if cursor > cursorCeil then cursorCeil else cursor
511 let matchingMsgs = case session of
512 Nothing {} -> []
513 Just {value: s} -> do
514 m <- messages
515 where m.toKey == s.pubkey
516 where m.receivedAt >= t - messageMaxAge
517 where m.receivedAt > clampedCursor
518 yield m
519 let limitedMsgs = fold (\acc m ->
520 if count acc >= maxPollEvents then acc
521 else union acc [m]) [] matchingMsgs
522 let authorized = case session of
523 Nothing {} -> False {}
524 _ -> True {}
525 let evts = do
526 m <- limitedMsgs
527 yield {id: m.messageId, eventType: "message",
528 payload: toJson {toKey: m.toKey, fromKey: m.fromKey,
529 encryptedBlob: m.encryptedBlob,
530 timestamp: m.timestamp}}
531 let lastReceivedAt = fold (\acc m -> if m.receivedAt > acc then m.receivedAt else acc) clampedCursor limitedMsgs
532 let newCursor = case session of
533 Nothing {} -> -1
534 _ -> lastReceivedAt
535 yield {events: evts, cursor: newCursor, authorized: authorized}
537pollLoop = \token cursor deadline -> do
538 result <- pollOnce token cursor
539 t <- now
540 if result.authorized && count result.events == 0 && t < deadline
541 then do
542 let remaining = deadline - t
543 let sleepMs = if remaining > 500 then 500 else remaining
544 sleep sleepMs
545 pollLoop token cursor deadline
546 else yield result
548-- ============================================
549-- Request Handler
550-- ============================================
552serve = \req -> case req of
554 ReqChallenge {pubkey, respond} -> do
555 let valid = not (pubkey == "") && not (tooLong pubkey maxPubkeyLen) && isValidHex pubkey
556 let pubkey = if isValidHex pubkey then normalizeHex pubkey else pubkey
557 challenge <- genToken
558 t <- now
559 if valid
560 then do
561 pruneResult <- atomic do
562 pr <- maybePrune t
563 challenges <- *challenges
564 let existingCount = count (do c <- challenges; where c.pubkey == pubkey; yield c)
565 let oldestChallengeExpiry = fold (\acc c2 -> if c2.pubkey == pubkey && (acc == -1 || c2.expiresAt < acc) then c2.expiresAt else acc) -1 challenges
566 let cappedChallenges = if existingCount >= maxChallengesPerKey && oldestChallengeExpiry >= 0
567 then fold (\acc c ->
568 if not acc.removed && c.pubkey == pubkey && c.expiresAt == oldestChallengeExpiry
569 then {removed: True {}, items: acc.items}
570 else {removed: acc.removed, items: union acc.items [c]}) {removed: False {}, items: []} challenges
571 else {removed: False {}, items: challenges}
572 let globalOldestExpiry = if count cappedChallenges.items >= maxChallenges
573 then fold (\acc c -> if acc == -1 || c.expiresAt < acc then c.expiresAt else acc) -1 cappedChallenges.items
574 else -1
575 let globalCapped = if globalOldestExpiry >= 0
576 then (fold (\acc c ->
577 if not acc.removed && c.expiresAt == globalOldestExpiry
578 then {removed: True {}, items: acc.items}
579 else {removed: acc.removed, items: union acc.items [c]}) {removed: False {}, items: []} cappedChallenges.items).items
580 else cappedChallenges.items
581 set *challenges = union
582 globalCapped
583 [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry}]
584 yield pr
585 respond {challenge: challenge}
586 broadcastOfflineGossip t pruneResult.peerServers pruneResult.expiredLocal
587 else respond {challenge: ""}
589 VerifyAuth {pubkey, challenge, signature, respond} -> do
590 let pubkey = if isValidHex pubkey then normalizeHex pubkey else pubkey
591 t <- now
592 token <- genToken
593 let expiry = t + sessionExpiry
594 result <- atomic do
595 pr <- maybePrune t
596 challenges <- *challenges
597 sessions <- *sessions
598 presence <- *presence
599 let matched = single (do
600 c <- challenges
601 where c.pubkey == pubkey
602 where c.challenge == challenge
603 where c.expiresAt > t
604 yield c)
605 let sigValid = case matched of
606 Nothing {} -> False {}
607 _ -> if tooLong signature maxSignatureLen
608 then False {}
609 else if not (isValidHex pubkey) || not (isValidHex signature)
610 then False {}
611 else verify (bytesFromHex pubkey) (textToBytes challenge) (bytesFromHex signature)
612 set *challenges = case matched of
613 Just {value: _} -> do
614 c <- challenges
615 where not (c.pubkey == pubkey && c.challenge == challenge)
616 yield c
617 Nothing {} -> challenges
618 let existingForKey = do
619 s <- sessions
620 where s.pubkey == pubkey
621 where s.expiresAt > t
622 yield s
623 let oldestExpiry = fold (\acc s -> if acc == -1 || s.expiresAt < acc then s.expiresAt else acc) -1 existingForKey
624 let cappedFold = fold (\acc s ->
625 if not acc.removed && s.pubkey == pubkey && s.expiresAt == oldestExpiry
626 then {removed: True {}, sessions: acc.sessions}
627 else {removed: acc.removed, sessions: union acc.sessions [s]}) {removed: False {}, sessions: []} sessions
628 let sessionsCapped = if count existingForKey >= maxSessionsPerKey && oldestExpiry >= 0
629 then cappedFold.sessions
630 else sessions
631 let globalOldestSessionExpiry = if count sessionsCapped >= maxSessions
632 then fold (\acc s -> if acc == -1 || s.expiresAt < acc then s.expiresAt else acc) -1 sessionsCapped
633 else -1
634 let globalSessionsCapped = if globalOldestSessionExpiry >= 0
635 then (fold (\acc s ->
636 if not acc.removed && s.expiresAt == globalOldestSessionExpiry
637 then {removed: True {}, items: acc.items}
638 else {removed: acc.removed, items: union acc.items [s]}) {removed: False {}, items: []} sessionsCapped).items
639 else sessionsCapped
640 set *sessions = if sigValid
641 then union globalSessionsCapped [{token: token, pubkey: pubkey, expiresAt: expiry}]
642 else sessions
643 let presenceWithout = do
644 p <- presence
645 where not (p.pubkey == pubkey && p.server == serverHostN)
646 yield p
647 let oldestPresenceExp = fold (\acc p -> if acc == -1 || p.expiresAt < acc then p.expiresAt else acc) -1 presenceWithout
648 let presenceEvicted = if count presenceWithout >= maxPresenceEntries && oldestPresenceExp >= 0
649 then (fold (\acc p ->
650 if not acc.removed && p.expiresAt == oldestPresenceExp
651 then {removed: True {}, items: acc.items}
652 else {removed: acc.removed, items: union acc.items [p]}) {removed: False {}, items: []} presenceWithout).items
653 else presenceWithout
654 let updatedPresence = if sigValid
655 then union presenceEvicted [{pubkey: pubkey, server: serverHostN, expiresAt: expiry, timestamp: t}]
656 else presence
657 set *presence = updatedPresence
658 let peerServers = if sigValid && not (federationSecret == "")
659 then fold (\acc p ->
660 if p.server == serverHostN then acc
661 else if count (do a <- acc; where a == p.server; yield a) > 0 then acc
662 else union acc [p.server]) [] updatedPresence
663 else []
664 yield {sigValid: sigValid, peerServers: peerServers, expiredLocal: pr.expiredLocal, offlinePeers: pr.peerServers}
665 if result.sigValid
666 then respond {token: token, expiresAt: expiry}
667 else respond {token: "", expiresAt: 0}
668 if result.sigValid && not (federationSecret == "")
669 then sendToEach result.peerServers (\server ->
670 fetch (federationScheme ++ server) (RecvGossip {
671 authorization: "Bearer " ++ federationSecret,
672 events: [{eventType: "online", pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl}], hopsRemaining: 2, fromServer: serverHostN}))
673 else yield {}
674 broadcastOfflineGossip t result.offlinePeers (do
675 p <- result.expiredLocal
676 where not (p.pubkey == pubkey)
677 yield p)
679 Poll {cursor, timeout, authorization, respond} -> do
680 t <- now
681 let clampedTimeout = if timeout < 0 then 0 else if timeout > 60000 then 60000 else timeout
682 let deadline = t + clampedTimeout
683 let token = extractToken authorization
684 result <- pollLoop token cursor deadline
685 respond {events: result.events, cursor: result.cursor, authorized: result.authorized}
686 tAfterPoll <- now
687 pruneResult <- atomic do
688 maybePrune tAfterPoll
689 broadcastOfflineGossip tAfterPoll pruneResult.peerServers pruneResult.expiredLocal
691 SendMessage {to, encryptedBlob, timestamp, authorization, respond} -> do
692 let validInput = not (to == "") && not (encryptedBlob == "") && not (tooLong to maxPubkeyLen) && not (tooLong encryptedBlob maxBlobLen) && isValidHex to && isValidHex encryptedBlob
693 let to = if isValidHex to then normalizeHex to else to
694 let encryptedBlob = if isValidHex encryptedBlob then normalizeHex encryptedBlob else encryptedBlob
695 t <- now
696 msgId <- genToken
697 let clampedTs = if timestamp > t then t else if timestamp < t - messageMaxAge then t - messageMaxAge else timestamp
698 result <- atomic do
699 pr <- maybePrune t
700 sessions <- *sessions
701 messages <- *messages
702 presence <- *presence
703 let session = lookupSession sessions (extractToken authorization) t
704 seqCounter <- *seqCounter
705 let seqTs = if t > seqCounter then t else seqCounter + 1
706 let recipientMsgCount = count (do
707 m <- messages
708 where m.toKey == to
709 where m.receivedAt >= t - messageMaxAge
710 yield m)
711 let shouldInsert = case session of
712 Nothing {} -> False {}
713 Just {value: _} -> validInput && recipientMsgCount < maxMessagesPerRecipient
714 set *seqCounter = if shouldInsert then seqTs else seqCounter
715 set *messages = case session of
716 Nothing {} -> messages
717 Just {value: s} -> if validInput && recipientMsgCount < maxMessagesPerRecipient
718 then union messages
719 [{messageId: msgId, toKey: to, fromKey: s.pubkey,
720 encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}]
721 else messages
722 let localOnline = case session of
723 Nothing {} -> False {}
724 Just {value: _} -> count (do
725 p <- presence
726 where p.pubkey == to
727 where p.server == serverHostN
728 where p.expiresAt > t
729 yield p) > 0
730 let remoteServers = case session of
731 Nothing {} -> []
732 Just {value: _} -> do
733 p <- presence
734 where p.pubkey == to
735 where not (p.server == serverHostN)
736 where p.expiresAt > t
737 yield p.server
738 let fromKey = case session of
739 Nothing {} -> ""
740 Just {value: s} -> s.pubkey
741 let status = case session of
742 Nothing {} -> "unauthorized"
743 _ -> if not validInput then "invalid"
744 else if localOnline then "delivered"
745 else if count remoteServers > 0 then "federated"
746 else "queued"
747 forwards <- *forwards
748 let forwardsAtCapacity = count forwards >= maxForwardsEntries
749 set *forwards = if shouldInsert && not forwardsAtCapacity && count remoteServers > 0
750 then union forwards (do s <- remoteServers; yield {messageId: msgId, toServer: s, createdAt: t, toKey: to, fromKey: fromKey, encryptedBlob: encryptedBlob, timestamp: clampedTs})
751 else forwards
752 yield {status: status, remoteServers: remoteServers, fromKey: fromKey, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, forwardsAtCapacity: forwardsAtCapacity}
753 let finalStatus = if result.status == "federated" && (federationSecret == "" || result.forwardsAtCapacity) then "queued" else result.status
754 respond {status: finalStatus, messageId: (if result.status == "unauthorized" then "" else if result.status == "invalid" then "" else msgId)}
755 if count result.remoteServers > 0 && not (result.status == "unauthorized") && not (result.status == "invalid") && not (federationSecret == "")
756 then sendToEach result.remoteServers (\server ->
757 fetch (federationScheme ++ server) (ForwardMessage {
758 toKey: to, fromKey: result.fromKey, encryptedBlob: encryptedBlob,
759 timestamp: clampedTs, originServer: serverHostN, messageId: msgId,
760 authorization: "Bearer " ++ federationSecret}))
761 else yield {}
762 if result.forwardsAtCapacity && count result.remoteServers > 0 && not (result.status == "unauthorized") && not (result.status == "invalid")
763 then println ("sendMessage: forwards at capacity (" ++ show maxForwardsEntries ++ "), skipping forward tracking")
764 else yield {}
765 broadcastOfflineGossip t result.peerServers result.expiredLocal
767 AckMessages {messageIds, authorization, respond} -> do
768 t <- now
769 let dedupedIds = fold (\acc mid ->
770 if mid == "" then acc
771 else if count acc >= maxMessageIds then acc
772 else if tooLong mid maxMessageIdLen then acc
773 else if not (isValidHex mid) then acc
774 else if count (do a <- acc; where a == normalizeHex mid; yield a) > 0 then acc
775 else union acc [normalizeHex mid]) [] messageIds
776 result <- atomic do
777 pr <- maybePrune t
778 sessions <- *sessions
779 messages <- *messages
780 let session = lookupSession sessions (extractToken authorization) t
781 let ackedMessages = case session of
782 Nothing {} -> []
783 Just {value: s} -> do
784 m <- messages
785 mid <- dedupedIds
786 where m.messageId == mid
787 where m.toKey == s.pubkey
788 yield m
789 set *messages = case session of
790 Nothing {} -> messages
791 Just {value: _} -> diff messages ackedMessages
792 let ackedFederated = do
793 m <- ackedMessages
794 where not (m.originServer == "")
795 where not (m.originServer == serverHostN)
796 yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer}
797 ackedIds <- *ackedIds
798 let combinedAckedIds = union ackedIds ackedFederated
799 let excessCount = count combinedAckedIds - maxAckedIds
800 let afterEviction = if excessCount <= 0
801 then combinedAckedIds
802 else evictOldestAckedIds combinedAckedIds excessCount
803 let evictedEntries = if excessCount <= 0
804 then []
805 else diff combinedAckedIds afterEviction
806 set *ackedIds = afterEviction
807 let ok = case session of
808 Nothing {} -> False {}
809 _ -> True {}
810 yield {ok: ok, ackedMessages: ackedMessages, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, evictedEntries: evictedEntries}
811 let fedMessages = do
812 m <- result.ackedMessages
813 where not (m.originServer == "")
814 where not (m.originServer == serverHostN)
815 yield m
816 let originServers = fold (\acc m ->
817 if count (do a <- acc; where a == m.originServer; yield a) > 0 then acc
818 else union acc [m.originServer]) [] fedMessages
819 respond {ok: result.ok}
820 if result.ok && count originServers > 0 && not (federationSecret == "")
821 then do
822 let pendingAcks = do
823 m <- fedMessages
824 yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer}
825 sendAcksAndClean pendingAcks originServers
826 else yield {}
827 if count result.evictedEntries > 0 && not (federationSecret == "")
828 then do
829 let evictedOriginServers = fold (\acc a ->
830 if count (do s <- acc; where s == a.originServer; yield s) > 0 then acc
831 else union acc [a.originServer]) [] result.evictedEntries
832 sendToEach evictedOriginServers (\server ->
833 fetch (federationScheme ++ server) (FederationAck {
834 authorization: "Bearer " ++ federationSecret,
835 messageIds: (do a <- result.evictedEntries; where a.originServer == server; yield a.messageId),
836 originServer: server, fromServer: serverHostN}))
837 else yield {}
838 broadcastOfflineGossip t result.peerServers result.expiredLocal
840 LookupPresence {pubkey, authorization, respond} -> do
841 let validPubkey = not (pubkey == "") && not (tooLong pubkey maxPubkeyLen) && isValidHex pubkey
842 let pubkey = if isValidHex pubkey then normalizeHex pubkey else pubkey
843 t <- now
844 result <- atomic do
845 pr <- maybePrune t
846 sessions <- *sessions
847 presence <- *presence
848 let session = lookupSession sessions (extractToken authorization) t
849 let authorized = case session of
850 Nothing {} -> False {}
851 _ -> True {}
852 let valid = do
853 p <- presence
854 where p.pubkey == pubkey
855 where p.expiresAt > t
856 yield p
857 let foundServer = case session of
858 Nothing {} -> ""
859 Just {value: _} ->
860 if not validPubkey || count valid == 0
861 then ""
862 else case single (do
863 p <- valid
864 where p.server == serverHostN
865 yield p) of
866 Just {value: p} -> p.server
867 Nothing {} -> (fold (\acc p -> if p.expiresAt > acc.best then {server: p.server, best: p.expiresAt} else acc) {server: "", best: 0} valid).server
868 yield {foundServer: foundServer, authorized: authorized, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers}
869 respond {online: not (result.foundServer == ""), server: result.foundServer, authorized: result.authorized}
870 broadcastOfflineGossip t result.peerServers result.expiredLocal
872 RecvGossip {events, hopsRemaining, fromServer, authorization, respond} -> do
873 t <- now
874 let authorized = verifyFederationAuth authorization
875 let cappedHops = if hopsRemaining > 5 then 5 else hopsRemaining
876 let fromServer = normalizeServerName fromServer
877 let validRequest = authorized && cappedHops >= 1 && not (fromServer == "") && not (fromServer == serverHostN) && not (tooLong fromServer maxServerNameLen) && not (isBadServerName fromServer) && count events <= maxGossipEvents
878 let onlineEntries = do
879 e <- events
880 where validRequest
881 where e.eventType == "online"
882 where not (e.server == "")
883 where not (e.pubkey == "")
884 where not (tooLong e.pubkey maxPubkeyLen)
885 where not (tooLong e.server maxServerNameLen)
886 where not (isBadServerName e.server)
887 where isValidHex e.pubkey
888 where not (normalizeServerName e.server == serverHostN)
889 where e.timestamp <= t + clockSkewTolerance
890 where e.timestamp + (clampTtl e.ttl) * 1000 > t
891 yield {pubkey: normalizeHex e.pubkey, server: normalizeServerName e.server, expiresAt: e.timestamp + (clampTtl e.ttl) * 1000, timestamp: e.timestamp}
892 let offlineEntries = do
893 e <- events
894 where validRequest
895 where e.eventType == "offline"
896 where not (e.server == "")
897 where not (e.pubkey == "")
898 where not (tooLong e.pubkey maxPubkeyLen)
899 where not (tooLong e.server maxServerNameLen)
900 where not (isBadServerName e.server)
901 where isValidHex e.pubkey
902 where not (normalizeServerName e.server == serverHostN)
903 where e.timestamp <= t + clockSkewTolerance
904 where t - e.timestamp < (clampTtl e.ttl) * 1000
905 yield {pubkey: normalizeHex e.pubkey, server: normalizeServerName e.server, timestamp: e.timestamp}
906 let filteredOnline = do
907 e <- onlineEntries
908 where count (do
909 o <- offlineEntries
910 where o.pubkey == e.pubkey
911 where o.server == e.server
912 where o.timestamp > e.timestamp
913 yield o) == 0
914 yield {pubkey: e.pubkey, server: e.server, expiresAt: e.expiresAt, timestamp: e.timestamp}
915 let affectedKeys = union
916 (do
917 e <- onlineEntries
918 yield {pubkey: e.pubkey, server: e.server})
919 (do
920 e <- offlineEntries
921 yield {pubkey: e.pubkey, server: e.server})
922 let deduped = fold (\acc e ->
923 if count (do
924 a <- acc
925 where a.pubkey == e.pubkey
926 where a.server == e.server
927 where a.expiresAt >= e.expiresAt
928 yield a) > 0
929 then acc
930 else union
931 (do
932 a <- acc
933 where not (a.pubkey == e.pubkey && a.server == e.server)
934 yield a)
935 [e]) [] filteredOnline
936 gossipResult <- atomic do
937 pr <- maybePrune t
938 presence <- *presence
939 let fromServerTrusted = count (do p <- presence; where p.server == fromServer; yield p) > 0
940 let trustedDeduped = do
941 e <- deduped
942 where e.server == fromServer || fromServerTrusted || count (do p <- presence; where p.server == e.server; yield p) > 0
943 yield e
944 let trustedOffline = do
945 e <- offlineEntries
946 where e.server == fromServer || fromServerTrusted || count (do p <- presence; where p.server == e.server; yield p) > 0
947 yield e
948 let appliedOnline = do
949 e <- trustedDeduped
950 where count (do
951 p <- presence
952 where p.pubkey == e.pubkey
953 where p.server == e.server
954 where p.expiresAt >= e.expiresAt
955 yield p) == 0
956 yield {pubkey: e.pubkey, server: e.server}
957 let merged = fold (\acc e ->
958 if count (do
959 p <- acc
960 where p.pubkey == e.pubkey
961 where p.server == e.server
962 where p.expiresAt >= e.expiresAt
963 yield p) > 0
964 then acc
965 else union
966 (do
967 p <- acc
968 where not (p.pubkey == e.pubkey && p.server == e.server)
969 yield p)
970 [e]) presence trustedDeduped
971 let offlineOnlyKeys = do
972 k <- affectedKeys
973 where count (do d <- trustedDeduped; where d.pubkey == k.pubkey; where d.server == k.server; yield d) == 0
974 yield k
975 let appliedOffline = do
976 k <- offlineOnlyKeys
977 where count (do
978 p <- presence
979 where p.pubkey == k.pubkey
980 where p.server == k.server
981 where count (do
982 o <- trustedOffline
983 where o.pubkey == k.pubkey
984 where o.server == k.server
985 where o.timestamp >= p.timestamp
986 yield o) > 0
987 yield p) > 0
988 yield {pubkey: k.pubkey, server: k.server}
989 let afterOffline = diff merged (do
990 p <- merged
991 k <- offlineOnlyKeys
992 where p.pubkey == k.pubkey
993 where p.server == k.server
994 where count (do
995 o <- trustedOffline
996 where o.pubkey == k.pubkey
997 where o.server == k.server
998 where o.timestamp >= p.timestamp
999 yield o) > 0
1000 yield p)
1001 let withinCapacity = count afterOffline <= maxPresenceEntries
1002 let updatesOnlyMerged = if withinCapacity then afterOffline
1003 else fold (\acc e ->
1004 if count (do
1005 p <- acc
1006 where p.pubkey == e.pubkey
1007 where p.server == e.server
1008 where p.expiresAt >= e.expiresAt
1009 yield p) > 0
1010 then acc
1011 else if count (do p <- presence; where p.pubkey == e.pubkey; where p.server == e.server; yield p) > 0
1012 then union
1013 (do
1014 p <- acc
1015 where not (p.pubkey == e.pubkey && p.server == e.server)
1016 yield p)
1017 [e]
1018 else acc) presence trustedDeduped
1019 let updatesOnlyAfterOffline = if withinCapacity then afterOffline
1020 else diff updatesOnlyMerged (do
1021 p <- updatesOnlyMerged
1022 k <- offlineOnlyKeys
1023 where p.pubkey == k.pubkey
1024 where p.server == k.server
1025 where count (do
1026 o <- trustedOffline
1027 where o.pubkey == k.pubkey
1028 where o.server == k.server
1029 where o.timestamp >= p.timestamp
1030 yield o) > 0
1031 yield p)
1032 let reclaimCapacity = if withinCapacity then 0 else maxPresenceEntries - count updatesOnlyAfterOffline
1033 let newEntries = if withinCapacity then []
1034 else do
1035 e <- trustedDeduped
1036 where count (do p <- presence; where p.pubkey == e.pubkey; where p.server == e.server; yield p) == 0
1037 where count (do p <- updatesOnlyAfterOffline; where p.pubkey == e.pubkey; where p.server == e.server; yield p) == 0
1038 yield e
1039 let reclaimedEntries = if withinCapacity then []
1040 else fold (\acc e ->
1041 if count acc >= reclaimCapacity then acc
1042 else union acc [e]) [] newEntries
1043 let afterReclaim = if withinCapacity then afterOffline
1044 else union updatesOnlyAfterOffline reclaimedEntries
1045 let appliedOnlineUpdates = if withinCapacity then appliedOnline
1046 else union
1047 (do
1048 e <- trustedDeduped
1049 where count (do p <- presence; where p.pubkey == e.pubkey; where p.server == e.server; yield p) > 0
1050 where count (do
1051 p <- presence
1052 where p.pubkey == e.pubkey
1053 where p.server == e.server
1054 where p.expiresAt >= e.expiresAt
1055 yield p) == 0
1056 yield {pubkey: e.pubkey, server: e.server})
1057 (do
1058 e <- reclaimedEntries
1059 yield {pubkey: e.pubkey, server: e.server})
1060 let finalPresence = if validRequest
1061 then if withinCapacity then afterOffline else afterReclaim
1062 else presence
1063 set *presence = finalPresence
1064 let gossipPeerServers = fold (\acc p ->
1065 if p.server == serverHostN then acc
1066 else if count (do a <- acc; where a == p.server; yield a) > 0 then acc
1067 else union acc [p.server]) [] finalPresence
1068 yield {appliedKeys: (if withinCapacity then union appliedOnline appliedOffline else union appliedOnlineUpdates appliedOffline), expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, gossipPeerServers: gossipPeerServers, withinCapacity: withinCapacity}
1069 let validEvents = do
1070 e <- events
1071 where validRequest
1072 where e.eventType == "online" || e.eventType == "offline"
1073 where not (e.server == "")
1074 where not (e.pubkey == "")
1075 where not (tooLong e.pubkey maxPubkeyLen)
1076 where not (tooLong e.server maxServerNameLen)
1077 where not (isBadServerName e.server)
1078 where isValidHex e.pubkey
1079 where not (normalizeServerName e.server == serverHostN)
1080 where e.timestamp <= t + clockSkewTolerance
1081 where e.timestamp + (clampTtl e.ttl) * 1000 > t
1082 where not (e.eventType == "online" && count (do
1083 o <- offlineEntries
1084 where o.pubkey == normalizeHex e.pubkey
1085 where o.server == normalizeServerName e.server
1086 where o.timestamp > e.timestamp
1087 yield o) > 0)
1088 where not (e.eventType == "offline" && count (do
1089 o <- onlineEntries
1090 where o.pubkey == normalizeHex e.pubkey
1091 where o.server == normalizeServerName e.server
1092 where o.timestamp > e.timestamp
1093 yield o) > 0)
1094 yield e
1095 let effectiveEvents = do
1096 e <- validEvents
1097 where count (do a <- gossipResult.appliedKeys; where a.pubkey == normalizeHex e.pubkey; where a.server == normalizeServerName e.server; yield a) > 0
1098 yield {eventType: e.eventType, pubkey: normalizeHex e.pubkey, server: normalizeServerName e.server, timestamp: e.timestamp, ttl: clampTtl e.ttl}
1099 respond {ok: validRequest}
1100 if not gossipResult.withinCapacity
1101 then println ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer)
1102 else yield {}
1103 if validRequest && cappedHops > 1 && not (federationSecret == "") && count effectiveEvents > 0
1104 then do
1105 let targets = do
1106 s <- gossipResult.gossipPeerServers
1107 where not (s == fromServer)
1108 yield s
1109 sendToEach targets (\server ->
1110 fetch (federationScheme ++ server) (RecvGossip {
1111 authorization: "Bearer " ++ federationSecret,
1112 events: effectiveEvents, hopsRemaining: cappedHops - 1, fromServer: serverHostN}))
1113 else yield {}
1114 broadcastOfflineGossip t gossipResult.peerServers gossipResult.expiredLocal
1116 ForwardMessage {toKey, fromKey, encryptedBlob, timestamp, originServer, messageId, authorization, respond} -> do
1117 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
1118 let toKey = if isValidHex toKey then normalizeHex toKey else toKey
1119 let fromKey = if isValidHex fromKey then normalizeHex fromKey else fromKey
1120 let encryptedBlob = if isValidHex encryptedBlob then normalizeHex encryptedBlob else encryptedBlob
1121 let messageId = if isValidHex messageId then normalizeHex messageId else messageId
1122 let originServer = normalizeServerName originServer
1123 t <- now
1124 let authorized = verifyFederationAuth authorization
1125 let validOrigin = not (originServer == "") && not (originServer == serverHostN) && not (tooLong originServer maxServerNameLen) && not (isBadServerName originServer)
1126 let clampedTs = if timestamp > t then t else if timestamp < t - messageMaxAge then t - messageMaxAge else timestamp
1127 result <- atomic do
1128 pr <- maybePrune t
1129 messages <- *messages
1130 seqCounter <- *seqCounter
1131 let seqTs = if t > seqCounter then t else seqCounter + 1
1132 let recipientMsgCount = count (do
1133 m <- messages
1134 where m.toKey == toKey
1135 where m.receivedAt >= t - messageMaxAge
1136 yield m)
1137 let existsInMessages = count (do
1138 m <- messages
1139 where m.messageId == messageId
1140 where m.toKey == toKey
1141 yield m) > 0
1142 ackedIds <- *ackedIds
1143 let existsInAcked = count (do
1144 a <- ackedIds
1145 where a.messageId == messageId
1146 where a.toKey == toKey
1147 yield a) > 0
1148 let exists = existsInMessages || existsInAcked
1149 let shouldInsert = authorized && validOrigin && validFields && not exists && recipientMsgCount < maxMessagesPerRecipient
1150 set *seqCounter = if shouldInsert then seqTs else seqCounter
1151 set *messages = if shouldInsert
1152 then union messages
1153 [{messageId: messageId, toKey: toKey, fromKey: fromKey,
1154 encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs}]
1155 else messages
1156 yield {peerServers: pr.peerServers, expiredLocal: pr.expiredLocal, inserted: shouldInsert, alreadyExists: exists}
1157 respond {ok: result.inserted || (authorized && validOrigin && validFields && result.alreadyExists)}
1158 broadcastOfflineGossip t result.peerServers result.expiredLocal
1160 FederationAck {messageIds, originServer, fromServer, authorization, respond} -> do
1161 let fromServer = normalizeServerName fromServer
1162 let originServer = normalizeServerName originServer
1163 t <- now
1164 let authorized = verifyFederationAuth authorization
1165 let dedupedIds = fold (\acc mid ->
1166 if mid == "" then acc
1167 else if count acc >= maxMessageIds then acc
1168 else if tooLong mid maxMessageIdLen then acc
1169 else if not (isValidHex mid) then acc
1170 else if count (do a <- acc; where a == normalizeHex mid; yield a) > 0 then acc
1171 else union acc [normalizeHex mid]) [] messageIds
1172 let validAckFields = authorized && originServer == serverHostN && not (fromServer == "") && not (fromServer == serverHostN) && not (tooLong fromServer maxServerNameLen) && not (isBadServerName fromServer)
1173 pruneResult <- atomic do
1174 pr <- maybePrune t
1175 forwards <- *forwards
1176 let validAck = validAckFields
1177 let forwardsToRemove = do
1178 f <- forwards
1179 mid <- dedupedIds
1180 where f.messageId == mid
1181 where f.toServer == fromServer
1182 yield f
1183 set *forwards = if validAck
1184 then diff forwards forwardsToRemove
1185 else forwards
1186 yield {peerServers: pr.peerServers, expiredLocal: pr.expiredLocal, validAck: validAck}
1187 respond {ok: pruneResult.validAck}
1188 broadcastOfflineGossip t pruneResult.peerServers pruneResult.expiredLocal
1190 Prune {authorization, respond} -> do
1191 t <- now
1192 let authorized = verifyMaintenanceAuth authorization
1193 if authorized
1194 then do
1195 result <- atomic do
1196 set *lastPrunedAt = t
1197 pr <- pruneExpired t
1198 forwards <- *forwards
1199 let pendingForwards = do
1200 f <- forwards
1201 yield {toServer: f.toServer, toKey: f.toKey, fromKey: f.fromKey, encryptedBlob: f.encryptedBlob, timestamp: f.timestamp, messageId: f.messageId}
1202 ackedIds <- *ackedIds
1203 let pendingAcks = do
1204 a <- ackedIds
1205 where not (a.originServer == "")
1206 where not (a.originServer == serverHostN)
1207 yield a
1208 yield {expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, pendingForwards: pendingForwards, pendingAcks: pendingAcks}
1209 respond {ok: True {}}
1210 broadcastOfflineGossip t result.peerServers result.expiredLocal
1211 if not (federationSecret == "") && count result.pendingForwards > 0
1212 then retryPendingForwards result.pendingForwards
1213 else yield {}
1214 if not (federationSecret == "") && count result.pendingAcks > 0
1215 then do
1216 let ackOriginServers = fold (\acc a ->
1217 if count (do s <- acc; where s == a.originServer; yield s) > 0 then acc
1218 else union acc [a.originServer]) [] result.pendingAcks
1219 sendAcksAndClean result.pendingAcks ackOriginServers
1220 else yield {}
1221 else respond {ok: False {}}
1223-- ============================================
1224-- Entry Point
1225-- ============================================
1227backgroundPrune = do
1228 sleep 30000
1229 t <- now
1230 catch (do
1231 result <- atomic do
1232 set *lastPrunedAt = t
1233 pr <- pruneExpired t
1234 forwards <- *forwards
1235 let pendingForwards = do
1236 f <- forwards
1237 yield {toServer: f.toServer, toKey: f.toKey, fromKey: f.fromKey, encryptedBlob: f.encryptedBlob, timestamp: f.timestamp, messageId: f.messageId}
1238 ackedIds <- *ackedIds
1239 let pendingAcks = do
1240 a <- ackedIds
1241 where not (a.originServer == "")
1242 where not (a.originServer == serverHostN)
1243 yield a
1244 yield {expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, pendingForwards: pendingForwards, pendingAcks: pendingAcks}
1245 broadcastOfflineGossip t result.peerServers result.expiredLocal
1246 if not (federationSecret == "") && count result.pendingForwards > 0
1247 then retryPendingForwards result.pendingForwards
1248 else yield {}
1249 if not (federationSecret == "") && count result.pendingAcks > 0
1250 then do
1251 let ackOriginServers = fold (\acc a ->
1252 if count (do s <- acc; where s == a.originServer; yield s) > 0 then acc
1253 else union acc [a.originServer]) [] result.pendingAcks
1254 sendAcksAndClean result.pendingAcks ackOriginServers
1255 else yield {}) (\err ->
1256 println ("background prune: " ++ show err))
1257 backgroundPrune
1259main = do
1260 println "Skrepka relay server starting on port 8080..."
1261 spawn backgroundPrune
1262 listen 8080 serve