git.iko.soy/skrepka
on d41c2bf help sign in
skrepka/server.knot
1098 lines 47 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
36offlineGossipMinTtl = 60 -- minimum TTL for offline gossip events in seconds
37maxForwardRetriesPerServer = 100
38federationEnabled = not (federationSecret == "")
40-- ============================================
41-- Types
42-- ============================================
44type PollEventType = Text where \s -> s == "message"
46type PollEvent = {id: MessageIdHex, eventType: PollEventType, payload: {encryptedBlob: BlobHex, fromKey: PubkeyHex, timestamp: Timestamp, toKey: PubkeyHex}}
48type GossipEventType = Text where \s -> s == "online" || s == "offline"
50type GossipEvent = {eventType: GossipEventType, pubkey: Text, server: Text, timestamp: Timestamp, ttl: Int<S>}
52type ValidGossipEntry = {eventType: GossipEventType, pubkey: PubkeyHex, server: ServerName, expiresAt: Timestamp, timestamp: Timestamp, ttl: Int<S>}
54type MessageStatus = Text where \s -> s == "unauthorized" || s == "rejected" || s == "delivered" || s == "federated" || s == "queued"
56type PubkeyHex = Text where \s -> isValidHex s && not (tooLong s maxPubkeyLen)
57type BlobHex = Text where \s -> isValidHex s && not (tooLong s maxBlobLen)
58type SignatureHex = Text where \s -> isValidHex s && not (tooLong s maxSignatureLen)
59type MessageIdHex = Text where \s -> isValidHex s && not (tooLong s maxMessageIdLen)
60type ServerName = Text where \s -> not (tooLong s maxServerNameLen) && not (isBadServerName s)
61type OptionalServerName = Text where \s -> s == "" || (not (tooLong s maxServerNameLen) && not (isBadServerName s))
63type Timestamp = Int<Ms>
65type Challenge = {pubkey: PubkeyHex, challenge: Text, expiresAt: Timestamp}
66type Session = {token: Text, pubkey: PubkeyHex, expiresAt: Timestamp}
67type Message = {messageId: MessageIdHex, toKey: PubkeyHex, fromKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: OptionalServerName, receivedAt: Timestamp}
68type PresenceEntry = {pubkey: PubkeyHex, server: Text, expiresAt: Timestamp, timestamp: Timestamp}
69type ForwardEntry = {messageId: MessageIdHex, toServer: ServerName, createdAt: Timestamp, toKey: PubkeyHex, fromKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp}
70type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: ServerName}
71type AuthResult = Authorized {pubkey: PubkeyHex} | Unauthorized {}
72type PruneResult = {expiredLocal: [PresenceEntry], peerServers: [ServerName]}
73type PollResult = PollUnauthorized {} | PollAuthorized {events: [PollEvent], cursor: Timestamp}
74type DeliveryStatus = Text where \s -> s == "delivered" || s == "federated" || s == "queued"
75type SendResult = SendUnauthorized {} | SendRejected {} | SendDone {status: DeliveryStatus, remoteServers: [ServerName], fromKey: PubkeyHex}
76type AckResult = AckUnauthorized {} | AckDone {pendingAcks: [AckedEntry], evictedEntries: [AckedEntry]}
79-- ============================================
80-- Persisted State
81-- ============================================
83*challenges : [Challenge]
85*sessions : [Session]
87*messages : [Message]
89*presence : [PresenceEntry]
91*seqCounter : Timestamp
93*forwards : [ForwardEntry]
95*ackedIds : [AckedEntry]
97*lastPrunedAt : Timestamp
99-- ============================================
100-- Routes
101-- ============================================
103route AuthApi where
104 /auth
105 POST {pubkey: PubkeyHex} /challenge
106 -> {challenge: Text}
107 = ReqChallenge
108 POST {pubkey: PubkeyHex, challenge: Text, signature: SignatureHex} /verify
109 -> {token: Text, expiresAt: Timestamp}
110 = VerifyAuth
112route ClientApi where
113 POST {cursor: Timestamp, timeout: Int} /poll headers {authorization: Text}
114 -> {events: [PollEvent], cursor: Timestamp, authorized: Bool}
115 = Poll
116 /messages
117 POST {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp} / headers {authorization: Text}
118 -> {status: MessageStatus, messageId: Text}
119 = SendMessage
120 POST {messageIds: [MessageIdHex]} /ack headers {authorization: Text}
121 -> {ok: Bool}
122 = AckMessages
123 GET /lookup/{pubkey: PubkeyHex} headers {authorization: Text}
124 -> {online: Bool, server: Text, authorized: Bool}
125 = LookupPresence
127route FederationApi where
128 /federation
129 POST {events: [GossipEvent], hopsRemaining: Int, fromServer: ServerName} /gossip headers {authorization: Text}
130 -> {ok: Bool}
131 = RecvGossip
132 POST {toKey: PubkeyHex, fromKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: ServerName, messageId: MessageIdHex} /forward headers {authorization: Text}
133 -> {ok: Bool}
134 = ForwardMessage
135 POST {messageIds: [MessageIdHex], originServer: ServerName, fromServer: ServerName} /ack headers {authorization: Text}
136 -> {ok: Bool}
137 = FederationAck
139route MaintenanceApi where
140 POST {} /maintenance/prune headers {authorization: Text}
141 -> {ok: Bool}
142 = Prune
144route Api = AuthApi | ClientApi | FederationApi | MaintenanceApi
146-- ============================================
147-- Helpers
148-- ============================================
150any = \xs pred -> fold (\acc x -> if acc then acc else pred x) (False {}) xs
151none = \xs pred -> not (any xs pred)
152when = \cond action -> if cond then action else yield {}
153whenOk = \result action -> case result of
154 Ok _ -> action
155 Err _ -> yield {}
156clamp = \lo hi x -> if x < lo then lo else if x > hi then hi else x
157headOf = \items -> fold (\acc x -> case acc of
158 Nothing {} -> Just {value: x}
159 _ -> acc) Nothing {} items
161hexDigit = \n ->
162 let n = if n < 0 then 0 else if n > 15 then 15 else n in
163 if n == 0 then "0" else if n == 1 then "1" else if n == 2 then "2" else if n == 3 then "3"
164 else if n == 4 then "4" else if n == 5 then "5" else if n == 6 then "6" else if n == 7 then "7"
165 else if n == 8 then "8" else if n == 9 then "9" else if n == 10 then "a" else if n == 11 then "b"
166 else if n == 12 then "c" else if n == 13 then "d" else if n == 14 then "e" else "f"
168byteToHex = \n -> hexDigit (n / 16) ++ hexDigit (n - (n / 16) * 16)
170intTo3Hex = \n ->
171 let b0 = n / 65536 in
172 let r0 = n - b0 * 65536 in
173 let b1 = r0 / 256 in
174 let b2 = r0 - b1 * 256 in
175 byteToHex b0 ++ byteToHex b1 ++ byteToHex b2
177genTokenN = \n -> if n <= 0 then yield "" else do
178 chunk <- randomInt 16777216
179 rest <- genTokenN (n - 1)
180 yield (intTo3Hex chunk ++ rest)
182genToken = genTokenN 8
184tooLong = \s maxLen -> not (drop maxLen s == "")
186isHexChar = \c ->
187 c == "0" || c == "1" || c == "2" || c == "3" || c == "4" || c == "5" || c == "6" || c == "7" || c == "8" || c == "9" ||
188 c == "a" || c == "b" || c == "c" || c == "d" || c == "e" || c == "f" ||
189 c == "A" || c == "B" || c == "C" || c == "D" || c == "E" || c == "F"
191isValidHexPairs = \s ->
192 if s == "" then True {}
193 else if drop 1 s == "" then False {}
194 else if isHexChar (take 1 s) && isHexChar (take 1 (drop 1 s)) then isValidHexPairs (drop 2 s)
195 else False {}
197isValidHex = \s -> not (s == "") && isValidHexPairs s
199normalizeHex = \s -> bytesToHex (bytesFromHex s)
201isDigitChar = \c -> c == "0" || c == "1" || c == "2" || c == "3" || c == "4" || c == "5" || c == "6" || c == "7" || c == "8" || c == "9"
203isAllowedServerChar = \c ->
204 isDigitChar c || c == "-" || c == "." || c == ":"
205 || c == "a" || c == "b" || c == "c" || c == "d" || c == "e" || c == "f"
206 || c == "g" || c == "h" || c == "i" || c == "j" || c == "k" || c == "l"
207 || c == "m" || c == "n" || c == "o" || c == "p" || c == "q" || c == "r"
208 || c == "s" || c == "t" || c == "u" || c == "v" || c == "w" || c == "x"
209 || c == "y" || c == "z"
211containsBadUrlChar = \s ->
212 if s == "" then False {}
213 else if not (isAllowedServerChar (take 1 s)) then True {}
214 else containsBadUrlChar (drop 1 s)
216isCgnatRange = \s ->
217 if not (take 4 s == "100.") then False {}
218 else if (take 1 (drop 4 s) == "7" || take 1 (drop 4 s) == "8" || take 1 (drop 4 s) == "9")
219 && isDigitChar (take 1 (drop 5 s)) && take 1 (drop 6 s) == "."
220 && isAllDigitsOrDotsUntilColonOrEnd (drop 7 s)
221 then True {}
222 else if take 1 (drop 4 s) == "6"
223 && (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")
224 && take 1 (drop 6 s) == "."
225 && isAllDigitsOrDotsUntilColonOrEnd (drop 7 s)
226 then True {}
227 else if take 1 (drop 4 s) == "1" && (take 1 (drop 5 s) == "0" || take 1 (drop 5 s) == "1")
228 && isDigitChar (take 1 (drop 6 s)) && take 1 (drop 7 s) == "."
229 && isAllDigitsOrDotsUntilColonOrEnd (drop 8 s)
230 then True {}
231 else if take 1 (drop 4 s) == "1" && take 1 (drop 5 s) == "2"
232 && (take 1 (drop 6 s) == "0" || take 1 (drop 6 s) == "1" || take 1 (drop 6 s) == "2" || take 1 (drop 6 s) == "3"
233 || take 1 (drop 6 s) == "4" || take 1 (drop 6 s) == "5" || take 1 (drop 6 s) == "6" || take 1 (drop 6 s) == "7")
234 && take 1 (drop 7 s) == "."
235 && isAllDigitsOrDotsUntilColonOrEnd (drop 8 s)
236 then True {}
237 else False {}
239containsColon = \s ->
240 if s == "" then False {}
241 else if take 1 s == ":" then True {}
242 else containsColon (drop 1 s)
244containsMultipleColons = \s ->
245 if s == "" then False {}
246 else if take 1 s == ":" then containsColon (drop 1 s)
247 else containsMultipleColons (drop 1 s)
249hasOctalOctet = \s ->
250 if take 1 s == "0" && isDigitChar (take 1 (drop 1 s)) && isAllDigitsOrDotsUntilColonOrEnd (drop 2 s) then True {}
251 else hasOctalOctetAfterDot s
253hasOctalOctetAfterDot = \s ->
254 if s == "" then False {}
255 else if take 1 s == "." && take 1 (drop 1 s) == "0" && isDigitChar (take 1 (drop 2 s)) && isAllDigitsOrDotsUntilColonOrEnd (drop 3 s) then True {}
256 else hasOctalOctetAfterDot (drop 1 s)
258hasHexPrefix = \s ->
259 if take 2 s == "0x" || take 2 s == "0X" then True {}
260 else hasHexPrefixAfterDot s
262hasHexPrefixAfterDot = \s ->
263 if s == "" then False {}
264 else if take 1 s == "." && (take 2 (drop 1 s) == "0x" || take 2 (drop 1 s) == "0X") then True {}
265 else hasHexPrefixAfterDot (drop 1 s)
267isAllDigitsOrDotsUntilColonOrEnd = \s ->
268 if s == "" then True {}
269 else if take 1 s == ":" then True {}
270 else if isDigitChar (take 1 s) || take 1 s == "." then isAllDigitsOrDotsUntilColonOrEnd (drop 1 s)
271 else False {}
273toLowerChar = \c ->
274 if c == "A" then "a" else if c == "B" then "b" else if c == "C" then "c"
275 else if c == "D" then "d" else if c == "E" then "e" else if c == "F" then "f"
276 else if c == "G" then "g" else if c == "H" then "h" else if c == "I" then "i"
277 else if c == "J" then "j" else if c == "K" then "k" else if c == "L" then "l"
278 else if c == "M" then "m" else if c == "N" then "n" else if c == "O" then "o"
279 else if c == "P" then "p" else if c == "Q" then "q" else if c == "R" then "r"
280 else if c == "S" then "s" else if c == "T" then "t" else if c == "U" then "u"
281 else if c == "V" then "v" else if c == "W" then "w" else if c == "X" then "x"
282 else if c == "Y" then "y" else if c == "Z" then "z"
283 else c
285toLower = \s ->
286 if s == "" then ""
287 else toLowerChar (take 1 s) ++ toLower (drop 1 s)
289isTrailingDots = \s ->
290 if s == "" then True {}
291 else if take 1 s == ":" then True {}
292 else if take 1 s == "." then isTrailingDots (drop 1 s)
293 else False {}
295stripTrailingDot = \s ->
296 if s == "" then ""
297 else if take 1 s == "." && isTrailingDots (drop 1 s) then stripTrailingDot (drop 1 s)
298 else take 1 s ++ stripTrailingDot (drop 1 s)
300normalizeServerName = \s -> toLower (stripTrailingDot s)
302serverHostN = normalizeServerName serverHost
304containsSuffix = \suffix suffixLen s ->
305 if s == "" then False {}
306 else if take suffixLen s == suffix && (drop suffixLen s == "" || take 1 (drop suffixLen s) == ":") then True {}
307 else containsSuffix suffix suffixLen (drop 1 s)
309containsLocalhostSuffix = \s -> containsSuffix ".localhost" 10 s
310containsLocalSuffix = \s -> containsSuffix ".local" 6 s
311containsInternalSuffix = \s -> containsSuffix ".internal" 9 s
313is172Private = \sn ->
314 take 4 sn == "172."
315 && let d1 = take 1 (drop 4 sn) in
316 let d2 = take 1 (drop 5 sn) in
317 take 1 (drop 6 sn) == "."
318 && ((d1 == "1" && (d2 == "6" || d2 == "7" || d2 == "8" || d2 == "9"))
319 || (d1 == "2" && isDigitChar d2)
320 || (d1 == "3" && (d2 == "0" || d2 == "1")))
322isHighOctetPrefix = \sn ->
323 take 1 sn == "2"
324 && let d2 = take 1 (drop 1 sn) in
325 let d3 = take 1 (drop 2 sn) in
326 take 1 (drop 3 sn) == "."
327 && ((d2 == "2" && (d3 == "4" || d3 == "5" || d3 == "6" || d3 == "7" || d3 == "8" || d3 == "9"))
328 || (d2 == "3" && isDigitChar d3)
329 || (d2 == "4" && isDigitChar d3)
330 || (d2 == "5" && (d3 == "0" || d3 == "1" || d3 == "2" || d3 == "3" || d3 == "4" || d3 == "5")))
332matchesHostname = \sn name nameLen ->
333 take nameLen sn == name && (drop nameLen sn == "" || take 1 (drop nameLen sn) == ":")
335isBadServerName = \s ->
336 let sn = toLower (stripTrailingDot s) in
337 sn == ""
338 || take 1 sn == "."
339 || take 1 sn == ":"
340 || containsBadUrlChar sn
341 || matchesHostname sn "localhost" 9
342 || matchesHostname sn "localhost.localdomain" 21
343 || containsLocalhostSuffix sn
344 || matchesHostname sn "local" 5
345 || containsLocalSuffix sn
346 || matchesHostname sn "internal" 8
347 || containsInternalSuffix sn
348 || matchesHostname sn "metadata" 8
349 || take 4 sn == "127." || take 3 sn == "10."
350 || take 8 sn == "192.168." || take 8 sn == "169.254."
351 || take 2 sn == "0." || take 1 sn == "["
352 || is172Private sn
353 || take 7 sn == "198.18." || take 7 sn == "198.19."
354 || take 8 sn == "192.0.0."
355 || take 8 sn == "192.0.2."
356 || take 10 sn == "192.88.99."
357 || take 11 sn == "198.51.100."
358 || take 10 sn == "203.0.113."
359 || take 2 sn == "::"
360 || isCgnatRange sn
361 || containsMultipleColons sn
362 || hasOctalOctet sn
363 || hasHexPrefix sn
364 || (isDigitChar (take 1 sn) && isAllDigitsOrDotsUntilColonOrEnd sn)
365 || isHighOctetPrefix sn
367authenticateSession : [Session] -> Text -> Timestamp -> AuthResult
368authenticateSession = \sessions token t ->
369 case headOf (do s <- sessions; where safeCompare s.token token && s.expiresAt > t; yield s) of
370 Nothing {} -> Unauthorized {}
371 Just {value: s} -> Authorized {pubkey: s.pubkey}
373authenticateRequest = \sessions authorization t ->
374 authenticateSession sessions (extractToken authorization) t
376extractToken = \authHeader ->
377 let token = drop 7 authHeader in
378 if toLower (take 7 authHeader) == "bearer " && not (tooLong token maxTokenLen)
379 then token
380 else ""
382verifySecret = \token secret ->
383 not (secret == "") && safeCompare token secret
385verifyFederationAuth = \authorization ->
386 verifySecret (extractToken authorization) federationSecret
388verifyMaintenanceAuth = \authorization ->
389 let token = extractToken authorization in
390 verifySecret token federationSecret || verifySecret token maintenanceSecret
392clampTtl = clamp 1 maxPresenceTtl
394clampTimestamp = \timestamp t -> clamp (t - messageMaxAge) t timestamp
396nextSeqTs = \t seqCounter -> if t > seqCounter then t else seqCounter + 1
398isFederatedOrigin = \originServer -> not (originServer == "") && not (originServer == serverHostN)
400takeN = \items n -> fold (\acc e -> if count acc >= n then acc else union acc [e]) [] items
402dedupMessageIds = \messageIds ->
403 fold (\acc mid ->
404 let nmid = normalizeHex mid in
405 if count acc >= maxMessageIds then acc
406 else if any acc (\a -> a == nmid) then acc
407 else union acc [nmid]) [] messageIds
409boundedSmallest = \items n getKey ->
410 (fold (\acc a ->
411 let k = getKey a in
412 if acc.size < n
413 then
414 let newHeap = union acc.heap [a] in
415 let newSize = count newHeap in
416 if newSize > acc.size
417 then {size: newSize, heap: newHeap, heapMax: case acc.heapMax of
418 Nothing {} -> Just {value: k}
419 Just {value: v} -> if k > v then Just {value: k} else acc.heapMax}
420 else acc
421 else case acc.heapMax of
422 Nothing {} -> acc
423 Just {value: maxVal} ->
424 if k < maxVal
425 then
426 let removed = fold (\r x ->
427 if not r.done && getKey x == maxVal
428 then {done: True {}, items: r.items}
429 else {done: r.done, items: union r.items [x]}) {done: False {}, items: []} acc.heap in
430 let newHeap = union removed.items [a] in
431 let newMax = fold (\best x -> case best of
432 Nothing {} -> Just {value: getKey x}
433 Just {value: v} -> if getKey x > v then Just {value: getKey x} else best) Nothing {} newHeap in
434 {size: acc.size, heap: newHeap, heapMax: newMax}
435 else acc) {size: 0, heap: [], heapMax: Nothing {}} items).heap
437evictOldestAckedIds = \items n ->
438 if n <= 0 then items
439 else diff items (boundedSmallest items n (\a -> a.ackedAt))
441safeCompareLoop = \s1 s2 n result ->
442 if n <= 0 && s1 == "" && s2 == ""
443 then result
444 else safeCompareLoop (drop 1 s1) (drop 1 s2) (n - 1) (result + (if take 1 s1 == take 1 s2 then 0 else 1))
446safeCompare = \a b ->
447 safeCompareLoop a b maxTokenLen 0 + (if a == "" then 1 else 0) + (if b == "" then 1 else 0) == 0
449evictOneOldest = \items getKey ->
450 let oldest = fold (\acc x -> let k = getKey x in case acc of
451 Nothing {} -> Just {value: k}
452 Just {value: v} -> if k < v then Just {value: k} else acc) Nothing {} items in
453 case oldest of
454 Nothing {} -> items
455 Just {value: oldestKey} ->
456 (fold (\acc x ->
457 if not acc.removed && getKey x == oldestKey
458 then {removed: True {}, items: acc.items}
459 else {removed: acc.removed, items: union acc.items [x]}) {removed: False {}, items: []} items).items
461capPerKey = \items matchKey maxPerKey getExpiry t ->
462 let active = do x <- items; where getExpiry x > t; yield x in
463 let forKey = do x <- active; where matchKey x; yield x in
464 if count forKey >= maxPerKey
465 then let others = do x <- active; where not (matchKey x); yield x in
466 union others (evictOneOldest forKey getExpiry)
467 else active
469capGlobal = \items maxCount getKey ->
470 if count items >= maxCount
471 then evictOneOldest items getKey
472 else items
474collectPeerServers = \items getServer ->
475 fold (\acc item ->
476 let s = getServer item in
477 if s == serverHostN then acc
478 else if any acc (\a -> a == s) then acc
479 else union acc [s]) [] items
481samePubkeyServer = \a b -> a.pubkey == b.pubkey && a.server == b.server
483mergeByKey = \sameKey initial items ->
484 fold (\acc e ->
485 if any acc (\a -> sameKey a e && a.expiresAt >= e.expiresAt)
486 then acc
487 else union
488 (do a <- acc; where not (sameKey a e); yield a)
489 [e]) initial items
491mergeByPubkeyServer = mergeByKey samePubkeyServer
493removeOffline = \presenceSet offlineOnlyKeys trustedOffline t ->
494 diff presenceSet (do
495 p <- presenceSet
496 k <- offlineOnlyKeys
497 where samePubkeyServer p k && any trustedOffline (\o -> samePubkeyServer o k && (if o.timestamp > t then t else o.timestamp) > p.timestamp)
498 yield p)
500countMessagesFor = \messages toKey t ->
501 count (do
502 m <- messages
503 where m.toKey == toKey && m.receivedAt >= t - messageMaxAge
504 yield m)
507validGossipEvents : [GossipEvent] -> Timestamp -> [ValidGossipEntry]
508validGossipEvents = \events t -> do
509 e <- events
510 where isValidHex e.pubkey && not (tooLong e.pubkey maxPubkeyLen) && not (tooLong e.server maxServerNameLen) && not (isBadServerName e.server)
511 let server = normalizeServerName e.server
512 where not (server == serverHostN) && e.timestamp <= t + clockSkewTolerance
513 let ttl = clampTtl e.ttl
514 let expiresAt = e.timestamp + ttl * 1000
515 where expiresAt > t
516 yield {eventType: e.eventType, pubkey: normalizeHex e.pubkey, server: server, expiresAt: expiresAt, timestamp: e.timestamp, ttl: ttl}
518sendToEach : [ServerName] -> (ServerName -> (a {})) -> a {}
519sendToEach = \servers action ->
520 case headOf servers of
521 Nothing {} -> yield {}
522 Just {value: server} -> do
523 let rest = do s <- servers; where not (s == server); yield s
524 fork (action server)
525 sendToEach rest action
527pruneExpired = \t -> do
528 set *challenges = do
529 c <- *challenges
530 where c.expiresAt > t
531 yield c
532 set *sessions = do
533 s <- *sessions
534 where s.expiresAt > t
535 yield s
536 presenceBefore <- *presence
537 let peerServersBefore = collectPeerServers presenceBefore (\p -> p.server)
538 let presenceAfter = do
539 p <- presenceBefore
540 where p.expiresAt > t
541 yield p
542 set *presence = presenceAfter
543 set *messages = do
544 m <- *messages
545 where m.receivedAt >= t - messageMaxAge
546 yield m
547 set *forwards = do
548 f <- *forwards
549 where f.createdAt >= t - messageMaxAge
550 yield f
551 set *ackedIds = do
552 a <- *ackedIds
553 where a.ackedAt >= t - messageMaxAge
554 yield a
555 yield {expiredLocal: (do
556 p <- presenceBefore
557 where p.expiresAt <= t && p.server == serverHostN && none presenceAfter (\q -> q.pubkey == p.pubkey && q.server == serverHostN)
558 yield p), peerServers: peerServersBefore}
560maybePrune = \t -> do
561 lastPruned <- *lastPrunedAt
562 if t - lastPruned < 1000
563 then yield {expiredLocal: [], peerServers: []}
564 else do
565 set *lastPrunedAt = t
566 pruneExpired t
568forEachBatch = \items batchSize process ->
569 if count items == 0 then yield {}
570 else do
571 let batch = takeN items batchSize
572 let rest = diff items batch
573 process batch
574 forEachBatch rest batchSize process
576sendBatchedGossipToServer : ServerName -> [GossipEvent] -> Int -> IO {network} {}
577sendBatchedGossipToServer = \server allEvents hops ->
578 forEachBatch allEvents maxGossipEvents (\batch -> do
579 _ <- fetch (federationScheme ++ server) (RecvGossip {
580 authorization: "Bearer " ++ federationSecret,
581 events: batch, hopsRemaining: hops, fromServer: serverHostN})
582 yield {})
584sendBatchedGossip : [ServerName] -> [GossipEvent] -> Int -> IO {network} {}
585sendBatchedGossip = \peerServers allEvents hops ->
586 sendToEach peerServers (\server -> sendBatchedGossipToServer server allEvents hops)
588broadcastOfflineGossip = \t peerServers expiredLocal ->
589 when (federationEnabled && count peerServers > 0 && count expiredLocal > 0) (do
590 let expiredEntries = mergeByKey (\a b -> a.pubkey == b.pubkey) [] (do
591 p <- expiredLocal
592 let ttl = clamp offlineGossipMinTtl maxPresenceTtl ((p.expiresAt - p.timestamp) / 1000)
593 yield {pubkey: p.pubkey, ttl: ttl, expiresAt: p.expiresAt})
594 let gossipEvents = do
595 e <- expiredEntries
596 yield {eventType: "offline", pubkey: e.pubkey, server: serverHostN, timestamp: e.expiresAt, ttl: e.ttl}
597 sendBatchedGossip peerServers gossipEvents 2)
599forkBroadcastOffline = \t result ->
600 fork (broadcastOfflineGossip t result.peerServers result.expiredLocal)
602maybePruneAndBroadcast = \t -> do
603 pruneResult <- atomic do maybePrune t
604 forkBroadcastOffline t pruneResult
606removeForward = \mid srv -> atomic do
607 forwards <- *forwards
608 set *forwards = do
609 x <- forwards
610 where not (x.messageId == mid && x.toServer == srv)
611 yield x
613retryForwardsForServer : [ForwardEntry] -> Int -> IO {network} {}
614retryForwardsForServer = \entries remaining ->
615 if remaining <= 0
616 then yield {}
617 else case headOf entries of
618 Nothing {} -> yield {}
619 Just {value: f} -> do
620 let rest = do x <- entries; where not (x.messageId == f.messageId); yield x
621 result <- fetch (federationScheme ++ f.toServer) (ForwardMessage {
622 toKey: f.toKey, fromKey: f.fromKey, encryptedBlob: f.encryptedBlob,
623 timestamp: f.timestamp, originServer: serverHostN, messageId: f.messageId,
624 authorization: "Bearer " ++ federationSecret})
625 whenOk result (removeForward f.messageId f.toServer)
626 retryForwardsForServer rest (remaining - 1)
628retryPendingForwards : [ForwardEntry] -> IO {network} {}
629retryPendingForwards = \pending -> do
630 let servers = collectPeerServers pending (\x -> x.toServer)
631 sendToEach servers (\server ->
632 retryForwardsForServer (do x <- pending; where x.toServer == server; yield x) maxForwardRetriesPerServer)
633 yield {}
635sendBatchedAcksToServer : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: ServerName, | a}] -> ServerName -> IO {network} {}
636sendBatchedAcksToServer = \serverAcks server ->
637 forEachBatch serverAcks maxMessageIds (\batch -> do
638 result <- fetch (federationScheme ++ server) (FederationAck {
639 authorization: "Bearer " ++ federationSecret,
640 messageIds: (do a <- batch; yield a.messageId),
641 originServer: server, fromServer: serverHostN})
642 whenOk result (atomic do
643 ackedIds <- *ackedIds
644 set *ackedIds = do
645 a <- ackedIds
646 where not (a.originServer == server && any batch (\sa -> sa.messageId == a.messageId && sa.toKey == a.toKey))
647 yield a))
649sendAcksAndClean : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: ServerName, | a}] -> [ServerName] -> IO {network} {}
650sendAcksAndClean = \pendingAcks servers ->
651 sendToEach servers (\server ->
652 sendBatchedAcksToServer (do a <- pendingAcks; where a.originServer == server; yield a) server)
654retryEvictedAcks = \evictedEntries pendingAcks ->
655 let evictedToRetry = do
656 a <- evictedEntries
657 where none pendingAcks (\a2 -> a2.messageId == a.messageId && a2.toKey == a.toKey)
658 yield a
659 in when (count evictedToRetry > 0) (do
660 let evictedOriginServers = collectPeerServers evictedToRetry (\a -> a.originServer)
661 sendToEach evictedOriginServers (\server -> do
662 fetchResult <- fetch (federationScheme ++ server) (FederationAck {
663 authorization: "Bearer " ++ federationSecret,
664 messageIds: (do a <- evictedToRetry; where a.originServer == server; yield a.messageId),
665 originServer: server, fromServer: serverHostN})
666 case fetchResult of
667 Err _ -> do
668 dropped <- atomic do
669 ackedIds <- *ackedIds
670 let reinsert = do a <- evictedToRetry; where a.originServer == server; yield a
671 let combined = union ackedIds reinsert
672 let excess = count combined - maxAckedIds
673 let afterEviction = if excess <= 0 then combined
674 else evictOldestAckedIds combined excess
675 set *ackedIds = afterEviction
676 yield excess
677 when (dropped > 0)
678 (println ("ackMessages: ackedIds at capacity (" ++ show maxAckedIds ++ "), " ++ show dropped ++ " ack entries permanently lost for " ++ server))
679 _ -> yield {}))
681pruneAndRetry = \t -> do
682 result <- atomic do
683 set *lastPrunedAt = t
684 pr <- pruneExpired t
685 forwards <- *forwards
686 ackedIds <- *ackedIds
687 let pendingAcks = do a <- ackedIds; where isFederatedOrigin a.originServer; yield a
688 yield {expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, pendingForwards: forwards, pendingAcks: pendingAcks}
689 broadcastOfflineGossip t result.peerServers result.expiredLocal
690 when (federationEnabled && count result.pendingForwards > 0)
691 (retryPendingForwards result.pendingForwards)
692 when (federationEnabled && count result.pendingAcks > 0) (do
693 let ackOriginServers = collectPeerServers result.pendingAcks (\a -> a.originServer)
694 sendAcksAndClean result.pendingAcks ackOriginServers)
696pollOnce = \token cursor -> do
697 t <- now
698 atomic do
699 sessions <- *sessions
700 let auth = authenticateSession sessions token t
701 case auth of
702 Unauthorized {} ->
703 yield PollUnauthorized {}
704 Authorized {pubkey} -> do
705 seqCounter <- *seqCounter
706 messages <- *messages
707 let cursorCeil = if seqCounter > t then seqCounter else t
708 let clampedCursor = clamp 0 cursorCeil cursor
709 let matchingMsgs = do
710 m <- messages
711 where m.toKey == pubkey && m.receivedAt >= t - messageMaxAge && m.receivedAt > clampedCursor
712 yield m
713 let limitedMsgs = boundedSmallest matchingMsgs maxPollEvents (\m -> m.receivedAt)
714 let evts = do
715 m <- limitedMsgs
716 yield {id: m.messageId, eventType: "message",
717 payload: {toKey: m.toKey, fromKey: m.fromKey,
718 encryptedBlob: m.encryptedBlob,
719 timestamp: m.timestamp}}
720 let lastReceivedAt = fold (\acc m -> if m.receivedAt > acc then m.receivedAt else acc) clampedCursor limitedMsgs
721 yield PollAuthorized {events: evts, cursor: lastReceivedAt}
723pollLoop = \token cursor deadline -> do
724 result <- pollOnce token cursor
725 t <- now
726 case result of
727 PollUnauthorized {} -> yield result
728 PollAuthorized {events} ->
729 if count events == 0 && t < deadline
730 then do
731 let remaining = deadline - t
732 let sleepMs = if remaining > 500 then 500<Ms> else remaining
733 sleep sleepMs
734 pollLoop token cursor deadline
735 else yield result
737-- ============================================
738-- Request Handler
739-- ============================================
741serve = \req -> case req of
743 ReqChallenge {pubkey, respond} -> do
744 let pubkey = normalizeHex pubkey
745 challenge <- genToken
746 t <- now
747 pruneResult <- atomic do
748 pr <- maybePrune t
749 challenges <- *challenges
750 let activeChallenges = capPerKey challenges (\c -> c.pubkey == pubkey) maxChallengesPerKey (\c -> c.expiresAt) t
751 let globalCapped = capGlobal activeChallenges maxChallenges (\c -> c.expiresAt)
752 set *challenges = union
753 globalCapped
754 [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry}]
755 yield pr
756 forkBroadcastOffline t pruneResult
757 yield (respond {challenge: challenge})
759 VerifyAuth {pubkey, challenge, signature, respond} -> do
760 let pubkey = normalizeHex pubkey
761 t <- now
762 token <- genToken
763 let expiry = t + sessionExpiry
764 result <- atomic do
765 pr <- maybePrune t
766 challenges <- *challenges
767 sessions <- *sessions
768 presence <- *presence
769 let hasMatch = any challenges (\c -> c.pubkey == pubkey && c.challenge == challenge && c.expiresAt > t)
770 let sigValid = if hasMatch then verify (bytesFromHex pubkey) (textToBytes challenge) (bytesFromHex signature) else False {}
771 if not sigValid
772 then yield {pr: pr, verified: False {}, peerServers: []}
773 else do
774 set *challenges = do
775 c <- challenges
776 where not (c.pubkey == pubkey && c.challenge == challenge)
777 yield c
778 let activeSessions = capPerKey sessions (\s -> s.pubkey == pubkey) maxSessionsPerKey (\s -> s.expiresAt) t
779 let globalSessionsCapped = capGlobal activeSessions maxSessions (\s -> s.expiresAt)
780 set *sessions = union globalSessionsCapped [{token: token, pubkey: pubkey, expiresAt: expiry}]
781 let presenceWithout = do
782 p <- presence
783 where not (p.pubkey == pubkey && p.server == serverHostN)
784 yield p
785 let presenceEvicted = capGlobal presenceWithout maxPresenceEntries (\p -> p.expiresAt)
786 let updatedPresence = union presenceEvicted [{pubkey: pubkey, server: serverHostN, expiresAt: expiry, timestamp: t}]
787 set *presence = updatedPresence
788 let peerServers = if federationEnabled
789 then collectPeerServers updatedPresence (\p -> p.server)
790 else []
791 yield {pr: pr, verified: True {}, peerServers: peerServers}
792 let expiredLocal = if result.verified
793 then do p <- result.pr.expiredLocal; where not (p.pubkey == pubkey); yield p
794 else result.pr.expiredLocal
795 fork (do
796 when (result.verified && federationEnabled)
797 (sendBatchedGossip result.peerServers [{eventType: "online", pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl}] 2)
798 broadcastOfflineGossip t result.pr.peerServers expiredLocal)
799 yield (if result.verified
800 then respond {token: token, expiresAt: expiry}
801 else respond {token: "", expiresAt: 0})
803 Poll {cursor, timeout, authorization, respond} -> do
804 t <- now
805 let clampedTimeout = clamp 0 60000 timeout
806 let deadline = t + clampedTimeout
807 let token = extractToken authorization
808 result <- pollLoop token cursor deadline
809 tAfterPoll <- now
810 maybePruneAndBroadcast tAfterPoll
811 yield (case result of
812 PollUnauthorized {} -> respond {events: [], cursor: -1, authorized: False {}}
813 PollAuthorized {events, cursor: c} -> respond {events: events, cursor: c, authorized: True {}})
815 SendMessage {to, encryptedBlob, timestamp, authorization, respond} -> do
816 let to = normalizeHex to
817 let encryptedBlob = normalizeHex encryptedBlob
818 t <- now
819 msgId <- genToken
820 let clampedTs = clampTimestamp timestamp t
821 result <- atomic do
822 pr <- maybePrune t
823 sessions <- *sessions
824 let auth = authenticateRequest sessions authorization t
825 case auth of
826 Unauthorized {} ->
827 yield {pr: pr, send: SendUnauthorized {}}
828 Authorized {pubkey: fromKey} -> do
829 messages <- *messages
830 if countMessagesFor messages to t >= maxMessagesPerRecipient
831 then yield {pr: pr, send: SendRejected {}}
832 else do
833 presence <- *presence
834 seqCounter <- *seqCounter
835 let seqTs = nextSeqTs t seqCounter
836 set *seqCounter = seqTs
837 set *messages = union messages
838 [{messageId: msgId, toKey: to, fromKey: fromKey,
839 encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}]
840 let toPresence = do p <- presence; where p.pubkey == to && p.expiresAt > t; yield p
841 let localOnline = any toPresence (\p -> p.server == serverHostN)
842 let remoteServers = do p <- toPresence; where not (p.server == serverHostN); yield p.server
843 forwards <- *forwards
844 let forwardServers = takeN remoteServers (maxForwardsEntries - count forwards)
845 let shouldForward = federationEnabled && count forwardServers > 0
846 let status = if localOnline then "delivered"
847 else if shouldForward then "federated"
848 else "queued"
849 set *forwards = if shouldForward
850 then union forwards (do s <- forwardServers; yield {messageId: msgId, toServer: s, createdAt: t, toKey: to, fromKey: fromKey, encryptedBlob: encryptedBlob, timestamp: clampedTs})
851 else forwards
852 yield {pr: pr, send: SendDone {status: status, remoteServers: remoteServers, fromKey: fromKey}}
853 fork (do
854 case result.send of
855 SendDone {status, remoteServers, fromKey} ->
856 when (count remoteServers > 0 && federationEnabled) (do
857 sendToEach remoteServers (\server -> do
858 fwdResult <- fetch (federationScheme ++ server) (ForwardMessage {
859 toKey: to, fromKey: fromKey, encryptedBlob: encryptedBlob,
860 timestamp: clampedTs, originServer: serverHostN, messageId: msgId,
861 authorization: "Bearer " ++ federationSecret})
862 whenOk fwdResult (removeForward msgId server))
863 when (status == "queued")
864 (println ("sendMessage: forwards at capacity (" ++ show maxForwardsEntries ++ "), skipping forward tracking")))
865 _ -> yield {}
866 broadcastOfflineGossip t result.pr.peerServers result.pr.expiredLocal)
867 yield (case result.send of
868 SendUnauthorized _ -> respond {status: "unauthorized", messageId: ""}
869 SendRejected _ -> respond {status: "rejected", messageId: ""}
870 SendDone {status} -> respond {status: status, messageId: msgId})
872 AckMessages {messageIds, authorization, respond} -> do
873 t <- now
874 let dedupedIds = dedupMessageIds messageIds
875 result <- atomic do
876 pr <- maybePrune t
877 sessions <- *sessions
878 let auth = authenticateRequest sessions authorization t
879 case auth of
880 Unauthorized {} ->
881 yield {pr: pr, ack: AckUnauthorized {}}
882 Authorized {pubkey} -> do
883 messages <- *messages
884 let ackedMessages = do
885 m <- messages
886 mid <- dedupedIds
887 where m.messageId == mid && m.toKey == pubkey
888 yield m
889 set *messages = diff messages ackedMessages
890 let ackedFederated = do
891 m <- ackedMessages
892 where isFederatedOrigin m.originServer
893 yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer}
894 ackedIds <- *ackedIds
895 let combinedAckedIds = union ackedIds ackedFederated
896 let evictedEntries = let excessCount = count combinedAckedIds - maxAckedIds in
897 if excessCount <= 0 then [] else boundedSmallest combinedAckedIds excessCount (\a -> a.ackedAt)
898 set *ackedIds = diff combinedAckedIds evictedEntries
899 yield {pr: pr, ack: AckDone {pendingAcks: ackedFederated, evictedEntries: evictedEntries}}
900 fork (do
901 case result.ack of
902 AckDone {pendingAcks, evictedEntries} ->
903 when federationEnabled (do
904 let originServers = collectPeerServers pendingAcks (\a -> a.originServer)
905 when (count originServers > 0)
906 (sendAcksAndClean pendingAcks originServers)
907 retryEvictedAcks evictedEntries pendingAcks)
908 _ -> yield {}
909 broadcastOfflineGossip t result.pr.peerServers result.pr.expiredLocal)
910 yield (case result.ack of
911 AckUnauthorized _ -> respond {ok: False {}}
912 AckDone _ -> respond {ok: True {}})
914 LookupPresence {pubkey, authorization, respond} -> do
915 let pubkey = normalizeHex pubkey
916 t <- now
917 result <- atomic do
918 pr <- maybePrune t
919 sessions <- *sessions
920 let auth = authenticateRequest sessions authorization t
921 case auth of
922 Unauthorized {} ->
923 yield {pr: pr, authorized: False {}, foundServer: ""}
924 Authorized _ -> do
925 presence <- *presence
926 let valid = do
927 p <- presence
928 where p.pubkey == pubkey && p.expiresAt > t
929 yield p
930 let foundServer = (fold (\acc p ->
931 if acc.server == serverHostN then acc
932 else if p.server == serverHostN then {server: serverHostN, best: p.expiresAt}
933 else if p.expiresAt > acc.best then {server: p.server, best: p.expiresAt}
934 else acc) {server: "", best: 0} valid).server
935 yield {pr: pr, authorized: True {}, foundServer: foundServer}
936 forkBroadcastOffline t result.pr
937 yield (if result.authorized
938 then respond {online: not (result.foundServer == ""), server: result.foundServer, authorized: True {}}
939 else respond {online: False {}, server: "", authorized: False {}})
941 RecvGossip {events, hopsRemaining, fromServer, authorization, respond} -> do
942 t <- now
943 let authorized = verifyFederationAuth authorization
944 let cappedHops = clamp 0 5 hopsRemaining
945 let fromServer = normalizeServerName fromServer
946 if not authorized || cappedHops < 1 || fromServer == serverHostN || count events > maxGossipEvents
947 then do
948 maybePruneAndBroadcast t
949 yield (respond {ok: False {}})
950 else do
951 let allValidEntries = validGossipEvents events t
952 let onlineEntries = do e <- allValidEntries; where e.eventType == "online"; yield e
953 let offlineEntries = do e <- allValidEntries; where e.eventType == "offline"; yield e
954 let filteredOnline = do
955 e <- onlineEntries
956 where none offlineEntries (\o -> o.server == fromServer && samePubkeyServer o e && o.timestamp > e.timestamp)
957 yield e
958 let filteredOffline = do
959 e <- offlineEntries
960 where none onlineEntries (\o -> samePubkeyServer o e && o.timestamp > e.timestamp)
961 yield e
962 let trustedOffline = do
963 e <- offlineEntries
964 where e.server == fromServer
965 yield e
966 let deduped = mergeByPubkeyServer [] filteredOnline
967 gossipResult <- atomic do
968 pr <- maybePrune t
969 presence <- *presence
970 let trustedDeduped = do
971 e <- deduped
972 where e.server == fromServer || any presence (\p -> p.server == e.server && p.expiresAt > t)
973 yield e
974 let appliedOnline = do
975 e <- trustedDeduped
976 where none presence (\p -> samePubkeyServer p e && p.expiresAt >= e.expiresAt)
977 yield {pubkey: e.pubkey, server: e.server}
978 let merged = mergeByPubkeyServer presence trustedDeduped
979 let offlineOnlyKeys = do
980 e <- allValidEntries
981 where none trustedDeduped (\d -> samePubkeyServer d e)
982 yield e
983 let appliedOffline = do
984 k <- offlineOnlyKeys
985 where any presence (\p -> samePubkeyServer p k && any trustedOffline (\o -> samePubkeyServer o k && (if o.timestamp > t then t else o.timestamp) > p.timestamp))
986 yield {pubkey: k.pubkey, server: k.server}
987 let afterOffline = removeOffline merged offlineOnlyKeys trustedOffline t
988 let withinCapacity = count afterOffline <= maxPresenceEntries
989 let capResult = if withinCapacity
990 then {finalPresence: afterOffline, appliedOnline: appliedOnline}
991 else
992 let updatesOnly = do e <- trustedDeduped; where any presence (\p -> samePubkeyServer p e); yield e in
993 let updatesOnlyMerged = mergeByPubkeyServer presence updatesOnly in
994 let updatesOnlyAfterOffline = removeOffline updatesOnlyMerged offlineOnlyKeys trustedOffline t in
995 let reclaimSlots = maxPresenceEntries - count updatesOnlyAfterOffline in
996 let newEntries = do e <- trustedDeduped; where none presence (\p -> samePubkeyServer p e) && none updatesOnlyAfterOffline (\p -> samePubkeyServer p e); yield e in
997 let reclaimedEntries = takeN newEntries reclaimSlots in
998 {finalPresence: union updatesOnlyAfterOffline reclaimedEntries,
999 appliedOnline: union
1000 (do e <- updatesOnly; where none presence (\p -> samePubkeyServer p e && p.expiresAt >= e.expiresAt); yield {pubkey: e.pubkey, server: e.server})
1001 (do e <- reclaimedEntries; yield {pubkey: e.pubkey, server: e.server})}
1002 set *presence = capResult.finalPresence
1003 let gossipPeerServers = collectPeerServers capResult.finalPresence (\p -> p.server)
1004 yield {appliedKeys: union capResult.appliedOnline appliedOffline, pr: pr, gossipPeerServers: gossipPeerServers, withinCapacity: withinCapacity}
1005 let effectiveEvents = do
1006 e <- union filteredOnline filteredOffline
1007 where any gossipResult.appliedKeys (\a -> samePubkeyServer a e)
1008 yield {eventType: e.eventType, pubkey: e.pubkey, server: e.server, timestamp: e.timestamp, ttl: e.ttl}
1009 when (not gossipResult.withinCapacity)
1010 (println ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer))
1011 fork (do
1012 when (cappedHops > 1 && federationEnabled && count effectiveEvents > 0) (
1013 let targets = do
1014 s <- gossipResult.gossipPeerServers
1015 where not (s == fromServer)
1016 yield s
1017 in sendBatchedGossip targets effectiveEvents (cappedHops - 1))
1018 let pr = gossipResult.pr
1019 broadcastOfflineGossip t pr.peerServers pr.expiredLocal)
1020 yield (respond {ok: True {}})
1022 ForwardMessage {toKey, fromKey, encryptedBlob, timestamp, originServer, messageId, authorization, respond} -> do
1023 let toKey = normalizeHex toKey
1024 let fromKey = normalizeHex fromKey
1025 let encryptedBlob = normalizeHex encryptedBlob
1026 let messageId = normalizeHex messageId
1027 let originServer = normalizeServerName originServer
1028 t <- now
1029 let authorized = verifyFederationAuth authorization
1030 let validOrigin = isFederatedOrigin originServer
1031 if not (authorized && validOrigin)
1032 then do
1033 maybePruneAndBroadcast t
1034 yield (respond {ok: False {}})
1035 else do
1036 let clampedTs = clampTimestamp timestamp t
1037 result <- atomic do
1038 pr <- maybePrune t
1039 messages <- *messages
1040 ackedIds <- *ackedIds
1041 if any messages (\m -> m.messageId == messageId && m.toKey == toKey)
1042 || any ackedIds (\a -> a.messageId == messageId && a.toKey == toKey)
1043 then yield {pr: pr, ok: True {}}
1044 else if countMessagesFor messages toKey t >= maxMessagesPerRecipient
1045 then yield {pr: pr, ok: False {}}
1046 else do
1047 seqCounter <- *seqCounter
1048 let seqTs = nextSeqTs t seqCounter
1049 set *seqCounter = seqTs
1050 set *messages = union messages
1051 [{messageId: messageId, toKey: toKey, fromKey: fromKey,
1052 encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs}]
1053 yield {pr: pr, ok: True {}}
1054 forkBroadcastOffline t result.pr
1055 yield (respond {ok: result.ok})
1057 FederationAck {messageIds, originServer, fromServer, authorization, respond} -> do
1058 let fromServer = normalizeServerName fromServer
1059 let originServer = normalizeServerName originServer
1060 t <- now
1061 let authorized = verifyFederationAuth authorization
1062 if not authorized || not (originServer == serverHostN) || fromServer == serverHostN
1063 then do
1064 maybePruneAndBroadcast t
1065 yield (respond {ok: False {}})
1066 else do
1067 let dedupedIds = dedupMessageIds messageIds
1068 pruneResult <- atomic do
1069 pr <- maybePrune t
1070 forwards <- *forwards
1071 set *forwards = do
1072 f <- forwards
1073 where not (f.toServer == fromServer && any dedupedIds (\mid -> f.messageId == mid))
1074 yield f
1075 yield pr
1076 forkBroadcastOffline t pruneResult
1077 yield (respond {ok: True {}})
1079 Prune {authorization, respond} -> do
1080 t <- now
1081 let authorized = verifyMaintenanceAuth authorization
1082 when authorized (fork (pruneAndRetry t))
1083 yield (respond {ok: authorized})
1085-- ============================================
1086-- Entry Point
1087-- ============================================
1089backgroundPrune = do
1090 sleep 30000
1091 t <- now
1092 pruneAndRetry t
1093 fork backgroundPrune
1095main = do
1096 println "Skrepka relay server starting on port 8080..."
1097 fork backgroundPrune
1098 listen 8080 serve