@@ -99,12 +99,6 @@ maxSendRatePerKey = 20
99-- Window for the per-key send rate limit (1 s).
99-- Window for the per-key send rate limit (1 s).
100sendRateWindowMs : Int<Ms>
100sendRateWindowMs : Int<Ms>
101sendRateWindowMs = 1000
101sendRateWindowMs = 1000
102−-- Receive/poll requests permitted per key per window.
103−maxRecvRatePerKey : Int
104−maxRecvRatePerKey = 10
105−-- Window for the per-key receive rate limit (1 s).
106−recvRateWindowMs : Int<Ms>
107−recvRateWindowMs = 1000
108-- Requests permitted per source IP per window (covers unauthenticated traffic).
102-- Requests permitted per source IP per window (covers unauthenticated traffic).
109maxIpRatePerIp : Int
103maxIpRatePerIp : Int
110maxIpRatePerIp = 10
104maxIpRatePerIp = 10
@@ -268,8 +262,6 @@ type RateEntry = {key: Text, windowStart: Timestamp, count: Int}
269-- Per-key fixed-window counters for /messages send-side rate limiting.
263-- Per-key fixed-window counters for /messages send-side rate limiting.
270*sendRates : [RateEntry]
264*sendRates : [RateEntry]
271−-- Per-recipient counters; protects a popular key from being flooded by many senders.
272−*recvRates : [RateEntry]
273-- Per-key /auth/challenge counters (separate from `*challenges` capacity gate).
265-- Per-key /auth/challenge counters (separate from `*challenges` capacity gate).
274*challengeRates : [RateEntry]
266*challengeRates : [RateEntry]
275-- Per-source-IP counters; gate for unauthenticated traffic.
267-- Per-source-IP counters; gate for unauthenticated traffic.
@@ -470,7 +462,7 @@ isBadServerName = \sn -> sn == "" || sn == "localhost" || not all (chars sn) (\c
470-- Bearer token extraction, session lookup, and the `gateAuth` combinator that
462-- Bearer token extraction, session lookup, and the `gateAuth` combinator that
471-- every client handler funnels through (global rate → session).
463-- every client handler funnels through (global rate → session).
472-- Federation requests are unauthenticated: anyone can hit /federation/*.
464-- Federation requests are unauthenticated: anyone can hit /federation/*.
473−-- Per-recipient `recvRate` and `no_presence` checks bound the spam ceiling.
465+-- The `no_presence` check bounds federated forwards to currently-online keys.
475-- Plain ip match (sessions with empty ip skip the check).
467-- Plain ip match (sessions with empty ip skip the check).
476matchIp : Text -> Text -> Bool
468matchIp : Text -> Text -> Bool
@@ -913,7 +905,7 @@ pollOnce = \authorization cursor clientIp t -> do
913-- some other transaction commits a change to `*pollWakeup`. Because
905-- some other transaction commits a change to `*pollWakeup`. Because
914-- `pollHeartbeat` bumps `*pollWakeup` every `pollTimeoutMs`, no caller waits
906-- `pollHeartbeat` bumps `*pollWakeup` every `pollTimeoutMs`, no caller waits
915-- longer than one heartbeat period before getting at least an empty response.
907-- longer than one heartbeat period before getting at least an empty response.
916−doPoll : {clampedCursor: Int, pubkey: PubkeyHex, startWakeup: Timestamp} -> Int<Ms> -> IO {r *pollWakeup, clock} {cursor: Int<Ms>, events: [{encryptedBlob: BlobHex}]}
908+doPoll : {clampedCursor: Int, pubkey: PubkeyHex, startWakeup: Timestamp | a} -> Int<Ms> -> IO {r *pollWakeup, clock} {cursor: Int<Ms>, events: [{encryptedBlob: BlobHex}]}
917doPoll = \s t -> do
909doPoll = \s t -> do
918 let collectAt = \tnow -> atomic (do
910 let collectAt = \tnow -> atomic (do
919 msgs <- *messages
911 msgs <- *messages
@@ -954,17 +946,17 @@ ackPolledMessages = \pubkey cursor -> do
954-- either succeeds (every message enqueued) or fails with a non-2xx HTTP
946-- either succeeds (every message enqueued) or fails with a non-2xx HTTP
955-- status carrying the rejection code ("rate_limited" | "self_send" | ...).
947-- status carrying the rejection code ("rate_limited" | "self_send" | ...).
956-- Duplicates are silently treated as success so retries are idempotent.
948-- Duplicates are silently treated as success so retries are idempotent.
957−-- Send is gated by global rate and the per-sender / per-pair / per-recipient
949+-- Send is gated by global rate and the per-sender rate limit charged inside
958−-- rate limits charged together inside `checkSendRate`.
950+-- `checkSendRate`.
959-- Federated forwards are kicked off in background forks after the atomic.
951-- Federated forwards are kicked off in background forks after the atomic.
961-- POST /messages: send a list of ciphertexts atomically. Every message is
953-- POST /messages: send a list of ciphertexts atomically. Every message is
962−-- validated (self-send + send/recv rate windows) inside a single atomic
954+-- validated (self-send + per-sender rate window) inside a single atomic
963-- block; if any one would be rejected, none are enqueued and the call returns
955-- block; if any one would be rejected, none are enqueued and the call returns
964−-- a non-2xx HTTP status. Rate counters are still charged for every attempted
956+-- a non-2xx HTTP status. The send-rate counter is still charged for every
965−-- message even when the batch is rejected, so a sender can't probe limits
957+-- attempted message even when the batch is rejected, so a sender can't probe
966−-- cheaply (e.g. by sending to themselves). Federation forwards are kicked
958+-- limits cheaply (e.g. by sending to themselves). Federation forwards are
967−-- per message after commit.
959+-- kicked per message after commit.
968handleSendMessages : [{encryptedBlob: BlobHex, to: PubkeyHex | a}] -> Text -> Text -> IO {rw *globalRateCount, rw *globalRateWindowStart, r *sessions, clock} (Result HttpError {})
960handleSendMessages : [{encryptedBlob: BlobHex, to: PubkeyHex | a}] -> Text -> Text -> IO {rw *globalRateCount, rw *globalRateWindowStart, r *sessions, clock} (Result HttpError {})
969handleSendMessages = \messages authorization clientIp -> do
961handleSendMessages = \messages authorization clientIp -> do
@@ -987,23 +979,20 @@ handleSendMessages = \messages authorization clientIp -> do
987 when federationEnabled (forEach checked (\c -> fork (forkForwardsTo c.to c.encryptedBlob c.remoteServers)))
979 when federationEnabled (forEach checked (\c -> fork (forkForwardsTo c.to c.encryptedBlob c.remoteServers)))
988 yield (Ok {value: {}}))
980 yield (Ok {value: {}}))
990−-- Charge send/recv rate counters for one (sender, recipient) pair and
982+-- Charge the send rate counter for the sender and decide whether the message
991−-- decide whether the message would be accepted. Mutates only the rate
983+-- would be accepted. Mutates only the rate counter, never the mailbox; the
992−-- counters, never the mailbox; the caller commits via `commitSendMessage`
984+-- caller commits via `commitSendMessage` only after every message in the
993−-- only after every message in the batch has passed.
994−-- Both rate counters are charged unconditionally (even on a failure)
985+-- batch has passed. The counter is charged unconditionally (even on a
995−-- so that a sender can't probe limits cheaply by sending to themselves.
986+-- failure) so that a sender can't probe limits cheaply by sending to
996−checkSendRate : PubkeyHex -> PubkeyHex -> Int<Ms> -> IO {r *presence, rw *recvRates, rw *sendRates} {remoteServers: [Text], error: Text}
988+checkSendRate : PubkeyHex -> PubkeyHex -> Int<Ms> -> IO {r *presence, rw *sendRates} {remoteServers: [Text], error: Text}
997checkSendRate = \senderKey to t -> do
989checkSendRate = \senderKey to t -> do
998 sRates <- *sendRates
990 sRates <- *sendRates
999 let sR = checkRate sRates senderKey maxSendRatePerKey sendRateWindowMs t
991 let sR = checkRate sRates senderKey maxSendRatePerKey sendRateWindowMs t
1000 *sendRates = sR.entries
992 *sendRates = sR.entries
1001− rRates <- *recvRates
1002− let rR = checkRate rRates to maxRecvRatePerKey recvRateWindowMs t
1003− *recvRates = rR.entries
1004 if senderKey == to
993 if senderKey == to
1005 then yield {error: "self_send", remoteServers: []}
994 then yield {error: "self_send", remoteServers: []}
1006− else if sR.limited || rR.limited
995+ else if sR.limited
1007 then yield {error: "rate_limited", remoteServers: []}
996 then yield {error: "rate_limited", remoteServers: []}
1009 pres <- *presence
998 pres <- *presence
@@ -1070,8 +1059,8 @@ forkForwardsTo = \to encryptedBlob servers -> forEach servers (\server -> do
1070-- Handler: federation receive
1059-- Handler: federation receive
1071-- ============================================
1060-- ============================================
1072-- Inbound peer-to-peer endpoints. Federation is open and unauthenticated:
1061-- Inbound peer-to-peer endpoints. Federation is open and unauthenticated:
1073−-- the per-recipient `recvRate` and `no_presence` checks in `storeFederated`
1062+-- the `no_presence` check in `storeFederated` bounds forwards to currently-
1074−-- do the actual spam-bounding.
1063+-- online local keys.
1076-- POST /federation/gossip: ingest peer presence events. The transport-level
1065-- POST /federation/gossip: ingest peer presence events. The transport-level
1077-- `fromServer` is the home server for every event. Drops every (pubkey,
1066-- `fromServer` is the home server for every event. Drops every (pubkey,
@@ -1107,43 +1096,37 @@ handleRecvGossip = \events fromServer -> do
1107 yield (Ok {value: {}})
1096 yield (Ok {value: {}})
1109-- Store a federated/forwarded message into the local mailbox. Caller wraps in `atomic`.
1098-- Store a federated/forwarded message into the local mailbox. Caller wraps in `atomic`.
1110−-- Returns "" on success, otherwise an error code: "no_presence" | "rate_limited".
1099+-- Returns "" on success, otherwise the error code "no_presence".
1112-- The presence gate requires the recipient to be **currently locally online**
1101-- The presence gate requires the recipient to be **currently locally online**
1113-- (not just visible to gossip somewhere). Without that, an A → B forward based
1102-- (not just visible to gossip somewhere). Without that, an A → B forward based
1114-- on stale gossip about R could land on B with no actively-polling client and
1103-- on stale gossip about R could land on B with no actively-polling client and
1115-- get stuck. If B refuses, A retries via `retryPendingForwards` against the
1104-- get stuck. If B refuses, A retries via `retryPendingForwards` against the
1116-- next presence target.
1105-- next presence target.
1117−storeFederated : PubkeyHex -> BlobHex -> Int<Ms> -> IO {rw *messages, rw *pollWakeup, r *presence, rw *recvRates, rw *seqCounter} Text
1106+storeFederated : PubkeyHex -> BlobHex -> Int<Ms> -> IO {rw *messages, rw *pollWakeup, r *presence, rw *seqCounter} Text
1118storeFederated = \toKey encryptedBlob t -> do
1107storeFederated = \toKey encryptedBlob t -> do
1119 pres <- *presence
1108 pres <- *presence
1120 msgs <- *messages
1109 msgs <- *messages
1121 if not any pres (\p -> p.pubkey == toKey && isLocalServer p.server && p.expiresAt > t)
1110 if not any pres (\p -> p.pubkey == toKey && isLocalServer p.server && p.expiresAt > t)
1122 then yield "no_presence"
1111 then yield "no_presence"
1124− rRates <- *recvRates
1125− let r = checkRate rRates toKey maxRecvRatePerKey recvRateWindowMs t
1126− *recvRates = r.entries
1128− then yield "rate_limited"
1130− seq <- *seqCounter
1113+ seq <- *seqCounter
1131− let seqTs = nextSeqTs t seq
1114+ let seqTs = nextSeqTs t seq
1132− *seqCounter = seqTs
1115+ *seqCounter = seqTs
1133− *messages = union msgs [{toKey, encryptedBlob, receivedAt: seqTs}]
1116+ *messages = union msgs [{toKey, encryptedBlob, receivedAt: seqTs}]
1134− *pollWakeup = seqTs
1117+ *pollWakeup = seqTs
1135− yield ""
1118+ yield ""
1137-- POST /federation/forward: receive a single federated message destined for
1120-- POST /federation/forward: receive a single federated message destined for
1138-- a local key. Delegates the actual store decision to `storeFederated`,
1121-- a local key. Delegates the actual store decision to `storeFederated`,
1139-- which returns "" on success.
1122-- which returns "" on success.
1140−handleForwardMessage : PubkeyHex -> BlobHex -> IO {rw *messages, rw *pollWakeup, r *presence, rw *recvRates, rw *seqCounter, clock} (Result HttpError {})
1123+handleForwardMessage : PubkeyHex -> BlobHex -> IO {rw *messages, rw *pollWakeup, r *presence, rw *seqCounter, clock} (Result HttpError {})
1141handleForwardMessage = \toKey encryptedBlob -> do
1124handleForwardMessage = \toKey encryptedBlob -> do
1142 t <- now
1125 t <- now
1143 result <- atomic storeFederated toKey encryptedBlob t
1126 result <- atomic storeFederated toKey encryptedBlob t
1144 if result == ""
1127 if result == ""
1145 then yield (Ok {value: {}})
1128 then yield (Ok {value: {}})
1146− else yield (Err {error: {status: if result == "no_presence" then 404 else 429, message: result}})
1129+ else yield (Err {error: {status: 404, message: result}})
1148-- ============================================
1131-- ============================================
1149-- Dispatch
1132-- Dispatch