git.iko.soy/skrepka
on 2026.06.23.0618 help sign in
skrepka/server.knot
1181 lines 63 kB history raw
1-- Skrepka Relay Server (maximally simplified)
2-- Single-server implementation of the Skrepka protocol (v0.1).
3-- Every collection is a flat list; rate limiting is delegated to the
4-- compiler's per-endpoint `rateLimit` clauses on each route.
5-- Token comparison is plain `==` (timing-attack vulnerable), retry backoff is linear,
6-- and many helpers are inlined. Performance suffers; clarity is the only goal.
8-- ============================================
9-- Configuration
10-- ============================================
11-- Top-level constants: bind address, federation identity, capacity ceilings,
12-- per-endpoint rate-limit budgets, and expiry/backoff/TTL durations.
13-- Values written here are defaults; every scalar (Int/Float/Text/Bool/Maybe ...)
14-- top-level constant can be overridden at startup via `--name=value` on the
15-- command line (see Knot's `knot_override_lookup`). Run with `--help` for the
16-- full list.
18-- This server's canonical federation identity. Used as the `server` field in
19-- presence/gossip events and for self-checks (`isLocalServer`). Not a bind
20-- address — the listener accepts connections on
21-- every interface regardless of this value. May also differ from the URL
22-- clients use to reach the server (e.g. behind a reverse proxy or alternate
23-- DNS name); peers must resolve this name to reach us for federation.
24--
25-- Must be unique per deployment. Every operator participating in federation
26-- has to recompile with their own public hostname here — peers reject events
27-- whose origin fails `isBadServerName` (which blocks the default "localhost"
28-- and other private/reserved names), so two default-built servers cannot
29-- federate with each other. The default is only suitable for a single-node,
30-- non-federated deployment.
31serverHost : ServerName
32serverHost = "localhost"
33-- Network interface the HTTP listener binds to. "0.0.0.0" accepts connections
34-- on every IPv4 interface (the typical default behind a reverse proxy);
35-- "127.0.0.1" restricts the listener to localhost-only; "::" / "::1" do the
36-- IPv6 equivalents. Override with `--bindHost=...` at startup. Unrelated to
37-- `serverHost` above — this is the kernel-level bind address; that is the
38-- federation identity peers see.
39bindHost : Text
40bindHost = "0.0.0.0"
41-- TCP port for incoming HTTP connections.
42listenPort : Int
43listenPort = 8080
44-- URL scheme used when contacting peer federation servers.
45federationScheme : Text
46federationScheme = "https://"
48-- Storage retention for messages and dedup-id records. Mailbox rows live this
49-- long after `receivedAt`; expired rows are pruned by `pruneExpired` (30 days).
50messageMaxAge : Int<Ms>
51messageMaxAge = 2592000000
52-- Max length of a hex-encoded message body / opaque blob (40 MiB hex ≈ 20 MiB bytes).
53maxBlobLen : Int
54maxBlobLen = 41943040
55-- Cap on items in a batched send request.
56maxBatchSize : Int
57maxBatchSize = 100
58-- Cap on gossip events accepted or emitted in a single federation exchange.
59maxGossipEvents : Int
60maxGossipEvents = 100
61-- Cap on events returned by one long-poll response.
62maxPollEvents : Int
63maxPollEvents = 50
64-- Per-peer ceiling on outstanding forward attempts.
65maxForwardsPerServer : Int
66maxForwardsPerServer = 2000
67-- Max accepted length of a server hostname.
68maxServerNameLen : Int
69maxServerNameLen = 128
70-- Ceiling on outstanding auth challenges held in memory.
71maxChallenges : Int
72maxChallenges = 10000
73-- Ceiling on concurrent active sessions.
74maxSessions : Int
75maxSessions = 10000
76-- Per-pubkey ceiling on outstanding challenges (prevents one key from exhausting the pool).
77maxChallengesPerKey : Int
78maxChallengesPerKey = 5
79-- Per-pubkey ceiling on concurrent sessions.
80maxSessionsPerKey : Int
81maxSessionsPerKey = 10
82-- Validity window for an auth challenge before it must be re-issued (1 min).
83challengeExpiry : Int<Ms>
84challengeExpiry = 60000
85-- How long a session token remains valid after issue (1 hour).
86sessionExpiry : Int<Ms>
87sessionExpiry = 3600000
88-- How long a received "online" presence gossip is considered fresh (90 min).
89-- Receivers anchor the TTL to their own wall clock at receive time, so this
90-- value isn't carried on the wire.
91onlineGossipTtl : Int<Ms>
92onlineGossipTtl = 5400000
93-- Times to retry a federation forward before giving up.
94maxForwardRetries : Int
95maxForwardRetries = 10
97-- Base delay for federation retry backoff (linear, not exponential — 30 s).
98retryBackoffBase : Int<Ms>
99retryBackoffBase = 30000
100-- Cap on federation retry backoff (8 min).
101retryBackoffMax : Int<Ms>
102retryBackoffMax = 480000
103-- How long a peer's failure record survives past its most recent failure (24 h).
104-- After this, accumulated backoff resets — the next attempt is treated as fresh.
105forwardFailureMaxAge : Int<Ms>
106forwardFailureMaxAge = 86400000
108-- Cadence of the background prune/cleanup task (30 s).
109backgroundPruneInterval : Int<Ms>
110backgroundPruneInterval = 30000
111-- Long-poll hold timeout before returning empty to the client (25 s).
112pollTimeoutMs : Int<Ms>
113pollTimeoutMs = 25000
115-- Rolling window for every per-endpoint rate-limit bucket (1 min). All routes
116-- share this window; only the per-window request budget varies per endpoint.
117rateLimitWindow : Int<Ms>
118rateLimitWindow = 60000
119-- Per-pubkey budget for /auth/challenge and /auth/verify per `rateLimitWindow`.
120-- The normal handshake is one challenge + one verify, so a healthy flow uses 2.
121authRateLimit : Int
122authRateLimit = 10
123-- Per-client-IP budget for /poll per `rateLimitWindow`. Clients reconnect
124-- after `pollTimeoutMs`, so steady state is ~3/min per client; the budget
125-- leaves room for several clients behind one NAT. (Keyed on IP, not bearer
126-- token — token-keyed buckets let an attacker mint a fresh bucket per
127-- invented bearer value.)
128pollRateLimit : Int
129pollRateLimit = 60
130-- Per-client-IP budget for /messages per `rateLimitWindow`. Each call may
131-- carry up to `maxBatchSize` ciphertexts, so the effective message ceiling
132-- is higher.
133sendRateLimit : Int
134sendRateLimit = 60
135-- Per-peer-IP budget for /federation/gossip per `rateLimitWindow`.
136gossipRateLimit : Int
137gossipRateLimit = 60
138-- Per-peer-IP budget for /federation/forward per `rateLimitWindow`. Higher than
139-- gossip because peers fan out one call per buffered ciphertext.
140forwardRateLimit : Int
141forwardRateLimit = 600
142-- Whether rate limiting may trust X-Forwarded-For for the client address.
143-- True matches the install.sh deployment (Caddy in front, which appends the
144-- real client address as the header's final entry — the only entry a trusted
145-- proxy vouches for). Set --trustForwardedFor=False when clients connect
146-- directly, otherwise a client could spoof the header to mint fresh
147-- rate-limit buckets.
148trustForwardedFor : Bool
149trustForwardedFor = True {}
151-- Master switch for federation; when False the server runs in single-node mode.
152federationEnabled : Bool
153federationEnabled = True {}
154federationUrl : Text -> Text
155federationUrl = \server -> federationScheme ++ server
157-- ============================================
158-- Types
159-- ============================================
160-- Wire/state shapes plus refinement aliases for hex/server-name validation.
161-- Hex types are bounded length + lowercase-hex; server names are lowercased
162-- and stripped of trailing dots. The SSRF deny-list (`isBadServerName`) is
163-- applied at federation egress, not in the type predicate.
165-- One event in a /poll response: the opaque cipher blob addressed to the
166-- polling pubkey. The server cannot read the blob.
167type PollEvent = {encryptedBlob: BlobHex}
169-- One presence change exchanged between federated servers. A flat wire record
170-- (`{"eventType": "online"|"offline", "pubkey": "<hex>"}`) rather than an ADT:
171-- the built-in record JSON round-trips it directly (a `Just` pubkey encodes as
172-- the bare hex string, `Nothing` as `null`/an absent field). `eventType` is
173-- free `Text`, so an unrecognized future event kind decodes cleanly instead of
174-- failing the whole batch, and `pubkey` is optional so a future pubkey-less
175-- kind still parses. `handleRecvGossip` ignores any event whose `eventType` it
176-- doesn't recognize or that carries no pubkey — that is the forward-compat
177-- fallback (the old `GossipUnknown` variant).
178type GossipEvent = {eventType: Text, pubkey: Maybe PubkeyHex}
180-- /messages is all-or-nothing per call: either every message in the batch is
181-- enqueued (delivered, federated, or buffered for federation), or none are
182-- and the call returns a non-2xx HTTP status. The per-message rejection code
183-- "self_send" surfaces in the HTTP error body. Per-client-IP rate limiting is
184-- enforced by the route's `rateLimit` clause, which short-circuits with 429
185-- before the handler runs.
186-- The sender's UI treats every success the same and learns "they got it"
187-- only via the recipient's E2E ack.
189-- True when `s` is non-empty lowercase hex of even length, with `maxLen` ceiling.
190isValidBoundedHex : Int -> Text -> Bool
191isValidBoundedHex = \maxLen s -> length s <= maxLen && isValidHex s
192-- True when `s` is lowercase hex of exactly `exactLen` characters (even by definition).
193isValidExactHex : Int -> Text -> Bool
194isValidExactHex = \exactLen s -> length s == exactLen && isValidHex s
195-- Hex-encoded Ed25519 public key (32 bytes → 64 chars exact).
196type PubkeyHex = Text where isValidExactHex 64
197-- Hex-encoded ciphertext or opaque blob; capped at `maxBlobLen` (~20 MiB).
198type BlobHex = Text where isValidBoundedHex maxBlobLen
199-- Hex-encoded Ed25519 signature (64 bytes → 128 chars exact).
200type SignatureHex = Text where isValidExactHex 128
201-- Normalized hostname (lowercased, no trailing dot, length-bounded). The SSRF
202-- deny-list (`isBadServerName`) is enforced separately at federation egress
203-- (`handleRecvGossip`); keeping it out of the type predicate lets the default
204-- `serverHost = "localhost"` flow through presence rows in a non-federated
205-- single-node deployment.
206isValidServerName : Text -> Bool
207isValidServerName = \s -> length s <= maxServerNameLen && s == toLower (stripTrailingDot s)
208type ServerName = Text where isValidServerName
210-- Wall-clock time in milliseconds since epoch.
211type Timestamp = Int<Ms>
213-- Outstanding nonce issued by /auth/challenge; consumed by /auth/verify.
214type Challenge = {pubkey: PubkeyHex, challenge: Text, expiresAt: Timestamp, ip: Text}
215-- An issued bearer token bound to (pubkey, ip, expiry).
216type Session = {token: Text, pubkey: PubkeyHex, expiresAt: Timestamp, ip: Text}
217-- Mailbox row: a stored message awaiting recipient pickup or federation forward.
218-- Storage TTL is fixed at `messageMaxAge`, so expiry is derived from
219-- `receivedAt`. Any send time the sender wants to convey lives inside
220-- `encryptedBlob` (the server cannot read it). `deliveredTo` records peers
221-- that have already accepted a federated forward of this blob, so a later
222-- presence re-announce of `toKey` doesn't trigger a duplicate delivery.
223-- (Single-field record rows, not bare names: the Knot runtime cannot persist
224-- a nested relation of scalars.)
225type Message = {toKey: PubkeyHex, encryptedBlob: BlobHex, receivedAt: Timestamp, deliveredTo: [{server: ServerName}]}
226-- Last-known location of a pubkey, either local (`server == serverHost`) or
227-- learned from a peer via gossip. Multiple entries per pubkey are allowed.
228type PresenceEntry = {pubkey: PubkeyHex, server: ServerName, expiresAt: Timestamp}
229-- A message queued for cross-server delivery to `toServer`. `retries` is
230-- incremented by `retryPendingForwards` until `maxForwardRetries` is hit.
231-- Identity is derived from `encryptedBlob` (see `Message`).
232type ForwardEntry = {toServer: ServerName, toKey: PubkeyHex, encryptedBlob: BlobHex, retries: Int}
233-- Running failure counter for a peer; keyed by `server`. Drives `shouldRetryServer`.
234-- `failedAt` is the start time of the last recorded attempt; `failures: 0` is
235-- the post-success tombstone (no backoff, kept so the timestamp can outrank
236-- slower concurrent failures).
237type ForwardFailure = {server: Text, failedAt: Timestamp, failures: Int}
239-- ============================================
240-- Persisted State (all flat lists, no sharding)
241-- ============================================
242-- Mutable refs (`*x`) holding the entire server state in memory. Every list
243-- scan is O(n); scaling is the first thing the production version replaces.
244-- Grouped here: auth (challenges/sessions), mailbox (messages/forwards),
245-- presence, and bookkeeping counters. Per-endpoint rate-limit buckets live
246-- in the runtime's hidden `_knot_rate_limits` SQLite table and don't appear
247-- here.
248-- The Knot runtime persists every `*ref` declared here through restarts (see
249-- the project memory note about SQLite-backed refs); nothing in this file is
250-- truly volatile despite being typed as plain in-memory lists.
252-- Outstanding /auth/challenge nonces awaiting /auth/verify. Capped overall and per-pubkey.
253*challenges : [Challenge]
254-- Live bearer tokens. Source of truth for `withSessionAuth`; trimmed by `pruneExpired`.
255*sessions : [Session]
256-- Mailbox: ciphertexts addressed to local recipients, awaiting poll or federation forward.
257*messages : [Message]
258-- Last-known location of each pubkey (local sessions and gossiped remote presence).
259*presence : [PresenceEntry]
260-- Cross-server outbox: messages queued for delivery to a peer, with retry counts.
261*forwards : [ForwardEntry]
262-- Per-peer failure counters; drives `shouldRetryServer`.
263*forwardFailures : [ForwardFailure]
265-- Strictly-monotonic sequence used as `Message.receivedAt`; guarantees
266-- poll-cursor ordering. Held as a 0-or-1-row wrapped relation, not a bare
267-- `Timestamp` scalar source: the runtime now panics when a scalar source's
268-- backing table is empty (it treats a missing row as corruption rather than
269-- defaulting to 0), which would 500 the very first /poll or /messages on a
270-- fresh database. Read through `currentSeq` (0 default). The `{value: ...}`
271-- record wrapper is required because a relation of bare scalars (`[Timestamp]`)
272-- also can't be persisted — same reason `Message.deliveredTo` is wrapped.
273*seqCounter : [{value: Timestamp}]
275-- ============================================
276-- Rate limiting
277-- ============================================
278-- Per-endpoint token-bucket configuration. Auth endpoints key on the submitted
279-- pubkey (so one keypair can't drain global capacity); client and federation
280-- endpoints key on the client IP (last X-Forwarded-For hop behind a trusted
281-- proxy, else the connection address). The bearer token is deliberately not a
282-- bucket key: the limiter runs before the handler, so it can't tell valid
283-- sessions from invented ones, and an attacker varying the token would mint a
284-- fresh bucket per value. Each route has an independent bucket store, so endpoints
285-- using the same key strategy don't share budget across routes. Buckets persist
286-- in the runtime's `_knot_rate_limits` SQLite table; rejection responds 429
287-- before the handler runs.
289-- Suffix of `s` after its last comma, trimmed. X-Forwarded-For is a
290-- comma-separated chain that each proxy appends to; only the final entry —
291-- the one written by the nearest trusted proxy — is reliable for bucketing.
292lastForwardedHop : Text -> Text
293lastForwardedHop = \s -> if contains "," s
294then lastForwardedHop (drop 1 s)
295else trim s
297-- The client address for rate-limit bucketing: the last X-Forwarded-For hop
298-- when `trustForwardedFor` is set (and the header is present and non-empty),
299-- otherwise the TCP connection address.
300pickClientIp : Maybe Text -> Text -> Text
301pickClientIp = \h connIp -> case h of
302 Just {value: ips} -> if trustForwardedFor && lastForwardedHop ips != ""
303 then lastForwardedHop ips
304 else connIp
305 Nothing {} -> connIp
307byPubkey = \{pubkey} _ -> Just {value: pubkey}
308byClientIp = \_ ctx -> Just {value: pickClientIp (ctx.header "X-Forwarded-For") ctx.clientIp}
310authLimit = {key: byPubkey, limit: {requests: authRateLimit, window: rateLimitWindow}}
311pollLimit = {key: byClientIp, limit: {requests: pollRateLimit, window: rateLimitWindow}}
312sendLimit = {key: byClientIp, limit: {requests: sendRateLimit, window: rateLimitWindow}}
313gossipLimit = {key: byClientIp, limit: {requests: gossipRateLimit, window: rateLimitWindow}}
314forwardLimit = {key: byClientIp, limit: {requests: forwardRateLimit, window: rateLimitWindow}}
316-- ============================================
317-- Routes
318-- ============================================
319-- Public HTTP surface, grouped by audience:
320-- AuthApi -- challenge/verify (client identity)
321-- ClientApi -- /poll, /messages (send)
322-- FederationApi -- inter-server gossip, forward (single + batch)
323-- Handlers return `Result HttpError T`: `Ok` produces a 200 with the
324-- declared body shape, `Err` produces a non-2xx response whose body is
325-- `{"error": "<code>"}`.
327route AuthApi where
328 POST {pubkey: PubkeyHex} /auth/challenge headers {xForwardedFor: Maybe Text} -> {challenge: Text, expiresAt: Timestamp} rateLimit authLimit = ReqChallenge
329 POST {pubkey: PubkeyHex, challenge: Text, signature: SignatureHex, revokeOthers: Bool} /auth/verify headers {xForwardedFor: Maybe Text} -> {token: Text, expiresAt: Timestamp} rateLimit authLimit = VerifyAuth
331route ClientApi where
332 POST {cursor: Timestamp} /poll headers {authorization: Text, xForwardedFor: Maybe Text} -> {events: [PollEvent], cursor: Timestamp} rateLimit pollLimit = Poll
333 POST {messages: [{to: PubkeyHex, encryptedBlob: BlobHex}]} /messages headers {authorization: Text, xForwardedFor: Maybe Text} -> {} rateLimit sendLimit = SendMessages
335route FederationApi where
336 POST {events: [GossipEvent], fromServer: ServerName} /federation/gossip -> {} rateLimit gossipLimit = RecvGossip
337 POST {toKey: PubkeyHex, encryptedBlob: BlobHex} /federation/forward -> {} rateLimit forwardLimit = ForwardMessage
339route Api = AuthApi | ClientApi | FederationApi
341-- ============================================
342-- Generic helpers
343-- ============================================
344-- List/predicate combinators (filter/all/findFirst/...), token generation,
345-- and small text/server-name utilities used throughout the rest of the file.
346-- Everything is intentionally O(n) over flat lists; performance is sacrificed
347-- for clarity per the file header.
349-- Cap to maxN by dropping the soonest-to-expire entries (LRU-style). Always
350-- sorts by `expiresAt` so the oldest/most-stale entries are evicted first.
351capList : [{expiresAt: Timestamp | r}] -> Int -> [{expiresAt: Timestamp | r}]
352capList = \items maxN -> drop (max 0 (count items - maxN)) (sortBy (\x -> x.expiresAt) items)
354-- Cap entries matching `pubkey` to maxN-1 (caller adds one more). Splits the
355-- list into "different pubkey" (untouched) and "same pubkey" (capped +
356-- LRU-evicted) so a single key can't exhaust a global pool — see
357-- `maxChallengesPerKey` and `maxSessionsPerKey`.
358capPerKey : [{pubkey: PubkeyHex, expiresAt: Timestamp | r}] -> PubkeyHex -> Int -> [{pubkey: PubkeyHex, expiresAt: Timestamp | r}]
359capPerKey = \items pubkey maxN -> union (filter (\x -> not (x.pubkey == pubkey)) items) (capList (filter (\x -> x.pubkey == pubkey) items) (maxN - 1))
361-- Random hex token: one random hex char per recursive call. `n` is the
362-- desired length in characters (so n=48 yields 24 random bytes of entropy).
363-- Session bearers, /auth challenges, and federation auth nonces all use n=48
364-- (192 bits of entropy, well beyond birthday bound).
365genToken : Int -> IO {random} Text
366genToken = \n -> if n <= 0
367then yield ""
368else do
369 d <- randomInt 16
370 rest <- genToken (n - 1)
371 yield (take 1 (drop d "0123456789abcdef") ++ rest)
373-- True when `s` is non-empty, even-length, and only lowercase hex digits.
374-- `chars` yields the *distinct* character set (relations dedupe), which is
375-- fine for the membership check; the length checks must use `length`.
376isValidHex : Text -> Bool
377isValidHex = \s -> do
378 let n = length s
379 n > 0 && n / 2 * 2 == n && all (\c -> contains c "0123456789abcdef") (chars s)
381-- Recursively strip every trailing "." from `s` (DNS root form).
382stripTrailingDot : Text -> Text
383stripTrailingDot = \s -> if hasSuffix "." s
384then stripTrailingDot (take (length s - 1) s)
385else s
387-- True when `s` refers to this server.
388isLocalServer : Text -> Bool
389isLocalServer = \s -> s == serverHost
391-- True when `s` starts with `pre`.
392hasPrefix : Text -> Text -> Bool
393hasPrefix = \pre s -> take (length pre) s == pre
395-- True when `s` ends with `suffix`.
396hasSuffix : Text -> Text -> Bool
397hasSuffix = \suffix s -> drop (max 0 (length s - length suffix)) s == suffix
399-- Occurrences of the single character `c` in `s`. (`chars` can't be used for
400-- this — it returns the distinct character *set*.)
401countChar : Text -> Text -> Int
402countChar = \c s -> if s == ""
403then 0
404else (if take 1 s == c then 1 else 0) + countChar c (drop 1 s)
406-- The host portion of `sn` — everything before the first `:`. With the
407-- one-colon ceiling enforced in `isBadServerName`, this strips an optional
408-- `:port` so `isNumericHost` sees just the host.
409beforeColon : Text -> Text
410beforeColon = \s -> if s == "" || take 1 s == ":"
411then ""
412else take 1 s ++ beforeColon (drop 1 s)
414-- True when the host part is a bare integer (no dots) — a decimal or octal
415-- "dword" IP like `2130706433` or `017700000001` (both = 127.0.0.1). These
416-- carry no dotted octets, so the prefix list below never matches them, yet
417-- they resolve to the same internal addresses.
418isNumericHost : Text -> Bool
419isNumericHost = \sn -> do
420 let host = beforeColon sn
421 host != "" && all (\c -> contains c "0123456789") (chars host)
423-- Decimal value of a single hex/ascii digit character.
424digitValue : Text -> Int
425digitValue = \c -> if c == "0" then 0 else if c == "1" then 1 else if c == "2" then 2 else if c == "3" then 3 else if c == "4" then 4 else if c == "5" then 5 else if c == "6" then 6 else if c == "7" then 7 else if c == "8" then 8 else 9
427-- Value of an all-digits string (e.g. "172" -> 172); -1 when empty or
428-- non-numeric. `chars` is fine for the membership guard (set semantics don't
429-- matter there); the actual parse walks the string in order via take/drop.
430decimalValue : Text -> Int
431decimalValue = \s -> if s == "" || not all (\c -> contains c "0123456789") (chars s)
432then -1
433else decimalValueGo 0 s
434decimalValueGo : Int -> Text -> Int
435decimalValueGo = \acc s -> if s == ""
436then acc
437else decimalValueGo (acc * 10 + digitValue (take 1 s)) (drop 1 s)
439-- Substring of `s` before / after its first ".". `afterDot` is empty when
440-- there is no dot. Used to pull the first two octets of a dotted IPv4 literal.
441beforeDot : Text -> Text
442beforeDot = \s -> if s == "" || take 1 s == "."
443then ""
444else take 1 s ++ beforeDot (drop 1 s)
445afterDot : Text -> Text
446afterDot = \s -> if s == ""
447then ""
448else if take 1 s == "."
449then drop 1 s
450else afterDot (drop 1 s)
452-- True when `host` is composed solely of digits and dots, i.e. a bare IPv4
453-- literal rather than a hostname. Gates the numeric range checks below so a
454-- legitimately-named peer like "224.example.com" is not misread as multicast.
455isDottedNumeric : Text -> Bool
456isDottedNumeric = \host -> host != "" && contains "." host && all (\c -> contains c "0123456789.") (chars host)
458-- Reserved/non-routable IPv4 ranges the dotted-octet prefix list misses:
459-- CGNAT 100.64/10, benchmarking 198.18/15, and multicast + reserved/broadcast
460-- (224/4 and 240/4, i.e. any first octet >= 224). Applied only to bare IPv4
461-- literals (`isDottedNumeric`) so hostnames are never caught. Shorthand forms
462-- like "100.64" still match — only the first two octets are inspected.
463isReservedIPv4 : Text -> Bool
464isReservedIPv4 = \host -> do
465 let o1 = decimalValue (beforeDot host)
466 let o2 = decimalValue (beforeDot (afterDot host))
467 isDottedNumeric host && (o1 >= 224 || (o1 == 100 && o2 >= 64 && o2 <= 127) || (o1 == 198 && (o2 == 18 || o2 == 19)))
469-- True when a dotted-numeric host carries an octet with a redundant leading
470-- zero (e.g. "0177", "0012"). Canonical IPv4 literals never do, but the system
471-- resolver behind `fetch` (ureq -> getaddrinfo / inet_aton) reads a leading-zero
472-- octet as OCTAL: "0012.0.0.1" dials 10.0.0.1 and "0177.0.0.1" dials 127.0.0.1,
473-- while the decimal octet parse in `isReservedIPv4` (and the dotted-prefix list)
474-- sees 12.0.0.1 / 177.0.0.1 and waves them through — an SSRF bypass of the same
475-- class as the non-dotted "dword"/`0x` forms. Reject the ambiguous octal form
476-- outright. Gated by `isDottedNumeric` so a digit-led hostname label like
477-- "0foo.example.com" is never caught (it isn't all-digits-and-dots).
478hasLeadingZeroOctet : Text -> Bool
479hasLeadingZeroOctet = \host -> isDottedNumeric host && anyOctetLeadingZero host
480anyOctetLeadingZero : Text -> Bool
481anyOctetLeadingZero = \host -> do
482 let octet = beforeDot host
483 let rest = afterDot host
484 (length octet > 1 && take 1 octet == "0") || (rest != "" && anyOctetLeadingZero rest)
486-- Block private/reserved/internal hostnames (SSRF defense). Skips percent-encoding
487-- tricks but covers the high-value cases: `localhost`, RFC1918 ranges (10/8,
488-- 172.16/12, 192.168/16) plus link-local
489-- 169.254/16, the IPv4 zero net, CGNAT 100.64/10, benchmarking 198.18/15,
490-- the TEST-NET ranges 192.0.2/24 and 203.0.113/24, and multicast/reserved
491-- 224/4+240/4 (any first octet >= 224, via `isReservedIPv4`), bare IPv6
492-- literals (any string with more than one `:` — covers loopback
493-- `::`/`::1`, link-local `fe80::/10`, ULA `fc00::/7`, globals like `2001:db8::1`),
494-- bracketed IPv6 literals, integer/octal "dword" IPv4 (`isNumericHost`), the
495-- `0x`-prefixed hex form, octal dotted-octet IPv4 (any leading-zero octet, via
496-- `hasLeadingZeroOctet` — the resolver reads "0177.0.0.1" as 127.0.0.1), and the
497-- `.local` / `.internal` / `.arpa` / `.onion`
498-- suffixes. Enforced on inbound requests (`handleRecvGossip`, so a peer can't
499-- claim a bogus `fromServer`) and again at every federation egress site
500-- (`sendBatchedGossip`, `retryPendingForwards`, `forkForwardsTo`) as
501-- defense-in-depth — state persisted before a denylist tightening, or any
502-- future ingress gap, must not steer a `fetch` at an internal host.
503-- A legitimate `host:port` server name has exactly one `:`, so the multi-colon
504-- check leaves that case intact.
505isBadServerName : Text -> Bool
506isBadServerName = \sn -> sn == "" || sn == "localhost" || hasPrefix "localhost:" sn || isNumericHost sn || isReservedIPv4 (beforeColon sn) || hasLeadingZeroOctet (beforeColon sn) || hasPrefix "0x" sn || any (\c -> not contains c "0123456789abcdefghijklmnopqrstuvwxyz-.:") (chars sn) || countChar ":" sn > 1 || any (\p -> hasPrefix p sn) [
507 ".",
508 ":",
509 "[",
510 "127.",
511 "10.",
512 "192.168.",
513 "169.254.",
514 "192.0.2.",
515 "203.0.113.",
516 "0.",
517 "172.16.",
518 "172.17.",
519 "172.18.",
520 "172.19.",
521 "172.20.",
522 "172.21.",
523 "172.22.",
524 "172.23.",
525 "172.24.",
526 "172.25.",
527 "172.26.",
528 "172.27.",
529 "172.28.",
530 "172.29.",
531 "172.30.",
532 "172.31."
533] || any (\s -> hasSuffix s sn) [".localhost", ".local", ".internal", ".arpa", ".onion"]
535-- ============================================
536-- Auth helpers
537-- ============================================
538-- Bearer token extraction and session lookup. Authenticated handlers funnel
539-- through `authedPubkey`; per-endpoint rate limiting is enforced by the
540-- compiler-generated `rateLimit` clauses on each route.
541-- Federation requests are unauthenticated: anyone can hit /federation/*.
542-- The `no_presence` check bounds federated forwards to currently-online keys.
544-- Plain ip match (sessions with empty ip skip the check).
545matchIp : Text -> Text -> Bool
546matchIp = \clientIp sessionIp -> sessionIp == "" || clientIp == sessionIp
548-- The client IP a session/challenge is bound to: the last (trusted) hop of the
549-- X-Forwarded-For chain, matching the rate limiter's `pickClientIp`. Taking the
550-- last hop rather than the whole header means client-supplied spoofed hops don't
551-- become part of the binding (a trusted proxy appends the real address last).
552-- Empty when the header is absent or `trustForwardedFor` is off — handlers have
553-- no access to the raw connection address, and an empty bound IP makes `matchIp`
554-- skip the check (so IP binding is simply disabled without a trusted proxy).
555clientIpOf : Maybe Text -> Text
556clientIpOf = \h -> case h of
557 Just {value: ips} -> if trustForwardedFor then lastForwardedHop ips else ""
558 Nothing {} -> ""
560-- Resolve a Bearer token to its session pubkey if the session is unexpired and
561-- the client IP matches. Returns Nothing on missing/invalid token.
562authedPubkey : Text -> Int<Ms> -> Text -> IO {r *sessions} (Maybe PubkeyHex)
563authedPubkey = \authorization t clientIp -> do
564 ses <- *sessions
565 if not hasPrefix "Bearer " authorization
566 then yield (Nothing {})
567 else do
568 let token = drop 7 authorization
569 yield (case findFirst ses (\s -> s.expiresAt > t && s.token == token && matchIp clientIp s.ip) of
570 Just {value: s} -> Just {value: s.pubkey}
571 _ -> Nothing {})
573-- ============================================
574-- Federation failure tracking (simple, no circuit breaker)
575-- ============================================
576-- Linear backoff per peer: each consecutive failure multiplies the delay by
577-- `retryBackoffBase` (capped at `retryBackoffMax`). `shouldRetryServer` is the
578-- single read-side gate; `recordFailure`/`clearFailure` are the write side.
579-- There is no half-open state — a single success clears all accumulated failures.
580-- Concurrent fetches to the same peer can complete out of order, so both
581-- writers take the attempt-start time and ignore updates older than the
582-- recorded one (latest-started attempt wins). `clearFailure` therefore writes
583-- a zero-count row rather than deleting — the timestamp must survive so a
584-- slower failure can be recognized as stale. Zero-count rows impose no
585-- backoff and age out of `*forwardFailures` via `pruneExpired`.
587-- True when enough time has passed since the last failure for `server` to be
588-- worth retrying (or no failure recorded at all). Linear backoff:
589-- `retryBackoffBase * failures`, capped at `retryBackoffMax`.
590shouldRetryServer : Text -> Timestamp -> [ForwardFailure] -> Bool
591shouldRetryServer = \server t failures -> not any (\f -> f.server == server && t - f.failedAt < min retryBackoffMax (retryBackoffBase * (f.failures : Int))) failures
593-- Bump the failure counter for `server`. `attemptAt` is when the failed call
594-- started; the update is skipped when a more recently started attempt has
595-- already written its outcome.
596recordFailure : Text -> Timestamp -> IO {rw *forwardFailures} {}
597recordFailure = \server attemptAt -> atomic (do
598 failures <- *forwardFailures
599 let prev = findFirst failures (\f -> f.server == server)
600 let stale = case prev of
601 Just {value: f} -> f.failedAt > attemptAt
602 _ -> False {}
603 when (not stale) (do
604 let prevCount = case prev of
605 Just {value: f} -> f.failures
606 _ -> 0
607 *forwardFailures = union (filter (\f -> f.server != server) failures) [
608 {server, failedAt: attemptAt, failures: prevCount + 1}
609 ]))
611-- Reset the failure counter for `server` to zero (called after any successful
612-- peer call; single-success recovery — there is no half-open state). Keeps a
613-- zero-count row so its timestamp can outrank slower concurrent failures.
614clearFailure : Text -> Timestamp -> IO {rw *forwardFailures} {}
615clearFailure = \server attemptAt -> atomic (do
616 failures <- *forwardFailures
617 let stale = any (\f -> f.server == server && f.failedAt > attemptAt) failures
618 when (not stale) (do
619 *forwardFailures = union (filter (\f -> f.server != server) failures) [
620 {server, failedAt: attemptAt, failures: 0}
621 ]))
623-- ============================================
624-- Common helpers
625-- ============================================
626-- Helpers shared by send/poll/federation paths: monotonic sequence
627-- generation, presence queries, and atomic mailbox/presence writes
628-- (callers supply the surrounding `atomic` block).
630-- True when `pk` has a live local presence row (i.e. is currently online here).
631hasLocalPresence : [PresenceEntry] -> PubkeyHex -> Timestamp -> Bool
632hasLocalPresence = \pres pk t -> any (\p -> p.pubkey == pk && isLocalServer p.server && p.expiresAt > t) pres
634-- Project items through `getServer`, drop our own hostname, and dedupe.
635-- Used to derive peer recipient lists from forwards/presence/etc.
636collectPeers : [a] -> (a -> Text) -> [Text]
637collectPeers = \items getServer -> filter (\s -> not isLocalServer s) (map getServer items)
639-- Current sequence value, or 0 before the first message seeds it. Reads the
640-- 0-or-1-row `*seqCounter` relation and folds out the value with a 0 default,
641-- rather than unwrapping a (possibly empty) scalar source — which the runtime
642-- panics on.
643currentSeq : [{value: Timestamp}] -> Timestamp
644currentSeq = \rows -> fold (\acc r -> max acc r.value) (0 : Timestamp) rows
646-- Append one ciphertext to the local mailbox under a fresh monotonic
647-- sequence. Row-level STM retry on `*messages` wakes any parked /poll
648-- transaction whose filter matches the new row. Caller wraps in `atomic`.
649appendMessage : PubkeyHex -> BlobHex -> Int<Ms> -> IO {rw *messages, rw *seqCounter} {}
650appendMessage = \toKey encryptedBlob t -> do
651 msgs <- *messages
652 seqRows <- *seqCounter
653 let seqTs = max t (currentSeq seqRows + 1)
654 *seqCounter = [{value: seqTs}]
655 *messages = union msgs [{toKey, encryptedBlob, receivedAt: seqTs, deliveredTo: []}]
657-- True when `f` is the queued forward of `blob` to `server`. Used to dedupe
658-- per-(blob, server) entries on the forwards outbox.
659forwardMatches : BlobHex -> ServerName -> ForwardEntry -> Bool
660forwardMatches = \blob server f -> f.encryptedBlob == blob && f.toServer == server
662-- Replace any local presence rows for `pubkey` with one fresh row at
663-- `expiry`, and return the post-update peer broadcast set. Caller wraps in
664-- `atomic`.
665refreshLocalPresence : PubkeyHex -> Timestamp -> IO {rw *presence} [Text]
666refreshLocalPresence = \pubkey expiry -> do
667 pres <- *presence
668 *presence = union (filter (\p -> not (p.pubkey == pubkey && isLocalServer p.server)) pres) [
669 {pubkey, server: serverHost, expiresAt: expiry}
670 ]
671 yield (collectPeers pres (\p -> p.server))
673-- ============================================
674-- Pruning
675-- ============================================
676-- Single sweep that drops expired entries from every state ref in one atomic
677-- block. Returns the locally-expired presence rows whose pubkey has no other
678-- live local session, so the caller can broadcast offline gossip for them.
680-- One sweep over every state ref: drops expired rows in a single atomic block.
681-- Returns the locally-expired presence rows whose pubkey has no other live
682-- local session, so the caller can broadcast offline gossip for them.
683pruneExpired : Int<Ms> -> IO {rw *challenges, rw *forwardFailures, rw *messages, rw *presence, rw *sessions} [PubkeyHex]
684pruneExpired = \t -> atomic do
685 ch <- *challenges
686 *challenges = filter (\x -> x.expiresAt > t) ch
687 ses <- *sessions
688 *sessions = filter (\x -> x.expiresAt > t) ses
689 msgs <- *messages
690 *messages = filter (\m -> t - m.receivedAt < messageMaxAge) msgs
691 ff <- *forwardFailures
692 *forwardFailures = filter (\f -> t - f.failedAt < forwardFailureMaxAge) ff
693 -- presence: remember locally-expired keys for offline gossip
694 pres <- *presence
695 *presence = filter (\p -> p.expiresAt > t) pres
696 let liveLocalPubkeys = map (\p -> p.pubkey) (filter (\p -> isLocalServer p.server && p.expiresAt > t) pres)
697 yield (map (\p -> p.pubkey) (filter (\p -> p.expiresAt <= t && isLocalServer p.server && not elem p.pubkey liveLocalPubkeys) pres))
699-- ============================================
700-- Federation gossip
701-- ============================================
702-- Construction and outbound dispatch of presence gossip events. Failures from
703-- a peer record into `forwardFailures` (shared with forward retries) so both
704-- subsystems back off together.
706-- Fan out one bundle of gossip events to each peer, skipping denylisted
707-- peers and peers in backoff. No-op when `events` is empty so callers don't
708-- need to guard. Records or clears federation-failure counters per call.
709sendBatchedGossip : [Text] -> [GossipEvent] -> IO {rw *forwardFailures, network, clock} {}
710sendBatchedGossip = \peers events -> when (count events > 0) (do
711 t <- now
712 failures <- *forwardFailures
713 forEach peers (\server -> when (not (isBadServerName server) && shouldRetryServer server t failures) (do
714 result <- fetch (federationUrl server) (RecvGossip {events, fromServer: serverHost})
715 case result of
716 Err _ -> recordFailure server t
717 Ok _ -> clearFailure server t)))
719-- ============================================
720-- Forward retries
721-- ============================================
722-- `retryPendingForwards` walks queued cross-server forwards in batches per
723-- destination, increments retry counts, and gives up after `maxForwardRetries`.
724-- `forwardQueuedToServer` is the catch-up path triggered when a remote key
725-- comes online: any messages already buffered for that pubkey are enqueued
726-- as forwards to the newly-known server.
728-- Walk pending forwards grouped by destination, retry each (up to maxBatchSize
729-- per peer) using `/federation/forward`. Drops successful entries and records
730-- the peer on the message's `deliveredTo`; for failures bumps the retry count
731-- and gives up after `maxForwardRetries`.
732retryPendingForwards : IO {rw *forwardFailures, rw *forwards, rw *messages, network, clock} {}
733retryPendingForwards = do
734 t <- now
735 fwds <- *forwards
736 failures <- *forwardFailures
737 forEach (collectPeers fwds (\f -> f.toServer)) (\server -> if isBadServerName server
738 -- Denylisted destination (state persisted before a denylist tightening):
739 -- its forwards would never be attempted, so drop them instead of letting
740 -- them sit in the outbox forever. The denylist check stays outside the
741 -- atomic predicate — only the plain equality filter is pushed down.
742 then atomic (do
743 cur <- *forwards
744 *forwards = filter (\x -> x.toServer != server) cur)
745 -- Order the per-peer backlog deterministically before `take maxBatchSize`:
746 -- `[a]` is an unordered set, so an unsorted `take` returns an
747 -- implementation-defined subset that could keep retrying the same entries
748 -- while starving others past the cap. A stable key drains a fixed prefix as
749 -- entries succeed (removed) or hit `maxForwardRetries` (dropped). (Same class
750 -- of fix as `forwardQueuedToServer`'s `sortBy receivedAt`; `ForwardEntry`
751 -- carries no timestamp, so key on the unique blob.)
752 else when (shouldRetryServer server t failures) (forEach (take maxBatchSize (sortBy (\f -> f.encryptedBlob) (filter (\f -> f.toServer == server) fwds))) (\f -> do
753 result <- fetch (federationUrl server) (ForwardMessage {f.toKey, f.encryptedBlob})
754 success <- case result of
755 Err _ -> do
756 recordFailure server t
757 yield (False {})
758 Ok _ -> do
759 clearFailure server t
760 yield (True {})
761 let isMatch = forwardMatches f.encryptedBlob server
762 let isDone = \x -> isMatch x && (success || x.retries + 1 >= maxForwardRetries)
763 atomic (do
764 cur <- *forwards
765 *forwards = map (\x -> if isMatch x
766 then {x.toServer, x.toKey, x.encryptedBlob, retries: x.retries + 1}
767 else x) (filter (\x -> not isDone x) cur)
768 when success (do
769 msgs <- *messages
770 *messages = map (\m -> if m.toKey == f.toKey && m.encryptedBlob == f.encryptedBlob && not any (\d -> d.server == server) m.deliveredTo
771 then {m.toKey, m.encryptedBlob, m.receivedAt, deliveredTo: union m.deliveredTo [{server}]}
772 else m) msgs)))))
774-- Forward queued messages to a server when a remote key comes online.
775-- Skips messages already delivered to `server` (tracked via Message.deliveredTo)
776-- so a presence re-announce after a TTL lapse doesn't duplicate delivery.
777forwardQueuedToServer : PubkeyHex -> ServerName -> IO {rw *forwards, r *messages} {}
778forwardQueuedToServer = \pubkey server -> atomic (do
779 msgs <- *messages
780 fwds <- *forwards
781 let existing = count (filter (\f -> f.toServer == server) fwds)
782 let budget = max 0 (maxForwardsPerServer - existing)
783 let candidates = filter (\m -> m.toKey == pubkey && not any (forwardMatches m.encryptedBlob server) fwds && not any (\d -> d.server == server) m.deliveredTo) msgs
784 let newFwds = map (\m -> {toServer: server, m.toKey, m.encryptedBlob, retries: 0}) (take budget (sortBy (\m -> m.receivedAt) candidates))
785 *forwards = union fwds newFwds)
787-- ============================================
788-- Handler: challenge / verify
789-- ============================================
790-- Identity is a public key; there are no accounts. The flow is:
791-- 1. Client requests a fresh nonce (`handleChallenge`).
792-- 2. Client signs the nonce with its private key and submits the signature
793-- together with the original challenge (`handleVerifyAuth`).
794-- 3. Server returns a Bearer token bound to (pubkey, ip, expiry); the new
795-- session also installs a local presence row and gossips an `online`
796-- event to known peers.
797-- Sessions expire on their own (`sessionExpiry`); there is no client-driven
798-- revoke/logout.
800-- POST /auth/challenge: issue a fresh signed-nonce challenge for `pubkey`.
801-- Capped per-key (maxChallengesPerKey); per-pubkey rate limiting is enforced
802-- by the route's `rateLimit` clause. The IP is recorded so /auth/verify can
803-- require the same client.
804handleChallenge : PubkeyHex -> Text -> IO {rw *challenges, clock, random} (Result HttpError {challenge: Text, expiresAt: Timestamp})
805handleChallenge = \pubkey clientIp -> do
806 t <- now
807 challenge <- genToken 48
808 let expiresAt = t + challengeExpiry
809 atomic (do
810 ch <- *challenges
811 let cappedActive = capPerKey (filter (\c -> c.expiresAt > t) ch) pubkey maxChallengesPerKey
812 if count cappedActive >= maxChallenges
813 then yield (Err {error: {status: 503, message: "capacity"}})
814 else do
815 *challenges = union cappedActive [{pubkey, challenge, expiresAt, ip: clientIp}]
816 yield (Ok {value: {challenge, expiresAt}}))
818-- POST /auth/verify: validate `signature` over `challenge` for `pubkey`.
819-- On success, mints a session token, records local presence (`pubkey` is now
820-- live on this server), and gossips an `online` event to known peers. If
821-- `revokeOthers` is set, all other sessions for the same pubkey are dropped.
822handleVerifyAuth : PubkeyHex -> Text -> Text -> Bool -> Text -> IO {rw *challenges, rw *forwardFailures, rw *presence, rw *sessions, network, clock, random} (Result HttpError {expiresAt: Timestamp, token: Text})
823handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp -> do
824 t <- now
825 let expiry = t + sessionExpiry
826 token <- genToken 48
827 -- The client signs (target host, challenge) under a domain-separation tag so a
828 -- relay server cannot replay the signature to a different server. We verify
829 -- against our own configured hostname (port stripped), which a relay cannot forge.
830 let signedText = "skrepka-auth-v1:" ++ beforeColon serverHost ++ ":" ++ challenge
831 let sigValid = case bytesFromHex pubkey of
832 Just {value: pk} -> case bytesFromHex signature of
833 Just {value: sigBytes} -> verify pk (textToBytes signedText) sigBytes
834 _ -> False {}
835 _ -> False {}
836 result <- atomic (do
837 ch <- *challenges
838 let same = \c -> c.pubkey == pubkey && c.challenge == challenge
839 let valid = any (\c -> same c && c.expiresAt > t && matchIp clientIp c.ip) ch && sigValid
840 if not valid
841 then yield (Nothing {})
842 else do
843 *challenges = filter (\c -> not same c) ch
844 ses <- *sessions
845 let pruned = if revokeOthers
846 then filter (\s -> s.pubkey != pubkey) ses
847 else capPerKey ses pubkey maxSessionsPerKey
848 let newSession = {token, pubkey, expiresAt: expiry, ip: clientIp}
849 *sessions = capList (union pruned [newSession]) maxSessions
850 peers <- refreshLocalPresence pubkey expiry
851 yield (Just {value: peers}))
852 case result of
853 Nothing {} -> yield (Err {error: {status: 401, message: "invalid"}})
854 Just {value: peers} -> do
855 when federationEnabled (fork (sendBatchedGossip peers [{eventType: "online", pubkey: Just {value: pubkey}}]))
856 yield (Ok {value: {token, expiresAt: expiry}})
858-- ============================================
859-- Handler: poll
860-- ============================================
861-- Long-poll. The client supplies a cursor (last-seen `receivedAt`). We:
862-- * Ack-and-drop everything <= the cursor (treating poll progress as ack).
863-- * Race a row-level STM wait on `*messages` against `pollTimeoutMs` and
864-- return new events plus piggybacked sender-side expirations/deliveries.
866-- POST /poll: long-poll body. Treats the cursor advance as an implicit ack
867-- (`ackPolledMessages`), refreshes the caller's local presence row so peers
868-- don't time us out mid-session, then runs `doPoll` to collect events (or
869-- block until a matching message arrives or the timeout elapses) and
870-- assembles the response.
871handlePoll : Int -> Text -> Text -> IO {r *seqCounter, r *sessions, rw *forwardFailures, rw *forwards, rw *messages, rw *presence, network, clock} (Result HttpError {cursor: Timestamp, events: [PollEvent]})
872handlePoll = \cursor authorization clientIp -> do
873 t <- now
874 setup <- atomic (do
875 auth <- authedPubkey authorization t clientIp
876 case auth of
877 Nothing {} -> yield (Nothing {})
878 Just {value: pubkey} -> do
879 -- Clamp the client cursor to the highest assigned sequence
880 -- (`*seqCounter`), not wall-clock `t`. Under sustained high throughput
881 -- `receivedAt` (= `seqTs`) can run ahead of `t`; clamping to `t` would
882 -- drag a legitimately-returned cursor backward and re-deliver the
883 -- in-between messages. `max t curSeq` keeps the old behaviour when the
884 -- counter is idle/behind, while still refusing to ack beyond what's
885 -- been issued.
886 curSeqRows <- *seqCounter
887 let curSeq = currentSeq curSeqRows
888 let clampedCursor = max 0 (min (max t curSeq) cursor)
889 when (clampedCursor > 0) (ackPolledMessages pubkey clampedCursor)
890 pres <- *presence
891 let stale = not any (\p -> p.pubkey == pubkey && isLocalServer p.server && p.expiresAt - t >= sessionExpiry / 2) pres
892 refreshPeers <- if stale then refreshLocalPresence pubkey (t + sessionExpiry) else yield []
893 yield (Just {value: {pubkey, clampedCursor, refreshPeers}}))
894 case setup of
895 Nothing {} -> yield (Err {error: {status: 401, message: "unauthorized"}})
896 Just {value: s} -> do
897 when federationEnabled (fork (sendBatchedGossip s.refreshPeers [{eventType: "online", pubkey: Just {value: s.pubkey}}]))
898 pollResult <- doPoll s t
899 yield (Ok {value: pollResult})
901-- One pass: collect this pubkey's unexpired messages newer than the cursor,
902-- sort by receivedAt, take up to `maxPollEvents`. The returned cursor is the
903-- max `receivedAt` in the page (or the input cursor when the page is empty),
904-- which the client echoes on the next /poll to advance — and which the
905-- server treats as an implicit ack.
906-- If the page is empty, race a row-level STM wait against `pollTimeoutMs`:
907-- the wait transaction reads `*messages` and `retry`s until a row matching
908-- this pubkey/cursor is committed; the timer wins on idle. Either way we
909-- re-collect at the new wall time, so a timeout returns an empty page.
910doPoll : {clampedCursor: Timestamp, pubkey: PubkeyHex | a} -> Timestamp -> IO {r *messages, clock} {cursor: Timestamp, events: [{encryptedBlob: BlobHex}]}
911doPoll = \s t -> do
912 let collectAt = \tnow -> atomic (do
913 msgs <- *messages
914 let fresh = filter (\m -> m.toKey == s.pubkey && tnow - m.receivedAt < messageMaxAge && m.receivedAt > s.clampedCursor) msgs
915 let limited = take maxPollEvents (sortBy (\m -> m.receivedAt) fresh)
916 yield {
917 events: map (\m -> {m.encryptedBlob}) limited,
918 cursor: fold (\acc m -> max acc m.receivedAt) s.clampedCursor limited
919 })
920 initial <- collectAt t
921 if count initial.events > 0
922 then yield initial
923 else do
924 -- `race` requires both arms to share a single effect row, so each arm
925 -- carries the union {r *messages, clock} — the STM wait gets a `now`,
926 -- the timer gets a no-op `*messages` read.
927 race
928 (do
929 _ <- now
930 atomic (do
931 msgs <- *messages
932 let fresh = filter (\m -> m.toKey == s.pubkey && m.receivedAt > s.clampedCursor) msgs
933 if count fresh > 0 then yield {} else retry))
934 (do
935 atomic (do _ <- *messages; yield {})
936 sleep pollTimeoutMs)
937 t2 <- now
938 collectAt t2
940-- Treat the supplied cursor as an ack: drop all messages with
941-- receivedAt <= cursor, plus their pending forwards.
942ackPolledMessages : PubkeyHex -> Int -> IO {rw *forwards, rw *messages} {}
943ackPolledMessages = \pubkey cursor -> do
944 msgs <- *messages
945 let matches = \m -> m.toKey == pubkey && m.receivedAt <= cursor
946 let toAck = filter matches msgs
947 let blobs = map (\m -> m.encryptedBlob) toAck
948 *messages = filter (\m -> not matches m) msgs
949 fwds <- *forwards
950 *forwards = filter (\f -> not (f.toKey == pubkey && elem f.encryptedBlob blobs)) fwds
952-- ============================================
953-- Handler: send
954-- ============================================
955-- Accept a batch of messages for recipient pubkeys. Placement (live local
956-- session, federated forward, or held-for-presence) is internal; the call
957-- either succeeds (every message enqueued) or fails with a non-2xx HTTP
958-- status carrying the rejection code ("self_send" | ...).
959-- Duplicates are silently treated as success so retries are idempotent.
960-- Per-client-IP send rate is enforced by the route's `rateLimit` clause.
961-- Federated forwards are kicked off in background forks after the atomic.
963-- POST /messages: send a list of ciphertexts atomically. Self-send is
964-- rejected up-front; the remaining batch is committed inside one atomic
965-- block (all-or-nothing) which reads presence to pre-stage federation
966-- forwards. Federation forwards are kicked per message after commit.
967handleSendMessages : [{encryptedBlob: BlobHex, to: PubkeyHex | a}] -> Text -> Text -> IO {r *presence, r *sessions, rw *forwardFailures, rw *forwards, rw *messages, rw *seqCounter, network, clock} (Result HttpError {})
968handleSendMessages = \messages authorization clientIp -> do
969 t <- now
970 auth <- authedPubkey authorization t clientIp
971 case auth of
972 Nothing {} -> yield (Err {error: {status: 401, message: "unauthorized"}})
973 Just {value: senderKey} -> do
974 if count messages > maxBatchSize
975 then yield (Err {error: {status: 413, message: "batch_too_large"}})
976 -- Refined-type validation at the route boundary only fires for top-level
977 -- scalar body fields; `to`/`encryptedBlob` are nested inside the `messages`
978 -- list, so the runtime does not check them. Enforce the `PubkeyHex` /
979 -- `BlobHex` contracts here so a malformed key or over-`maxBlobLen` / non-hex
980 -- blob is rejected instead of stored.
981 else if any (\m -> not (isValidExactHex 64 m.to && isValidBoundedHex maxBlobLen m.encryptedBlob)) messages
982 then yield (Err {error: {status: 400, message: "invalid_message"}})
983 else if any (\m -> m.to == senderKey) messages
984 then yield (Err {error: {status: 400, message: "self_send"}})
985 else do
986 committed <- atomic (do
987 pres <- *presence
988 let staged = map (\m -> {
989 m.to,
990 m.encryptedBlob,
991 remoteServers: collectPeers (filter (\p -> p.pubkey == m.to && p.expiresAt > t) pres) (\p -> p.server)
992 }) messages
993 forEach staged (\c -> do
994 -- Dedupe like the federation ingress path (`handleForwardMessage`):
995 -- a client retry after a lost response carries the identical
996 -- ciphertext, and `appendMessage`'s fresh `seqTs` would otherwise
997 -- make `union` keep both rows and double-deliver. Re-reading
998 -- `*messages` each iteration also collapses duplicate blobs that
999 -- appear twice within one batch.
1000 existingMsgs <- *messages
1001 when (not any (\m -> m.toKey == c.to && m.encryptedBlob == c.encryptedBlob) existingMsgs) (appendMessage c.to c.encryptedBlob t)
1002 -- Peers that already hold this blob — from an earlier send of the
1003 -- identical ciphertext that was forwarded then pruned from
1004 -- `*forwards`. A brand-new message has empty `deliveredTo`, so this
1005 -- is `[]` on the normal path; on a client resend it carries the real
1006 -- set. Mirrors `forwardQueuedToServer`'s guard so we don't re-stage a
1007 -- forward to a peer that already accepted this blob.
1008 let delivered = fold (\acc m -> if m.toKey == c.to && m.encryptedBlob == c.encryptedBlob then union acc (map (\d -> d.server) m.deliveredTo) else acc) [] existingMsgs
1009 -- Stage outbox rows for peers under their `maxForwardsPerServer`
1010 -- ceiling. Peers at the cap still get the immediate post-commit
1011 -- delivery attempt (`forkForwardsTo`) — only the retry-queue row
1012 -- is skipped.
1013 when federationEnabled (do
1014 fwds <- *forwards
1015 let newFwds = map (\s -> {toServer: s, toKey: c.to, c.encryptedBlob, retries: 0}) (filter (\s -> not (any (forwardMatches c.encryptedBlob s) fwds) && not elem s delivered && countWhere (\f -> f.toServer == s) fwds < maxForwardsPerServer) c.remoteServers)
1016 *forwards = union fwds newFwds))
1017 yield staged)
1018 when federationEnabled (forEach committed (\c -> fork (forkForwardsTo c.to c.encryptedBlob c.remoteServers)))
1019 yield (Ok {value: {}})
1021-- Fire one /federation/forward call per pre-staged server. On success the
1022-- corresponding row in `forwards` is pruned and the peer is appended to the
1023-- message's `deliveredTo` set. The backing `*messages` row itself stays —
1024-- cleaned up by `ackPolledMessages` when the recipient polls here, or by
1025-- `pruneExpired` once `messageMaxAge` lapses. Keeping the row lets
1026-- `forwardQueuedToServer` re-forward to peers learned after send time (so a
1027-- multi-homed recipient who appears on a new peer post-send still receives
1028-- the message), and `deliveredTo` prevents re-forwards to peers that already
1029-- have the message. On failure the forwards row stays so
1030-- `retryPendingForwards` can re-attempt with backoff.
1031forkForwardsTo : PubkeyHex -> BlobHex -> [Text] -> IO {rw *forwardFailures, rw *forwards, rw *messages, network, clock} {}
1032forkForwardsTo = \to encryptedBlob servers -> forEach servers (\server -> when (not isBadServerName server) (do
1033 t <- now
1034 result <- fetch (federationUrl server) (ForwardMessage {toKey: to, encryptedBlob})
1035 let same = forwardMatches encryptedBlob server
1036 case result of
1037 Err _ -> recordFailure server t
1038 Ok _ -> do
1039 clearFailure server t
1040 atomic (do
1041 fwds <- *forwards
1042 *forwards = filter (\x -> not same x) fwds
1043 msgs <- *messages
1044 *messages = map (\m -> if m.toKey == to && m.encryptedBlob == encryptedBlob && not any (\d -> d.server == server) m.deliveredTo
1045 then {m.toKey, m.encryptedBlob, m.receivedAt, deliveredTo: union m.deliveredTo [{server}]}
1046 else m) msgs)))
1048-- ============================================
1049-- Handler: federation receive
1050-- ============================================
1051-- Inbound peer-to-peer endpoints. Federation is open and unauthenticated:
1052-- the `no_presence` check in `handleForwardMessage` bounds forwards to
1053-- currently-online local keys. Both handlers refuse with 403 when
1054-- `federationEnabled` is off, matching the outbound paths — a single-node
1055-- deployment accepts no peer writes to `*presence`/`*messages`.
1057-- POST /federation/gossip: ingest peer presence events. The transport-level
1058-- `fromServer` is the home server for every event. Drops every (pubkey,
1059-- fromServer) row touched by this batch, then re-adds fresh-TTL rows for
1060-- pubkeys with an `online` event (so Online wins on collision). For keys
1061-- that transition offline → online, flushes any locally-buffered messages via
1062-- `forwardQueuedToServer`. Events with an unrecognized `eventType` or no
1063-- `pubkey` are skipped (forward-compat) rather than rejecting the batch.
1064handleRecvGossip : [GossipEvent] -> Text -> IO {r *messages, rw *forwards, rw *presence, clock} (Result HttpError {})
1065handleRecvGossip = \events fromServer -> do
1066 t <- now
1067 if not federationEnabled
1068 then yield (Err {error: {status: 403, message: "federation_disabled"}})
1069 else if isLocalServer fromServer || isBadServerName fromServer || count events > maxGossipEvents
1070 then yield (Err {error: {status: 400, message: "invalid_request"}})
1071 else do
1072 newlyOnline <- atomic (do
1073 pres <- *presence
1074 let collected = fold (\acc e -> case e.pubkey of
1075 -- `pubkey` is nested inside the `events` list, so the route-boundary
1076 -- refinement never validates it (same nested-list gap that `/messages`
1077 -- works around with an explicit `isValidExactHex 64` check). Reject a
1078 -- malformed key here so a peer can't poison `*presence` with non-hex
1079 -- rows that then spawn bogus-`toKey` forward entries.
1080 Just {value: pubkey} -> if not isValidExactHex 64 pubkey
1081 then acc
1082 else if e.eventType == "online"
1083 then {
1084 touched: union acc.touched [pubkey],
1085 added: union acc.added [{pubkey, server: fromServer, expiresAt: t + onlineGossipTtl}]
1087 else if e.eventType == "offline"
1088 then {acc.added, touched: union acc.touched [pubkey]}
1089 else acc
1090 Nothing {} -> acc) {touched: [], added: []} events
1091 let kept = filter (\p -> p.server != fromServer || not elem p.pubkey collected.touched) pres
1092 *presence = union kept collected.added
1093 yield (filter (\pk -> not any (\p -> p.pubkey == pk && p.server == fromServer && p.expiresAt > t) pres) (map (\p -> p.pubkey) collected.added)))
1094 fork (forEach newlyOnline (\pk -> forwardQueuedToServer pk fromServer))
1095 yield (Ok {value: {}})
1097-- POST /federation/forward: receive a single federated message destined for
1098-- a local key. The presence gate requires the recipient to be **currently
1099-- locally online** (not just visible to gossip somewhere). Without that, an
1100-- A → B forward based on stale gossip about R could land on B with no
1101-- actively-polling client and get stuck. If B refuses, A retries via
1102-- `retryPendingForwards` against the next presence target.
1103-- Re-delivery of an already-stored (toKey, blob) pair is treated as success
1104-- without a second append, so a sender whose success response was lost can
1105-- retry without duplicating the message in the recipient's mailbox.
1106handleForwardMessage : PubkeyHex -> BlobHex -> IO {rw *messages, r *presence, rw *seqCounter, clock} (Result HttpError {})
1107handleForwardMessage = \toKey encryptedBlob -> do
1108 t <- now
1109 if not federationEnabled
1110 then yield (Err {error: {status: 403, message: "federation_disabled"}})
1111 else atomic (do
1112 pres <- *presence
1113 if not hasLocalPresence pres toKey t
1114 then yield (Err {error: {status: 404, message: "no_presence"}})
1115 else do
1116 msgs <- *messages
1117 when (not any (\m -> m.toKey == toKey && m.encryptedBlob == encryptedBlob) msgs) (appendMessage toKey encryptedBlob t)
1118 yield (Ok {value: {}}))
1120-- ============================================
1121-- Dispatch
1122-- ============================================
1123-- Single match from generated request constructor (one per route entry above)
1124-- to the corresponding `handleXxx` function. Adding a new route means adding
1125-- a `route ... = NewReq` line, a `handleNewReq` definition, and a clause here.
1127-- Top-level request dispatcher: maps each generated request constructor to
1128-- its `handleXxx` function. Every route entry above must have a clause here.
1129api : Server Api _
1130api = serve Api where
1131 ReqChallenge = \{pubkey, xForwardedFor} -> handleChallenge pubkey (clientIpOf xForwardedFor)
1132 VerifyAuth = \{pubkey, challenge, signature, revokeOthers, xForwardedFor} -> handleVerifyAuth pubkey challenge signature revokeOthers (clientIpOf xForwardedFor)
1133 Poll = \{cursor, authorization, xForwardedFor} -> handlePoll cursor authorization (clientIpOf xForwardedFor)
1134 SendMessages = \{messages, authorization, xForwardedFor} -> handleSendMessages messages authorization (clientIpOf xForwardedFor)
1135 RecvGossip = \{events, fromServer} -> handleRecvGossip events fromServer
1136 ForwardMessage = \{toKey, encryptedBlob} -> handleForwardMessage toKey encryptedBlob
1138-- ============================================
1139-- Background loops
1140-- ============================================
1141-- `backgroundPrune` runs the prune sweep every `backgroundPruneInterval` ms.
1142-- `backgroundRetryForwards` runs the forward-retry sweep on the same cadence
1143-- but as its own sequential loop so two retries can't race.
1145-- Endless tick: prune expired state and gossip offline for locally-expired
1146-- keys. First sweep happens immediately so stale rows persisted by the
1147-- previous process clear before the next interval elapses.
1148backgroundPrune : IO {rw *challenges, rw *forwardFailures, rw *messages, rw *presence, rw *sessions, network, clock} a
1149backgroundPrune = do
1150 t <- now
1151 expiredLocal <- pruneExpired t
1152 when federationEnabled (fork (do
1153 pres <- *presence
1154 sendBatchedGossip (collectPeers pres (\p -> p.server)) (map (\pk -> {eventType: "offline", pubkey: Just {value: pk}}) expiredLocal)))
1155 sleep backgroundPruneInterval
1156 backgroundPrune
1158-- Endless tick: retry pending cross-server forwards. Calls
1159-- `retryPendingForwards` inline (not via `fork`) so a slow run blocks the
1160-- next iteration instead of racing with itself — two concurrent runs would
1161-- see the same `*forwards` snapshot and double-count `recordFailure`,
1162-- inflating the per-peer backoff multiplier.
1163backgroundRetryForwards : IO {rw *forwardFailures, rw *forwards, rw *messages, network, clock} a
1164backgroundRetryForwards = do
1165 retryPendingForwards
1166 sleep backgroundPruneInterval
1167 backgroundRetryForwards
1169-- ============================================
1170-- Entry Point
1171-- ============================================
1172-- Fork the background prune loop (which runs an immediate sweep on its
1173-- first tick) and start the HTTP listener. The forward-retry loop only
1174-- runs when federation is enabled.
1175main : IO _ {}
1176main = do
1177 logInfo ("Skrepka relay server starting on " ++ bindHost ++ ":" ++ show listenPort ++ " ...")
1178 logInfo (" federation: " ++ (if federationEnabled then "enabled" else "disabled"))
1179 fork backgroundPrune
1180 when federationEnabled (fork backgroundRetryForwards)
1181 listenOn bindHost listenPort api