crates/admit/src/visitors.rs
//! **How many requests, and how many distinct consumers — without keeping an
//! address.**
//!
//! The flow drawing (holger's `flow`) prints, on every door and every repository,
//! how many requests it answered in the current window and how many DISTINCT
//! clients sent them. This module is where those two numbers are kept.
//!
//! ## What `/metrics` already had, and why it is not reused for this
//!
//! holger's `metrics` is a set of process-lifetime `AtomicU64` counters: one per
//! verb, server-wide, since the process started. There is no window, no per-door
//! or per-repository split, and no notion of "distinct". So the window here is
//! new, and it is stated on the picture rather than implied.
//!
//! ## The window: one UTC day, tumbling
//!
//! `[00:00, 24:00)` UTC. Tumbling rather than rolling, and that is forced by the
//! privacy rule below, not chosen for convenience: a distinct count is only
//! right if the SAME client maps to the SAME token for the whole window, so the
//! key may only rotate at a window boundary. A rolling 24 h window would need a
//! key that outlives 24 h — i.e. a key that never rotates.
//!
//! The day that just closed is kept as final numbers only (its key is gone), so
//! the picture right after midnight still says what yesterday looked like. A
//! restart mid-day starts a fresh window and the picture says "since HH:MM",
//! because a count that silently lost the morning would read as a quiet morning.
//!
//! ## The privacy rule: a keyed hash that dies with its window
//!
//! No address is stored. At the start of every window a fresh 32-byte key is
//! drawn from the OS RNG; a client is reduced to
//! `blake3::keyed_hash(key, canonical address bytes)[..16]` and only those 16
//! bytes go into the per-scope set. At the window boundary the sets are cleared,
//! the key is overwritten, and a new one is drawn.
//!
//! * **Right count**: within a window one client is one token, so a client that
//! sends a thousand requests is one consumer. 128 bits makes a collision
//! between two real clients negligible.
//! * **Nobody identifiable after the window**: the key never leaves this
//! process's memory and is destroyed at rollover. A token from yesterday cannot
//! be tested against a candidate address by anyone, because the one input that
//! would make that possible no longer exists.
//! * **The honest limit**: DURING the window, someone who can read this
//! process's memory holds the key and could test candidate addresses one by
//! one (the IPv4 space is small). That attacker can also read the socket table
//! directly, so the tally adds nothing to what they already have.
//!
//! This is the same construction as a salted daily hash in privacy-preserving
//! web analytics; nothing here is persisted, so nothing survives a restart.
//!
//! ## Who is a client
//!
//! The transport peer — unless the peer is loopback, which on holger.rs is the
//! reverse proxy (Caddy) standing in front of the HTTP door. Then the client is
//! the RIGHT-MOST `X-Forwarded-For` entry, the one address the proxy itself
//! wrote. A loopback peer with no `X-Forwarded-For` is the box talking to
//! itself — a health probe, the console's own gRPC calls — and it is not
//! counted at all, the same rule holger's `flow::is_loopback_ident` applies to
//! the puller band.
//!
//! ## Bounded
//!
//! Scopes are registered at boot from the configuration (every door, every
//! repository); a name nobody configured is ignored, so a caller cannot grow the
//! table. Each scope's set holds at most [`MAX_DISTINCT`] tokens — past that the
//! count is reported as `≥ N`, never as a number that stopped. Sets are
//! allocated once and `clear()`ed at rollover, which keeps their capacity: the
//! steady state allocates nothing.
use std::collections::{BTreeMap, HashMap, HashSet};
use std::net::IpAddr;
use std::sync::Mutex;
/// The window length: one UTC day.
pub const WINDOW_SECS: u64 = 86_400;
/// Most distinct tokens one scope holds in one window. 16 bytes each, so a full
/// scope is ~1.6 MB plus the table's own overhead.
pub const MAX_DISTINCT: usize = 100_000;
/// Initial set capacity per scope, allocated once at boot.
const PREALLOC: usize = 256;
/// What a client is reduced to.
type Token = [u8; 16];
/// One window's numbers for one door or one repository.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct Count {
/// Requests answered, whatever their status.
pub requests: u64,
/// Distinct clients that sent them.
pub consumers: u64,
/// `true` when the distinct set hit [`MAX_DISTINCT`]: `consumers` is then a
/// floor, and the picture prints it as `≥ N`.
pub saturated: bool,
}
/// A client address, borrowed — never copied into anything that is kept.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Client<'a> {
/// A parsed address. IPv4-mapped IPv6 is folded to IPv4, so one host is one
/// client whichever stack it arrived on.
Ip(IpAddr),
/// An `X-Forwarded-For` entry that does not parse as an address (a proxy
/// that writes `unknown`, an obfuscated identifier). Hashed as written.
Raw(&'a str),
}
impl<'a> Client<'a> {
fn from_text(s: &'a str) -> Client<'a> {
let t = s.trim().trim_start_matches('[').trim_end_matches(']');
match t.parse::<IpAddr>() {
Ok(ip) => Client::Ip(canonical(ip)),
Err(_) => Client::Raw(s.trim()),
}
}
}
fn canonical(ip: IpAddr) -> IpAddr {
match ip {
IpAddr::V6(v6) => match v6.to_ipv4_mapped() {
Some(v4) => IpAddr::V4(v4),
None => IpAddr::V6(v6),
},
v4 => v4,
}
}
fn is_loopback(ip: IpAddr) -> bool {
canonical(ip).is_loopback()
}
/// Strip the `:port` off a `host:port` peer string (`1.2.3.4:5`, `[::1]:5`).
fn peer_ip(peer: &str) -> Option<IpAddr> {
if let Ok(sa) = peer.parse::<std::net::SocketAddr>() {
return Some(sa.ip());
}
peer.parse::<IpAddr>().ok()
}
/// **Who sent this request**, or `None` when it is the box itself.
///
/// `peer` is the transport peer as the listener saw it (`ip:port`);
/// `forwarded_for` is every `X-Forwarded-For` line joined in order. A
/// forwarded-for header is believed ONLY from a loopback peer — the reverse
/// proxy on the same box — because from anywhere else it is a value the caller
/// chose, and one header per request would make every request a new consumer.
pub fn client_of<'a>(peer: &'a str, forwarded_for: Option<&'a str>) -> Option<Client<'a>> {
client_of_ip(peer_ip(peer), forwarded_for)
}
/// [`client_of`] for a transport that hands over the peer as an address
/// already (gRPC's `remote_addr`), so nothing is formatted to be parsed back.
pub fn client_of_ip(ip: Option<IpAddr>, forwarded_for: Option<&str>) -> Option<Client<'_>> {
match ip {
Some(ip) if is_loopback(ip) => {
let last = forwarded_for?
.rsplit(',')
.map(str::trim)
.find(|s| !s.is_empty())?;
let c = Client::from_text(last);
match c {
// A proxy forwarding the box's own loopback call is still the box.
Client::Ip(ip) if is_loopback(ip) => None,
other => Some(other),
}
}
Some(ip) => Some(Client::Ip(canonical(ip))),
// The transport exposed nothing parseable. Not a client we can count.
None => None,
}
}
/// One scope's live tally.
struct Tally {
requests: u64,
clients: HashSet<Token>,
saturated: bool,
}
impl Tally {
fn new() -> Tally {
Tally { requests: 0, clients: HashSet::with_capacity(PREALLOC), saturated: false }
}
fn add(&mut self, t: Token) {
self.requests += 1;
if self.clients.len() < MAX_DISTINCT {
self.clients.insert(t);
} else if !self.clients.contains(&t) {
self.saturated = true;
}
}
fn count(&self) -> Count {
Count {
requests: self.requests,
consumers: self.clients.len() as u64,
saturated: self.saturated,
}
}
/// Back to zero, KEEPING the allocation.
fn reset(&mut self) {
self.requests = 0;
self.clients.clear();
self.saturated = false;
}
}
/// A closed window: final numbers, no tokens, no key.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct Closed {
/// Which UTC day (`unix / WINDOW_SECS`).
pub day: u64,
pub doors: BTreeMap<String, Count>,
pub repos: BTreeMap<String, Count>,
}
/// What the picture reads.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct Snapshot {
/// The current UTC day (`unix / WINDOW_SECS`).
pub day: u64,
/// When counting in this window began: the day's start, or a later restart.
pub counting_since_unix: u64,
pub doors: BTreeMap<String, Count>,
pub repos: BTreeMap<String, Count>,
/// The day before, as final numbers. `None` when this process was not
/// running at the last midnight.
pub previous: Option<Closed>,
}
struct Inner {
day: u64,
counting_since_unix: u64,
key: [u8; 32],
doors: HashMap<String, Tally>,
repos: HashMap<String, Tally>,
previous: Option<Closed>,
}
/// The per-door, per-repository tally. One per server, shared by every door.
pub struct Visitors {
inner: Mutex<Inner>,
}
fn fresh_key() -> [u8; 32] {
let mut k = [0u8; 32];
if getrandom::getrandom(&mut k).is_err() {
// No OS RNG: derive from the clock and this allocation's address. Weaker
// against an attacker who can guess both, but the key never leaves the
// process and dies at the window's end either way.
let mut h = blake3::Hasher::new();
h.update(&now_nanos().to_le_bytes());
h.update(&(std::ptr::addr_of!(k) as usize).to_le_bytes());
h.update(&std::process::id().to_le_bytes());
k = *h.finalize().as_bytes();
}
k
}
fn token(key: &[u8; 32], client: Client<'_>) -> Token {
let digest = match client {
Client::Ip(IpAddr::V4(v4)) => blake3::keyed_hash(key, &v4.octets()),
Client::Ip(IpAddr::V6(v6)) => blake3::keyed_hash(key, &v6.octets()),
Client::Raw(s) => blake3::keyed_hash(key, s.as_bytes()),
};
let mut t = [0u8; 16];
t.copy_from_slice(&digest.as_bytes()[..16]);
t
}
/// Nanoseconds since the epoch: the fallback key's clock input when the OS RNG
/// is absent (in holger this was `crate::audit::now_nanos`).
fn now_nanos() -> u128 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_nanos())
.unwrap_or(0)
}
/// Seconds since the epoch.
pub fn now_unix() -> u64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0)
}
impl Visitors {
/// A tally for exactly these doors and repositories, starting at `now_unix`.
pub fn new<'a>(
doors: impl IntoIterator<Item = &'a str>,
repos: impl IntoIterator<Item = &'a str>,
now_unix: u64,
) -> Visitors {
Visitors {
inner: Mutex::new(Inner {
day: now_unix / WINDOW_SECS,
counting_since_unix: now_unix,
key: fresh_key(),
doors: doors.into_iter().map(|d| (d.to_owned(), Tally::new())).collect(),
repos: repos.into_iter().map(|r| (r.to_owned(), Tally::new())).collect(),
previous: None,
}),
}
}
/// A tally that counts nothing — no scope is registered.
pub fn empty() -> Visitors {
Visitors::new([], [], now_unix())
}
/// **Count one request.** `door` and `repo` are the configured names it was
/// answered through; either may be absent. `None` client ⇒ nothing counted.
pub fn record(&self, now_unix: u64, door: Option<&str>, repo: Option<&str>, client: Option<Client<'_>>) {
let Some(client) = client else { return };
let Ok(mut g) = self.inner.lock() else { return };
g.roll(now_unix);
let t = token(&g.key, client);
if let Some(d) = door {
if let Some(tally) = g.doors.get_mut(d) {
tally.add(t);
}
}
if let Some(r) = repo {
if let Some(tally) = g.repos.get_mut(r) {
tally.add(t);
}
}
}
/// The numbers, as of `now_unix`.
pub fn snapshot(&self, now_unix: u64) -> Snapshot {
let Ok(mut g) = self.inner.lock() else { return Snapshot::default() };
g.roll(now_unix);
Snapshot {
day: g.day,
counting_since_unix: g.counting_since_unix,
doors: g.doors.iter().map(|(k, v)| (k.clone(), v.count())).collect(),
repos: g.repos.iter().map(|(k, v)| (k.clone(), v.count())).collect(),
previous: g.previous.clone(),
}
}
/// **The name a puller is DRAWN under**: `client ` + the first 8 hex digits
/// of the SAME keyed-hash token [`Visitors::record`] counts with — the one
/// key, rotated at the window boundary. Within a UTC day one client is one
/// label (so the rows can be told apart and followed); on another day it is
/// a different label, which is the point: yesterday's picture cannot be
/// joined to today's. `ident` is an address as the audit log wrote it
/// (`1.2.3.4`, `[::1]`, or anything else, hashed as written).
pub fn puller_label(&self, ident: &str, now_unix: u64) -> String {
let Ok(mut g) = self.inner.lock() else {
return "client ?".to_string();
};
g.roll(now_unix);
let t = token(&g.key, Client::from_text(ident));
format!("client {:02x}{:02x}{:02x}{:02x}", t[0], t[1], t[2], t[3])
}
/// Test seam: is any byte sequence equal to `needle` held in this tally's
/// state? Used to prove no address is kept.
#[doc(hidden)]
pub fn holds_bytes(&self, needle: &[u8]) -> bool {
let Ok(g) = self.inner.lock() else { return false };
let hay = |b: &[u8]| b.windows(needle.len()).any(|w| w == needle);
g.doors
.values()
.chain(g.repos.values())
.flat_map(|t| t.clients.iter())
.any(|tok| hay(tok))
|| g.doors.keys().chain(g.repos.keys()).any(|k| hay(k.as_bytes()))
}
}
impl Inner {
/// Close the window if `now_unix` is past it.
fn roll(&mut self, now_unix: u64) {
let day = now_unix / WINDOW_SECS;
if day <= self.day {
// Same window — or the clock went backwards, in which case the
// current window keeps counting rather than reopening a closed one.
return;
}
let consecutive = day == self.day + 1;
let closed = Closed {
day: day - 1,
doors: self
.doors
.iter()
.map(|(k, v)| (k.clone(), if consecutive { v.count() } else { Count::default() }))
.collect(),
repos: self
.repos
.iter()
.map(|(k, v)| (k.clone(), if consecutive { v.count() } else { Count::default() }))
.collect(),
};
self.previous = Some(closed);
for t in self.doors.values_mut().chain(self.repos.values_mut()) {
t.reset();
}
// The old key is destroyed before the new one exists: from here on no
// token from the closed window can be recomputed by anyone.
self.key.fill(0);
self.key = fresh_key();
self.day = day;
self.counting_since_unix = day * WINDOW_SECS;
}
}
#[cfg(test)]
mod tests {
use super::*;
const DAY0: u64 = 20_000 * WINDOW_SECS; // some UTC midnight
fn v() -> Visitors {
Visitors::new(["doors"], ["crates", "oci"], DAY0 + 10)
}
/// ★ One client sending many requests is ONE consumer; two clients are two.
#[test]
fn one_client_is_one_consumer_however_often_it_asks() {
let v = v();
for _ in 0..50 {
v.record(DAY0 + 20, Some("doors"), Some("crates"), client_of("203.0.113.7:4000", None));
}
// Same host, different source port: still the same client.
v.record(DAY0 + 21, Some("doors"), Some("crates"), client_of("203.0.113.7:4999", None));
// …and the same host arriving IPv4-mapped over an IPv6 socket.
v.record(DAY0 + 22, Some("doors"), Some("crates"), client_of("[::ffff:203.0.113.7]:1", None));
v.record(DAY0 + 23, Some("doors"), Some("oci"), client_of("198.51.100.9:1", None));
let s = v.snapshot(DAY0 + 30);
assert_eq!(s.repos["crates"], Count { requests: 52, consumers: 1, saturated: false });
assert_eq!(s.repos["oci"], Count { requests: 1, consumers: 1, saturated: false });
assert_eq!(s.doors["doors"], Count { requests: 53, consumers: 2, saturated: false });
}
/// ★ Behind the proxy, the client is the forwarded address, and the proxy's
/// loopback peer is not a client of its own.
#[test]
fn behind_the_proxy_the_forwarded_address_is_the_client() {
assert_eq!(
client_of("127.0.0.1:5555", Some("198.51.100.1, 203.0.113.7")),
Some(Client::Ip("203.0.113.7".parse().unwrap())),
"the right-most entry is the one the proxy wrote"
);
assert_eq!(client_of("127.0.0.1:5555", None), None, "loopback with no header is the box itself");
assert_eq!(client_of("127.0.0.1:5555", Some("127.0.0.1")), None);
// From a non-loopback peer the header is the caller's own word, ignored.
assert_eq!(
client_of("203.0.113.7:1", Some("1.1.1.1")),
Some(Client::Ip("203.0.113.7".parse().unwrap()))
);
}
/// ★ **No address is held.** Neither the text nor the octets of any client
/// address appear anywhere in the tally's state.
#[test]
fn no_raw_address_is_stored() {
let v = v();
let addrs = ["203.0.113.7", "198.51.100.9", "2001:db8::42"];
for a in addrs {
let peer = if a.contains(':') { format!("[{a}]:1") } else { format!("{a}:1") };
v.record(DAY0 + 20, Some("doors"), Some("crates"), client_of(&peer, None));
}
assert_eq!(v.snapshot(DAY0 + 21).doors["doors"].consumers, 3);
for a in addrs {
let ip: IpAddr = a.parse().unwrap();
let octets: Vec<u8> = match ip {
IpAddr::V4(x) => x.octets().to_vec(),
IpAddr::V6(x) => x.octets().to_vec(),
};
assert!(!v.holds_bytes(a.as_bytes()), "the text of {a} is held");
assert!(!v.holds_bytes(&octets), "the octets of {a} are held");
}
}
/// ★ The window closes at UTC midnight: the new day starts at zero with a new
/// key, and yesterday's final numbers survive as numbers only.
#[test]
fn the_window_rolls_at_midnight_and_keeps_yesterday_as_numbers() {
let v = v();
v.record(DAY0 + 100, Some("doors"), Some("crates"), client_of("203.0.113.7:1", None));
let before = v.inner.lock().unwrap().key;
let s = v.snapshot(DAY0 + WINDOW_SECS + 5);
assert_ne!(before, v.inner.lock().unwrap().key, "the key rotates with the window");
assert_eq!(s.repos["crates"], Count::default(), "a new day starts at zero");
assert_eq!(s.counting_since_unix, DAY0 + WINDOW_SECS);
let prev = s.previous.expect("yesterday is kept");
assert_eq!(prev.repos["crates"].consumers, 1);
assert_eq!(prev.repos["crates"].requests, 1);
// Two idle days later, "yesterday" is an honest zero, not a stale copy.
let s = v.snapshot(DAY0 + 3 * WINDOW_SECS);
assert_eq!(s.previous.unwrap().repos["crates"], Count::default());
}
/// A name nobody configured is ignored: a caller cannot grow the table.
#[test]
fn an_unconfigured_scope_is_not_counted() {
let v = v();
v.record(DAY0 + 20, Some("nope"), Some("nope"), client_of("203.0.113.7:1", None));
let s = v.snapshot(DAY0 + 21);
assert!(!s.doors.contains_key("nope") && !s.repos.contains_key("nope"));
}
/// ★ **A puller keeps its label for the whole day, and loses it at
/// midnight.** Same client (whatever port, v4 or v4-mapped) ⇒ same label
/// within the window; two clients ⇒ two labels; the next day ⇒ a new label,
/// because the one key rotated. The label never contains the address.
#[test]
fn a_puller_keeps_its_label_within_a_day_and_not_across_days() {
let v = v();
let a = v.puller_label("203.0.113.7", DAY0 + 20);
assert!(a.starts_with("client ") && a.len() == "client ".len() + 8, "{a}");
assert_eq!(a, v.puller_label("203.0.113.7", DAY0 + 80_000), "same client, same day, same label");
assert_eq!(a, v.puller_label("::ffff:203.0.113.7", DAY0 + 30), "one host whichever stack");
assert_ne!(a, v.puller_label("198.51.100.9", DAY0 + 40), "two clients, two labels");
assert!(!a.contains("203"), "the label carries the address: {a}");
// …and it is the SAME key the visitor tally counts with: recording the
// client and labelling it cannot use two different secrets.
v.record(DAY0 + 50, Some("doors"), None, client_of("203.0.113.7:9", None));
let key_token = {
let g = v.inner.lock().unwrap();
token(&g.key, Client::Ip("203.0.113.7".parse().unwrap()))
};
assert_eq!(a, format!("client {:02x}{:02x}{:02x}{:02x}", key_token[0], key_token[1], key_token[2], key_token[3]));
assert_ne!(a, v.puller_label("203.0.113.7", DAY0 + WINDOW_SECS + 1), "a new day is a new label");
}
}