crates/gunnar-ssh/tests/anonymous_ceiling.rs
//! **The anonymous ceiling**: a stranger's share of the box is
//! small, hard, and never grows.
//!
//! # The hole this closes
//!
//! `crate::admission` already bounds git transfers three ways — a governed
//! server-wide limit, a memory budget, and a per-principal max-min share. All
//! three were designed for callers gunnar can *name*: the per-principal share
//! is keyed on the principal id precisely because gunnar authenticates keys.
//!
//! Anonymous read then opened a door that has no key behind it. Every unauthenticated
//! reader arrives as the one principal
//! [`gunnar_core::ANONYMOUS_PRINCIPAL_ID`], so the whole internet shares a
//! single row in the per-principal map — which sounds like the fix and is not,
//! for one reason: **the max-min share is deliberately work-conserving.** Below
//! [`gunnar_ssh::admission::CONTENTION_PERCENT`] of the server-wide limit the
//! share does not bind at all, and the ceiling for any principal — including
//! `anonymous` — is the *entire server*. That rule was measured and is right
//! for a keyed identity: refusing a lone team its idle appliance is the exact
//! mistake the admission gate was rewritten to stop making.
//!
//! It is wrong for a stranger. An idle box plus an anonymous flood means
//! `anonymous` legitimately takes all 256 slots, and the operator who then
//! shows up does not get refused — worse, they *wait*, because nothing evicts a
//! transfer already in flight. "The console is unusable while somebody scrapes
//! us" is not a limit failing; it is the absence of a limit for the one caller
//! that has no accountability at all.
//!
//! # Seen red
//!
//! Written before the ceiling existed, against the gate as it shipped. The
//! first run does not fail an assertion — it does not compile, because
//! `Admission::with_anonymous_ceiling`, `Refusal::AnonymousAtCapacity` and
//! `AdmissionStats::refused_anonymous` were not there. With the ceiling stubbed
//! in as `None` (the "wired but off" arm, which is the honest red once the
//! types exist) every assertion below fails in the informative direction:
//! `the_anonymous_ceiling_binds_on_an_idle_server` admits all 64 instead of 5,
//! and `refused_anonymous` stays 0. See the report in the module docs of
//! `gunnar_ssh::admission`.
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
use gunnar_ssh::admission::DEFAULT_MAX_ANONYMOUS_TRANSFERS;
use gunnar_ssh::config::DEFAULT_MAX_TRANSFERS_PER_PRINCIPAL;
use gunnar_ssh::governor::OpClass;
use gunnar_ssh::gunnar_core::ANONYMOUS_PRINCIPAL_ID;
use gunnar_ssh::{Admission, AdmissionGuard, Refusal, SshConfig};
/// A gate with room to spare server-wide, so that anything the anonymous
/// ceiling refuses was refused *by the anonymous ceiling* and not by a full
/// box. That is the whole point: on a busy server every limit refuses and the
/// test proves nothing.
fn roomy_gate(anonymous: usize) -> Arc<Admission> {
Arc::new(
Admission::new(Some(64), Some(DEFAULT_MAX_TRANSFERS_PER_PRINCIPAL))
.with_anonymous_ceiling(Some(anonymous))
// No wait: this file is about *who* is refused, and a wait would
// turn every refusal into a timing question. The wait itself is
// tested by `an_anonymous_refusal_is_never_parked`.
.with_wait(Duration::ZERO),
)
}
/// The ceiling binds on an **idle** server, which is the case the per-principal
/// share deliberately does not cover.
///
/// 64 slots free, one anonymous caller, and the answer must still be 5. Under
/// the max-min rule alone this principal's ceiling here is 64 — the server is
/// nowhere near `CONTENTION_PERCENT` — so a gate without the anonymous ceiling
/// admits every one of these and this test reads 64 where it wants 5.
#[test]
fn the_anonymous_ceiling_binds_on_an_idle_server() {
let gate = roomy_gate(5);
let mut held: Vec<AdmissionGuard> = Vec::new();
for i in 0..5 {
held.push(
gate.try_admit(ANONYMOUS_PRINCIPAL_ID, "team", OpClass::Clone)
.unwrap_or_else(|refusal| {
panic!("anonymous transfer {i} must be admitted, and was refused: {refusal}")
}),
);
}
let refusal = gate
.try_admit(ANONYMOUS_PRINCIPAL_ID, "team", OpClass::Clone)
.expect_err(
"the sixth anonymous transfer must be refused on an EMPTY server; if it was admitted \
the max-min share is the only ceiling and a stranger's ceiling is the whole box",
);
assert_eq!(
refusal,
Refusal::AnonymousAtCapacity { limit: 5 },
"the refusal must name the anonymous ceiling, not the server-wide limit: an operator who \
reads `at capacity` on a box with 59 free slots raises the wrong number"
);
let stats = gate.stats();
assert_eq!(stats.in_flight, 5, "exactly the ceiling is in flight");
assert_eq!(
stats.peak_in_flight, 5,
"and the HIGH-WATER MARK is the cap"
);
assert_eq!(
stats.refused_anonymous, 1,
"the refusal must be COUNTED under its own reason; folded into \
`refused_per_principal` it is indistinguishable from a keyed client hitting its share"
);
assert_eq!(
stats.refused_per_principal, 0,
"and it must NOT be counted as a per-principal refusal"
);
assert_eq!(
stats.refused(),
1,
"it is a refusal, so the total moves too"
);
// A limit, not a budget: the slot comes back.
drop(held.pop().expect("five were held"));
gate.try_admit(ANONYMOUS_PRINCIPAL_ID, "team", OpClass::Clone)
.expect("a freed anonymous slot is reusable");
assert_eq!(gate.stats().peak_in_flight, 5, "and the peak never moved");
}
/// The bound holds under **real concurrency**, not just in a straight line.
///
/// 200 anonymous askers racing on a multi-threaded runtime, each holding its
/// permit across an await point. The assertion is on the observed maximum
/// number simultaneously inside the guarded region — counted by the test
/// itself, not read from the gate's own counter, so a gate that lied about
/// `peak_in_flight` could not make this pass.
#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
async fn two_hundred_anonymous_readers_never_exceed_the_ceiling() {
const CEILING: usize = 5;
let gate = roomy_gate(CEILING);
let inside = Arc::new(AtomicUsize::new(0));
let observed_peak = Arc::new(AtomicUsize::new(0));
let admitted = Arc::new(AtomicUsize::new(0));
let mut tasks = tokio::task::JoinSet::new();
for _ in 0..200 {
let gate = Arc::clone(&gate);
let inside = Arc::clone(&inside);
let observed_peak = Arc::clone(&observed_peak);
let admitted = Arc::clone(&admitted);
tasks.spawn(async move {
// Every task retries, so the 195 that are refused at first do not
// simply vanish — they keep pressing on the ceiling, which is what
// an anonymous flood actually does.
for _ in 0..40 {
match gate.try_admit(ANONYMOUS_PRINCIPAL_ID, "team", OpClass::Clone) {
Ok(guard) => {
let now = inside.fetch_add(1, Ordering::SeqCst) + 1;
observed_peak.fetch_max(now, Ordering::SeqCst);
admitted.fetch_add(1, Ordering::SeqCst);
// Hold it across a yield: a permit released before the
// next task ever asks measures the scheduler, not the
// gate.
tokio::time::sleep(Duration::from_millis(2)).await;
inside.fetch_sub(1, Ordering::SeqCst);
drop(guard);
return;
}
Err(_) => tokio::time::sleep(Duration::from_millis(1)).await,
}
}
});
}
while tasks.join_next().await.is_some() {}
let peak = observed_peak.load(Ordering::SeqCst);
assert!(
peak > 0 && peak <= CEILING,
"the observed maximum number of anonymous transfers in flight at once was {peak}, and the \
ceiling is {CEILING}; a bound that is exceeded under load is not a bound"
);
assert_eq!(
peak, CEILING,
"the ceiling must actually be REACHED as well as not exceeded; a test that only ever saw \
one transfer at a time would pass the bound above while proving nothing"
);
assert!(
admitted.load(Ordering::SeqCst) > CEILING,
"the gate must keep SERVING anonymous readers as slots free up, not serve five and shut; \
a ceiling that never recycles is an outage with a nicer name"
);
// `peak_in_flight` is allowed ONE above the ceiling, and this is not
// slack — it is a documented property of that counter that this test found
// and that is worth pinning rather than papering over.
//
// The number the gate enforces under its lock is the per-principal map
// entry, and that is never above `CEILING`. `in_flight` is a separate,
// lock-free global: `Admission::release` decrements the map entry INSIDE
// the lock and the counter AFTER it, so between those two instants a
// freed slot has already been handed to a new admission whose own
// `fetch_add` can land first. `peak_in_flight` therefore reads
// `CEILING + 1` on a busy multi-core run — a transfer that had finished
// being counted beside the one that replaced it, not six live transfers.
// The bound that matters is the one the test measured above.
let counter_peak = gate.stats().peak_in_flight as usize;
assert!(
counter_peak <= CEILING + 1,
"the gate's lock-free high-water mark read {counter_peak} against a ceiling of {CEILING}; \
one above is the release/admit interleave described here, more than one is the ceiling \
itself failing"
);
}
/// **The asymmetry, which is the entire point.**
///
/// Anonymous saturates its ceiling and keeps hammering; a keyed principal walks
/// past it and takes its full share without waiting for a single anonymous
/// transfer to finish. If this ever fails, the limit has stopped being a limit
/// on strangers and become a limit on the server.
#[test]
fn an_authenticated_caller_is_never_held_up_by_an_anonymous_flood() {
let gate = roomy_gate(5);
// Anonymous takes everything it is allowed and is then refused, repeatedly.
let _anon: Vec<AdmissionGuard> = (0..5)
.map(|_| {
gate.try_admit(ANONYMOUS_PRINCIPAL_ID, "team", OpClass::Clone)
.expect("anonymous fills its ceiling")
})
.collect();
for _ in 0..50 {
gate.try_admit(ANONYMOUS_PRINCIPAL_ID, "team", OpClass::Clone)
.expect_err("and every further anonymous ask is refused");
}
// The operator arrives. Nothing anonymous has finished.
let mut operator = Vec::new();
for i in 0..DEFAULT_MAX_TRANSFERS_PER_PRINCIPAL {
operator.push(
gate.try_admit("alice", "team", OpClass::Clone)
.unwrap_or_else(|refusal| {
panic!(
"a keyed principal's transfer {i} was refused while anonymous held only 5 of \
64 slots: {refusal}"
)
}),
);
}
assert_eq!(
gate.stats().in_flight,
5 + DEFAULT_MAX_TRANSFERS_PER_PRINCIPAL as u64,
"anonymous kept its five and the operator got its whole floor, at the same time"
);
assert_eq!(
gate.stats().refused_per_principal,
0,
"and the operator was never refused for any reason"
);
}
/// The permit comes back **when the transfer panics**, not only when it returns.
///
/// The release is `Drop`, so unwinding runs it; this pins that, because a gate
/// that leaks one permit per panicking clone is a server that refuses
/// everything an hour after start-up and blames the ceiling.
#[test]
fn a_panicking_anonymous_transfer_gives_its_permit_back() {
let gate = roomy_gate(1);
// `AssertUnwindSafe` because the gate may now hold a flow ledger, whose
// control store is a `dyn` trait object and therefore not `RefUnwindSafe`.
// The assertion is sound for what this test does: the gate under test was
// built by `roomy_gate` and has no ledger at all, and the permit release
// this is about is a `Drop` on the guard, which unwinding runs whatever the
// gate is holding.
let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe({
let gate = Arc::clone(&gate);
move || {
let _guard = gate
.try_admit(ANONYMOUS_PRINCIPAL_ID, "team", OpClass::Clone)
.expect("the only anonymous slot");
assert_eq!(gate.stats().in_flight, 1);
panic!("a clone that dies mid-pack");
}
}));
assert!(result.is_err(), "the closure must actually have panicked");
assert_eq!(
gate.stats().in_flight,
0,
"the permit must be released by unwinding; a manual decrement on the happy path would \
read 1 here and the ceiling would shrink by one per panic until it was zero"
);
gate.try_admit(ANONYMOUS_PRINCIPAL_ID, "team", OpClass::Clone)
.expect("and the ceiling is whole again after a panic");
}
/// An anonymous refusal is **final and immediate**: it is never parked in
/// [`Admission::admit_within`]'s wait.
///
/// The wait exists because a keyed client that is refused reconnects, and the
/// reconnect costs more than parking one future. For a stranger the arithmetic
/// inverts — parking is what holds the socket, the session and the key material
/// a flood is trying to exhaust — so the anonymous ceiling refuses on the spot
/// while the keyed refusal beside it still waits out its bound.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn an_anonymous_refusal_is_never_parked() {
let wait = Duration::from_millis(400);
// `hard_capped`, so the keyed control below is genuinely refused on an idle
// box. Under the ordinary max-min rule its ceiling here would be the whole
// 64 — the server is nowhere near `CONTENTION_PERCENT` — and the control
// would be admitted rather than parked, which is precisely the
// work-conserving behaviour the anonymous ceiling exists to *not* inherit.
let gate = Arc::new(
Admission::hard_capped(Some(64), Some(2))
.with_anonymous_ceiling(Some(1))
.with_wait(wait),
);
let _anon = gate
.try_admit(ANONYMOUS_PRINCIPAL_ID, "team", OpClass::Clone)
.expect("the only anonymous slot");
let _keyed = [
gate.try_admit("alice", "team", OpClass::Clone)
.expect("first"),
gate.try_admit("alice", "team", OpClass::Clone)
.expect("second"),
];
let started = Instant::now();
let refusal = gate
.admit_within(ANONYMOUS_PRINCIPAL_ID, "team", OpClass::Clone)
.await
.expect_err("anonymous is over its ceiling");
let anonymous_took = started.elapsed();
assert!(
matches!(refusal, Refusal::AnonymousAtCapacity { limit: 1 }),
"refused by the anonymous ceiling, got {refusal:?}"
);
assert!(
anonymous_took < wait / 4,
"an anonymous caller over the ceiling waited {anonymous_took:?} of a {wait:?} bound; it \
must be refused immediately, because parking it is what holds the socket a flood wants"
);
// The control: the same gate, the same instant, a KEYED principal over its
// hard-ish share — that one does park, which is what makes the assertion
// above a statement about anonymity and not about a broken wait.
let started = Instant::now();
gate.admit_within("alice", "team", OpClass::Clone)
.await
.expect_err("the keyed principal is over its share too");
let keyed_took = started.elapsed();
assert!(
keyed_took >= wait,
"the keyed refusal waited only {keyed_took:?} of its {wait:?} bound, so the anonymous \
comparison above proves nothing"
);
}
/// The shipped default is **small**, and small relative to the keyed share
/// rather than merely a round number.
///
/// Pinned as a decision, in the same spirit as
/// `the_default_admission_limits_are_a_decision_and_are_ordered`: a default
/// that drifted up to the per-principal floor would silently re-open the hole
/// this file exists to close, and nothing else would notice.
#[test]
fn the_anonymous_default_is_a_decision_and_is_far_below_the_keyed_share() {
assert_eq!(
DEFAULT_MAX_ANONYMOUS_TRANSFERS, 5,
"the number is a decision; change it here and in the doc comment that argues for it"
);
const {
assert!(
DEFAULT_MAX_ANONYMOUS_TRANSFERS < DEFAULT_MAX_TRANSFERS_PER_PRINCIPAL,
"a stranger must never be able to hold as much as a keyed identity is GUARANTEED"
)
}
let config = SshConfig::default();
assert_eq!(
config.max_anonymous_transfers,
Some(DEFAULT_MAX_ANONYMOUS_TRANSFERS),
"the appliance default must HAVE an anonymous ceiling: the door is open by default, so \
the door has to arrive with the limit already on it"
);
assert_eq!(
config.without_anonymous_access().max_anonymous_transfers,
Some(DEFAULT_MAX_ANONYMOUS_TRANSFERS),
"closing the door must not silently clear the ceiling; re-opening it later would then \
re-open it unlimited"
);
}
/// A ceiling of `None` is genuinely off — the operator's escape hatch — and
/// says so by admitting past the default.
///
/// Tested because "off" is the arm that was red before the fix, and an
/// implementation that ignored the `Option` would pass every test above.
#[test]
fn the_ceiling_can_be_switched_off_entirely() {
let gate = Arc::new(
Admission::new(Some(64), Some(DEFAULT_MAX_TRANSFERS_PER_PRINCIPAL))
.with_anonymous_ceiling(None)
.with_wait(Duration::ZERO),
);
let held: Vec<AdmissionGuard> = (0..DEFAULT_MAX_TRANSFERS_PER_PRINCIPAL)
.map(|i| {
gate.try_admit(ANONYMOUS_PRINCIPAL_ID, "team", OpClass::Clone)
.unwrap_or_else(|e| panic!("with the ceiling off, anonymous {i} is admitted: {e}"))
})
.collect();
assert_eq!(held.len(), DEFAULT_MAX_TRANSFERS_PER_PRINCIPAL);
assert_eq!(gate.stats().refused_anonymous, 0);
assert_eq!(
gate.max_anonymous(),
None,
"and the gate reports the ceiling it is actually enforcing"
);
}