openmoat_proxy/server/
capacity.rs

1//! Connections refused at accept for capacity, in the audit log.
2//!
3//! A connection over `max_connections`, or over the share of those not yet
4//! authenticated, is answered 503 before its head is read. Each is a
5//! refusal whether or not it is recorded; recording every one would let a
6//! flood fill the log, so the first refusal of a kind is recorded at once and
7//! the ones after it are counted and recorded as one row per [`INTERVAL`],
8//! when the next connection is accepted or refused after it.
9
10use std::sync::{Mutex, PoisonError};
11use std::time::{Duration, Instant};
12
13use openmoat_core::{Decision, Verdict};
14
15use crate::audit::{Connection, Recorder};
16
17/// Rule id for a connection refused at accept because the proxy, or the
18/// connections not yet authenticated, hold all the slots they may.
19pub const RULE_CAPACITY: &str = "proxy-capacity";
20
21/// The shortest time between two rows of one kind.
22pub(super) const INTERVAL: Duration = Duration::from_secs(10);
23
24/// Why a connection was refused at accept.
25#[derive(Debug, Clone, Copy, PartialEq, Eq)]
26pub(super) enum Full {
27    /// Every slot is taken.
28    Slots,
29    /// The connections not yet authenticated hold their share.
30    Pending,
31}
32
33impl Full {
34    /// What the 503 says.
35    pub(super) fn reply(self) -> &'static str {
36        match self {
37            Self::Slots => "too many connections",
38            Self::Pending => "too many connections waiting to authenticate",
39        }
40    }
41}
42
43/// Per kind: when a row was last written, and the refusals not in a row yet.
44#[derive(Debug, Default)]
45struct Kind {
46    last: Option<Instant>,
47    unrecorded: u64,
48}
49
50/// The refusals at accept of one [`Proxy::serve`](crate::Proxy::serve).
51#[derive(Debug, Default)]
52pub(super) struct Refusals {
53    kinds: Mutex<[Kind; 2]>,
54}
55
56impl Refusals {
57    /// Count one refusal, and record the count once [`INTERVAL`] has passed
58    /// since the last row of its kind.
59    pub(super) fn refused(&self, full: Full, recorder: &dyn Recorder, now: Instant) {
60        let due = {
61            let mut kinds = self.kinds.lock().unwrap_or_else(PoisonError::into_inner);
62            let kind = &mut kinds[full as usize];
63            kind.unrecorded += 1;
64            take_due(kind, now)
65        };
66        if let Some(count) = due {
67            write(full, count, recorder);
68        }
69    }
70
71    /// Record the refusals counted since the last row, where [`INTERVAL`] has
72    /// passed: called as connections are accepted, so a burst that ends is
73    /// still recorded.
74    pub(super) fn flush(&self, recorder: &dyn Recorder, now: Instant) {
75        let due: Vec<(Full, u64)> = {
76            let mut kinds = self.kinds.lock().unwrap_or_else(PoisonError::into_inner);
77            [Full::Slots, Full::Pending]
78                .into_iter()
79                .filter_map(|full| {
80                    let kind = &mut kinds[full as usize];
81                    (kind.unrecorded > 0)
82                        .then(|| take_due(kind, now))
83                        .flatten()
84                        .map(|n| (full, n))
85                })
86                .collect()
87        };
88        for (full, count) in due {
89            write(full, count, recorder);
90        }
91    }
92}
93
94/// The count to record now, if [`INTERVAL`] has passed since the last row.
95fn take_due(kind: &mut Kind, now: Instant) -> Option<u64> {
96    let due = kind
97        .last
98        .is_none_or(|last| now.saturating_duration_since(last) >= INTERVAL);
99    if !due {
100        return None;
101    }
102    kind.last = Some(now);
103    Some(std::mem::take(&mut kind.unrecorded))
104}
105
106fn write(full: Full, count: u64, recorder: &dyn Recorder) {
107    let plural = if count == 1 { "" } else { "s" };
108    let decision = Decision::single(
109        Verdict::Deny,
110        RULE_CAPACITY,
111        format!(
112            "{count} connection{plural} refused before a request was read: {}; later ones \
113             within {} s are counted in the next row",
114            full.reply(),
115            INTERVAL.as_secs()
116        ),
117    );
118    let connection = Connection {
119        session: None,
120        method: None,
121        destination: None,
122        decision: &decision,
123        latency_us: 0,
124    };
125    // The connections are refused either way; a row that cannot be written
126    // is reported where the recorder reports failures.
127    if let Err(error) = recorder.record(&connection) {
128        recorder.unrecorded(&error);
129    }
130}
131
132#[cfg(test)]
133mod tests;
openmoat_proxy/server/capacity.rs

openmoat_proxy/server/
capacity.rs
