Skip to main content

max / makenotwork

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