@@ -15,6 +15,7 @@
1515
1616-- Server identity and federation credentials
1717serverHost = "localhost:8080"
18+listenPort = 8080
1819-- TODO: Support runtime secret rotation: accept both old and new secrets during a
1920-- transition window, so operators can rotate without restarting. E.g., add a
2021-- maintenanceSecretPrevious field that's accepted for a configurable overlap period.
@@ -170,6 +171,8 @@ maxGossipEventRateEntries = 100
170171maxGlobalFederationConcurrency = 30
171172-- Peer blocklist: max blocked servers tracked at runtime (managed via /maintenance/block, /maintenance/unblock)
172173maxBlockedPeers = 100
174+-- Peer allowlist: when non-empty, only these servers can federate with us (managed via /maintenance/allow, /maintenance/unallow)
175+maxAllowedPeers = 100
173176
174177-- Drain timeout: max time for drain flush operations before force-completing
175178drainTimeoutMs : Int<Ms>
@@ -245,6 +248,12 @@ maxPendingAckCleanups = 10000
245248-- Per-key deferred ack cleanup cap (prevents one noisy key from dominating the queue)
246249maxPendingAckCleanupsPerKey = 100
247250
251+-- Message expiration notifications: track expired locally-originated messages so senders
252+-- can learn their message was never read (returned in poll response).
253+maxSenderExpirations = 10000
254+maxSenderExpirationsPerKey = 100
255+-- Max pubkeys in a batch lookup request
256+maxBatchLookupKeys = 50
248257-- Health endpoint rate limiting (global, since /health has no clientIp)
249258maxHealthRate = 100 -- max health checks per window
250259healthRateWindowMs : Int<Ms>
@@ -328,7 +337,9 @@ type Session = {token: Text, pubkey: PubkeyHex, expiresAt: Timestamp, ip: Text,
328337-- Stored message; originServer is "" for locally-originated messages, set to sender server for federated forwards.
329338-- expiresAt: computed from sender-specified TTL or default unackedMessageMaxAge.
330339-- Messages with expiresAt <= now are pruned and excluded from polls.
331−type Message = {messageId: MessageIdHex, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: Text, receivedAt: Timestamp, expiresAt: Timestamp}
340+-- senderKey: pubkey of the authenticated sender (locally-originated only; "" for federation forwards).
341+-- Used by /messages/cancel to verify the caller is the actual sender.
342+type Message = {messageId: MessageIdHex, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: Text, receivedAt: Timestamp, expiresAt: Timestamp, senderKey: Text}
332343-- Presence: tracks that a pubkey is online at a given server, with expiry
333344type PresenceEntry = {pubkey: PubkeyHex, server: AnyServerName, expiresAt: Timestamp, timestamp: Timestamp}
334345-- Queued federation forward: message waiting to be delivered to a remote server.
@@ -375,6 +386,10 @@ type ForwardNonce = {messageId: MessageIdHex, toKey: PubkeyHex, originServer: Te
375386type BatchMessage = {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: OptionalMessageId, messageTtl: Int}
376387type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
377388
389+-- Sender expiration notification: tracks when a locally-originated message expired unread.
390+-- Pruned after messageMaxAge. Included in the sender's next poll response.
391+type SenderExpiration = {senderKey: Text, messageId: MessageIdHex, toKey: PubkeyHex, expiredAt: Timestamp}
392+
378393-- ============================================
379394-- Persisted State
380395-- ============================================
@@ -694,6 +709,10 @@ type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
694709-- Blocked federation peers (managed at runtime via /maintenance/block and /maintenance/unblock)
695710*blockedPeers : [Text]
696711
712+-- Allowed federation peers (when non-empty, only these servers can federate with us)
713+-- Managed via /maintenance/allow and /maintenance/unallow. Empty = open federation.
714+*allowedPeers : [Text]
715+
697716-- Deferred ack cleanups: origin message reclamation queued by federation ack handlers,
698717-- processed by the prune cycle after ackGracePeriod elapses (replaces fork+sleep pattern).
699718*pendingAckCleanups : [{messageId: MessageIdHex, toKey: PubkeyHex, cleanupAfter: Timestamp}]
@@ -702,6 +721,10 @@ type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
702721-- (key-rotation abuse bypassing per-key rate limits). Pruned with other rate limiters.
703722*uniqueKeyIpRates : [UniqueKeyIpEntry]
704723
724+-- Sender expiration notifications: messages that expired unread, queued for the sender's next poll.
725+-- Populated during prune for locally-originated messages (senderKey != "").
726+*senderExpirations : [SenderExpiration]
727+
705728-- Runtime-configurable rate limit overrides (0 = use compile-time default).
706729-- Adjusted via /maintenance/rate-limits POST endpoint without recompile+restart.
707730*rateOverrideGlobalRequest : Int
@@ -765,7 +788,7 @@ route AuthApi where
765788route ClientApi where
766789 -- Long-poll for new messages; cursor tracks last-seen receivedAt timestamp
767790 POST {cursor: Timestamp, generation: Int, watchKeys: [PubkeyHex]} /poll headers {authorization: Text, clientIp: Text}
768− -> {events: [PollEvent], cursor: Timestamp, authorized: Bool, rateLimited: Bool, draining: Bool, generation: Int, presenceUpdates: [{pubkey: PubkeyHex, online: Bool}], heartbeat: Bool, retryAfterMs: Int, cursorReset: Bool}
791+ -> {events: [PollEvent], cursor: Timestamp, authorized: Bool, rateLimited: Bool, draining: Bool, generation: Int, presenceUpdates: [{pubkey: PubkeyHex, online: Bool}], heartbeat: Bool, retryAfterMs: Int, cursorReset: Bool, messageExpirations: [{messageId: MessageIdHex, toKey: PubkeyHex, expiredAt: Timestamp}]}
769792 = Poll
770793 /messages
771794 -- Send an E2E encrypted message to a recipient pubkey.
@@ -806,6 +829,10 @@ route ClientApi where
806829 GET /lookup/{pubkey: PubkeyHex} headers {authorization: Text, clientIp: Text}
807830 -> {server: AnyServerName, authorized: Bool}
808831 = LookupPresence
832+ -- Batch lookup: look up presence for multiple pubkeys in one request
833+ POST {pubkeys: [PubkeyHex]} /lookup/batch headers {authorization: Text, clientIp: Text}
834+ -> {results: [{pubkey: PubkeyHex, server: AnyServerName}], authorized: Bool}
835+ = BatchLookupPresence
809836
810837-- Server-to-server federation API (authenticated via domain callback verification)
811838-- correlationId: optional trace ID for cross-server request tracing (empty = not provided).
@@ -896,6 +923,22 @@ route MaintenanceApi where
896923 POST {globalRequest: Int, ipRate: Int, sendRate: Int, federationRate: Int, authGlobal: Int, recvRate: Int} /maintenance/rate-limits headers {authorization: Text}
897924 -> {ok: Bool}
898925 = SetRateLimits
926+ -- Flush forward nonces (recovery from corruption or capacity issues)
927+ POST {} /maintenance/nonces/flush headers {authorization: Text}
928+ -> {ok: Bool, flushedCount: Int}
929+ = FlushNonces
930+ -- Allow a federation peer (when allowlist is non-empty, only allowed peers can federate)
931+ POST {server: ServerName} /maintenance/allow headers {authorization: Text}
932+ -> {ok: Bool, allowed: Int}
933+ = AllowPeer
934+ -- Remove a server from the federation allowlist
935+ POST {server: ServerName} /maintenance/unallow headers {authorization: Text}
936+ -> {ok: Bool, removed: Bool}
937+ = UnallowPeer
938+ -- List currently allowed federation peers (empty = open federation)
939+ GET /maintenance/allowed headers {authorization: Text}
940+ -> {peers: [Text], ok: Bool}
941+ = ListAllowed
899942
900943route HealthApi where
901944 GET /health headers {authorization: Text} -> {ok: Bool, pruneHealthy: Bool, heartbeatHealthy: Bool, capacityHealthy: Bool, federationHealthy: Bool, snapshotHealthy: Bool, watchdogHealthy: Bool, draining: Bool} = Health
@@ -1224,6 +1267,10 @@ verifyMaintenanceAuth = verifyAuthWith (\token -> verifySecret token maintenance
12241267isBlockedPeer : Text -> [Text] -> Bool
12251268isBlockedPeer = \server blocked -> any blocked (\b -> b == server)
12261269
1270+-- Check if a server passes the allowlist (empty allowlist = all allowed)
1271+isAllowedByAllowlist : Text -> [Text] -> Bool
1272+isAllowedByAllowlist = \server allowed -> count allowed == 0 || any allowed (\a -> a == server)
1273+
12271274-- Generate a single-use token for outbound federation auth.
12281275-- The receiving server calls back to /federation/verify to confirm we issued it.
12291276-- scope binds the token to a request type ("gossip", "forward", "ack") so a token
@@ -1746,6 +1793,8 @@ uniqueBucketsFor = \items getKey ->
17461793-- Each bucket prune decrements totalMessageCount and its rcBucket by exactly
17471794-- what was removed, preserving concurrent modifications from poll-ack or send.
17481795-- Uses per-message expiresAt (respects sender-specified TTL).
1796+-- Also queues expiration notifications for locally-originated messages (senderKey != "")
1797+-- so the sender can learn their message expired unread via the next poll response.
17491798pruneMessageBucket = \b t -> atomic do
17501799 bucket <- readMessageBucket b
17511800 let kept = do m <- bucket; where m.expiresAt > t; yield m
@@ -1757,6 +1806,14 @@ pruneMessageBucket = \b t -> atomic do
17571806 rc <- readRcBucket b
17581807 let removed = do m <- bucket; where m.expiresAt <= t; yield m
17591808 writeRcBucket b (batchDecrementRecipientCounts rc (countByRecipient removed))
1809+ -- Queue expiration notifications for locally-originated messages with known sender
1810+ let localExpired = do m <- removed; where m.senderKey != "" && m.originServer == ""; yield {senderKey: m.senderKey, messageId: m.messageId, toKey: m.toKey, expiredAt: t}
1811+ when (count localExpired > 0) (do
1812+ se <- *senderExpirations
1813+ let combined = union se localExpired
1814+ set *senderExpirations = if count combined > maxSenderExpirations
1815+ then evictOneOldest combined (\e -> e.expiredAt)
1816+ else combined)
17601817 else yield {}
17611818
17621819-- Prune all 16 message buckets with concurrency: fork first 15, run last in foreground.
@@ -2112,6 +2169,20 @@ computeForwardStatsInner = \b t acc ->
21122169 let newMaxDepth = fold (\mx s -> max mx s.count) acc.maxDepth updatedServerCounts
21132170 computeForwardStatsInner (b + 1) t {totalCount: acc.totalCount + bucketCount, oldestAge: max acc.oldestAge bucketOldest, serverCount: count updatedServerCounts, maxDepth: newMaxDepth, serverCounts: updatedServerCounts}
21142171
2172+-- Remove forward entries matching a server and message IDs, scanning per-bucket.
2173+-- Only rewrites buckets that actually changed (avoids writeAllForwards repartitioning).
2174+-- Returns all removed forward entries across all buckets.
2175+removeForwardsByServer = \server messageIds b acc ->
2176+ if b > 15 then yield acc
2177+ else do
2178+ removed <- atomic do
2179+ bucket <- readFwdBucket b
2180+ let matched = do f <- bucket; where f.toServer == server && any messageIds (\mid -> f.messageId == mid); yield f
2181+ when (count matched > 0)
2182+ (writeFwdBucket b (do f <- bucket; where not (f.toServer == server && any messageIds (\mid -> f.messageId == mid)); yield f))
2183+ yield matched
2184+ removeForwardsByServer server messageIds (b + 1) (if count removed > 0 then union acc removed else acc)
2185+
21152186-- Read only the forward buckets at the given indices (avoids reading all 16)
21162187readSelectedFwdBuckets : [Int] -> IO [ForwardEntry]
21172188readSelectedFwdBuckets = \buckets ->
@@ -2669,15 +2740,16 @@ prunePresenceBucketN = \b t -> atomic do
26692740 yield {expiredLocal: expiredLocal, keptPeerServers: collectPeerServers kept (\p -> p.server), allPeerServers: collectPeerServers bucket (\p -> p.server), removedCount: count expired}
26702741
26712742-- Recursively prune all 16 presence buckets and aggregate results.
2743+-- Collects raw server lists from each bucket; deduplication is done once
2744+-- at the end by the caller (pruneExpired) to avoid O(n*k) per-bucket merges.
26722745prunePresenceBuckets = \b t ->
26732746 if b > 15 then yield {expiredLocal: [], keptPeerServers: [], allPeerServers: [], totalRemoved: 0}
26742747 else do
26752748 r <- prunePresenceBucketN b t
26762749 rest <- prunePresenceBuckets (b + 1) t
2677− let mergePeers = \a ps -> fold (\acc s -> if any acc (\x -> x == s) then acc else union acc [s]) a ps
26782750 yield {expiredLocal: union r.expiredLocal rest.expiredLocal,
2679− keptPeerServers: mergePeers r.keptPeerServers rest.keptPeerServers,
2680− allPeerServers: mergePeers r.allPeerServers rest.allPeerServers,
2751+ keptPeerServers: union r.keptPeerServers rest.keptPeerServers,
2752+ allPeerServers: union r.allPeerServers rest.allPeerServers,
26812753 totalRemoved: r.removedCount + rest.totalRemoved}
26822754
26832755-- Remove all expired entries from every collection. Returns expired local presence
@@ -2687,12 +2759,17 @@ pruneExpired = \t -> do
26872759 -- Each bucket is independent: all entries for a pubkey hash to the same bucket,
26882760 -- so per-bucket before/after comparison is equivalent to the global check.
26892761 presenceResults <- prunePresenceBuckets 0 t
2762+ -- Dedup peer servers once after collecting from all 16 buckets (O(n) total
2763+ -- vs O(n*k) if deduped per-bucket during aggregation).
2764+ let dedupServers = \servers -> fold (\acc s -> if any acc (\x -> x == s) then acc else union acc [s]) [] servers
2765+ let keptPeers = dedupServers presenceResults.keptPeerServers
2766+ let allPeers = dedupServers presenceResults.allPeerServers
26902767 atomic do
26912768 tpc <- *totalPresenceCount
26922769 set *totalPresenceCount = max 0 (tpc - presenceResults.totalRemoved)
2693− set *knownPeerServers = presenceResults.keptPeerServers
2770+ set *knownPeerServers = keptPeers
26942771 when (presenceResults.totalRemoved > 0) (set *presenceVersion = t)
2695− let presenceResult = {expiredLocal: presenceResults.expiredLocal, peerServers: presenceResults.allPeerServers}
2772+ let presenceResult = {expiredLocal: presenceResults.expiredLocal, peerServers: allPeers}
26962773 -- Prune auth state (challenges in their own block; sessions counted during transform)
26972774 atomic do
26982775 challenges <- *challenges
@@ -2797,6 +2874,9 @@ pruneExpired = \t -> do
27972874 -- Prune stale deferred ack cleanups (older than messageMaxAge are never useful)
27982875 pendingCleanups <- *pendingAckCleanups
27992876 set *pendingAckCleanups = do c <- pendingCleanups; where t - c.cleanupAfter < messageMaxAge; yield c
2877+ -- Prune old sender expiration notifications (keep for one poll cycle + grace)
2878+ senderExp <- *senderExpirations
2879+ set *senderExpirations = do e <- senderExp; where t - e.expiredAt < sessionExpiry; yield e
28002880 -- Reconcile global federation connection counter with actual in-flight state
28012881 -- (self-heals drift from leaked slots where a forked task died without decrementing)
28022882 inFlightRec <- *federationInFlight
@@ -2968,13 +3048,19 @@ checkPreAuth = \clientIp t -> do
29683048 else yield ""
29693049
29703050withFederationAuth = \authorization t respond fromServer scope requestTimestamp precondition clearOnAuth action -> do
2971− -- Check blocklist before spending resources on verification callback
3051+ -- Check blocklist and allowlist before spending resources on verification callback
29723052 blocked <- *blockedPeers
29733053 if isBlockedPeer fromServer blocked
29743054 then do
29753055 logWarn ("federation: blocked peer " ++ fromServer ++ " rejected")
29763056 yield (respond {ok: False {}, error: "blocked"})
29773057 else do
3058+ allowed <- *allowedPeers
3059+ if not (isAllowedByAllowlist fromServer allowed)
3060+ then do
3061+ logWarn ("federation: peer " ++ fromServer ++ " not in allowlist, rejected")
3062+ yield (respond {ok: False {}, error: "not_allowed"})
3063+ else do
29783064 verified <- verifyFederationCallback fromServer authorization scope
29793065 if not verified
29803066 then yield (respond {ok: False {}, error: "unauthorized"})
@@ -4017,10 +4103,10 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
40174103 let cursorWasReset = generationMismatch && cursor > 0
40184104 let cappedWatchKeys = takeRelation maxWatchKeys watchKeys
40194105 dp <- *drainPhase
4020− if dp >= 2 then yield (respond {events: [], cursor: 0, authorized: True {}, rateLimited: False {}, draining: True {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: 0, cursorReset: cursorWasReset})
4106+ if dp >= 2 then yield (respond {events: [], cursor: 0, authorized: True {}, rateLimited: False {}, draining: True {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: 0, cursorReset: cursorWasReset, messageExpirations: []})
40214107 else do
40224108 globalRateLimited <- checkGlobalRate t
4023− if globalRateLimited then yield (respond {events: [], cursor: 0, authorized: False {}, rateLimited: True {}, draining: False {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: globalRateWindowMs, cursorReset: cursorWasReset})
4109+ if globalRateLimited then yield (respond {events: [], cursor: 0, authorized: False {}, rateLimited: True {}, draining: False {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: globalRateWindowMs, cursorReset: cursorWasReset, messageExpirations: []})
40244110 else do
40254111 -- Snapshot presenceVersion before long-poll wait to skip unchanged lookups
40264112 startPresenceVersion <- *presenceVersion
@@ -4043,14 +4129,32 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
40434129 sendAcksAndClean result.pendingAcks
40444130 retryEvictedAcks result.evictedEntries result.pendingAcks))
40454131 let isHeartbeat = result.authorized && count result.events == 0
4046− yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized, rateLimited: False {}, draining: False {}, generation: serverGen, presenceUpdates: presenceUpdates, heartbeat: isHeartbeat, retryAfterMs: 0, cursorReset: cursorWasReset})
4132+ -- Deliver message expiration notifications: atomically read and drain expirations
4133+ -- for this sender's pubkey so each notification is delivered exactly once.
4134+ expirations <- if not result.authorized then yield []
4135+ else case extractToken authorization of
4136+ Nothing {} -> yield []
4137+ Just {value: token} -> do
4138+ let bucket = tokenBucket token
4139+ sessions <- readSessionBucket bucket
4140+ t3 <- now
4141+ case authenticateSession sessions token t3 of
4142+ Unauthorized {} -> yield []
4143+ Authorized {pubkey} -> atomic do
4144+ se <- *senderExpirations
4145+ let forKey = do e <- se; where e.senderKey == pubkey; yield {messageId: e.messageId, toKey: e.toKey, expiredAt: e.expiredAt}
4146+ let remaining = do e <- se; where e.senderKey != pubkey; yield e
4147+ set *senderExpirations = remaining
4148+ let capped = takeRelation maxSenderExpirationsPerKey forKey
4149+ yield capped
4150+ yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized, rateLimited: False {}, draining: False {}, generation: serverGen, presenceUpdates: presenceUpdates, heartbeat: isHeartbeat, retryAfterMs: 0, cursorReset: cursorWasReset, messageExpirations: expirations})
40474151
40484152-- Core send logic: dedup, rate limiting, store message, create forwards.
40494153-- Called inside an atomic block by both handleSendMessage and sendBatchInner.
40504154-- Caller handles auth, drain checks, and messageId generation.
40514155-- messageTtl: sender-specified TTL in ms (0 = default unackedMessageMaxAge).
40524156sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId messageTtl t ->
4053− if senderKey == to then yield {status: MsRejected {}, remoteServers: [], retryAfterMs: 0}
4157+ if senderKey == to then yield {status: MsRejected {}, remoteServers: [], retryAfterMs: 0, msgExpiresAt: 0}
40544158 else do
40554159 let msgBucket = pubkeyBucket to
40564160 bucket <- readMessageBucket msgBucket
@@ -4061,31 +4165,31 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId messageTt
40614165 -- expire but before the nonce expires.
40624166 nonceBkt <- readNonceBucket msgBucket
40634167 let isDup = checkMessageDup bucket effectiveMsgId to || hasMessage effectiveMsgId to ackedBkt || hasMessage effectiveMsgId to delivBkt || any nonceBkt (\n -> n.messageId == effectiveMsgId && n.toKey == to)
4064− if isDup then yield {status: MsDuplicate {}, remoteServers: [], retryAfterMs: 0}
4168+ if isDup then yield {status: MsDuplicate {}, remoteServers: [], retryAfterMs: 0, msgExpiresAt: 0}
40654169 else do
40664170 sendRates <- *sendRates
40674171 sendRate <- effectiveSendRate
40684172 let rateResult = checkPubkeyRate sendRates senderKey sendRate sendRateWindowMs maxSendRateEntries t
4069− if rateResult.limited then yield {status: MsRateLimited {}, remoteServers: [], retryAfterMs: rateResult.retryAfterMs}
4173+ if rateResult.limited then yield {status: MsRateLimited {}, remoteServers: [], retryAfterMs: rateResult.retryAfterMs, msgExpiresAt: 0}
40704174 else do
40714175 set *sendRates = rateResult.entries
40724176 pairRates <- *pairRates
40734177 let pairResult = checkPairRate pairRates senderKey to maxPairRatePerPair pairRateWindowMs maxPairRateEntries t
40744178 set *pairRates = pairResult.entries
4075− if pairResult.limited then yield {status: MsRateLimited {}, remoteServers: [], retryAfterMs: pairResult.retryAfterMs}
4179+ if pairResult.limited then yield {status: MsRateLimited {}, remoteServers: [], retryAfterMs: pairResult.retryAfterMs, msgExpiresAt: 0}
40764180 else do
40774181 recvRates <- *recvRates
40784182 recvRate <- effectiveRecvRate
40794183 let recvResult = checkPubkeyRate recvRates to recvRate recvRateWindowMs maxRecvRateEntries t
4080− if recvResult.limited then yield {status: MsRateLimited {}, remoteServers: [], retryAfterMs: recvResult.retryAfterMs}
4184+ if recvResult.limited then yield {status: MsRateLimited {}, remoteServers: [], retryAfterMs: recvResult.retryAfterMs, msgExpiresAt: 0}
40814185 else do
40824186 set *recvRates = recvResult.entries
40834187 totalMsgCount <- *totalMessageCount
4084− if totalMsgCount >= maxMessages then yield {status: MsRejected {}, remoteServers: [], retryAfterMs: 0}
4188+ if totalMsgCount >= maxMessages then yield {status: MsRejected {}, remoteServers: [], retryAfterMs: 0, msgExpiresAt: 0}
40854189 else do
40864190 recipientCounts <- readRcBucket msgBucket
40874191 let recipientCount = getRecipientCount to recipientCounts
4088− if recipientCount >= maxMessagesPerRecipient then yield {status: MsRejected {}, remoteServers: [], retryAfterMs: 0}
4192+ if recipientCount >= maxMessagesPerRecipient then yield {status: MsRejected {}, remoteServers: [], retryAfterMs: 0, msgExpiresAt: 0}
40894193 else do
40904194 recipientPresence <- readPresenceBucket (pubkeyBucket to)
40914195 seqCounter <- *seqCounter
@@ -4094,7 +4198,7 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId messageTt
40944198 let msgExpiresAt = computeMessageExpiresAt seqTs messageTtl
40954199 writeMessageBucket msgBucket (union bucket
40964200 [{messageId: effectiveMsgId, toKey: to,
4097− encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs, expiresAt: msgExpiresAt}])
4201+ encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs, expiresAt: msgExpiresAt, senderKey: senderKey}])
40984202 set *totalMessageCount = totalMsgCount + 1
40994203 writeRcBucket msgBucket (adjustRecipientCount recipientCounts to 1)
41004204 signalMessageForKey to seqTs
@@ -4126,10 +4230,12 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId messageTt
41264230 let status = if localOnline then MsDelivered {}
41274231 else if federationEnabled && forwardsEnqueued then MsFederated {}
41284232 else MsQueued {}
4129− yield {status: status, remoteServers: remoteServers, retryAfterMs: 0}
4233+ yield {status: status, remoteServers: remoteServers, retryAfterMs: 0, msgExpiresAt: msgExpiresAt}
41304234
4131−-- Fork federation forwards for messages that need to be delivered to remote servers
4132−forkSendForwards = \effectiveMsgId to encryptedBlob clampedTs messageTtl remoteServers ->
4235+-- Fork federation forwards for messages that need to be delivered to remote servers.
4236+-- msgExpiresAt: the stored message's expiration timestamp (used to compute remaining TTL
4237+-- at forward time, ensuring the remote server gets the correct remaining lifetime).
4238+forkSendForwards = \effectiveMsgId to encryptedBlob clampedTs msgExpiresAt remoteServers ->
41334239 when (count remoteServers > 0 && federationEnabled) (
41344240 fork (sendToEach remoteServers (\server -> do
41354241 stillPending <- atomic do
@@ -4139,8 +4245,10 @@ forkSendForwards = \effectiveMsgId to encryptedBlob clampedTs messageTtl remoteS
41394245 fwdT <- now
41404246 fwdAuth <- generateFederationAuth "forward"
41414247 fwdCorrId <- genTokenN 1
4142− -- Compute remaining TTL at forward time (may have decreased since enqueue)
4143− let fwdTtl = if messageTtl > 0 then max 0 (computeMessageExpiresAt fwdT messageTtl - fwdT) else 0
4248+ -- Compute remaining TTL from the stored message's expiration (not the original TTL).
4249+ -- This ensures the remote server gets the correct remaining lifetime, even if
4250+ -- the fork is delayed by network issues or server load.
4251+ let fwdTtl = if msgExpiresAt > 0 then max 0 (msgExpiresAt - fwdT) else 0
41444252 fwdResult <- fetchFederation server (ForwardMessage {
41454253 toKey: to, encryptedBlob: encryptedBlob,
41464254 timestamp: clampedTs, originServer: serverHostN, messageId: effectiveMsgId,
@@ -4171,11 +4279,13 @@ sendBatchForwardsSequential = \msgs server ->
41714279 fwdT <- now
41724280 fwdAuth <- generateFederationAuth "forward"
41734281 fwdCorrId <- genTokenN 1
4282+ -- Use stored msgExpiresAt to compute remaining TTL (not original messageTtl)
4283+ let fwdTtl = if r.msgExpiresAt > 0 then max 0 (r.msgExpiresAt - fwdT) else 0
41744284 fwdResult <- fetchFederation server (ForwardMessage {
41754285 toKey: r.toKey, encryptedBlob: r.blob,
41764286 timestamp: r.ts, originServer: serverHostN, messageId: r.messageId,
41774287 fromServer: serverHostN, requestTimestamp: fwdT,
4178− authorization: fwdAuth, correlationId: fwdCorrId, messageTtl: r.messageTtl})
4288+ authorization: fwdAuth, correlationId: fwdCorrId, messageTtl: fwdTtl})
41794289 case fwdResult of
41804290 Err e -> do
41814291 logWarn ("federation: forward to " ++ server ++ " failed for " ++ r.messageId ++ ": " ++ show e)
@@ -4207,15 +4317,15 @@ handleSendMessage = \to encryptedBlob timestamp messageId messageTtl authorizati
42074317 let clampedTs = clampTimestamp timestamp t
42084318 result <- atomic do
42094319 dp <- *drainPhase
4210− if dp >= 2 then yield {status: MsDraining {}, remoteServers: [], retryAfterMs: 0}
4320+ if dp >= 2 then yield {status: MsDraining {}, remoteServers: [], retryAfterMs: 0, msgExpiresAt: 0}
42114321 else withSessionAuth authorization t clientIp
4212− (yield {status: MsUnauthorized {}, remoteServers: [], retryAfterMs: 0})
4322+ (yield {status: MsUnauthorized {}, remoteServers: [], retryAfterMs: 0, msgExpiresAt: 0})
42134323 (\senderKey -> sendMessageCore senderKey to encryptedBlob clampedTs effectiveMsgId messageTtl t)
42144324 case result.status of
42154325 MsRateLimited {} -> logWarn ("sendMessage: rate limited for " ++ effectiveMsgId)
42164326 MsRejected {} -> when (result.retryAfterMs == 0) (logWarn ("sendMessage: rejected for " ++ effectiveMsgId ++ " (capacity or mailbox full)"))
42174327 _ -> yield {}
4218− forkSendForwards effectiveMsgId to encryptedBlob clampedTs messageTtl result.remoteServers
4328+ forkSendForwards effectiveMsgId to encryptedBlob clampedTs result.msgExpiresAt result.remoteServers
42194329 yield (case result.status of
42204330 MsDuplicate {} -> respond {status: MsDuplicate {}, messageId: effectiveMsgId, retryAfterMs: 0}
42214331 MsDelivered {} -> respond {status: MsDelivered {}, messageId: effectiveMsgId, retryAfterMs: 0}
@@ -4251,6 +4361,37 @@ handleLookupPresence = \pubkey authorization clientIp respond -> do
42514361 yield {server: maskedServer, authorized: True {}})
42524362 yield (respond {server: result.server, authorized: result.authorized})
42534363
4364+-- POST /lookup/batch: Look up presence for multiple pubkeys in one request.
4365+-- Reads only the needed presence buckets (not all 16), reducing round trips
4366+-- vs sequential single-key lookups. Uses one rate limit check for the whole batch.
4367+handleBatchLookupPresence = \pubkeys authorization clientIp respond -> do
4368+ t <- now
4369+ globalRateLimited <- checkGlobalRate t
4370+ if globalRateLimited then yield (respond {results: [], authorized: False {}})
4371+ else do
4372+ dp <- *drainPhase
4373+ if dp >= 2 then yield (respond {results: [], authorized: False {}})
4374+ else
4375+ withSessionAuth authorization t clientIp
4376+ (yield (respond {results: [], authorized: False {}}))
4377+ (\senderKey -> do
4378+ lookupRateLimited <- atomic do
4379+ lookupRates <- *lookupRates
4380+ let rateResult = checkPubkeyRate lookupRates senderKey maxLookupRatePerKey lookupRateWindowMs maxLookupRateEntries t
4381+ set *lookupRates = rateResult.entries
4382+ yield rateResult.limited
4383+ if lookupRateLimited then yield (respond {results: [], authorized: True {}})
4384+ else do
4385+ let cappedKeys = takeRelation maxBatchLookupKeys pubkeys
4386+ result <- atomic do
4387+ let neededBuckets = uniqueBucketsFor cappedKeys (\k -> k)
4388+ relevantPresence <- readSelectedPresenceBuckets neededBuckets
4389+ yield (do key <- cappedKeys; yield (
4390+ let foundServer = bestServerForKey (presenceForKey relevantPresence key t)
4391+ in let maskedServer = if foundServer == "" then "" else if isLocalServer foundServer then foundServer else "remote"
4392+ in {pubkey: key, server: maskedServer}))
4393+ yield (respond {results: result, authorized: True {}}))
4394+
42544395-- POST /messages/status: Check delivery status of a previously sent message.
42554396-- Caller must provide both messageId and recipient pubkey (proves they know the recipient).
42564397-- Returns: "pending" (in mailbox), "forwarding" (federation in progress),
@@ -4614,7 +4755,7 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
46144755 let fwdExpiresAt = computeMessageExpiresAt seqTs messageTtl
46154756 writeMessageBucket bkt (union bucket
46164757 [{messageId: messageId, toKey: toKey,
4617− encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs, expiresAt: fwdExpiresAt}])
4758+ encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs, expiresAt: fwdExpiresAt, senderKey: ""}])
46184759 set *totalMessageCount = totalMsgCount + 1
46194760 writeRcBucket bkt (adjustRecipientCount recipientCounts toKey 1)
46204761 signalMessageForKey toKey seqTs
@@ -4657,11 +4798,9 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp corre
46574798 let dedupedIds = dedupMessageIds messageIds
46584799 -- Clear forward entries for the acking server. Extract removed forwards' toKey
46594800 -- so we can reclaim origin messages that are no longer needed.
4660− removedForwards <- atomic do
4661− forwards <- readAllForwards
4662− let removed = do f <- forwards; where f.toServer == fromServer && any dedupedIds (\mid -> f.messageId == mid); yield f
4663− writeAllForwards (forwardsExcluding forwards fromServer dedupedIds)
4664− yield removed
4801+ -- Optimized: scan all 16 forward buckets but only rewrite those that changed.
4802+ -- Avoids the O(n) writeAllForwards repartitioning when only a few buckets are affected.
4803+ removedForwards <- removeForwardsByServer fromServer dedupedIds 0 []
46654804 let ackedCount = count removedForwards
46664805 -- Guard: warn if peer acks messages we had no forward entries for. This can indicate
46674806 -- a peer acking messages it never received (e.g., the forward response was lost but
@@ -5010,13 +5149,10 @@ handleStatsPeers = \authorization respond -> do
50105149 if authorized
50115150 then do
50125151 failures <- *forwardFailures
5013− forwards <- readAllForwards
5014− -- Single-pass aggregation: accumulate {server, count} pairs in one fold over forwards.
5015− -- O(n*k) where k = unique servers (vs previous O(n*k + k*n) two-pass approach).
5016− let fwdCounts = fold (\acc f ->
5017− if any acc (\c -> c.server == f.toServer)
5018− then do c <- acc; yield (if c.server == f.toServer then {server: c.server, count: c.count + 1} else c)
5019− else union acc [{server: f.toServer, count: 1}]) [] forwards
5152+ -- Per-bucket aggregation: compute forward counts without materializing all 16 buckets.
5153+ -- Reuses computeForwardStats which reads one bucket at a time.
5154+ fwdStats <- computeForwardStats 0 t
5155+ let fwdCounts = fwdStats.serverCounts
50205156 let fwdCountFor = \server -> fold (\acc c -> if c.server == server then c.count else acc) 0 fwdCounts
50215157 let peers = do f <- failures; yield {server: f.server, failures: f.failures, lastFailedAt: f.failedAt, firstFailedAt: f.firstFailedAt, backoffMs: retryBackoffFor f.failures f.server, retriable: shouldRetryServer f.server t failures, pendingForwards: fwdCountFor f.server}
50225158 -- Include servers with pending forwards but no failure records
@@ -5337,9 +5473,9 @@ handleDeleteMessages = \messageIds authorization clientIp respond -> do
53375473 yield (respond {ok: True {}, authorized: True {}, deletedCount: deleteResult.deletedCount}))
53385474
53395475-- POST /messages/cancel: Retract a previously sent message before the recipient reads it.
5340−-- Caller provides messageId and recipient pubkey. Knowing the messageId is proof of being
5341−-- the sender (either client-generated or returned only in the send response).
5342−-- Removes from recipient mailbox, cancels pending forwards, sends federation acks.
5476+-- Only works for locally-originated messages where the caller's pubkey matches the
5477+-- stored senderKey. Federation forwards cannot be cancelled via this endpoint.
5478+-- Removes from recipient mailbox and cancels pending forwards.
53435479handleCancelMessage = \messageId to authorization clientIp respond -> do
53445480 t <- now
53455481 dp <- *drainPhase
@@ -5365,7 +5501,13 @@ handleCancelMessage = \messageId to authorization clientIp respond -> do
53655501 let toCancel = headOf (do m <- bucket; where m.messageId == messageId && m.toKey == to; yield m)
53665502 case toCancel of
53675503 Nothing {} -> yield {cancelled: False {}, pendingAcks: [], evictedEntries: []}
5368− Just {value: msg} -> do
5504+ Just {value: msg} ->
5505+ -- Only allow cancelling locally-originated messages (originServer == "")
5506+ -- where the caller is the actual sender. Federation forwards have no
5507+ -- senderKey and should not be cancellable via this endpoint.
5508+ if msg.originServer != "" then yield {cancelled: False {}, pendingAcks: [], evictedEntries: []}
5509+ else if msg.senderKey != "" && msg.senderKey != senderKey then yield {cancelled: False {}, pendingAcks: [], evictedEntries: []}
5510+ else do
53695511 writeMessageBucket b (do m <- bucket; where not (m.messageId == messageId && m.toKey == to); yield m)
53705512 tc <- *totalMessageCount
53715513 set *totalMessageCount = max 0 (tc - 1)
@@ -5374,16 +5516,9 @@ handleCancelMessage = \messageId to authorization clientIp respond -> do
53745516 -- Cancel pending forwards
53755517 fwdBkt <- readFwdBucket b
53765518 writeFwdBucket b (do f <- fwdBkt; where not (f.messageId == messageId && f.toKey == to); yield f)
5377− -- Create ack entries for federated messages so origin server learns about cancel
5378− if msg.originServer != "" then do
5379− let ackedFederated = [{messageId: msg.messageId, toKey: msg.toKey, ackedAt: t, originServer: msg.originServer, retries: 0}]
5380− ackedBkt <- readAckedBucket b
5381− let combinedAcked = union ackedBkt ackedFederated
5382− let excessCount = max 0 (count combinedAcked - maxAckedIds)
5383− let eviction = evictSmallest combinedAcked excessCount (\a -> a.ackedAt)
5384− writeAckedBucket b eviction.kept
5385− yield {cancelled: True {}, pendingAcks: ackedFederated, evictedEntries: eviction.evicted}
5386− else yield {cancelled: True {}, pendingAcks: [], evictedEntries: []}
5519+ -- Only locally-originated messages reach this point (originServer == ""),
5520+ -- so no federation acks are needed.
5521+ yield {cancelled: True {}, pendingAcks: [], evictedEntries: []}
53875522 fork (when (federationEnabled && count cancelResult.pendingAcks > 0) (do
53885523 sendAcksAndClean cancelResult.pendingAcks
53895524 retryEvictedAcks cancelResult.evictedEntries cancelResult.pendingAcks))
@@ -5441,18 +5576,18 @@ sendBatchInner = \senderKey messages t acc ->
54415576 when (msg.timestamp < t - messageMaxAge / 2 && t - msg.timestamp <= messageMaxAge)
54425577 (logWarn ("sendBatch: stale timestamp (" ++ show ((t - msg.timestamp) / 86400000) ++ " days old) for " ++ effectiveMsgId))
54435578 result <- if t - msg.timestamp > messageMaxAge
5444− then yield {messageId: effectiveMsgId, status: MsRejected {}, remoteServers: [], toKey: "", blob: "", ts: 0, messageTtl: 0, retryAfterMs: 0}
5579+ then yield {messageId: effectiveMsgId, status: MsRejected {}, remoteServers: [], toKey: "", blob: "", ts: 0, msgExpiresAt: 0, retryAfterMs: 0}
54455580 else do
54465581 let clampedTs = clampTimestamp msg.timestamp t
54475582 sendResult <- atomic do
54485583 dp <- *drainPhase
5449− if dp >= 2 then yield {status: MsDraining {}, remoteServers: [], retryAfterMs: 0}
5584+ if dp >= 2 then yield {status: MsDraining {}, remoteServers: [], retryAfterMs: 0, msgExpiresAt: 0}
54505585 else sendMessageCore senderKey msg.to msg.encryptedBlob clampedTs effectiveMsgId msg.messageTtl t
54515586 case sendResult.status of
54525587 MsRateLimited {} -> logWarn ("sendBatch: rate limited for " ++ effectiveMsgId)
54535588 MsRejected {} -> logWarn ("sendBatch: rejected for " ++ effectiveMsgId ++ " (capacity or mailbox full)")
54545589 _ -> yield {}
5455− yield {messageId: effectiveMsgId, status: sendResult.status, remoteServers: sendResult.remoteServers, toKey: msg.to, blob: msg.encryptedBlob, ts: clampedTs, messageTtl: msg.messageTtl, retryAfterMs: sendResult.retryAfterMs}
5590+ yield {messageId: effectiveMsgId, status: sendResult.status, remoteServers: sendResult.remoteServers, toKey: msg.to, blob: msg.encryptedBlob, ts: clampedTs, msgExpiresAt: sendResult.msgExpiresAt, retryAfterMs: sendResult.retryAfterMs}
54565591 sendBatchInner senderKey split.rest t (union acc [result])
54575592
54585593-- POST /maintenance/block: Add a server to the federation blocklist.
@@ -5497,10 +5632,55 @@ handleListBlocked = \authorization respond -> do
54975632 blocked <- *blockedPeers
54985633 yield (respond {peers: blocked, ok: True {}})
54995634
5635+-- POST /maintenance/allow: Add a server to the federation allowlist.
5636+-- When the allowlist is non-empty, only servers on the list can federate with us.
5637+-- Adding the first server transitions from open federation to allowlist-only mode.
5638+handleAllowPeer = \server authorization respond -> do
5639+ let authorized = verifyMaintenanceAuth authorization
5640+ if not authorized then yield (respond {ok: False {}, allowed: 0})
5641+ else do
5642+ result <- atomic do
5643+ allowed <- *allowedPeers
5644+ if any allowed (\a -> a == server) then yield {alreadyAllowed: True {}, total: count allowed}
5645+ else do
5646+ let updated = if count allowed >= maxAllowedPeers
5647+ then case headOf allowed of
5648+ Nothing {} -> [server]
5649+ Just {value: oldest} -> union (do a <- allowed; where a != oldest; yield a) [server]
5650+ else union allowed [server]
5651+ set *allowedPeers = updated
5652+ yield {alreadyAllowed: False {}, total: count updated}
5653+ when (not result.alreadyAllowed) (logInfo ("allow: added " ++ server ++ " to allowlist (" ++ show result.total ++ " total)"))
5654+ yield (respond {ok: True {}, allowed: result.total})
5655+
5656+-- POST /maintenance/unallow: Remove a server from the federation allowlist.
5657+-- Removing the last server transitions back to open federation mode.
5658+handleUnallowPeer = \server authorization respond -> do
5659+ let authorized = verifyMaintenanceAuth authorization
5660+ if not authorized then yield (respond {ok: False {}, removed: False {}})
5661+ else do
5662+ removed <- atomic do
5663+ allowed <- *allowedPeers
5664+ let had = any allowed (\a -> a == server)
5665+ set *allowedPeers = do a <- allowed; where a != server; yield a
5666+ yield had
5667+ when removed (logInfo ("unallow: removed " ++ server ++ " from allowlist"))
5668+ yield (respond {ok: True {}, removed: removed})
5669+
5670+-- GET /maintenance/allowed: List currently allowed federation peers.
5671+-- Empty list = open federation (all peers allowed).
5672+handleListAllowed = \authorization respond -> do
5673+ let authorized = verifyMaintenanceAuth authorization
5674+ if not authorized then yield (respond {peers: [], ok: False {}})
5675+ else do
5676+ allowed <- *allowedPeers
5677+ yield (respond {peers: allowed, ok: True {}})
5678+
55005679-- POST /maintenance/ping: Test federation connectivity to a specific peer.
5501−-- Sends an empty gossip request via fetchFederation, triggering a full verify callback
5502−-- round-trip (peer calls back to our /federation/verify to authenticate the request).
5503−-- This exercises the real federation auth path and tracks in-flight/latency metrics.
5680+-- Sends an empty gossip (0 events, hopsRemaining=1) to the peer via the full
5681+-- federation auth path. The peer calls back to our /federation/verify to authenticate,
5682+-- then processes the empty gossip (no mutations, no re-broadcast since hops=1).
5683+-- This exercises the real bidirectional connectivity and auth flow.
55045684handlePingPeer = \server authorization respond -> do
55055685 let authorized = verifyMaintenanceAuth authorization
55065686 if not authorized then yield (respond {ok: False {}, latencyMs: 0, error: "unauthorized"})
@@ -5612,6 +5792,21 @@ handleAdminPurgeMessages = \pubkey authorization respond -> do
56125792 retryEvictedAcks purgeResult.evictedEntries purgeResult.pendingAcks))
56135793 yield (respond {ok: True {}, purgedCount: purgeResult.purgedCount})
56145794
5795+-- POST /maintenance/nonces/flush: Clear all forward nonces (recovery from corruption).
5796+-- Nonces provide extended replay protection beyond message/ack dedup lifetime.
5797+-- Flushing temporarily reopens a replay window for messages older than 30 days but
5798+-- younger than 60 days. Use only for recovery, not routine maintenance.
5799+handleFlushNonces = \authorization respond -> do
5800+ let authorized = verifyMaintenanceAuth authorization
5801+ if not authorized then yield (respond {ok: False {}, flushedCount: 0})
5802+ else do
5803+ flushedCount <- atomic do
5804+ total <- countAllNonces 0
5805+ writeAllNonces []
5806+ yield total
5807+ logInfo ("nonces/flush: flushed " ++ show flushedCount ++ " forward nonces")
5808+ yield (respond {ok: True {}, flushedCount: flushedCount})
5809+
56155810-- GET /maintenance/rate-limits: Return current effective rate limits.
56165811handleGetRateLimits = \authorization respond -> do
56175812 let authorized = verifyMaintenanceAuth authorization
@@ -5658,6 +5853,7 @@ requestPath = \req -> case req of
56585853 DeleteMessages {} -> "POST /messages/delete"
56595854 CancelMessage {} -> "POST /messages/cancel"
56605855 LookupPresence {} -> "GET /lookup"
5856+ BatchLookupPresence {} -> "POST /lookup/batch"
56615857 RecvGossip {} -> "POST /federation/gossip"
56625858 ForwardMessage {} -> "POST /federation/forward"
56635859 FederationAck {} -> "POST /federation/ack"
@@ -5677,6 +5873,10 @@ requestPath = \req -> case req of
56775873 AdminPurgeMessages {} -> "POST /maintenance/messages/purge"
56785874 GetRateLimits {} -> "GET /maintenance/rate-limits"
56795875 SetRateLimits {} -> "POST /maintenance/rate-limits"
5876+ FlushNonces {} -> "POST /maintenance/nonces/flush"
5877+ AllowPeer {} -> "POST /maintenance/allow"
5878+ UnallowPeer {} -> "POST /maintenance/unallow"
5879+ ListAllowed {} -> "GET /maintenance/allowed"
56805880 Health {} -> "GET /health"
56815881 Stats {} -> "GET /stats"
56825882 StatsPeers {} -> "GET /stats/peers"
@@ -5699,6 +5899,7 @@ requestClientIp = \req -> case req of
56995899 DeleteMessages {clientIp} -> clientIp
57005900 CancelMessage {clientIp} -> clientIp
57015901 LookupPresence {clientIp} -> clientIp
5902+ BatchLookupPresence {clientIp} -> clientIp
57025903 VerifyToken {clientIp} -> clientIp
57035904 RecvGossip {} -> ""
57045905 ForwardMessage {} -> ""
@@ -5718,6 +5919,10 @@ requestClientIp = \req -> case req of
57185919 AdminPurgeMessages {} -> ""
57195920 GetRateLimits {} -> ""
57205921 SetRateLimits {} -> ""
5922+ FlushNonces {} -> ""
5923+ AllowPeer {} -> ""
5924+ UnallowPeer {} -> ""
5925+ ListAllowed {} -> ""
57215926 Health {} -> ""
57225927 Stats {} -> ""
57235928 StatsPeers {} -> ""
@@ -5743,6 +5948,7 @@ serveInner = \req -> case req of
57435948 DeleteMessages {messageIds, authorization, clientIp, respond} -> handleDeleteMessages messageIds authorization clientIp respond
57445949 CancelMessage {messageId, to, authorization, clientIp, respond} -> handleCancelMessage messageId to authorization clientIp respond
57455950 LookupPresence {pubkey, authorization, clientIp, respond} -> handleLookupPresence pubkey authorization clientIp respond
5951+ BatchLookupPresence {pubkeys, authorization, clientIp, respond} -> handleBatchLookupPresence pubkeys authorization clientIp respond
57465952 RecvGossip {events, hopsRemaining, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleRecvGossip events hopsRemaining fromServer requestTimestamp correlationId authorization respond
57475953 ForwardMessage {toKey, encryptedBlob, timestamp, originServer, messageId, fromServer, requestTimestamp, correlationId, messageTtl, authorization, respond} -> handleForwardMessage toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp correlationId messageTtl authorization respond
57485954 FederationAck {messageIds, originServer, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleFederationAck messageIds originServer fromServer requestTimestamp correlationId authorization respond
@@ -5762,6 +5968,10 @@ serveInner = \req -> case req of
57625968 AdminPurgeMessages {pubkey, authorization, respond} -> handleAdminPurgeMessages pubkey authorization respond
57635969 GetRateLimits {authorization, respond} -> handleGetRateLimits authorization respond
57645970 SetRateLimits {globalRequest, ipRate, sendRate, federationRate, authGlobal, recvRate, authorization, respond} -> handleSetRateLimits globalRequest ipRate sendRate federationRate authGlobal recvRate authorization respond
5971+ FlushNonces {authorization, respond} -> handleFlushNonces authorization respond
5972+ AllowPeer {server, authorization, respond} -> handleAllowPeer server authorization respond
5973+ UnallowPeer {server, authorization, respond} -> handleUnallowPeer server authorization respond
5974+ ListAllowed {authorization, respond} -> handleListAllowed authorization respond
57655975 Health {authorization, respond} -> handleHealth authorization respond
57665976 Stats {authorization, respond} -> handleStats authorization respond
57675977 StatsPeers {authorization, respond} -> handleStatsPeers authorization respond
@@ -6040,7 +6250,7 @@ resetFederationLatencyMetrics = atomic do
60406250 set *federationLatencyResetAt = t
60416251
60426252main = do
6043− let configOk = maxMessages > 0 && maxMessagesPerRecipient > 0 && maxPresenceEntries > 0
6253+ let configOk = listenPort > 0 && maxMessages > 0 && maxMessagesPerRecipient > 0 && maxPresenceEntries > 0
60446254 && maxSessions > 0 && maxChallenges > 0 && backgroundPruneInterval > 0
60456255 && pollTimeoutMs > 0 && challengeExpiry > 0 && sessionExpiry > 0
60466256 && maxSessionLifetime > 0
@@ -6094,7 +6304,7 @@ main = do
60946304 logInfo "IP binding is active — sessions created with a ClientIp header will reject requests without one"
60956305 when federationEnabled
60966306 (logWarn "SECURITY: DNS rebinding risk — federation URLs resolve DNS at fetch time, not at validation time. A peer hostname passing isBadServerName could rebind to a private IP between validation and fetch. Mitigate by blocking outbound connections to RFC1918/link-local/CGNAT ranges at the firewall level.")
6097− logInfo "Skrepka relay server starting on port 8080..."
6307+ logInfo ("Skrepka relay server starting on port " ++ show listenPort ++ "...")
60986308 logInfo (" federation: " ++ if federationEnabled then "enabled" else "disabled")
60996309 logInfo (" prune interval: " ++ show backgroundPruneInterval ++ "ms")
61006310 logInfo (" poll timeout: " ++ show pollTimeoutMs ++ "ms")
@@ -6113,6 +6323,7 @@ main = do
61136323 logInfo (" ack retry limit: " ++ show maxAckRetries ++ " per entry")
61146324 when (federationCircuitBreakerThreshold > 0)
61156325 (logInfo (" circuit breaker: " ++ show federationCircuitBreakerThreshold ++ " failing peers"))
6326+ logInfo (" federation allowlist: " ++ show maxAllowedPeers ++ " max (empty = open federation)")
61166327 when (autoSnapshotInterval > 0)
61176328 (logInfo (" auto-snapshot: " ++ show autoSnapshotInterval ++ "ms"))
61186329 when (autoSnapshotExportInterval > 0)
@@ -6129,4 +6340,4 @@ main = do
61296340 logInfo "startup: probing federation peers..."
61306341 startupProbePeers
61316342 resetFederationLatencyMetrics))
6132− listen 8080 serve
6343+ listen listenPort serve