didbot_serve/
refusal.rs

1//! Refused OAuth and XRPC requests, as the journal says them.
2//!
3//! A refusal is one line at `info` or `warn`: the endpoint, the client, the
4//! account when the refusing code knows it, the error the caller was sent,
5//! and the reason. A token, a code or an assertion is never one of those
6//! fields.
7//!
8//! Refusals are volume a caller drives, so repeats collapse the way
9//! `didbot_pds::evaluation_log` collapses denials. The first refusal of a
10//! kind is written at once. The rest within [`WINDOW`] are counted, and the
11//! count is written as one `the same refusal repeated` line when the window
12//! closes. A kind is the endpoint, client, account, error and reason
13//! together.
14//!
15//! A caller who varies the kind on every request meets two more bounds:
16//! [`MAX_OPEN`] kinds are tracked at once, the least recent closing first,
17//! and at most [`FRESH_PER_MINUTE`] new kinds of each class are written each
18//! minute. A class is the severity, the endpoint and the error together.
19//! Past that, refusals are counted, and the count is written when the
20//! minute ends.
21//!
22//! Each class has its own budget, so a flood fills only its own. A caller
23//! with no grant can send `/oauth/par` a new `client_id` on every request,
24//! and fill the `warn` budget for `invalid_client` there. The `warn` lines
25//! for a reused refresh token or a mismatched DPoP key come from
26//! `/oauth/token` as `invalid_grant` and `invalid_dpop_proof`, which only a
27//! real grant reaches, so they are still written.
28
29use std::collections::HashMap;
30use std::num::NonZeroUsize;
31use std::sync::{LazyLock, Mutex, PoisonError};
32use std::time::{Duration, Instant};
33
34use lru::LruCache;
35
36/// How long repeats of one kind collapse into the line that opened it.
37pub(crate) const WINDOW: Duration = Duration::from_secs(5 * 60);
38
39/// How many kinds are tracked at once.
40pub(crate) const MAX_OPEN: usize = 1024;
41
42/// How many refusals of a kind not already open are written in a minute, for
43/// each class.
44pub(crate) const FRESH_PER_MINUTE: u32 = 60;
45
46/// How many classes get a budget of their own in a minute. Error names come
47/// from this server's code, so this is far more than are ever met. Past it,
48/// the rest at an endpoint share one budget.
49const MAX_CLASSES: usize = 64;
50
51/// The longest a client id, account or reason is written. Each comes from
52/// the caller in some part, and a line is not the place to store a
53/// megabyte of it.
54const MAX_FIELD_CHARS: usize = 256;
55
56const MINUTE: Duration = Duration::from_secs(60);
57
58/// How loudly a refusal is written.
59#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
60pub(crate) enum Severity {
61    /// What a working client meets in ordinary use, or a caller's mistake.
62    Info,
63    /// What the operator has to act on: a client whose metadata or
64    /// authentication no longer works, a server that is full, or a
65    /// presentation that looks like a stolen credential.
66    Warn,
67}
68
69/// One refused request, as the code that refused it describes it.
70#[derive(Debug, Clone, Copy)]
71pub(crate) struct Refusal<'a> {
72    /// `par`, `authorize`, `token`, or `xrpc`. An XRPC line's request span
73    /// names the method.
74    pub(crate) endpoint: &'static str,
75    pub(crate) client_id: Option<&'a str>,
76    pub(crate) account: Option<&'a str>,
77    /// The error name the caller was sent.
78    pub(crate) error: &'a str,
79    pub(crate) reason: &'a str,
80    pub(crate) severity: Severity,
81}
82
83/// Writes `refusal` to the journal, or counts it against a line already
84/// written.
85pub(crate) fn refused(refusal: Refusal<'_>) {
86    REFUSALS.record(&refusal, Instant::now());
87}
88
89/// Writes the count of every window that has closed, and of a minute's
90/// unwritten refusals once it has ended. The health tick calls this, so a
91/// burst that stops is still counted in the journal.
92pub(crate) fn flush() {
93    REFUSALS.flush(Instant::now());
94}
95
96static REFUSALS: LazyLock<RefusalLog> = LazyLock::new(|| RefusalLog::new(MAX_OPEN));
97
98/// What [`RefusalLog::record`] did with one refusal.
99#[derive(Debug, Clone, Copy, PartialEq, Eq)]
100pub(crate) enum Recorded {
101    /// Written as a line of its own.
102    Written,
103    /// Counted against an open window of its kind.
104    Collapsed,
105    /// Counted against this minute's unwritten refusals.
106    Unwritten,
107}
108
109#[derive(Debug, Clone, PartialEq, Eq, Hash)]
110struct Kind {
111    endpoint: &'static str,
112    client_id: Option<String>,
113    account: Option<String>,
114    error: String,
115    reason: String,
116}
117
118impl Kind {
119    fn of(refusal: &Refusal<'_>) -> Self {
120        Self {
121            endpoint: refusal.endpoint,
122            client_id: refusal.client_id.map(clipped),
123            account: refusal.account.map(clipped),
124            error: clipped(refusal.error),
125            reason: clipped(refusal.reason),
126        }
127    }
128}
129
130fn clipped(value: &str) -> String {
131    value.chars().take(MAX_FIELD_CHARS).collect()
132}
133
134#[derive(Debug, Clone, Copy)]
135struct Window {
136    severity: Severity,
137    opened: Instant,
138    repeats: u64,
139}
140
141/// What shares one budget: the severity, the endpoint and the error.
142/// `error` is `None` for the errors at an endpoint past [`MAX_CLASSES`].
143#[derive(Debug, Clone, PartialEq, Eq, Hash)]
144struct Class {
145    severity: Severity,
146    endpoint: &'static str,
147    error: Option<String>,
148}
149
150/// A line to write once the lock is released.
151enum Line {
152    Fresh(Kind, Severity),
153    Repeated(Kind, Window),
154    Unwritten(Class, u64),
155}
156
157/// One class's share of a minute.
158#[derive(Default)]
159struct Budget {
160    written: u32,
161    unwritten: u64,
162}
163
164struct State {
165    open: LruCache<Kind, Window>,
166    minute_began: Option<Instant>,
167    budgets: HashMap<Class, Budget>,
168}
169
170impl State {
171    /// Ends the current minute if it is over, carrying its unwritten count
172    /// out as a line.
173    fn roll_minute(&mut self, now: Instant, lines: &mut Vec<Line>) {
174        let over = self
175            .minute_began
176            .is_none_or(|began| now.duration_since(began) >= MINUTE);
177        if !over {
178            return;
179        }
180        for (class, budget) in self.budgets.drain() {
181            if budget.unwritten > 0 {
182                lines.push(Line::Unwritten(class, budget.unwritten));
183            }
184        }
185        self.minute_began = Some(now);
186    }
187
188    fn budget(&mut self, refusal: &Refusal<'_>, kind: &Kind) -> &mut Budget {
189        let mut class = Class {
190            severity: refusal.severity,
191            endpoint: kind.endpoint,
192            error: Some(kind.error.clone()),
193        };
194        if self.budgets.len() >= MAX_CLASSES && !self.budgets.contains_key(&class) {
195            class.error = None;
196        }
197        self.budgets.entry(class).or_default()
198    }
199}
200
201/// The collapsing state behind [`refused`]. Its own type so a test can run
202/// one against a clock it controls.
203pub(crate) struct RefusalLog {
204    state: Mutex<State>,
205}
206
207impl RefusalLog {
208    pub(crate) fn new(max_open: usize) -> Self {
209        Self {
210            state: Mutex::new(State {
211                open: LruCache::new(NonZeroUsize::new(max_open).unwrap_or(NonZeroUsize::MIN)),
212                minute_began: None,
213                budgets: HashMap::new(),
214            }),
215        }
216    }
217
218    pub(crate) fn record(&self, refusal: &Refusal<'_>, now: Instant) -> Recorded {
219        let kind = Kind::of(refusal);
220        let mut lines = Vec::new();
221        let recorded = {
222            let mut state = self.state.lock().unwrap_or_else(PoisonError::into_inner);
223            match state.open.get_mut(&kind) {
224                Some(window) if now.duration_since(window.opened) < WINDOW => {
225                    window.repeats += 1;
226                    Recorded::Collapsed
227                }
228                _ => {
229                    if let Some(window) = state.open.pop(&kind) {
230                        lines.push(Line::Repeated(kind.clone(), window));
231                    }
232                    state.roll_minute(now, &mut lines);
233                    let budget = state.budget(refusal, &kind);
234                    if budget.written >= FRESH_PER_MINUTE {
235                        budget.unwritten += 1;
236                        Recorded::Unwritten
237                    } else {
238                        budget.written += 1;
239                        lines.push(Line::Fresh(kind.clone(), refusal.severity));
240                        let window = Window {
241                            severity: refusal.severity,
242                            opened: now,
243                            repeats: 0,
244                        };
245                        if let Some((evicted, window)) = state.open.push(kind, window) {
246                            lines.push(Line::Repeated(evicted, window));
247                        }
248                        Recorded::Written
249                    }
250                }
251            }
252        };
253        write(lines);
254        recorded
255    }
256
257    pub(crate) fn flush(&self, now: Instant) {
258        let mut lines = Vec::new();
259        {
260            let mut state = self.state.lock().unwrap_or_else(PoisonError::into_inner);
261            let closed: Vec<Kind> = state
262                .open
263                .iter()
264                .filter(|(_, window)| now.duration_since(window.opened) >= WINDOW)
265                .map(|(kind, _)| kind.clone())
266                .collect();
267            for kind in closed {
268                if let Some(window) = state.open.pop(&kind) {
269                    lines.push(Line::Repeated(kind, window));
270                }
271            }
272            state.roll_minute(now, &mut lines);
273        }
274        write(lines);
275    }
276}
277
278macro_rules! at {
279    ($severity:expr, $($line:tt)+) => {
280        match $severity {
281            Severity::Warn => tracing::warn!($($line)+),
282            Severity::Info => tracing::info!($($line)+),
283        }
284    };
285}
286
287fn write(lines: Vec<Line>) {
288    for line in lines {
289        match line {
290            Line::Fresh(kind, severity) => at!(
291                severity,
292                endpoint = kind.endpoint,
293                client_id = kind.client_id.as_deref(),
294                account = kind.account.as_deref(),
295                error = kind.error,
296                reason = kind.reason,
297                "refused a request"
298            ),
299            // A window nothing repeated into says nothing the line that
300            // opened it did not.
301            Line::Repeated(_, window) if window.repeats == 0 => {}
302            Line::Repeated(kind, window) => at!(
303                window.severity,
304                endpoint = kind.endpoint,
305                client_id = kind.client_id.as_deref(),
306                account = kind.account.as_deref(),
307                error = kind.error,
308                reason = kind.reason,
309                repeats = window.repeats,
310                window_secs = WINDOW.as_secs(),
311                "the same refusal repeated"
312            ),
313            Line::Unwritten(class, count) => tracing::warn!(
314                severity = ?class.severity,
315                endpoint = class.endpoint,
316                error = class.error.as_deref(),
317                unwritten = count,
318                per_minute = FRESH_PER_MINUTE,
319                "refusals went unwritten: more new kinds than the journal takes in a minute"
320            ),
321        }
322    }
323}
324
325#[cfg(test)]
326mod tests {
327    use super::*;
328
329    fn clients(count: u32) -> Vec<&'static str> {
330        (0..count)
331            .map(|n| &*format!("https://app{n}.example/client.json").leak())
332            .collect()
333    }
334
335    fn refusal(client_id: &'static str) -> Refusal<'static> {
336        Refusal {
337            endpoint: "par",
338            client_id: Some(client_id),
339            account: None,
340            error: "invalid_request",
341            reason: "`redirect_uri` is a required parameter",
342            severity: Severity::Info,
343        }
344    }
345
346    #[test]
347    fn repeats_collapse_until_the_window_closes() {
348        let log = RefusalLog::new(MAX_OPEN);
349        let start = Instant::now();
350        let same = refusal("https://app.example/client.json");
351        assert_eq!(log.record(&same, start), Recorded::Written);
352        assert_eq!(
353            log.record(&same, start + Duration::from_secs(1)),
354            Recorded::Collapsed
355        );
356        assert_eq!(log.record(&same, start + WINDOW), Recorded::Written);
357    }
358
359    #[test]
360    fn new_kinds_past_the_minute_budget_go_unwritten_until_it_ends() {
361        let log = RefusalLog::new(MAX_OPEN);
362        let start = Instant::now();
363        let clients = clients(FRESH_PER_MINUTE + 1);
364        for client in &clients[..FRESH_PER_MINUTE as usize] {
365            assert_eq!(log.record(&refusal(client), start), Recorded::Written);
366        }
367        let last = refusal(clients[FRESH_PER_MINUTE as usize]);
368        assert_eq!(log.record(&last, start), Recorded::Unwritten);
369        assert_eq!(log.record(&last, start + MINUTE), Recorded::Written);
370    }
371
372    /// A caller who can only reach `info` refusals, sending a new kind on
373    /// every request, still leaves room for the `warn` line an operator has
374    /// to act on.
375    #[test]
376    fn a_flood_of_info_kinds_leaves_the_warn_budget_whole() {
377        let log = RefusalLog::new(MAX_OPEN);
378        let start = Instant::now();
379        let clients = clients(FRESH_PER_MINUTE * 2);
380        for client in &clients {
381            log.record(&refusal(client), start);
382        }
383        assert_eq!(
384            log.record(&refusal("https://late.example/client.json"), start),
385            Recorded::Unwritten
386        );
387        for client in &clients[..FRESH_PER_MINUTE as usize] {
388            let reused = Refusal {
389                error: "invalid_grant",
390                reason: "refresh token reused",
391                severity: Severity::Warn,
392                ..refusal(client)
393            };
394            assert_eq!(log.record(&reused, start), Recorded::Written);
395        }
396    }
397
398    /// A caller with no grant floods `/oauth/par` with a new `client_id` on
399    /// every request, each a `warn`. The `warn` lines only a real grant
400    /// reaches, at `/oauth/token`, are still written, and so is another
401    /// error at `/oauth/par`.
402    #[test]
403    fn a_flood_of_one_warn_class_leaves_every_other_whole() {
404        let log = RefusalLog::new(MAX_OPEN);
405        let start = Instant::now();
406        let unfetchable = |client_id| Refusal {
407            error: "invalid_client",
408            reason: "client metadata could not be fetched",
409            severity: Severity::Warn,
410            ..refusal(client_id)
411        };
412        for client in &clients(FRESH_PER_MINUTE * 2) {
413            log.record(&unfetchable(client), start);
414        }
415        assert_eq!(
416            log.record(&unfetchable("https://late.example/client.json"), start),
417            Recorded::Unwritten
418        );
419
420        let real = "https://real.example/client.json";
421        for (error, reason) in [
422            ("invalid_grant", "refresh token reused"),
423            (
424                "invalid_grant",
425                "the DPoP key is not the one the grant is bound to",
426            ),
427            ("invalid_dpop_proof", "the code was bound to another key"),
428        ] {
429            let stolen = Refusal {
430                endpoint: "token",
431                error,
432                reason,
433                severity: Severity::Warn,
434                ..refusal(real)
435            };
436            assert_eq!(log.record(&stolen, start), Recorded::Written, "{reason}");
437        }
438        let full = Refusal {
439            error: "access_denied",
440            reason: "too many pending sign-ins",
441            severity: Severity::Warn,
442            ..refusal(real)
443        };
444        assert_eq!(log.record(&full, start), Recorded::Written);
445    }
446
447    /// Past [`MAX_CLASSES`], the rest of an endpoint's errors share one
448    /// budget rather than each getting its own.
449    #[test]
450    fn classes_past_the_bound_share_a_budget() {
451        let log = RefusalLog::new(MAX_OPEN);
452        let start = Instant::now();
453        let errors: Vec<&'static str> = (0..MAX_CLASSES + FRESH_PER_MINUTE as usize + 1)
454            .map(|n| &*format!("error_{n}").leak())
455            .collect();
456        let client = "https://app.example/client.json";
457        for error in &errors[..MAX_CLASSES] {
458            let refused = Refusal {
459                error,
460                ..refusal(client)
461            };
462            assert_eq!(log.record(&refused, start), Recorded::Written);
463        }
464        let rest = &errors[MAX_CLASSES..];
465        for error in &rest[..FRESH_PER_MINUTE as usize] {
466            let refused = Refusal {
467                error,
468                ..refusal(client)
469            };
470            assert_eq!(log.record(&refused, start), Recorded::Written);
471        }
472        let last = Refusal {
473            error: rest[FRESH_PER_MINUTE as usize],
474            ..refusal(client)
475        };
476        assert_eq!(log.record(&last, start), Recorded::Unwritten);
477    }
478
479    #[test]
480    fn the_least_recent_kind_makes_way_past_the_bound() {
481        let log = RefusalLog::new(1);
482        let start = Instant::now();
483        let first = refusal("https://first.example/client.json");
484        let second = refusal("https://second.example/client.json");
485        assert_eq!(log.record(&first, start), Recorded::Written);
486        assert_eq!(log.record(&second, start), Recorded::Written);
487        assert_eq!(log.record(&first, start), Recorded::Written);
488    }
489}
didbot_serve/refusal.rs

didbot_serve/
refusal.rs
