Skip to main content

max / makenotwork

22.5 KB · 635 lines History Blame Raw
1 //! Alerting on security signals, so the logs are not the only place they land.
2 //!
3 //! A 5xx spike, a credential-stuffing run against login, a webhook arriving with
4 //! a bad Stripe signature: each is logged, and a log nobody reads is not a
5 //! control. This raises them where someone sees them.
6 //!
7 //! Deliberately not a monitoring product. No metrics stack, no time series, no
8 //! dashboard, no second thing to keep alive on the box. Counters live in this
9 //! process, in fixed windows, and cross a threshold at most once per window.
10 //! Delivery reuses what `POST /api/internal/alerts` already built: a row in
11 //! `admin_alerts` and mail to `ALERT_EMAIL`. That endpoint exists for external
12 //! agents (PoM, MT), so signals raised here call the same insert-and-mail path
13 //! in process rather than posting to ourselves over the loopback with a bearer
14 //! token. `ALERTS_INGEST_TOKEN` gates inbound requests from other machines and
15 //! is not a precondition for anything in this file.
16 //!
17 //! Thresholds are numbers, and every one of them is reversible: noisy, raise
18 //! it; quiet, lower it. Nothing here is structural.
19 //!
20 //! **Suppression is load-bearing.** Nothing dedups server-side --
21 //! `insert_alert` is a plain INSERT and `admin_alerts` has no unique constraint
22 //! on `dedup_key` (migration 169 says deduplication is the sending agent's job).
23 //! The once-per-window rule below is therefore the only thing standing between
24 //! one condition and a mailbox full of identical alerts, which is why
25 //! `crossing_twice_in_one_window_alerts_once` is a real test and not a
26 //! formality.
27 //!
28 //! <!-- wiki: mnw-server-overview -->
29
30 use std::collections::{HashMap, HashSet};
31 use std::sync::{Mutex, OnceLock};
32 use std::time::{Duration, Instant};
33
34 use crate::db::admin_alerts::{AlertKind, AlertSeverity, NewAlert};
35 use crate::email::EmailClient;
36
37 /// Counting window for everything except CSP. Long enough that a threshold
38 /// means a rate, short enough that the operator hears about it while it is
39 /// still happening.
40 const WINDOW: Duration = Duration::from_mins(5);
41
42 /// CSP violations are a page-level defect, not a rate: the same blocked URI
43 /// fires on every load until someone fixes the page. Once a day per distinct
44 /// URI says so without saying it four thousand times.
45 const CSP_WINDOW: Duration = Duration::from_hours(24);
46
47 /// Fraction of responses that must be 5xx before the error rate alerts.
48 const ERROR_RATE_THRESHOLD: f64 = 0.01;
49
50 /// Requests a window needs before its error rate is meaningful. Without this, a
51 /// quiet night with two requests and one error reads as a 50% error rate.
52 const ERROR_RATE_MIN_REQUESTS: u64 = 20;
53
54 /// Failed auth attempts from one address before it looks like stuffing rather
55 /// than a forgotten password.
56 const AUTH_FAILURES_PER_IP: u64 = 20;
57
58 /// Failed auth attempts across all addresses. Catches the distributed version,
59 /// which the per-IP threshold is blind to by construction.
60 const AUTH_FAILURES_SITE_WIDE: u64 = 100;
61
62 /// Rate-limit trips before the volume is worth hearing about. An individual
63 /// trip is the system working exactly as designed and must never alert.
64 const RATE_LIMIT_TRIPS: u64 = 50;
65
66 /// Everything needed to raise an alert, captured once at startup.
67 ///
68 /// A global rather than state threaded through every call site: the CSP report
69 /// handler takes a request body and nothing else, and the response middleware
70 /// runs outside the state extractors. Uninitialised (every unit test, and any
71 /// build that never calls [`install`]) means counting still happens and nothing
72 /// is sent, which is the correct degraded behaviour for a best-effort channel.
73 struct Sink {
74 db: sqlx::PgPool,
75 email: EmailClient,
76 }
77
78 static SINK: OnceLock<Sink> = OnceLock::new();
79
80 /// Point the signal counters at a database and a mailer. Call once at startup.
81 /// A second call is ignored rather than treated as an error, so the per-test
82 /// `build_app` does not have to care.
83 pub fn install(db: sqlx::PgPool, email: EmailClient) {
84 let _ = SINK.set(Sink { db, email });
85 }
86
87 /// A condition that has crossed its threshold and is worth one alert.
88 #[derive(Debug, Clone)]
89 struct Firing {
90 severity: AlertSeverity,
91 dedup_key: String,
92 title: String,
93 body: String,
94 }
95
96 /// Fixed-window counters. One mutex: every field is touched together on the
97 /// response path and the critical section is a handful of integer adds, so
98 /// splitting it would buy contention rather than remove it.
99 #[derive(Default)]
100 struct Counters {
101 window_started: Option<Instant>,
102 requests: u64,
103 server_errors: u64,
104 auth_failures: u64,
105 auth_failures_by_ip: HashMap<String, u64>,
106 rate_limit_trips: u64,
107 /// Conditions already alerted on in this window. Cleared with the window.
108 fired: HashSet<String>,
109
110 csp_window_started: Option<Instant>,
111 /// Blocked URIs already alerted on today. Bounded by [`CSP_URI_CAP`] so a
112 /// page generating unique blocked URIs cannot grow this without limit.
113 csp_fired: HashSet<String>,
114 }
115
116 /// Distinct blocked URIs tracked per CSP window. A violation carrying a
117 /// cache-busted or otherwise unique URI each time would otherwise make this set
118 /// a slow memory leak; past the cap, further distinct URIs are counted as
119 /// already-alerted rather than remembered.
120 const CSP_URI_CAP: usize = 256;
121
122 static COUNTERS: Mutex<Option<Counters>> = Mutex::new(None);
123
124 /// Run `f` against the counters, rolling the window first if it has expired.
125 ///
126 /// A poisoned mutex is not propagated. Every caller is on a request path that
127 /// has real work to do, and failing a checkout because an alert counter panicked
128 /// in another thread would make this module the outage it exists to report.
129 fn with_counters<T>(now: Instant, f: impl FnOnce(&mut Counters) -> T) -> Option<T> {
130 let mut guard = COUNTERS.lock().ok()?;
131 let counters = guard.get_or_insert_with(Counters::default);
132
133 match counters.window_started {
134 Some(started) if now.duration_since(started) < WINDOW => {}
135 _ => {
136 counters.window_started = Some(now);
137 counters.requests = 0;
138 counters.server_errors = 0;
139 counters.auth_failures = 0;
140 counters.auth_failures_by_ip.clear();
141 counters.rate_limit_trips = 0;
142 counters.fired.clear();
143 }
144 }
145
146 match counters.csp_window_started {
147 Some(started) if now.duration_since(started) < CSP_WINDOW => {}
148 _ => {
149 counters.csp_window_started = Some(now);
150 counters.csp_fired.clear();
151 }
152 }
153
154 Some(f(counters))
155 }
156
157 impl Counters {
158 /// Record that `key` has fired, returning false if it already had this
159 /// window. The whole suppression rule, in one place.
160 fn claim(&mut self, key: &str) -> bool {
161 self.fired.insert(key.to_string())
162 }
163 }
164
165 /// Observe one finished response.
166 ///
167 /// Called from the outermost middleware, so it sees every request the server
168 /// answered, including the ones rejected by a layer before any handler ran.
169 /// `ip` is the Cloudflare-derived client address where one is available.
170 pub fn note_response(status: u16, ip: Option<&str>) {
171 note_response_at(status, ip, Instant::now());
172 }
173
174 fn note_response_at(status: u16, ip: Option<&str>, now: Instant) {
175 let firing = with_counters(now, |c| {
176 c.requests += 1;
177
178 if status >= 500 {
179 c.server_errors += 1;
180 let rate = c.server_errors as f64 / c.requests as f64;
181 if c.requests >= ERROR_RATE_MIN_REQUESTS
182 && rate > ERROR_RATE_THRESHOLD
183 && c.claim("sec:5xx")
184 {
185 return Some(Firing {
186 severity: AlertSeverity::Critical,
187 dedup_key: "sec:5xx".to_string(),
188 title: "Server error rate is elevated".to_string(),
189 body: format!(
190 "{} of {} responses in the last {} minutes were 5xx ({:.1}%). \
191 Threshold is {:.0}% over at least {} requests.",
192 c.server_errors,
193 c.requests,
194 WINDOW.as_secs() / 60,
195 rate * 100.0,
196 ERROR_RATE_THRESHOLD * 100.0,
197 ERROR_RATE_MIN_REQUESTS,
198 ),
199 });
200 }
201 }
202
203 if status == 429 {
204 c.rate_limit_trips += 1;
205 if c.rate_limit_trips > RATE_LIMIT_TRIPS && c.claim("sec:ratelimit") {
206 return Some(Firing {
207 severity: AlertSeverity::Warning,
208 dedup_key: "sec:ratelimit".to_string(),
209 title: "Rate limiting is tripping in volume".to_string(),
210 body: format!(
211 "{} requests were rate limited in the last {} minutes (threshold {}). \
212 Individual trips are the limiter working; this many suggests a scraper, \
213 a stuck client, or a limit set too tight.",
214 c.rate_limit_trips,
215 WINDOW.as_secs() / 60,
216 RATE_LIMIT_TRIPS,
217 ),
218 });
219 }
220 }
221
222 let _ = ip;
223 None
224 })
225 .flatten();
226
227 dispatch(firing);
228 }
229
230 /// Record one failed authentication attempt.
231 ///
232 /// Called at the failure sites rather than inferred from status codes: a wrong
233 /// password re-renders the login form with a 200, so nothing about the response
234 /// says an attempt failed.
235 pub fn note_auth_failure(ip: Option<&str>) {
236 note_auth_failure_at(ip, Instant::now());
237 }
238
239 fn note_auth_failure_at(ip: Option<&str>, now: Instant) {
240 let firing = with_counters(now, |c| {
241 c.auth_failures += 1;
242
243 if let Some(ip) = ip {
244 // One entry per address per window, and the window clears it. A
245 // spray from many addresses is bounded by the same window rather
246 // than by a cap, which is what the site-wide counter is for.
247 let per_ip = c.auth_failures_by_ip.entry(ip.to_string()).or_insert(0);
248 *per_ip += 1;
249 let count = *per_ip;
250 let key = format!("sec:authfail:{ip}");
251 if count > AUTH_FAILURES_PER_IP && c.claim(&key) {
252 return Some(Firing {
253 severity: AlertSeverity::Critical,
254 dedup_key: key,
255 title: format!("Repeated auth failures from {ip}"),
256 body: format!(
257 "{count} failed authentication attempts from {ip} in the last {} minutes \
258 (threshold {AUTH_FAILURES_PER_IP}). Looks like credential stuffing rather \
259 than a forgotten password.",
260 WINDOW.as_secs() / 60,
261 ),
262 });
263 }
264 }
265
266 if c.auth_failures > AUTH_FAILURES_SITE_WIDE && c.claim("sec:authfail") {
267 return Some(Firing {
268 severity: AlertSeverity::Critical,
269 dedup_key: "sec:authfail".to_string(),
270 title: "Auth failures are elevated site-wide".to_string(),
271 body: format!(
272 "{} failed authentication attempts across all addresses in the last {} minutes \
273 (threshold {AUTH_FAILURES_SITE_WIDE}). A distributed attempt would look like \
274 this and would not trip the per-address threshold.",
275 c.auth_failures,
276 WINDOW.as_secs() / 60,
277 ),
278 });
279 }
280
281 None
282 })
283 .flatten();
284
285 dispatch(firing);
286 }
287
288 /// Record a webhook that arrived with a signature that did not verify.
289 ///
290 /// No threshold: there is no benign cause. Stripe is the caller and it signs
291 /// correctly, so one of these means either a misconfigured secret or someone
292 /// posting forged events at the billing path.
293 pub fn note_webhook_signature_failure(provider: &str) {
294 note_webhook_signature_failure_at(provider, Instant::now());
295 }
296
297 fn note_webhook_signature_failure_at(provider: &str, now: Instant) {
298 let provider = provider.to_string();
299 let firing = with_counters(now, |c| {
300 let key = format!("sec:webhook:{provider}");
301 if !c.claim(&key) {
302 return None;
303 }
304 Some(Firing {
305 severity: AlertSeverity::Critical,
306 dedup_key: key,
307 title: format!("{provider} webhook signature failed to verify"),
308 body: format!(
309 "A request to the {provider} webhook path carried a signature that did not \
310 verify. There is no benign cause: either the signing secret is wrong (in which \
311 case real events are being dropped) or someone is posting forged events."
312 ),
313 })
314 })
315 .flatten();
316
317 dispatch(firing);
318 }
319
320 /// Record a CSP violation report, alerting once per distinct blocked URI per
321 /// day.
322 pub fn note_csp_violation(blocked_uri: &str, directive: &str, document: &str) {
323 note_csp_violation_at(blocked_uri, directive, document, Instant::now());
324 }
325
326 fn note_csp_violation_at(blocked_uri: &str, directive: &str, document: &str, now: Instant) {
327 let firing = with_counters(now, |c| {
328 if c.csp_fired.len() >= CSP_URI_CAP || !c.csp_fired.insert(blocked_uri.to_string()) {
329 return None;
330 }
331 Some(Firing {
332 severity: AlertSeverity::Warning,
333 dedup_key: format!("sec:csp:{blocked_uri}"),
334 title: format!("CSP violation: {directive}"),
335 body: format!(
336 "A page reported a Content-Security-Policy violation.\n\n\
337 directive: {directive}\nblocked: {blocked_uri}\ndocument: {document}\n\n\
338 Either something on the page broke, or someone is probing. Reported once per \
339 blocked URI per day."
340 ),
341 })
342 })
343 .flatten();
344
345 dispatch(firing);
346 }
347
348 /// Persist and mail a firing, off the request path.
349 ///
350 /// Best effort in the strict sense: a failure here logs and goes no further. It
351 /// never blocks the request, never returns an error to a caller, and never
352 /// panics. An alerting channel that can take the site down is worse than no
353 /// alerting channel.
354 fn dispatch(firing: Option<Firing>) {
355 let Some(firing) = firing else { return };
356 let Some(sink) = SINK.get() else {
357 // No sink: unit tests and any build that skipped `install`. The
358 // threshold logic still ran, which is what those tests assert.
359 tracing::debug!(
360 dedup_key = %firing.dedup_key,
361 "security signal fired before the alert sink was installed"
362 );
363 return;
364 };
365
366 let db = sink.db.clone();
367 let email = sink.email.clone();
368 tokio::spawn(async move {
369 tracing::warn!(
370 target: "security_signal",
371 dedup_key = %firing.dedup_key,
372 severity = firing.severity.as_str(),
373 "{}",
374 firing.title
375 );
376
377 let id = match crate::db::admin_alerts::insert_alert(
378 &db,
379 &NewAlert {
380 source: "mnw",
381 kind: AlertKind::Security,
382 severity: firing.severity,
383 title: &firing.title,
384 body: &firing.body,
385 dedup_key: Some(&firing.dedup_key),
386 details: None,
387 },
388 )
389 .await
390 {
391 Ok(id) => id,
392 Err(e) => {
393 tracing::error!(error = ?e, "failed to persist security alert");
394 return;
395 }
396 };
397
398 crate::routes::api::internal::alerts::email_alert(
399 &db,
400 &email,
401 id,
402 "mnw",
403 AlertKind::Security,
404 firing.severity,
405 &firing.title,
406 &firing.body,
407 )
408 .await;
409 });
410 }
411
412 #[cfg(test)]
413 mod tests {
414 use super::*;
415
416 /// Every test drives the counters through the `_at` variants with an
417 /// explicit clock, and they share one global. Serialise them rather than
418 /// letting a stray count from a parallel test move a threshold.
419 fn lock() -> std::sync::MutexGuard<'static, ()> {
420 static SERIAL: Mutex<()> = Mutex::new(());
421 SERIAL
422 .lock()
423 .unwrap_or_else(std::sync::PoisonError::into_inner)
424 }
425
426 /// Force a fresh window. The counters are global, so a test that assumes an
427 /// empty window has to say so.
428 fn reset() {
429 if let Ok(mut guard) = COUNTERS.lock() {
430 *guard = None;
431 }
432 }
433
434 /// What `fired` holds, which is the observable form of "an alert was sent"
435 /// without a database behind it.
436 fn fired() -> HashSet<String> {
437 COUNTERS
438 .lock()
439 .unwrap()
440 .as_ref()
441 .map(|c| c.fired.clone())
442 .unwrap_or_default()
443 }
444
445 #[test]
446 fn error_rate_needs_volume_before_it_alerts() {
447 let _g = lock();
448 reset();
449 let t = Instant::now();
450
451 // Two requests, one of them a 500. That is a 50% error rate and it must
452 // not alert: below the minimum request count, the ratio is noise.
453 note_response_at(500, None, t);
454 note_response_at(200, None, t);
455 assert!(!fired().contains("sec:5xx"), "alerted on two requests");
456 }
457
458 #[test]
459 fn error_rate_alerts_once_past_the_threshold() {
460 let _g = lock();
461 reset();
462 let t = Instant::now();
463
464 for _ in 0..ERROR_RATE_MIN_REQUESTS {
465 note_response_at(200, None, t);
466 }
467 assert!(!fired().contains("sec:5xx"), "clean traffic alerted");
468
469 note_response_at(500, None, t);
470 assert!(fired().contains("sec:5xx"), "1 in 21 is over 1%");
471 }
472
473 #[test]
474 fn crossing_twice_in_one_window_alerts_once() {
475 let _g = lock();
476 reset();
477 let t = Instant::now();
478
479 for _ in 0..ERROR_RATE_MIN_REQUESTS {
480 note_response_at(200, None, t);
481 }
482 note_response_at(500, None, t);
483 assert!(fired().contains("sec:5xx"));
484
485 // Nothing dedups server-side, so this is the only suppression there is.
486 // Staying over the threshold must not produce a second alert.
487 let before = fired().len();
488 for _ in 0..50 {
489 note_response_at(500, None, t);
490 }
491 assert_eq!(fired().len(), before, "a sustained condition realerted");
492 }
493
494 #[test]
495 fn the_window_resets_and_can_alert_again() {
496 let _g = lock();
497 reset();
498 let t = Instant::now();
499
500 for _ in 0..ERROR_RATE_MIN_REQUESTS {
501 note_response_at(200, None, t);
502 }
503 note_response_at(500, None, t);
504 assert!(fired().contains("sec:5xx"));
505
506 let later = t + WINDOW + Duration::from_secs(1);
507 for _ in 0..ERROR_RATE_MIN_REQUESTS {
508 note_response_at(200, None, later);
509 }
510 assert!(!fired().contains("sec:5xx"), "counters did not roll");
511
512 note_response_at(500, None, later);
513 assert!(fired().contains("sec:5xx"), "a new window cannot alert");
514 }
515
516 #[test]
517 fn individual_rate_limit_trips_never_alert() {
518 let _g = lock();
519 reset();
520 let t = Instant::now();
521
522 for _ in 0..RATE_LIMIT_TRIPS {
523 note_response_at(429, None, t);
524 }
525 assert!(
526 !fired().contains("sec:ratelimit"),
527 "the limiter working is not an incident"
528 );
529
530 note_response_at(429, None, t);
531 assert!(fired().contains("sec:ratelimit"));
532 }
533
534 #[test]
535 fn auth_failures_alert_per_address() {
536 let _g = lock();
537 reset();
538 let t = Instant::now();
539
540 for _ in 0..AUTH_FAILURES_PER_IP {
541 note_auth_failure_at(Some("203.0.113.7"), t);
542 }
543 assert!(!fired().contains("sec:authfail:203.0.113.7"));
544
545 note_auth_failure_at(Some("203.0.113.7"), t);
546 assert!(fired().contains("sec:authfail:203.0.113.7"));
547
548 // A different address is its own condition and its own alert.
549 assert!(!fired().contains("sec:authfail:203.0.113.8"));
550 }
551
552 #[test]
553 fn auth_failures_alert_site_wide_when_spread_thin() {
554 let _g = lock();
555 reset();
556 let t = Instant::now();
557
558 // One attempt per address, so no per-address threshold is ever crossed.
559 // This is the distributed case the per-IP counter cannot see.
560 for i in 0..=AUTH_FAILURES_SITE_WIDE {
561 note_auth_failure_at(Some(&format!("198.51.100.{i}")), t);
562 }
563 assert!(
564 fired().iter().all(|k| k != "sec:authfail:198.51.100.1"),
565 "no single address should have tripped"
566 );
567 assert!(fired().contains("sec:authfail"));
568 }
569
570 #[test]
571 fn one_webhook_signature_failure_is_enough() {
572 let _g = lock();
573 reset();
574 let t = Instant::now();
575
576 note_webhook_signature_failure_at("stripe", t);
577 assert!(fired().contains("sec:webhook:stripe"));
578 }
579
580 #[test]
581 fn csp_alerts_once_per_blocked_uri() {
582 let _g = lock();
583 reset();
584 let t = Instant::now();
585
586 note_csp_violation_at("https://evil.example/x.js", "script-src", "/", t);
587 note_csp_violation_at("https://evil.example/x.js", "script-src", "/", t);
588 note_csp_violation_at("https://other.example/y.js", "script-src", "/", t);
589
590 let uris = COUNTERS.lock().unwrap().as_ref().unwrap().csp_fired.clone();
591 assert_eq!(
592 uris.len(),
593 2,
594 "one entry per distinct blocked URI: {uris:?}"
595 );
596 }
597
598 #[test]
599 fn csp_uri_tracking_is_bounded() {
600 let _g = lock();
601 reset();
602 let t = Instant::now();
603
604 // A page minting a unique blocked URI per load must not grow the set
605 // for a whole day.
606 for i in 0..(CSP_URI_CAP * 2) {
607 note_csp_violation_at(
608 &format!("https://evil.example/{i}.js"),
609 "script-src",
610 "/",
611 t,
612 );
613 }
614 let uris = COUNTERS.lock().unwrap().as_ref().unwrap().csp_fired.len();
615 assert_eq!(uris, CSP_URI_CAP);
616 }
617
618 #[test]
619 fn csp_survives_the_short_window_rolling() {
620 let _g = lock();
621 reset();
622 let t = Instant::now();
623
624 note_csp_violation_at("https://evil.example/x.js", "script-src", "/", t);
625
626 // A CSP entry is kept for a day, not for five minutes. Rolling the
627 // short window must not re-arm it.
628 let later = t + WINDOW + Duration::from_secs(1);
629 note_csp_violation_at("https://evil.example/x.js", "script-src", "/", later);
630
631 let uris = COUNTERS.lock().unwrap().as_ref().unwrap().csp_fired.len();
632 assert_eq!(uris, 1, "the daily CSP entry was cleared by the 5m window");
633 }
634 }
635