@@ -845,10 +845,11 @@ authenticateSession = \sessions token t ->
845845 Nothing {} -> Unauthorized {}
846846 Just {value: s} -> Authorized {pubkey: s.pubkey, ip: s.ip}
847847
848−-- Extract bearer token from "Authorization: Bearer <token>" header
848+-- Extract bearer token from "Authorization: Bearer <token>" header.
849+-- Rejects empty tokens early to avoid unnecessary session bucket scans.
849850extractToken = \authHeader ->
850851 let token = drop 7 authHeader in
851− if toLower (take 7 authHeader) == "bearer " && not (tooLong token maxTokenLen)
852+ if toLower (take 7 authHeader) == "bearer " && token != "" && not (tooLong token maxTokenLen)
852853 then Just {value: token}
853854 else Nothing {}
854855
@@ -885,6 +886,7 @@ generateFederationAuth = \scope -> do
885886-- Verify an inbound federation token by callback to the sender's /federation/verify endpoint.
886887-- Single-use: the sender consumes the token on verification.
887888-- scope must match the scope the token was generated with; empty scope accepts any (backward compat).
889+-- Retries once on network error to handle transient failures (e.g., peer restarting).
888890verifyFederationCallback : Text -> Text -> Text -> IO {network} Bool
889891verifyFederationCallback = \fromServer authorization scope ->
890892 case extractToken authorization of
@@ -893,8 +895,14 @@ verifyFederationCallback = \fromServer authorization scope ->
893895 result <- fetch (federationUrl fromServer) (VerifyToken {token: token, scope: scope})
894896 case result of
895897 Err e -> do
896− println ("federation: verify callback to " ++ fromServer ++ " failed: " ++ show e)
897− yield (False {})
898+ println ("federation: verify callback to " ++ fromServer ++ " failed (attempt 1): " ++ show e)
899+ sleep 1000
900+ result2 <- fetch (federationUrl fromServer) (VerifyToken {token: token, scope: scope})
901+ case result2 of
902+ Err e2 -> do
903+ println ("federation: verify callback to " ++ fromServer ++ " failed (attempt 2): " ++ show e2)
904+ yield (False {})
905+ Ok r -> yield r.ok
898906 Ok r -> yield r.ok
899907
900908-- ============================================
@@ -1417,11 +1425,15 @@ safeCompare = \a b ->
14171425 if take 1 a == "" || take 1 b == "" then False {}
14181426 else safeCompareLoop a b maxTokenLen 0 0 == 0
14191427
1420−-- Constant-time IP binding check: returns True if IPs match or either is empty (no binding).
1428+-- Constant-time IP binding check: returns True if IPs match or no binding is configured.
1429+-- When a session was created with an IP (behind a proxy), requests without an IP header
1430+-- are rejected — prevents bypass if the proxy is removed mid-session.
14211431-- Uses safeCompareLoop to prevent timing side-channels on IP comparison.
14221432matchIp : Text -> Text -> Bool
14231433matchIp = \clientIp sessionIp ->
1424− clientIp == "" || sessionIp == "" || safeCompareLoop clientIp sessionIp maxServerNameLen 0 0 == 0
1434+ if sessionIp == "" then True {} -- no binding configured for this session
1435+ else if clientIp == "" then False {} -- session has IP but request doesn't — proxy disappeared
1436+ else safeCompareLoop clientIp sessionIp maxServerNameLen 0 0 == 0
14251437
14261438-- Generic sliding-window rate limiter. Returns whether the request is rate-limited
14271439-- and the updated entries list. Callers provide match/create/update functions for their entry type.
@@ -1671,13 +1683,14 @@ pruneExpired = \t -> do
16711683 set *recentGossip = do r <- recentGossip; where t - r.seenAt < gossipDedupWindow; yield r
16721684 -- Clean up stale in-flight entries: remove zeroed entries and reset leaked slots
16731685 -- where the forked task died without calling clearFederationInFlight.
1674− -- Uses 2x timeout threshold (tighter than the skip-hung-server logic at 1x)
1675− -- so leaked slots are reclaimed promptly after the skip logic has already taken effect.
1686+ -- Uses 1.5x timeout threshold so leaked slots are reclaimed promptly
1687+ -- after the skip-hung-server logic (at 1x) has already taken effect.
16761688 inFlight <- *federationInFlight
1677− let staleInFlight = do r <- inFlight; where r.count > 0 && r.firstStartedAt > 0 && t - r.firstStartedAt >= federationFetchTimeoutMs * 2; yield r
1689+ let staleThreshold = federationFetchTimeoutMs * 3 / 2
1690+ let staleInFlight = do r <- inFlight; where r.count > 0 && r.firstStartedAt > 0 && t - r.firstStartedAt >= staleThreshold; yield r
16781691 when (count staleInFlight > 0)
16791692 (println ("pruneExpired: cleaning " ++ show (count staleInFlight) ++ " stale in-flight entries (leaked slots)"))
1680− set *federationInFlight = do r <- inFlight; where r.count > 0 && (r.firstStartedAt == 0 || t - r.firstStartedAt < federationFetchTimeoutMs * 2); yield r
1693+ set *federationInFlight = do r <- inFlight; where r.count > 0 && (r.firstStartedAt == 0 || t - r.firstStartedAt < staleThreshold); yield r
16811694 -- Prune stale federation replay dedup entries
16821695 recentFedReqs <- *recentFederationRequests
16831696 set *recentFederationRequests = do r <- recentFedReqs; where t - r.seenAt < clockSkewTolerance * 2; yield r
@@ -2437,8 +2450,10 @@ pollOnce = \token cursor clientIp -> do
24372450 r <- pollNewEvents s.pubkey s.clampedCursor t2
24382451 yield {authorized: True {}, events: r.events, cursor: r.cursor}
24392452 yield {authorized: result.authorized, events: result.events, cursor: result.cursor}
2440− -- Release poll slot. Safety net: if this is never reached (e.g., panic between
2441− -- acquire and release), pruneExpired cleans stale entries after pollTimeoutMs * 3.
2453+ -- Release poll slot. Safety: the code between acquire and release only contains
2454+ -- atomic reads and STM retry (no IO that can throw), so this path is always reached.
2455+ -- If it ever isn't (e.g., runtime panic), pruneExpired cleans stale entries
2456+ -- after pollTimeoutMs * 3 as a safety net.
24422457 atomic do
24432458 activePolls <- *activePolls
24442459 set *activePolls = removeOneForKey activePolls s.pubkey
@@ -2734,10 +2749,10 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
27342749 let effectiveCursor = if generation == serverGen then cursor else 0
27352750 let cappedWatchKeys = takeRelation maxWatchKeys watchKeys
27362751 draining <- *draining
2737− if draining then yield (respond {events: [], cursor: 0, authorized: True {}, rateLimited: False {}, draining: True {}, generation: serverGen, presenceUpdates: []})
2752+ if draining then yield (respond {events: [], cursor: 0, authorized: True {}, rateLimited: False {}, draining: True {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}})
27382753 else do
27392754 globalRateLimited <- checkGlobalRate t
2740− if globalRateLimited then yield (respond {events: [], cursor: 0, authorized: False {}, rateLimited: True {}, draining: False {}, generation: serverGen, presenceUpdates: []})
2755+ if globalRateLimited then yield (respond {events: [], cursor: 0, authorized: False {}, rateLimited: True {}, draining: False {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}})
27412756 else do
27422757 result <- case extractToken authorization of
27432758 Nothing {} -> yield {authorized: False {}, events: [], cursor: 0, pendingAcks: [], evictedEntries: []}
@@ -2752,7 +2767,8 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
27522767 fork (when federationEnabled (do
27532768 sendAcksAndClean result.pendingAcks
27542769 retryEvictedAcks result.evictedEntries result.pendingAcks))
2755− yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized, rateLimited: False {}, draining: False {}, generation: serverGen, presenceUpdates: presenceUpdates})
2770+ let isHeartbeat = result.authorized && count result.events == 0
2771+ yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized, rateLimited: False {}, draining: False {}, generation: serverGen, presenceUpdates: presenceUpdates, heartbeat: isHeartbeat})
27562772
27572773-- POST /messages: Send an encrypted message to a recipient.
27582774-- Deduplicates by messageId across messages, ackedIds, and deliveredIds.
@@ -2766,6 +2782,10 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp
27662782 if t - timestamp > messageMaxAge
27672783 then yield (respond {status: MsRejected {}, messageId: ""})
27682784 else do
2785+ when (timestamp > t + clockSkewTolerance)
2786+ (println ("sendMessage: future timestamp detected (+" ++ show (timestamp - t) ++ "ms ahead), clamping to now"))
2787+ when (timestamp < t - messageMaxAge / 2)
2788+ (println ("sendMessage: stale timestamp (" ++ show ((t - timestamp) / 86400000) ++ " days old) for " ++ effectiveMsgId))
27692789 let clampedTs = clampTimestamp timestamp t
27702790 result <- atomic do
27712791 draining <- *draining
@@ -2836,6 +2856,10 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp
28362856 let toPresence = presenceForKey recipientPresence to t
28372857 let localOnline = hasLocalPresence toPresence
28382858 let remoteServers = collectPeerServers toPresence (\p -> p.server)
2859+ -- Note: DsFederated means "queued for federation forwarding", not
2860+ -- "confirmed delivered to remote server". The actual HTTP forward
2861+ -- happens asynchronously in a fork below; retries are handled by
2862+ -- the background prune cycle if the initial attempt fails.
28392863 let delivery = if localOnline then DsDelivered {}
28402864 else if federationEnabled && count remoteServers > 0 then DsFederated {}
28412865 else DsQueued {}
@@ -3099,14 +3123,20 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp authorizati
30993123 let deduped = mergeByPubkeyServer [] filteredOnline
31003124 gossipResult <- atomic do
31013125 allPresence <- readAllPresence
3102− let gm = mergeGossipPresence allPresence deduped trustedOffline fromServer t
3126+ tpc <- *totalPresenceCount
3127+ -- Pre-filter: when near capacity, reject gossip entries for pubkeys not already in presence.
3128+ -- This prevents a malicious peer from spiking presence above maxPresenceEntries via large batches.
3129+ let capacityDeduped = if tpc >= maxPresenceEntries
3130+ then do d <- deduped; where any allPresence (\p -> p.pubkey == d.pubkey); yield d
3131+ else deduped
3132+ let gm = mergeGossipPresence allPresence capacityDeduped trustedOffline fromServer t
31033133 writeAllPresence gm.finalPresence
31043134 set *totalPresenceCount = count gm.finalPresence
31053135 set *knownPeerServers = gm.gossipPeerServers
3106− yield {gm: gm}
3136+ yield {gm: gm, preFiltered: count deduped - count capacityDeduped}
31073137 let effectiveEvents = union (effectiveGossipEvents filteredOnline gossipResult.gm.appliedKeys (GossipOnline {})) (effectiveGossipEvents filteredOffline gossipResult.gm.appliedKeys (GossipOffline {}))
3108− when (not gossipResult.gm.withinCapacity)
3109− (println ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer))
3138+ when (not gossipResult.gm.withinCapacity || gossipResult.preFiltered > 0)
3139+ (println ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer ++ " (pre-filtered: " ++ show gossipResult.preFiltered ++ ")"))
31103140 -- Suppress re-broadcast of events we've already gossiped recently (prevents amplification)
31113141 freshEvents <- atomic do
31123142 recentGossip <- *recentGossip
@@ -3417,9 +3447,10 @@ handleStats = \authorization respond -> do
34173447 let fwdPct = if maxForwardsEntries > 0 then result.forwards * 100 / maxForwardsEntries else 0
34183448 let sessPct = if maxSessions > 0 then result.sessions * 100 / maxSessions else 0
34193449 let fedAvg = if result.federationLatencyCount > 0 then result.federationLatencyTotal / result.federationLatencyCount else 0
3450+ let hungFedSlots = count (do r <- core.fi; where r.count > 0 && r.firstStartedAt > 0 && t - r.firstStartedAt >= federationFetchTimeoutMs; yield r)
34203451 let oldestFwdAge = case minBy core.f (\x -> x.createdAt) of Nothing {} -> 0; Just {value: x} -> t - x.createdAt
34213452 let oldestAckAge = case minBy core.a (\x -> x.ackedAt) of Nothing {} -> 0; Just {value: x} -> t - x.ackedAt
3422− yield (respond {sessions: result.sessions, sessionsMax: maxSessions, messages: result.messages, messagesMax: maxMessages, presence: result.presence, presenceMax: maxPresenceEntries, forwards: result.forwards, forwardsMax: maxForwardsEntries, challenges: result.challenges, challengesMax: maxChallenges, ackedIds: result.ackedIds, ackedIdsMax: maxAckedIds, ackOverflow: result.ackOverflow, ackOverflowMax: maxAckOverflow, deliveredIds: result.deliveredIds, deliveredIdsMax: maxDeliveredIds, forwardFailures: result.forwardFailures, activePolls: result.activePolls, activePollsMax: maxActivePolls, sendRates: result.sendRates, challengeRates: result.challengeRates, federationRates: result.federationRates, globalRateCount: result.globalRateCount, ipRates: result.ipRates, lookupRates: result.lookupRates, recvRates: result.recvRates, pairRates: result.pairRates, federationPairRates: result.federationPairRates, pruneCount: result.pruneCount, pruneRunning: result.pruneRunning, heartbeatCount: result.heartbeatCount, uptime: result.uptime, lastPruneAt: result.lastPruneAt, lastHeartbeatAt: result.lastHeartbeatAt, draining: result.draining, federationEnabled: federationEnabled, recentGossip: result.recentGossip, totalMessageCount: result.totalMessageCount, recipientCounts: result.recipientCounts, federationInFlight: result.federationInFlight, messagesPct: msgPct, presencePct: presPct, forwardsPct: fwdPct, sessionsPct: sessPct, federationAvgLatencyMs: fedAvg, federationMaxLatencyMs: result.federationLatencyMax, federationCallCount: result.federationLatencyCount, drainCompletedAt: result.drainCompletedAt, oldestForwardAge: oldestFwdAge, oldestAckAge: oldestAckAge, ok: True {}})
3453+ yield (respond {sessions: result.sessions, sessionsMax: maxSessions, messages: result.messages, messagesMax: maxMessages, presence: result.presence, presenceMax: maxPresenceEntries, forwards: result.forwards, forwardsMax: maxForwardsEntries, challenges: result.challenges, challengesMax: maxChallenges, ackedIds: result.ackedIds, ackedIdsMax: maxAckedIds, ackOverflow: result.ackOverflow, ackOverflowMax: maxAckOverflow, deliveredIds: result.deliveredIds, deliveredIdsMax: maxDeliveredIds, forwardFailures: result.forwardFailures, activePolls: result.activePolls, activePollsMax: maxActivePolls, sendRates: result.sendRates, challengeRates: result.challengeRates, federationRates: result.federationRates, globalRateCount: result.globalRateCount, ipRates: result.ipRates, lookupRates: result.lookupRates, recvRates: result.recvRates, pairRates: result.pairRates, federationPairRates: result.federationPairRates, pruneCount: result.pruneCount, pruneRunning: result.pruneRunning, heartbeatCount: result.heartbeatCount, uptime: result.uptime, lastPruneAt: result.lastPruneAt, lastHeartbeatAt: result.lastHeartbeatAt, draining: result.draining, federationEnabled: federationEnabled, recentGossip: result.recentGossip, totalMessageCount: result.totalMessageCount, recipientCounts: result.recipientCounts, federationInFlight: result.federationInFlight, hungFederationSlots: hungFedSlots, messagesPct: msgPct, presencePct: presPct, forwardsPct: fwdPct, sessionsPct: sessPct, federationAvgLatencyMs: fedAvg, federationMaxLatencyMs: result.federationLatencyMax, federationCallCount: result.federationLatencyCount, drainCompletedAt: result.drainCompletedAt, oldestForwardAge: oldestFwdAge, oldestAckAge: oldestAckAge, ok: True {}})
34233454 else yield (respond zeroStats)
34243455
34253456handleStatsPeers = \authorization respond -> do
@@ -3705,6 +3736,7 @@ main = do
37053736 (println "WARNING: maintenanceSecret is empty — /maintenance/* and /stats endpoints are disabled")
37063737 when (matchesHostnameOrSubdomain serverHostN "localhost")
37073738 (println ("WARNING: serverHost is " ++ serverHost ++ " — peers cannot reach this server for federation"))
3739+ println "NOTE: IP binding is active — sessions created with a ClientIp header will reject requests without one"
37083740 println "Skrepka relay server starting on port 8080..."
37093741 println (" federation: " ++ if federationEnabled then "enabled" else "disabled")
37103742 println (" prune interval: " ++ show backgroundPruneInterval ++ "ms")