@@ -29,9 +29,8 @@ federationScheme = "https://" -- scheme for federation HTTP calls
29-- Optional pre-shared key for federation auth. When non-empty and a peer's server name
29-- Optional pre-shared key for federation auth. When non-empty and a peer's server name
30-- matches, the key is used instead of callback verification — eliminates the verify
30-- matches, the key is used instead of callback verification — eliminates the verify
31-- round-trip for trusted peers. Format: "server1:key1,server2:key2".
31-- round-trip for trusted peers. Format: "server1:key1,server2:key2".
32−-- TODO: Implement pre-shared key parsing and lookup. Currently callback verification
32+-- Keys must be at least 16 hex characters. Peers with pre-shared keys skip the
33−-- is always used. Pre-shared keys would halve federation latency for configured peers
33+-- /federation/verify callback entirely, halving federation latency.
34−-- and remove the /federation/verify endpoint as an attack surface for those peers.
35federationPresharedKeys = ""
34federationPresharedKeys = ""
36-- NOTE: listenHost is not configurable — Knot's listen primitive only accepts a port.
35-- NOTE: listenHost is not configurable — Knot's listen primitive only accepts a port.
37-- To restrict the bind address (e.g., 127.0.0.1 behind a reverse proxy), use
36-- To restrict the bind address (e.g., 127.0.0.1 behind a reverse proxy), use
@@ -882,6 +881,12 @@ route FederationApi where
882 POST {messageIds: [MessageIdHex], originServer: ServerName, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text} /ack headers {authorization: Text}
881 POST {messageIds: [MessageIdHex], originServer: ServerName, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text} /ack headers {authorization: Text}
883 -> {ok: Bool, error: Text, ackedCount: Int}
882 -> {ok: Bool, error: Text, ackedCount: Int}
884 = FederationAck
883 = FederationAck
884+ -- Cancel (retract) a previously forwarded message. The origin server sends this
885+ -- when the sender cancels a message via /messages/cancel. The receiving server
886+ -- deletes the message if it hasn't been polled by the recipient yet.
887+ POST {messageId: MessageIdHex, toKey: PubkeyHex, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text} /cancel headers {authorization: Text}
888+ -> {ok: Bool, error: Text, cancelled: Bool}
886-- Federation callback verification: the receiving server calls back to the sender
891-- Federation callback verification: the receiving server calls back to the sender
887-- to confirm a request token is genuine. No shared secret needed.
892-- to confirm a request token is genuine. No shared secret needed.
@@ -1324,41 +1329,107 @@ isBlockedPeer = \server blocked -> any blocked (\b -> b == server)
1324isAllowedByAllowlist : Text -> [Text] -> Bool
1329isAllowedByAllowlist : Text -> [Text] -> Bool
1325isAllowedByAllowlist = \server allowed -> count allowed == 0 || any allowed (\a -> a == server)
1330isAllowedByAllowlist = \server allowed -> count allowed == 0 || any allowed (\a -> a == server)
1332+-- ============================================
1333+-- Pre-shared Federation Key Helpers
1334+-- ============================================
1335+-- Parse "server1:key1,server2:key2" into a lookup-ready list.
1336+-- Keys provide HMAC-style auth: the outbound side sends "PresharedKey <key>" in the
1337+-- Authorization header; the inbound side checks safeCompare against the stored key.
1338+-- No callback round-trip needed for configured peers.
1340+-- Parse one "server:key" entry from the head of a comma-separated string.
1341+-- Returns {server, key, rest} where rest is everything after the comma (or "").
1342+parsePskEntry : Text -> {server: Text, key: Text, rest: Text}
1343+parsePskEntry = \s -> parsePskServer s ""
1345+parsePskServer : Text -> Text -> {server: Text, key: Text, rest: Text}
1346+parsePskServer = \s acc ->
1347+ if s == "" then {server: acc, key: "", rest: ""}
1348+ else if take 1 s == ":" then parsePskKey (drop 1 s) acc ""
1349+ else if take 1 s == "," then {server: acc, key: "", rest: drop 1 s}
1350+ else parsePskServer (drop 1 s) (acc ++ take 1 s)
1352+parsePskKey : Text -> Text -> Text -> {server: Text, key: Text, rest: Text}
1353+parsePskKey = \s server acc ->
1354+ if s == "" then {server: server, key: acc, rest: ""}
1355+ else if take 1 s == "," then {server: server, key: acc, rest: drop 1 s}
1356+ else parsePskKey (drop 1 s) server (acc ++ take 1 s)
1358+-- Parse the full config string into a list of {server, key} pairs.
1359+-- Filters out entries with empty server/key or keys shorter than 16 chars.
1360+parsePresharedKeys : Text -> [{server: Text, key: Text}]
1361+parsePresharedKeys = \s ->
1364+ let entry = parsePskEntry s in
1365+ let rest = parsePresharedKeys entry.rest in
1366+ if entry.server == "" || entry.key == "" || not (tooLong entry.key 15)
1368+ else union [{server: normalizeServerName entry.server, key: entry.key}] rest
1370+-- Pre-parsed at startup (evaluated once since federationPresharedKeys is a constant)
1371+presharedKeyList = parsePresharedKeys federationPresharedKeys
1373+-- Look up a pre-shared key for a server. Returns "" if no key configured.
1374+lookupPresharedKey : Text -> Text
1375+lookupPresharedKey = \server ->
1376+ fold (\acc entry -> if acc != "" then acc else if entry.server == server then entry.key else "") "" presharedKeyList
1378+-- Check if an authorization header contains a valid pre-shared key for the given server.
1379+-- Uses constant-time comparison to prevent timing side-channels.
1380+verifyPresharedKey : Text -> Text -> Bool
1381+verifyPresharedKey = \server authorization ->
1382+ case extractToken authorization of
1383+ Nothing {} -> False {}
1384+ Just {value: token} ->
1385+ let psk = lookupPresharedKey server in
1386+ psk != "" && safeCompare token psk
1327-- Generate a single-use token for outbound federation auth.
1388-- Generate a single-use token for outbound federation auth.
1328−-- The receiving server calls back to /federation/verify to confirm we issued it.
1389+-- If a pre-shared key is configured for the target server, uses it directly
1390+-- (no token generation, no pending verification entry, no callback needed).
1391+-- Otherwise falls back to the callback verification flow.
1329-- scope binds the token to a request type ("gossip", "forward", "ack") so a token
1392-- scope binds the token to a request type ("gossip", "forward", "ack") so a token
1330-- issued for one operation cannot authorize a different one.
1393-- issued for one operation cannot authorize a different one.
1331−generateFederationAuth : Text -> IO {network} Text
1394+generateFederationAuth : Text -> Text -> IO {network} Text
1332−generateFederationAuth = \scope -> do
1395+generateFederationAuth = \scope server ->
1396+ -- Fast path: use pre-shared key if configured for this peer (no token generation,
1397+ -- no pending verification entry, no callback round-trip on the receiving side)
1398+ let psk = lookupPresharedKey server in
1399+ if psk != "" then yield ("Bearer " ++ psk)
1333− token <- genToken
1401+ token <- genToken
1334− t <- now
1402+ t <- now
1335− atomic do
1403+ atomic do
1336− pending <- *pendingVerifications
1404+ pending <- *pendingVerifications
1337− let updated = union pending [{token: token, createdAt: t, scope: scope}]
1405+ let updated = union pending [{token: token, createdAt: t, scope: scope}]
1338− set *pendingVerifications = if count updated > maxPendingVerifications
1406+ set *pendingVerifications = if count updated > maxPendingVerifications
1339− then evictOneOldest updated (\p -> p.createdAt)
1407+ then evictOneOldest updated (\p -> p.createdAt)
1340− else updated
1408+ else updated
1341− yield ("Bearer " ++ token)
1409+ yield ("Bearer " ++ token)
1411+-- Verify an inbound federation token. First checks if a pre-shared key is configured
1412+-- for the sender — if so, verifies via constant-time comparison (no network round-trip).
1343−-- Verify an inbound federation token by callback to the sender's /federation/verify endpoint.
1413+-- Otherwise falls back to callback verification via the sender's /federation/verify endpoint.
1344−-- Uses fetchFederation for in-flight tracking, connection counting, and slow-response logging,
1414+-- Pre-shared key auth halves federation latency and removes the verify endpoint as an
1345−-- preventing a malicious peer from slow-loris attacks on the verify endpoint.
1415+-- attack surface for configured peers.
1346−-- Single-use: the sender consumes the token on verification.
1347−-- scope must match the scope the token was generated with; empty scope accepts any (backward compat).
1348-- No retry: a failed verify rejects the request. The peer's federation rate limit + circuit breaker
1416-- No retry: a failed verify rejects the request. The peer's federation rate limit + circuit breaker
1349-- handle unreliable peers; retrying here doubles verify traffic and blocks with a 1s sleep per op,
1417-- handle unreliable peers; retrying here doubles verify traffic and blocks with a 1s sleep per op,
1350-- which a malicious peer can exploit for amplification.
1418-- which a malicious peer can exploit for amplification.
1351verifyFederationCallback : Text -> Text -> Text -> IO {network} Bool
1419verifyFederationCallback : Text -> Text -> Text -> IO {network} Bool
1352verifyFederationCallback = \fromServer authorization scope ->
1420verifyFederationCallback = \fromServer authorization scope ->
1421+ -- Fast path: pre-shared key verification (no network round-trip)
1422+ if verifyPresharedKey fromServer authorization then yield (True {})
1353− case extractToken authorization of
1424+ case extractToken authorization of
1354− Nothing {} -> yield (False {})
1425+ Nothing {} -> yield (False {})
1355− Just {value: token} -> do
1426+ Just {value: token} -> do
1356− result <- fetchFederation fromServer (VerifyToken {token: token, scope: scope})
1427+ result <- fetchFederation fromServer (VerifyToken {token: token, scope: scope})
1357− case result of
1428+ case result of
1358− Err e -> do
1429+ Err e -> do
1359− logWarn ("federation: verify callback to " ++ fromServer ++ " failed: " ++ show e)
1430+ logWarn ("federation: verify callback to " ++ fromServer ++ " failed: " ++ show e)
1360− yield (False {})
1431+ yield (False {})
1361− Ok r -> yield r.ok
1432+ Ok r -> yield r.ok
1363-- ============================================
1434-- ============================================
1364-- Session Bucket Helpers
1435-- Session Bucket Helpers
@@ -2726,11 +2797,9 @@ removeOneForKey = \items pubkey ->
2727-- Extract unique remote (non-local) server names from a list of items.
2798-- Extract unique remote (non-local) server names from a list of items.
2728-- O(n*k) where k = unique servers; acceptable since k is typically small (<20).
2799-- O(n*k) where k = unique servers; acceptable since k is typically small (<20).
2729−-- NOTE: `union acc [s]` copies acc (O(|acc|) per step, O(n*k) total). Using
2800+-- NOTE: `union [s] acc` prepends (O(1) per step) vs `union acc [s]` which copies
2730−-- `union [s] acc` would be O(1) per step but reverses insertion order. Since this
2731−-- is a set (order doesn't matter), prepend is preferred for large collections.
2801+-- acc (O(|acc|) per step). Since this is a set (order doesn't matter), prepend is
2732−-- TODO: Switch to `union [s] acc` across set-accumulating folds (collectPeerServers,
2802+-- preferred for large collections. All set-accumulating folds use this pattern.
2733−-- uniqueBucketsFor, dedupMessageIds) if Knot confirms union copies left operand.
2734collectPeerServers : [a] -> (a -> Text) -> [Text]
2803collectPeerServers : [a] -> (a -> Text) -> [Text]
2735collectPeerServers = \items getServer ->
2804collectPeerServers = \items getServer ->
2736 fold (\acc item ->
2805 fold (\acc item ->
@@ -2861,7 +2930,7 @@ pruneExpired = \t -> do
2861 presenceResults <- prunePresenceBuckets 0 t
2930 presenceResults <- prunePresenceBuckets 0 t
2862 -- Dedup peer servers once after collecting from all 16 buckets (O(n) total
2931 -- Dedup peer servers once after collecting from all 16 buckets (O(n) total
2863 -- vs O(n*k) if deduped per-bucket during aggregation).
2932 -- vs O(n*k) if deduped per-bucket during aggregation).
2864− let dedupServers = \servers -> fold (\acc s -> if any acc (\x -> x == s) then acc else union acc [s]) [] servers
2933+ let dedupServers = \servers -> fold (\acc s -> if any acc (\x -> x == s) then acc else union [s] acc) [] servers
2865 let keptPeers = dedupServers presenceResults.keptPeerServers
2934 let keptPeers = dedupServers presenceResults.keptPeerServers
2866 let allPeers = dedupServers presenceResults.allPeerServers
2935 let allPeers = dedupServers presenceResults.allPeerServers
2867 atomic do
2936 atomic do
@@ -3017,7 +3086,7 @@ sendBatchedGossipToServer : ServerName -> [GossipEvent] -> Int -> IO {network} {
3017sendBatchedGossipToServer = \server allEvents hops ->
3086sendBatchedGossipToServer = \server allEvents hops ->
3018 forEachBatch allEvents maxGossipEvents (\batch -> do
3087 forEachBatch allEvents maxGossipEvents (\batch -> do
3019 t <- now
3088 t <- now
3020− auth <- generateFederationAuth "gossip"
3089+ auth <- generateFederationAuth "gossip" server
3021 corrId <- genTokenN 1
3090 corrId <- genTokenN 1
3022 result <- fetchFederation server (RecvGossip {
3091 result <- fetchFederation server (RecvGossip {
3023 authorization: auth,
3092 authorization: auth,
@@ -3378,7 +3447,7 @@ retryForwardsForServer = \entries remaining ->
3378 Just {value: f} -> do
3447 Just {value: f} -> do
3379 let rest = do x <- entries; where x.messageId != f.messageId; yield x
3448 let rest = do x <- entries; where x.messageId != f.messageId; yield x
3380 t <- now
3449 t <- now
3381− auth <- generateFederationAuth "forward"
3450+ auth <- generateFederationAuth "forward" f.toServer
3382 corrId <- genTokenN 1
3451 corrId <- genTokenN 1
3383 result <- fetchFederation f.toServer (ForwardMessage {
3452 result <- fetchFederation f.toServer (ForwardMessage {
3384 toKey: f.toKey, encryptedBlob: f.encryptedBlob,
3453 toKey: f.toKey, encryptedBlob: f.encryptedBlob,
@@ -3456,7 +3525,7 @@ forwardQueuedMessages = \targets t -> do
3457fetchFederationAck = \origin messageIds -> do
3526fetchFederationAck = \origin messageIds -> do
3458 t <- now
3527 t <- now
3459− auth <- generateFederationAuth "ack"
3528+ auth <- generateFederationAuth "ack" origin
3460 corrId <- genTokenN 1
3529 corrId <- genTokenN 1
3461 fetchFederation origin (FederationAck {
3530 fetchFederation origin (FederationAck {
3462 authorization: auth,
3531 authorization: auth,
@@ -3684,7 +3753,7 @@ pruneAndRetry = \t -> do
3684 yield {extendedActive: [], nearExpiryActive: [], refreshPeers: knownPeers, shouldRefresh: True {}}
3753 yield {extendedActive: [], nearExpiryActive: [], refreshPeers: knownPeers, shouldRefresh: True {}}
3686 -- Check per-key session existence (avoids materializing all 16 session buckets)
3755 -- Check per-key session existence (avoids materializing all 16 session buckets)
3687− let nearExpiryPubkeys = fold (\acc p -> if any acc (\a -> a == p.pubkey) then acc else union acc [p.pubkey]) [] localNearExpiry
3756+ let nearExpiryPubkeys = fold (\acc p -> if any acc (\a -> a == p.pubkey) then acc else union [p.pubkey] acc) [] localNearExpiry
3688 activeKeys <- filterActiveSessionKeys nearExpiryPubkeys t
3757 activeKeys <- filterActiveSessionKeys nearExpiryPubkeys t
3689 let toExtend = do p <- localNearExpiry; where any activeKeys (\k -> k == p.pubkey); yield p
3758 let toExtend = do p <- localNearExpiry; where any activeKeys (\k -> k == p.pubkey); yield p
3690 let notExtended = do p <- localNearExpiry; where none activeKeys (\k -> k == p.pubkey); yield p
3759 let notExtended = do p <- localNearExpiry; where none activeKeys (\k -> k == p.pubkey); yield p
@@ -3730,6 +3799,10 @@ pruneAndRetry = \t -> do
3730 let refreshEvents = do p <- result.extendedActive; yield {eventType: GossipOnline {}, pubkey: p.pubkey, server: serverHostN, timestamp: t, ttl: clampTtl (sessionExpiry / 1000)}
3799 let refreshEvents = do p <- result.extendedActive; yield {eventType: GossipOnline {}, pubkey: p.pubkey, server: serverHostN, timestamp: t, ttl: clampTtl (sessionExpiry / 1000)}
3731 when (count result.refreshPeers > 0 && count refreshEvents > 0)
3800 when (count result.refreshPeers > 0 && count refreshEvents > 0)
3732 (fork (sendBatchedGossip result.refreshPeers refreshEvents gossipOriginHops)))
3801 (fork (sendBatchedGossip result.refreshPeers refreshEvents gossipOriginHops)))
3802+ -- Update cached health state so /health returns fresh data even when rate-limited.
3803+ -- This runs every prune cycle (~30s), keeping the cache within one interval of reality.
3804+ computeAndCacheHealth t
3734-- Map a single hex character to its integer value (0-15).
3807-- Map a single hex character to its integer value (0-15).
3735-- Handles both lowercase and uppercase hex to prevent hot-bucket skew from
3808-- Handles both lowercase and uppercase hex to prevent hot-bucket skew from
@@ -3857,7 +3930,7 @@ pollOnce = \token cursor clientIp -> do
3857 startSignal <- readMessageSignal pubkey
3930 startSignal <- readMessageSignal pubkey
3858 yield (Just {value: {pubkey: pubkey, clampedCursor: clampedCursor, startWakeup: startWakeup, startSignal: startSignal, pendingAcks: ackResult.pendingAcks, evictedEntries: ackResult.evictedEntries}})
3931 yield (Just {value: {pubkey: pubkey, clampedCursor: clampedCursor, startWakeup: startWakeup, startSignal: startSignal, pendingAcks: ackResult.pendingAcks, evictedEntries: ackResult.evictedEntries}})
3859 case setup of
3932 case setup of
3860− Nothing {} -> yield {authorized: False {}, events: [], cursor: 0, pendingAcks: [], evictedEntries: []}
3933+ Nothing {} -> yield {authorized: False {}, events: [], cursor: 0, pendingAcks: [], evictedEntries: [], pubkey: "", pollSlotFull: False {}}
3861 Just {value: s} -> do
3934 Just {value: s} -> do
3862 -- Acquire poll slot (limit concurrent polls per key)
3935 -- Acquire poll slot (limit concurrent polls per key)
3863 acquired <- atomic do
3936 acquired <- atomic do
@@ -3871,7 +3944,8 @@ pollOnce = \token cursor clientIp -> do
3871 else union activePolls [{pubkey: s.pubkey, startedAt: t}]
3944 else union activePolls [{pubkey: s.pubkey, startedAt: t}]
3872 yield (True {})
3945 yield (True {})
3873 if not acquired
3946 if not acquired
3947+ -- Poll slot exhausted: tell client to back off for one poll cycle instead of busy-retrying
3874− then yield {authorized: True {}, events: [], cursor: cursor, pendingAcks: s.pendingAcks, evictedEntries: s.evictedEntries}
3948+ then yield {authorized: True {}, events: [], cursor: cursor, pendingAcks: s.pendingAcks, evictedEntries: s.evictedEntries, pubkey: s.pubkey, pollSlotFull: True {}}
3876 pollResult <- do
3950 pollResult <- do
3877 -- Phase 1: check for existing messages (atomic read)
3951 -- Phase 1: check for existing messages (atomic read)
@@ -3916,7 +3990,7 @@ pollOnce = \token cursor clientIp -> do
3916 atomic do
3990 atomic do
3917 activePolls <- *activePolls
3991 activePolls <- *activePolls
3918 set *activePolls = removeOneForKey activePolls s.pubkey
3992 set *activePolls = removeOneForKey activePolls s.pubkey
3919− yield {authorized: pollResult.authorized, events: pollResult.events, cursor: pollResult.cursor, pendingAcks: s.pendingAcks, evictedEntries: s.evictedEntries}
3993+ yield {authorized: pollResult.authorized, events: pollResult.events, cursor: pollResult.cursor, pendingAcks: s.pendingAcks, evictedEntries: s.evictedEntries, pubkey: s.pubkey, pollSlotFull: False {}}
3921-- ============================================
3995-- ============================================
3922-- Derived Relations
3996-- Derived Relations
@@ -4249,7 +4323,7 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
4249 -- Snapshot presenceVersion before long-poll wait to skip unchanged lookups
4323 -- Snapshot presenceVersion before long-poll wait to skip unchanged lookups
4250 startPresenceVersion <- *presenceVersion
4324 startPresenceVersion <- *presenceVersion
4251 result <- case extractToken authorization of
4325 result <- case extractToken authorization of
4252− Nothing {} -> yield {authorized: False {}, events: [], cursor: 0, pendingAcks: [], evictedEntries: []}
4326+ Nothing {} -> yield {authorized: False {}, events: [], cursor: 0, pendingAcks: [], evictedEntries: [], pubkey: "", pollSlotFull: False {}}
4253 Just {value: token} -> pollOnce token effectiveCursor clientIp
4327 Just {value: token} -> pollOnce token effectiveCursor clientIp
4254 -- Check presence for watched keys (piggybacks on poll cycle; max latency = pollTimeoutMs).
4328 -- Check presence for watched keys (piggybacks on poll cycle; max latency = pollTimeoutMs).
4255 -- Skip the expensive bucket reads if presenceVersion hasn't changed since poll start.
4329 -- Skip the expensive bucket reads if presenceVersion hasn't changed since poll start.
@@ -4269,23 +4343,20 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
4269 let isHeartbeat = result.authorized && count result.events == 0
4343 let isHeartbeat = result.authorized && count result.events == 0
4270 -- Deliver message expiration notifications: atomically read and drain expirations
4344 -- Deliver message expiration notifications: atomically read and drain expirations
4271 -- for this sender's pubkey so each notification is delivered exactly once.
4345 -- for this sender's pubkey so each notification is delivered exactly once.
4346+ -- Uses result.pubkey from pollOnce (already authenticated) instead of re-authenticating,
4347+ -- avoiding a redundant safeCompare scan and preventing a race where a concurrent poll
4348+ -- could double-drain or lose expirations between the two separate atomic blocks.
4272− expirations <- if not result.authorized then yield []
4349+ expirations <- if not result.authorized || result.pubkey == "" then yield []
4273− else case extractToken authorization of
4274− Nothing {} -> yield []
4275− Just {value: token} -> do
4276− let bucket = tokenBucket token
4277− sessions <- readSessionBucket bucket
4279− case authenticateSession sessions token t3 of
4280− Unauthorized {} -> yield []
4281− Authorized {pubkey} -> atomic do
4350+ else atomic do
4282− se <- *senderExpirations
4351+ se <- *senderExpirations
4283− let forKey = do e <- se; where e.senderKey == pubkey; yield {messageId: e.messageId, toKey: e.toKey, expiredAt: e.expiredAt}
4352+ let forKey = do e <- se; where e.senderKey == result.pubkey; yield {messageId: e.messageId, toKey: e.toKey, expiredAt: e.expiredAt}
4284− let remaining = do e <- se; where e.senderKey != pubkey; yield e
4353+ let remaining = do e <- se; where e.senderKey != result.pubkey; yield e
4285− set *senderExpirations = remaining
4354+ set *senderExpirations = remaining
4286− let capped = takeRelation maxSenderExpirationsPerKey forKey
4355+ let capped = takeRelation maxSenderExpirationsPerKey forKey
4287− yield capped
4356+ yield capped
4357+ -- Poll slot exhaustion: signal clients to back off for one poll cycle
4358+ let pollRetryMs = if result.pollSlotFull then pollTimeoutMs else 0
4288− 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})
4359+ yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized, rateLimited: result.pollSlotFull, draining: False {}, generation: serverGen, presenceUpdates: presenceUpdates, heartbeat: isHeartbeat, retryAfterMs: pollRetryMs, cursorReset: cursorWasReset, messageExpirations: expirations})
4290-- Core send logic: dedup, rate limiting, store message, create forwards.
4361-- Core send logic: dedup, rate limiting, store message, create forwards.
4291-- Called inside an atomic block by both handleSendMessage and sendBatchInner.
4362-- Called inside an atomic block by both handleSendMessage and sendBatchInner.
@@ -4385,7 +4456,7 @@ forkSendForwards = \effectiveMsgId to encryptedBlob clampedTs msgExpiresAt remot
4385 yield (any fwdBkt (\f -> f.messageId == effectiveMsgId && f.toServer == server))
4456 yield (any fwdBkt (\f -> f.messageId == effectiveMsgId && f.toServer == server))
4386 when stillPending (do
4457 when stillPending (do
4387 fwdT <- now
4458 fwdT <- now
4388− fwdAuth <- generateFederationAuth "forward"
4459+ fwdAuth <- generateFederationAuth "forward" server
4389 fwdCorrId <- genTokenN 1
4460 fwdCorrId <- genTokenN 1
4390 -- Compute remaining TTL from the stored message's expiration (not the original TTL).
4461 -- Compute remaining TTL from the stored message's expiration (not the original TTL).
4391 -- This ensures the remote server gets the correct remaining lifetime, even if
4462 -- This ensures the remote server gets the correct remaining lifetime, even if
@@ -4419,7 +4490,7 @@ sendBatchForwardsSequential = \msgs server ->
4419 yield (any fwdBkt (\f -> f.messageId == r.messageId && f.toServer == server))
4490 yield (any fwdBkt (\f -> f.messageId == r.messageId && f.toServer == server))
4420 when stillPending (do
4491 when stillPending (do
4421 fwdT <- now
4492 fwdT <- now
4422− fwdAuth <- generateFederationAuth "forward"
4493+ fwdAuth <- generateFederationAuth "forward" server
4423 fwdCorrId <- genTokenN 1
4494 fwdCorrId <- genTokenN 1
4424 -- Use stored msgExpiresAt to compute remaining TTL (not original messageTtl)
4495 -- Use stored msgExpiresAt to compute remaining TTL (not original messageTtl)
4425 let fwdTtl = if r.msgExpiresAt > 0 then max 0 (r.msgExpiresAt - fwdT) else 0
4496 let fwdTtl = if r.msgExpiresAt > 0 then max 0 (r.msgExpiresAt - fwdT) else 0
@@ -4630,6 +4701,9 @@ handleRevoke = \targetToken authorization clientIp respond -> do
4630-- This gives faster offline detection than waiting for session expiry (1 hour).
4701-- This gives faster offline detection than waiting for session expiry (1 hour).
4631handleLogout = \authorization clientIp respond -> do
4702handleLogout = \authorization clientIp respond -> do
4632 t <- now
4703 t <- now
4705+ if dp >= 2 then yield (respond {ok: False {}, authorized: False {}})
4633 globalRateLimited <- checkGlobalRate t
4707 globalRateLimited <- checkGlobalRate t
4634 if globalRateLimited then yield (respond {ok: False {}, authorized: False {}})
4708 if globalRateLimited then yield (respond {ok: False {}, authorized: False {}})
@@ -4774,7 +4848,7 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
4774 -- Merge new peer servers with existing known peers (selected buckets may not contain all peers)
4848 -- Merge new peer servers with existing known peers (selected buckets may not contain all peers)
4775 knownPeers <- *knownPeerServers
4849 knownPeers <- *knownPeerServers
4776 let newPeers = gm.gossipPeerServers
4850 let newPeers = gm.gossipPeerServers
4777− set *knownPeerServers = fold (\acc s -> if any acc (\a -> a == s) then acc else union acc [s]) knownPeers newPeers
4851+ set *knownPeerServers = fold (\acc s -> if any acc (\a -> a == s) then acc else union [s] acc) knownPeers newPeers
4778 set *presenceVersion = t
4852 set *presenceVersion = t
4779 yield {gm: gm, preFiltered: count safeDeduped - count capacityDeduped, localClashCount: count localClashes}
4853 yield {gm: gm, preFiltered: count safeDeduped - count capacityDeduped, localClashCount: count localClashes}
4780 when (gossipResult.localClashCount > 0)
4854 when (gossipResult.localClashCount > 0)
@@ -4788,7 +4862,7 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
4788 let active = do r <- recentGossip; where t - r.seenAt < gossipDedupWindow; yield r
4862 let active = do r <- recentGossip; where t - r.seenAt < gossipDedupWindow; yield r
4789 -- Pre-filter active entries to only those matching (pubkey, server) pairs in this batch.
4863 -- Pre-filter active entries to only those matching (pubkey, server) pairs in this batch.
4790 -- Reduces the inner dedup scan from O(n*m) to O(n*k) where k << m (k = matching pairs).
4864 -- Reduces the inner dedup scan from O(n*m) to O(n*k) where k << m (k = matching pairs).
4791− let batchPairs = fold (\acc e -> if any acc (\a -> a.pubkey == e.pubkey && a.server == e.server) then acc else union acc [{pubkey: e.pubkey, server: e.server}]) [] effectiveEvents
4865+ let batchPairs = fold (\acc e -> if any acc (\a -> a.pubkey == e.pubkey && a.server == e.server) then acc else union [{pubkey: e.pubkey, server: e.server}] acc) [] effectiveEvents
4792 let relevantActive = do r <- active; where any batchPairs (\p -> p.pubkey == r.pubkey && p.server == r.server); yield r
4866 let relevantActive = do r <- active; where any batchPairs (\p -> p.pubkey == r.pubkey && p.server == r.server); yield r
4793 let fresh = do e <- effectiveEvents; where none relevantActive (\r -> r.pubkey == e.pubkey && r.server == e.server && r.timestamp == e.timestamp); yield e
4867 let fresh = do e <- effectiveEvents; where none relevantActive (\r -> r.pubkey == e.pubkey && r.server == e.server && r.timestamp == e.timestamp); yield e
4794 let newRecent = do e <- fresh; yield {pubkey: e.pubkey, server: e.server, timestamp: e.timestamp, seenAt: t}
4868 let newRecent = do e <- fresh; yield {pubkey: e.pubkey, server: e.server, timestamp: e.timestamp, seenAt: t}
@@ -4979,6 +5053,37 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp corre
4979 set *pendingAckCleanups = if count perKeyCapped > maxPendingAckCleanups then takeRelation maxPendingAckCleanups perKeyCapped else perKeyCapped)
5053 set *pendingAckCleanups = if count perKeyCapped > maxPendingAckCleanups then takeRelation maxPendingAckCleanups perKeyCapped else perKeyCapped)
4980 yield (respond {ok: True {}, error: "", ackedCount: ackedCount}))
5054 yield (respond {ok: True {}, error: "", ackedCount: ackedCount}))
5056+-- POST /federation/cancel: Cancel a previously forwarded message.
5057+-- Called by the origin server when the sender retracts a message via /messages/cancel.
5058+-- Deletes the message from the recipient's mailbox if it hasn't been polled yet.
5059+-- If the message was already polled, returns cancelled=False (no-op).
5060+handleFederationCancel = \messageId toKey fromServer requestTimestamp correlationId authorization respond -> do
5062+ let corrId = sanitizeCorrelationId correlationId
5063+ when (corrId != "") (logInfo ("[CORR] " ++ corrId ++ " cancel " ++ messageId ++ " from " ++ fromServer))
5064+ let cancelRespond = \r -> respond {ok: r.ok, error: r.error, cancelled: False {}}
5065+ withFederationAuth authorization t cancelRespond fromServer "cancel" requestTimestamp (not (isLocalServer fromServer)) (False {}) (do
5066+ result <- atomic do
5067+ let b = pubkeyBucket toKey
5068+ bucket <- readMessageBucket b
5069+ -- Only cancel messages that originated from the requesting server
5070+ let toCancel = headOf (do m <- bucket; where m.messageId == messageId && m.toKey == toKey && m.originServer == fromServer; yield m)
5072+ Nothing {} -> yield {cancelled: False {}}
5074+ writeMessageBucket b (do m <- bucket; where not (m.messageId == messageId && m.toKey == toKey); yield m)
5075+ tc <- *totalMessageCount
5076+ set *totalMessageCount = max 0 (tc - 1)
5077+ rc <- readRcBucket b
5078+ writeRcBucket b (adjustRecipientCount rc toKey (0 - 1))
5079+ -- Also cancel pending forwards for this message
5080+ fwdBkt <- readFwdBucket b
5081+ writeFwdBucket b (do f <- fwdBkt; where not (f.messageId == messageId && f.toKey == toKey); yield f)
5082+ yield {cancelled: True {}}
5083+ when result.cancelled
5084+ (logInfo ("federation/cancel: cancelled " ++ messageId ++ " for " ++ toKey ++ " (from " ++ fromServer ++ ")"))
5085+ yield (respond {ok: True {}, error: "", cancelled: result.cancelled}))
4982-- POST /maintenance/drain: Broadcast offline gossip for all local keys (graceful shutdown).
5087-- POST /maintenance/drain: Broadcast offline gossip for all local keys (graceful shutdown).
4983-- Call before stopping the server so peers learn keys went offline immediately.
5088-- Call before stopping the server so peers learn keys went offline immediately.
4984-- Gossip is sent via fork; allow a few seconds after response before process exit.
5089-- Gossip is sent via fork; allow a few seconds after response before process exit.
@@ -5120,26 +5225,10 @@ handlePrune = \authorization respond -> do
5120 when authorized (fork (pruneAndRetry t))
5225 when authorized (fork (pruneAndRetry t))
5121 yield (respond {ok: authorized})
5226 yield (respond {ok: authorized})
5123−handleHealth = \authorization respond -> do
5228+-- Compute health state and update the cached health values.
5124− t <- now
5229+-- Called from both handleHealth (request-driven) and pruneAndRetry (background-driven)
5125− -- Health checks exempt from global rate limiting (monitoring traffic shouldn't starve real requests)
5230+-- so the cache stays fresh even when /health is rate-limited under sustained monitoring load.
5126− -- but have their own dedicated rate limiter to prevent DDoS on /health from saturating reads.
5231+computeAndCacheHealth = \t -> do
5127− healthLimited <- checkHealthRate t
5128− if healthLimited then do
5129− -- Return cached last-known health (prevents DDoS on /health from masking real failures)
5130− cached <- atomic do
5131− cOk <- *cachedHealthOk
5132− cP <- *cachedHealthPrune
5133− cH <- *cachedHealthHeartbeat
5134− cC <- *cachedHealthCapacity
5135− cF <- *cachedHealthFederation
5136− cS <- *cachedHealthSnapshot
5137− cW <- *cachedHealthWatchdog
5139− yield {ok: cOk && dp == 0, pruneHealthy: cP, heartbeatHealthy: cH, capacityHealthy: cC, federationHealthy: cF, snapshotHealthy: cS, watchdogHealthy: cW, draining: dp > 0}
5140− yield (respond cached)
5142− let isAuthed = verifyMaintenanceAuth authorization
5143 result <- atomic do
5232 result <- atomic do
5144 lastPrune <- *lastPrunedAt
5233 lastPrune <- *lastPrunedAt
5145 lastHb <- *lastHeartbeatAt
5234 lastHb <- *lastHeartbeatAt
@@ -5153,30 +5242,22 @@ handleHealth = \authorization respond -> do
5153 lastFedSuccess <- *lastFederationSuccessAt
5242 lastFedSuccess <- *lastFederationSuccessAt
5154 forwardFailures <- *forwardFailures
5243 forwardFailures <- *forwardFailures
5155 pruneRun <- *pruneRunning
5244 pruneRun <- *pruneRunning
5156− -- lastSnapshotExportAt no longer used in health check (replaced by prunesSinceSnapshot)
5157 dp <- *drainPhase
5245 dp <- *drainPhase
5158 let warmup = t - sa < healthPruneMaxAge
5246 let warmup = t - sa < healthPruneMaxAge
5159 let pruneStuck = pruneRun > 0 && t - pruneRun > backgroundPruneInterval * 2
5247 let pruneStuck = pruneRun > 0 && t - pruneRun > backgroundPruneInterval * 2
5160− -- Capacity pressure: flag when any capped collection exceeds 90% full
5161 let capacityOk = tmc * 10 <= maxMessages * 9
5248 let capacityOk = tmc * 10 <= maxMessages * 9
5162 && presenceCount * 10 <= maxPresenceEntries * 9
5249 && presenceCount * 10 <= maxPresenceEntries * 9
5163 && forwardCount * 10 <= maxForwardsEntries * 9
5250 && forwardCount * 10 <= maxForwardsEntries * 9
5164 && sessionCount * 10 <= maxSessions * 9
5251 && sessionCount * 10 <= maxSessions * 9
5165 && count activePolls * 10 <= maxActivePolls * 9
5252 && count activePolls * 10 <= maxActivePolls * 9
5166− -- Federation: healthy if disabled, in warmup, no peer issues, or recent success
5167 let fedOk = not federationEnabled || warmup
5253 let fedOk = not federationEnabled || warmup
5168 || count forwardFailures == 0
5254 || count forwardFailures == 0
5169 || (lastFedSuccess > 0 && t - lastFedSuccess < backgroundPruneInterval * 5)
5255 || (lastFedSuccess > 0 && t - lastFedSuccess < backgroundPruneInterval * 5)
5170− -- Snapshot staleness: unhealthy if maintenance secret is configured but no snapshot
5171− -- has been exported within the staleness window (federation state will be lost on restart)
5172− -- Snapshot staleness: uses prune-cycle counter (clock-jump resilient).
5173− -- Stale when prunesSinceSnapshot exceeds threshold computed from staleness/prune interval.
5174 pss <- *prunesSinceSnapshot
5256 pss <- *prunesSinceSnapshot
5175 let staleThresholdCycles = if backgroundPruneInterval > 0 then snapshotStalenessWarning / backgroundPruneInterval else 0
5257 let staleThresholdCycles = if backgroundPruneInterval > 0 then snapshotStalenessWarning / backgroundPruneInterval else 0
5176 let snapOk = maintenanceSecret == "" || snapshotStalenessWarning == 0 || warmup
5258 let snapOk = maintenanceSecret == "" || snapshotStalenessWarning == 0 || warmup
5177 || staleThresholdCycles == 0 || pss <= staleThresholdCycles
5259 || staleThresholdCycles == 0 || pss <= staleThresholdCycles
5178 || (tmc == 0 && presenceCount == 0 && forwardCount == 0)
5260 || (tmc == 0 && presenceCount == 0 && forwardCount == 0)
5179− -- Watchdog: healthy if recently bumped or in warmup (watchdog starts after 10x prune interval)
5180 let wdOk = warmup || t - sa < healthWatchdogMaxAge || (lastWd > 0 && t - lastWd < healthWatchdogMaxAge)
5261 let wdOk = warmup || t - sa < healthWatchdogMaxAge || (lastWd > 0 && t - lastWd < healthWatchdogMaxAge)
5181 yield {pruneOk: warmup || (t - lastPrune < healthPruneMaxAge && not pruneStuck),
5262 yield {pruneOk: warmup || (t - lastPrune < healthPruneMaxAge && not pruneStuck),
5182 heartbeatOk: warmup || t - lastHb < healthHeartbeatMaxAge,
5263 heartbeatOk: warmup || t - lastHb < healthHeartbeatMaxAge,
@@ -5185,9 +5266,7 @@ handleHealth = \authorization respond -> do
5185 snapshotOk: snapOk,
5266 snapshotOk: snapOk,
5186 watchdogOk: wdOk,
5267 watchdogOk: wdOk,
5187 isDraining: dp > 0}
5268 isDraining: dp > 0}
5188− -- Draining servers report ok=False so load balancers stop routing traffic to them
5189 let allOk = not result.isDraining && result.pruneOk && result.heartbeatOk && result.capacityOk && result.federationOk && result.snapshotOk && result.watchdogOk
5269 let allOk = not result.isDraining && result.pruneOk && result.heartbeatOk && result.capacityOk && result.federationOk && result.snapshotOk && result.watchdogOk
5190− -- Cache health result for rate-limited responses
5191 atomic do
5270 atomic do
5192 set *cachedHealthOk = allOk
5271 set *cachedHealthOk = allOk
5193 set *cachedHealthPrune = result.pruneOk
5272 set *cachedHealthPrune = result.pruneOk
@@ -5196,6 +5275,30 @@ handleHealth = \authorization respond -> do
5196 set *cachedHealthFederation = result.federationOk
5275 set *cachedHealthFederation = result.federationOk
5197 set *cachedHealthSnapshot = result.snapshotOk
5276 set *cachedHealthSnapshot = result.snapshotOk
5198 set *cachedHealthWatchdog = result.watchdogOk
5277 set *cachedHealthWatchdog = result.watchdogOk
5280+handleHealth = \authorization respond -> do
5282+ -- Health checks exempt from global rate limiting (monitoring traffic shouldn't starve real requests)
5283+ -- but have their own dedicated rate limiter to prevent DDoS on /health from saturating reads.
5284+ healthLimited <- checkHealthRate t
5285+ if healthLimited then do
5286+ -- Return cached last-known health (prevents DDoS on /health from masking real failures).
5287+ -- Cache is kept fresh by computeAndCacheHealth called from both here and background prune.
5288+ cached <- atomic do
5289+ cOk <- *cachedHealthOk
5290+ cP <- *cachedHealthPrune
5291+ cH <- *cachedHealthHeartbeat
5292+ cC <- *cachedHealthCapacity
5293+ cF <- *cachedHealthFederation
5294+ cS <- *cachedHealthSnapshot
5295+ cW <- *cachedHealthWatchdog
5297+ yield {ok: cOk && dp == 0, pruneHealthy: cP, heartbeatHealthy: cH, capacityHealthy: cC, federationHealthy: cF, snapshotHealthy: cS, watchdogHealthy: cW, draining: dp > 0}
5298+ yield (respond cached)
5300+ let isAuthed = verifyMaintenanceAuth authorization
5301+ result <- computeAndCacheHealth t
5199 -- Unauthenticated: return real ok boolean so load balancers can route around unhealthy nodes,
5302 -- Unauthenticated: return real ok boolean so load balancers can route around unhealthy nodes,
5200 -- but hide subsystem breakdown to prevent targeted load-correlation attacks.
5303 -- but hide subsystem breakdown to prevent targeted load-correlation attacks.
5201 -- Draining status is always visible (load balancers need it for routing decisions).
5304 -- Draining status is always visible (load balancers need it for routing decisions).
@@ -5491,12 +5594,12 @@ handleRestore = \version presenceData ackedIdsData ackOverflowData deliveredIdsD
5491 writeAllNonces cappedNonces
5594 writeAllNonces cappedNonces
5492 -- Restore blocked peers (union with existing, cap at maxBlockedPeers)
5595 -- Restore blocked peers (union with existing, cap at maxBlockedPeers)
5493 blocked <- *blockedPeers
5596 blocked <- *blockedPeers
5494− let mergedBlocked = fold (\acc b -> if any acc (\x -> x == b) then acc else union acc [b]) blocked blockedPeersData
5597+ let mergedBlocked = fold (\acc b -> if any acc (\x -> x == b) then acc else union [b] acc) blocked blockedPeersData
5495 let cappedBlocked = takeRelation maxBlockedPeers mergedBlocked
5598 let cappedBlocked = takeRelation maxBlockedPeers mergedBlocked
5496 set *blockedPeers = cappedBlocked
5599 set *blockedPeers = cappedBlocked
5497 -- Restore allowed peers (union with existing, cap at maxAllowedPeers)
5600 -- Restore allowed peers (union with existing, cap at maxAllowedPeers)
5498 allowed <- *allowedPeers
5601 allowed <- *allowedPeers
5499− let mergedAllowed = fold (\acc a -> if any acc (\x -> x == a) then acc else union acc [a]) allowed allowedPeersData
5602+ let mergedAllowed = fold (\acc a -> if any acc (\x -> x == a) then acc else union [a] acc) allowed allowedPeersData
5500 let cappedAllowed = takeRelation maxAllowedPeers mergedAllowed
5603 let cappedAllowed = takeRelation maxAllowedPeers mergedAllowed
5501 set *allowedPeers = cappedAllowed
5604 set *allowedPeers = cappedAllowed
5502 yield {presenceCount: count cappedPresence, ackedIdsCount: count cappedAcked, deliveredIdsCount: count cappedDelivered}
5605 yield {presenceCount: count cappedPresence, ackedIdsCount: count cappedAcked, deliveredIdsCount: count cappedDelivered}
@@ -5672,28 +5775,41 @@ handleCancelMessage = \messageId to authorization clientIp respond -> do
5672 bucket <- readMessageBucket b
5775 bucket <- readMessageBucket b
5673 let toCancel = headOf (do m <- bucket; where m.messageId == messageId && m.toKey == to; yield m)
5776 let toCancel = headOf (do m <- bucket; where m.messageId == messageId && m.toKey == to; yield m)
5674 case toCancel of
5777 case toCancel of
5675− Nothing {} -> yield {cancelled: False {}, pendingAcks: [], evictedEntries: []}
5778+ Nothing {} -> yield {cancelled: False {}, cancelServers: []}
5676 Just {value: msg} ->
5779 Just {value: msg} ->
5677 -- Only allow cancelling locally-originated messages (originServer == "")
5780 -- Only allow cancelling locally-originated messages (originServer == "")
5678 -- where the caller is the actual sender. Federation forwards have no
5781 -- where the caller is the actual sender. Federation forwards have no
5679 -- senderKey and should not be cancellable via this endpoint.
5782 -- senderKey and should not be cancellable via this endpoint.
5680− if msg.originServer != "" then yield {cancelled: False {}, pendingAcks: [], evictedEntries: []}
5783+ if msg.originServer != "" then yield {cancelled: False {}, cancelServers: []}
5681− else if msg.senderKey != "" && msg.senderKey != senderKey then yield {cancelled: False {}, pendingAcks: [], evictedEntries: []}
5784+ else if msg.senderKey != "" && msg.senderKey != senderKey then yield {cancelled: False {}, cancelServers: []}
5683 writeMessageBucket b (do m <- bucket; where not (m.messageId == messageId && m.toKey == to); yield m)
5786 writeMessageBucket b (do m <- bucket; where not (m.messageId == messageId && m.toKey == to); yield m)
5684 tc <- *totalMessageCount
5787 tc <- *totalMessageCount
5685 set *totalMessageCount = max 0 (tc - 1)
5788 set *totalMessageCount = max 0 (tc - 1)
5686 rc <- readRcBucket b
5789 rc <- readRcBucket b
5687 writeRcBucket b (adjustRecipientCount rc to (0 - 1))
5790 writeRcBucket b (adjustRecipientCount rc to (0 - 1))
5688− -- Cancel pending forwards
5791+ -- Cancel pending forwards and collect servers that already received the message
5689 fwdBkt <- readFwdBucket b
5792 fwdBkt <- readFwdBucket b
5793+ let cancelledFwds = do f <- fwdBkt; where f.messageId == messageId && f.toKey == to; yield f
5690 writeFwdBucket b (do f <- fwdBkt; where not (f.messageId == messageId && f.toKey == to); yield f)
5794 writeFwdBucket b (do f <- fwdBkt; where not (f.messageId == messageId && f.toKey == to); yield f)
5691− -- Only locally-originated messages reach this point (originServer == ""),
5795+ -- Collect servers from recipient's presence that may have received the forward
5692− -- so no federation acks are needed.
5796+ -- (forwards that completed are no longer in fwdBucket, so check presence)
5797+ presenceBkt <- readPresenceBucket b
5798+ let remotePresence = collectPeerServers (presenceForKey presenceBkt to t) (\p -> p.server)
5693− yield {cancelled: True {}, pendingAcks: [], evictedEntries: []}
5799+ yield {cancelled: True {}, cancelServers: remotePresence}
5800+ -- Send federation cancel to remote servers that may have the message
5694− fork (when (federationEnabled && count cancelResult.pendingAcks > 0) (do
5801+ fork (when (federationEnabled && cancelResult.cancelled && count cancelResult.cancelServers > 0)
5695− sendAcksAndClean cancelResult.pendingAcks
5802+ (sendToEach cancelResult.cancelServers (\server -> do
5696− retryEvictedAcks cancelResult.evictedEntries cancelResult.pendingAcks))
5803+ cancelT <- now
5804+ cancelAuth <- generateFederationAuth "cancel" server
5805+ cancelCorrId <- genTokenN 1
5806+ cancelFedResult <- fetchFederation server (FederationCancel {
5807+ messageId: messageId, toKey: to, fromServer: serverHostN,
5808+ requestTimestamp: cancelT, correlationId: cancelCorrId,
5809+ authorization: cancelAuth})
5810+ case cancelFedResult of
5811+ Err e -> logWarn ("federation: cancel to " ++ server ++ " failed for " ++ messageId ++ ": " ++ show e)
5812+ Ok r -> when (not r.ok) (logWarn ("federation: cancel to " ++ server ++ " rejected for " ++ messageId ++ ": " ++ r.error)))))
5697 yield (respond {ok: True {}, authorized: True {}, cancelled: cancelResult.cancelled, error: ""}))
5813 yield (respond {ok: True {}, authorized: True {}, cancelled: cancelResult.cancelled, error: ""}))
5699-- POST /messages/batch: Send multiple messages in one request.
5815-- POST /messages/batch: Send multiple messages in one request.
@@ -5858,7 +5974,7 @@ handlePingPeer = \server authorization respond -> do
5858 if not authorized then yield (respond {ok: False {}, latencyMs: 0, error: "unauthorized"})
5974 if not authorized then yield (respond {ok: False {}, latencyMs: 0, error: "unauthorized"})
5860 startTime <- now
5976 startTime <- now
5861− auth <- generateFederationAuth "gossip"
5977+ auth <- generateFederationAuth "gossip" server
5862 pingCorrId <- genTokenN 1
5978 pingCorrId <- genTokenN 1
5863 result <- fetchFederation server (RecvGossip {
5979 result <- fetchFederation server (RecvGossip {
5864 authorization: auth,
5980 authorization: auth,
@@ -6028,6 +6144,7 @@ requestPath = \req -> case req of
6028 RecvGossip {} -> "POST /federation/gossip"
6144 RecvGossip {} -> "POST /federation/gossip"
6029 ForwardMessage {} -> "POST /federation/forward"
6145 ForwardMessage {} -> "POST /federation/forward"
6030 FederationAck {} -> "POST /federation/ack"
6146 FederationAck {} -> "POST /federation/ack"
6147+ FederationCancel {} -> "POST /federation/cancel"
6031 VerifyToken {} -> "POST /federation/verify"
6148 VerifyToken {} -> "POST /federation/verify"
6032 Prune {} -> "POST /maintenance/prune"
6149 Prune {} -> "POST /maintenance/prune"
6033 Drain {} -> "POST /maintenance/drain"
6150 Drain {} -> "POST /maintenance/drain"
@@ -6075,6 +6192,7 @@ requestClientIp = \req -> case req of
6075 RecvGossip {} -> ""
6192 RecvGossip {} -> ""
6076 ForwardMessage {} -> ""
6193 ForwardMessage {} -> ""
6077 FederationAck {} -> ""
6194 FederationAck {} -> ""
6195+ FederationCancel {} -> ""
6078 Prune {} -> ""
6196 Prune {} -> ""
6079 Drain {} -> ""
6197 Drain {} -> ""
6080 Shutdown {} -> ""
6198 Shutdown {} -> ""
@@ -6123,6 +6241,7 @@ serveInner = \req -> case req of
6123 RecvGossip {events, hopsRemaining, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleRecvGossip events hopsRemaining fromServer requestTimestamp correlationId authorization respond
6241 RecvGossip {events, hopsRemaining, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleRecvGossip events hopsRemaining fromServer requestTimestamp correlationId authorization respond
6124 ForwardMessage {toKey, encryptedBlob, timestamp, originServer, messageId, fromServer, requestTimestamp, correlationId, messageTtl, authorization, respond} -> handleForwardMessage toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp correlationId messageTtl authorization respond
6242 ForwardMessage {toKey, encryptedBlob, timestamp, originServer, messageId, fromServer, requestTimestamp, correlationId, messageTtl, authorization, respond} -> handleForwardMessage toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp correlationId messageTtl authorization respond
6125 FederationAck {messageIds, originServer, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleFederationAck messageIds originServer fromServer requestTimestamp correlationId authorization respond
6243 FederationAck {messageIds, originServer, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleFederationAck messageIds originServer fromServer requestTimestamp correlationId authorization respond
6244+ FederationCancel {messageId, toKey, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleFederationCancel messageId toKey fromServer requestTimestamp correlationId authorization respond
6126 VerifyToken {token, scope, clientIp, respond} -> handleVerifyToken token scope clientIp respond
6245 VerifyToken {token, scope, clientIp, respond} -> handleVerifyToken token scope clientIp respond
6127 Prune {authorization, respond} -> handlePrune authorization respond
6246 Prune {authorization, respond} -> handlePrune authorization respond
6128 Drain {phase, authorization, respond} -> handleDrain phase authorization respond
6247 Drain {phase, authorization, respond} -> handleDrain phase authorization respond
@@ -6474,6 +6593,8 @@ main = do
6474 logInfo "IP binding is active — sessions created with a ClientIp header will reject requests without one"
6593 logInfo "IP binding is active — sessions created with a ClientIp header will reject requests without one"
6475 when federationEnabled
6594 when federationEnabled
6476 (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.")
6595 (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.")
6596+ when (count presharedKeyList > 0)
6597+ (logInfo ("pre-shared federation keys configured for " ++ show (count presharedKeyList) ++ " peer(s) — callback verification bypassed for those peers"))
6477 logInfo ("Skrepka relay server starting on port " ++ show listenPort ++ "...")
6598 logInfo ("Skrepka relay server starting on port " ++ show listenPort ++ "...")
6478 logInfo (" federation: " ++ if federationEnabled then "enabled" else "disabled")
6599 logInfo (" federation: " ++ if federationEnabled then "enabled" else "disabled")
6479 logInfo (" prune interval: " ++ show backgroundPruneInterval ++ "ms")
6600 logInfo (" prune interval: " ++ show backgroundPruneInterval ++ "ms")