Skip to main content

max / makenotwork

12.3 KB · 412 lines History Blame Raw
1 //! The send path.
2 //!
3 //! Six things have to happen in a particular order, and the order is a security
4 //! property rather than a style preference, so it lives here once instead of in
5 //! each host.
6 //!
7 //! 1. **Validate the body.** Free, and rejects the largest class of junk.
8 //! 2. **Check room state.** Free, and a closed or read-only room ends it.
9 //! 3. **Take a rate token.** In-memory. Deliberately before authz, so a caller
10 //! hammering a room they cannot write to is throttled at memory speed instead
11 //! of putting a database query behind every attempt. The cost is that a
12 //! refused sender still spends budget, which is the right trade: the budget
13 //! exists to protect the room and the box, not to be fair to someone who is
14 //! already being refused.
15 //! 4. **Check authz.** The first step that touches the host's database.
16 //! 5. **Store.** The host renders through docengine and assigns the id.
17 //! 6. **Publish.** Only after the row is durable, so nobody sees a message that
18 //! a failed insert means will not survive a reconnect.
19 //!
20 //! Getting 5 and 6 backwards is the interesting mistake. Publishing first makes
21 //! chat feel faster and produces a room where a message is visible to everyone
22 //! present, absent from the backlog, and gone for anyone who reloads.
23
24 use crate::chat::SendRequest;
25 use crate::error::ChatError;
26 use crate::event::ChatEvent;
27 use crate::hub::Hub;
28 use crate::message::{Message, MessageBody};
29 use crate::rate_limit::{RateDecision, RateLimiter};
30 use crate::traits::{ChatAuthz, DenyReason, WriteAccess};
31
32 /// Why a send was refused, distinguishing the cases a caller must respond to
33 /// differently.
34 #[derive(Debug, Clone, PartialEq, Eq)]
35 pub enum SendRejection {
36 /// The body did not survive validation.
37 Invalid(String),
38 /// The room is not accepting messages.
39 RoomClosed,
40 RoomReadOnly,
41 /// Authz or rate limiting said no.
42 Denied(DenyReason),
43 }
44
45 /// Validate, authorize, store, and fan out one message.
46 ///
47 /// Reached through [`crate::Chat::send`]; the module docs above explain the
48 /// ordering it enforces.
49 ///
50 /// `store` receives the validated body and must render it (through docengine's
51 /// chat preset, never raw) and insert it, returning the stored [`Message`]. It is
52 /// called only after every check has passed.
53 pub(crate) async fn send<A, F, Fut>(
54 hub: &Hub,
55 limiter: &RateLimiter,
56 authz: &A,
57 req: SendRequest<'_>,
58 store: F,
59 ) -> Result<Message, ChatError>
60 where
61 A: ChatAuthz + ?Sized,
62 F: FnOnce(MessageBody) -> Fut,
63 Fut: Future<Output = Result<Message, ChatError>>,
64 {
65 let SendRequest {
66 room,
67 author,
68 body: raw_body,
69 nonce,
70 now,
71 } = req;
72
73 // 1. Free.
74 let body = MessageBody::parse(raw_body)
75 .map_err(|e| ChatError::Rejected(SendRejection::Invalid(e.to_string())))?;
76
77 // 2. Free.
78 if !room.state.allows_reads() {
79 return Err(ChatError::Rejected(SendRejection::RoomClosed));
80 }
81 if !room.state.allows_writes() {
82 return Err(ChatError::Rejected(SendRejection::RoomReadOnly));
83 }
84
85 // 3. In-memory, before anything that queries.
86 if let RateDecision::Deny { retry_after } = limiter.check(author, room.id, now) {
87 return Err(ChatError::Rejected(SendRejection::Denied(
88 DenyReason::RateLimited { retry_after },
89 )));
90 }
91
92 // 4. First database hit.
93 if let WriteAccess::Deny(reason) = authz.can_write(author, room).await? {
94 return Err(ChatError::Rejected(SendRejection::Denied(reason)));
95 }
96
97 // 5. Durable before visible.
98 let mut stored = store(body).await?;
99 stored.nonce = nonce;
100
101 // 6. Fan out. A room with nobody in it is normal and not a failure.
102 hub.publish(room.id, ChatEvent::Message(stored.clone()));
103
104 Ok(stored)
105 }
106
107 #[cfg(test)]
108 mod tests {
109 use super::*;
110 use crate::hub::HubLimits;
111 use crate::ids::{MessageId, Nonce, RoomId, UserId};
112 use crate::message::MAX_MESSAGE_LEN;
113 use crate::rate_limit::RateLimits;
114 use crate::retention::Retention;
115 use crate::room::{Room, RoomState};
116 use async_trait::async_trait;
117 use std::sync::Arc;
118 use std::sync::atomic::{AtomicUsize, Ordering};
119 use std::time::Instant;
120 use uuid::Uuid;
121
122 struct Authz {
123 decision: WriteAccess,
124 calls: Arc<AtomicUsize>,
125 }
126
127 #[async_trait]
128 impl ChatAuthz for Authz {
129 async fn can_read(&self, _v: Option<UserId>, room: &Room) -> Result<bool, ChatError> {
130 Ok(room.state.allows_reads())
131 }
132 async fn can_write(&self, _u: UserId, _r: &Room) -> Result<WriteAccess, ChatError> {
133 self.calls.fetch_add(1, Ordering::SeqCst);
134 Ok(self.decision.clone())
135 }
136 async fn is_moderator(&self, _u: UserId, _r: &Room) -> Result<bool, ChatError> {
137 Ok(false)
138 }
139 }
140
141 fn authz(decision: WriteAccess) -> (Authz, Arc<AtomicUsize>) {
142 let calls = Arc::new(AtomicUsize::new(0));
143 (
144 Authz {
145 decision,
146 calls: calls.clone(),
147 },
148 calls,
149 )
150 }
151
152 fn room(state: RoomState) -> Room {
153 Room {
154 id: RoomId(Uuid::new_v4()),
155 state,
156 retention: Retention::forum_default(),
157 }
158 }
159
160 fn stored(room: RoomId, author: UserId, body: &MessageBody) -> Message {
161 Message {
162 id: MessageId(1),
163 room_id: room,
164 author_id: author,
165 body_html: format!("<p>{}</p>", body.as_str()),
166 created_at: 0,
167 nonce: None,
168 }
169 }
170
171 fn parts() -> (Hub, RateLimiter) {
172 (
173 Hub::new(HubLimits::default()),
174 RateLimiter::new(RateLimits::default()),
175 )
176 }
177
178 fn req<'a>(room: &'a Room, author: UserId, body: &'a str) -> SendRequest<'a> {
179 SendRequest {
180 room,
181 author,
182 body,
183 nonce: None,
184 now: Instant::now(),
185 }
186 }
187
188 #[tokio::test]
189 async fn a_good_message_is_stored_then_published() {
190 let (hub, rl) = parts();
191 let (a, _) = authz(WriteAccess::Allow);
192 let r = room(RoomState::Open);
193 let author = UserId(Uuid::new_v4());
194 let mut listener = hub.subscribe(r.id, UserId(Uuid::new_v4())).unwrap();
195
196 let out = send(
197 &hub,
198 &rl,
199 &a,
200 SendRequest {
201 nonce: Nonce::parse("n1"),
202 ..req(&r, author, " hello ")
203 },
204 |body| async move { Ok(stored(r.id, author, &body)) },
205 )
206 .await
207 .unwrap();
208
209 assert_eq!(
210 out.body_html, "<p>hello</p>",
211 "body was trimmed and rendered"
212 );
213 assert_eq!(out.nonce, Nonce::parse("n1"));
214
215 match listener.recv().await {
216 Some(ChatEvent::Message(m)) => {
217 assert_eq!(m.id, MessageId(1));
218 assert_eq!(m.nonce, Nonce::parse("n1"), "sender can reconcile");
219 }
220 other => panic!("expected the message on the room, got {other:?}"),
221 }
222 }
223
224 #[tokio::test]
225 async fn nothing_is_published_when_the_store_fails() {
226 // The ordering this protects: a message visible to everyone present but
227 // absent from the backlog and gone on reload.
228 let (hub, rl) = parts();
229 let (a, _) = authz(WriteAccess::Allow);
230 let r = room(RoomState::Open);
231 let mut listener = hub.subscribe(r.id, UserId(Uuid::new_v4())).unwrap();
232
233 let result = send(
234 &hub,
235 &rl,
236 &a,
237 req(&r, UserId(Uuid::new_v4()), "hi"),
238 |_| async { Err(ChatError::host(std::io::Error::other("db down"))) },
239 )
240 .await;
241
242 assert!(result.is_err());
243 // Probe the room. If a phantom message had been published it would be
244 // ahead of this in the queue.
245 assert_eq!(hub.publish(r.id, ChatEvent::Gap), 1, "listener still live");
246 assert!(
247 matches!(listener.recv().await, Some(ChatEvent::Gap)),
248 "a failed store must publish nothing"
249 );
250 }
251
252 #[tokio::test]
253 async fn an_empty_body_never_reaches_authz() {
254 let (hub, rl) = parts();
255 let (a, calls) = authz(WriteAccess::Allow);
256 let r = room(RoomState::Open);
257
258 let result = send(
259 &hub,
260 &rl,
261 &a,
262 req(&r, UserId(Uuid::new_v4()), " "),
263 |_| async { panic!("store must not run") },
264 )
265 .await;
266
267 assert!(matches!(
268 result,
269 Err(ChatError::Rejected(SendRejection::Invalid(_)))
270 ));
271 assert_eq!(calls.load(Ordering::SeqCst), 0, "no query for a blank body");
272 }
273
274 #[tokio::test]
275 async fn an_overlong_body_is_refused() {
276 let (hub, rl) = parts();
277 let (a, _) = authz(WriteAccess::Allow);
278 let r = room(RoomState::Open);
279 let long = "x".repeat(MAX_MESSAGE_LEN + 1);
280
281 let result = send(
282 &hub,
283 &rl,
284 &a,
285 req(&r, UserId(Uuid::new_v4()), &long),
286 |_| async { panic!("store must not run") },
287 )
288 .await;
289
290 assert!(matches!(
291 result,
292 Err(ChatError::Rejected(SendRejection::Invalid(_)))
293 ));
294 }
295
296 #[tokio::test]
297 async fn a_read_only_room_refuses_before_querying() {
298 let (hub, rl) = parts();
299 let (a, calls) = authz(WriteAccess::Allow);
300 let r = room(RoomState::ReadOnly);
301
302 let result = send(
303 &hub,
304 &rl,
305 &a,
306 req(&r, UserId(Uuid::new_v4()), "hi"),
307 |_| async { panic!("store must not run") },
308 )
309 .await;
310
311 assert!(matches!(
312 result,
313 Err(ChatError::Rejected(SendRejection::RoomReadOnly))
314 ));
315 assert_eq!(calls.load(Ordering::SeqCst), 0);
316 }
317
318 #[tokio::test]
319 async fn a_closed_room_refuses_as_closed_not_read_only() {
320 let (hub, rl) = parts();
321 let (a, _) = authz(WriteAccess::Allow);
322 let r = room(RoomState::Closed);
323
324 let result = send(
325 &hub,
326 &rl,
327 &a,
328 req(&r, UserId(Uuid::new_v4()), "hi"),
329 |_| async { panic!("store must not run") },
330 )
331 .await;
332
333 assert!(matches!(
334 result,
335 Err(ChatError::Rejected(SendRejection::RoomClosed))
336 ));
337 }
338
339 #[tokio::test]
340 async fn a_denied_sender_gets_the_specific_reason() {
341 let (hub, rl) = parts();
342 let (a, _) = authz(WriteAccess::Deny(DenyReason::Muted));
343 let r = room(RoomState::Open);
344
345 let result = send(
346 &hub,
347 &rl,
348 &a,
349 req(&r, UserId(Uuid::new_v4()), "hi"),
350 |_| async { panic!("store must not run") },
351 )
352 .await;
353
354 assert!(matches!(
355 result,
356 Err(ChatError::Rejected(SendRejection::Denied(
357 DenyReason::Muted
358 )))
359 ));
360 }
361
362 #[tokio::test]
363 async fn the_rate_limit_bites_before_authz_is_queried() {
364 // A caller hammering a room is throttled in memory rather than putting a
365 // query behind every attempt.
366 let hub = Hub::new(HubLimits::default());
367 let rl = RateLimiter::new(RateLimits {
368 burst: 2,
369 sustain_per_min: 60,
370 });
371 let (a, calls) = authz(WriteAccess::Allow);
372 let r = room(RoomState::Open);
373 let author = UserId(Uuid::new_v4());
374 let t0 = Instant::now();
375
376 let at = |body| SendRequest {
377 room: &r,
378 author,
379 body,
380 nonce: None,
381 now: t0,
382 };
383
384 for _ in 0..2 {
385 send(&hub, &rl, &a, at("hi"), |body| async move {
386 Ok(stored(r.id, author, &body))
387 })
388 .await
389 .unwrap();
390 }
391 assert_eq!(calls.load(Ordering::SeqCst), 2);
392
393 for _ in 0..10 {
394 let result = send(&hub, &rl, &a, at("hi"), |_| async {
395 panic!("store must not run")
396 })
397 .await;
398 assert!(matches!(
399 result,
400 Err(ChatError::Rejected(SendRejection::Denied(
401 DenyReason::RateLimited { .. }
402 )))
403 ));
404 }
405 assert_eq!(
406 calls.load(Ordering::SeqCst),
407 2,
408 "throttled attempts must not reach the database"
409 );
410 }
411 }
412