Skip to main content

max / makenotwork

12.4 KB · 413 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 author: None,
169 }
170 }
171
172 fn parts() -> (Hub, RateLimiter) {
173 (
174 Hub::new(HubLimits::default()),
175 RateLimiter::new(RateLimits::default()),
176 )
177 }
178
179 fn req<'a>(room: &'a Room, author: UserId, body: &'a str) -> SendRequest<'a> {
180 SendRequest {
181 room,
182 author,
183 body,
184 nonce: None,
185 now: Instant::now(),
186 }
187 }
188
189 #[tokio::test]
190 async fn a_good_message_is_stored_then_published() {
191 let (hub, rl) = parts();
192 let (a, _) = authz(WriteAccess::Allow);
193 let r = room(RoomState::Open);
194 let author = UserId(Uuid::new_v4());
195 let mut listener = hub.subscribe(r.id, UserId(Uuid::new_v4())).unwrap();
196
197 let out = send(
198 &hub,
199 &rl,
200 &a,
201 SendRequest {
202 nonce: Nonce::parse("n1"),
203 ..req(&r, author, " hello ")
204 },
205 |body| async move { Ok(stored(r.id, author, &body)) },
206 )
207 .await
208 .unwrap();
209
210 assert_eq!(
211 out.body_html, "<p>hello</p>",
212 "body was trimmed and rendered"
213 );
214 assert_eq!(out.nonce, Nonce::parse("n1"));
215
216 match listener.recv().await {
217 Some(ChatEvent::Message(m)) => {
218 assert_eq!(m.id, MessageId(1));
219 assert_eq!(m.nonce, Nonce::parse("n1"), "sender can reconcile");
220 }
221 other => panic!("expected the message on the room, got {other:?}"),
222 }
223 }
224
225 #[tokio::test]
226 async fn nothing_is_published_when_the_store_fails() {
227 // The ordering this protects: a message visible to everyone present but
228 // absent from the backlog and gone on reload.
229 let (hub, rl) = parts();
230 let (a, _) = authz(WriteAccess::Allow);
231 let r = room(RoomState::Open);
232 let mut listener = hub.subscribe(r.id, UserId(Uuid::new_v4())).unwrap();
233
234 let result = send(
235 &hub,
236 &rl,
237 &a,
238 req(&r, UserId(Uuid::new_v4()), "hi"),
239 |_| async { Err(ChatError::host(std::io::Error::other("db down"))) },
240 )
241 .await;
242
243 assert!(result.is_err());
244 // Probe the room. If a phantom message had been published it would be
245 // ahead of this in the queue.
246 assert_eq!(hub.publish(r.id, ChatEvent::Gap), 1, "listener still live");
247 assert!(
248 matches!(listener.recv().await, Some(ChatEvent::Gap)),
249 "a failed store must publish nothing"
250 );
251 }
252
253 #[tokio::test]
254 async fn an_empty_body_never_reaches_authz() {
255 let (hub, rl) = parts();
256 let (a, calls) = authz(WriteAccess::Allow);
257 let r = room(RoomState::Open);
258
259 let result = send(
260 &hub,
261 &rl,
262 &a,
263 req(&r, UserId(Uuid::new_v4()), " "),
264 |_| async { panic!("store must not run") },
265 )
266 .await;
267
268 assert!(matches!(
269 result,
270 Err(ChatError::Rejected(SendRejection::Invalid(_)))
271 ));
272 assert_eq!(calls.load(Ordering::SeqCst), 0, "no query for a blank body");
273 }
274
275 #[tokio::test]
276 async fn an_overlong_body_is_refused() {
277 let (hub, rl) = parts();
278 let (a, _) = authz(WriteAccess::Allow);
279 let r = room(RoomState::Open);
280 let long = "x".repeat(MAX_MESSAGE_LEN + 1);
281
282 let result = send(
283 &hub,
284 &rl,
285 &a,
286 req(&r, UserId(Uuid::new_v4()), &long),
287 |_| async { panic!("store must not run") },
288 )
289 .await;
290
291 assert!(matches!(
292 result,
293 Err(ChatError::Rejected(SendRejection::Invalid(_)))
294 ));
295 }
296
297 #[tokio::test]
298 async fn a_read_only_room_refuses_before_querying() {
299 let (hub, rl) = parts();
300 let (a, calls) = authz(WriteAccess::Allow);
301 let r = room(RoomState::ReadOnly);
302
303 let result = send(
304 &hub,
305 &rl,
306 &a,
307 req(&r, UserId(Uuid::new_v4()), "hi"),
308 |_| async { panic!("store must not run") },
309 )
310 .await;
311
312 assert!(matches!(
313 result,
314 Err(ChatError::Rejected(SendRejection::RoomReadOnly))
315 ));
316 assert_eq!(calls.load(Ordering::SeqCst), 0);
317 }
318
319 #[tokio::test]
320 async fn a_closed_room_refuses_as_closed_not_read_only() {
321 let (hub, rl) = parts();
322 let (a, _) = authz(WriteAccess::Allow);
323 let r = room(RoomState::Closed);
324
325 let result = send(
326 &hub,
327 &rl,
328 &a,
329 req(&r, UserId(Uuid::new_v4()), "hi"),
330 |_| async { panic!("store must not run") },
331 )
332 .await;
333
334 assert!(matches!(
335 result,
336 Err(ChatError::Rejected(SendRejection::RoomClosed))
337 ));
338 }
339
340 #[tokio::test]
341 async fn a_denied_sender_gets_the_specific_reason() {
342 let (hub, rl) = parts();
343 let (a, _) = authz(WriteAccess::Deny(DenyReason::Muted));
344 let r = room(RoomState::Open);
345
346 let result = send(
347 &hub,
348 &rl,
349 &a,
350 req(&r, UserId(Uuid::new_v4()), "hi"),
351 |_| async { panic!("store must not run") },
352 )
353 .await;
354
355 assert!(matches!(
356 result,
357 Err(ChatError::Rejected(SendRejection::Denied(
358 DenyReason::Muted
359 )))
360 ));
361 }
362
363 #[tokio::test]
364 async fn the_rate_limit_bites_before_authz_is_queried() {
365 // A caller hammering a room is throttled in memory rather than putting a
366 // query behind every attempt.
367 let hub = Hub::new(HubLimits::default());
368 let rl = RateLimiter::new(RateLimits {
369 burst: 2,
370 sustain_per_min: 60,
371 });
372 let (a, calls) = authz(WriteAccess::Allow);
373 let r = room(RoomState::Open);
374 let author = UserId(Uuid::new_v4());
375 let t0 = Instant::now();
376
377 let at = |body| SendRequest {
378 room: &r,
379 author,
380 body,
381 nonce: None,
382 now: t0,
383 };
384
385 for _ in 0..2 {
386 send(&hub, &rl, &a, at("hi"), |body| async move {
387 Ok(stored(r.id, author, &body))
388 })
389 .await
390 .unwrap();
391 }
392 assert_eq!(calls.load(Ordering::SeqCst), 2);
393
394 for _ in 0..10 {
395 let result = send(&hub, &rl, &a, at("hi"), |_| async {
396 panic!("store must not run")
397 })
398 .await;
399 assert!(matches!(
400 result,
401 Err(ChatError::Rejected(SendRejection::Denied(
402 DenyReason::RateLimited { .. }
403 )))
404 ));
405 }
406 assert_eq!(
407 calls.load(Ordering::SeqCst),
408 2,
409 "throttled attempts must not reach the database"
410 );
411 }
412 }
413