gunnar-ssh 3.0.0

//! **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"
    );
}