crates/nornir-identity-sso/src/state.rs
//! **One listener's SSO memory** (moved from Gunnar's `sso/mod.rs`, A34 M-4):
//! the flows begun and not yet finished (single use, ten minutes), the
//! second-factor tickets (single use, five minutes), the SAML assertion ids
//! already consumed, the IdPs' discovery documents and keys (an hour, feature
//! `oidc`) and the rate windows. Every map is bounded and swept on use.
//!
//! **In memory, on purpose** (A34-40, accepted and written down): flows,
//! second-factor tickets and the SAML replay set live in ONE process and are
//! lost on a restart or a twin takeover. Nothing in flight is a credential yet,
//! so losing it costs a person one more click ("start again"), never access:
//! a state, a ticket or a SAML response from before the restart is refused
//! (its flow is gone, and a SAML response is accepted only as the answer to an
//! AuthnRequest a live flow holds, `InResponseTo`), so the replay set's loss
//! re-admits nothing. Keeping them in the product's shared store would make
//! every begin a replicated write for a sign-in that may never come back.
use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex, RwLock};
use crate::egress::Egress;
use crate::guard::PrivateNets;
/// How long a begun flow may wait for the IdP: ten minutes.
pub const FLOW_TTL_MS: i64 = 10 * 60_000;
/// How long a second-factor ticket lives: five minutes.
pub const SECOND_TTL_MS: i64 = 5 * 60_000;
/// The most flows a listener holds open at once.
pub const MAX_FLOWS: usize = 10_000;
/// The most flows ONE client (an IPv4 address, an IPv6 /64) holds open for one
/// account at once (A34-6: counted per client, never per account alone, so a
/// few addresses cannot fill an account's sign-ins).
pub const MAX_FLOWS_PER_CLIENT: usize = 20;
/// Clock skew tolerated on IdP timestamps: two minutes.
pub const SKEW_SECS: i64 = 120;
/// How long an IdP's discovery document and keys are trusted: an hour.
pub const IDP_CACHE_MS: i64 = 60 * 60_000;
/// The most bytes an IdP answer may be.
pub const MAX_IDP_BODY: usize = 1 << 20;
/// Logins one client begins for one account per minute (A34-6).
pub const BEGINS_PER_CLIENT_PER_MINUTE: u32 = 20;
/// SCIM requests per bearer per minute.
pub const SCIM_PER_MINUTE: u32 = 600;
/// Failed finishes one client may make for one account per minute before THAT
/// client is held back for the rest of the minute; never another client (A34-6).
pub const FAILURES_PER_CLIENT_PER_MINUTE: u32 = 10;
/// **The client a sign-in is counted for**: the relay's word for the browser's
/// address, partitioned as the passkey ceremonies are (an IPv4 address, an IPv6
/// /64; `nornir_auth_webauthn::ceremony::ClientKey`). A relay that names none is
/// one client, `?`.
pub fn client_key(peer: &str) -> String {
if peer.trim().is_empty() {
return "?".to_owned();
}
nornir_auth_webauthn::ceremony::ClientKey::of_peer(peer)
.as_str()
.to_owned()
}
/// The refusal while the operator has not turned SSO egress on (A34-16).
pub const EGRESS_OFF: &str = "[sso-egress-off] this server reaches no identity provider: its operator has not turned single sign-on's network on (airgap)";
/// The refusal when the build or the product wired no network at all.
pub const NO_EGRESS: &str =
"[sso-no-egress] this server reaches no identity provider (no egress is wired into this build)";
/// A begun flow, waiting for the IdP.
#[derive(Clone)]
pub struct Flow {
pub account: String,
pub saml: bool,
pub console_origin: String,
pub expires_ms: i64,
/// OIDC: the nonce and the PKCE verifier (never logged).
pub nonce: String,
pub verifier: String,
pub redirect_uri: String,
/// SAML: the AuthnRequest's ID (`InResponseTo`).
pub request_id: String,
/// The client that began it ([`client_key`]).
pub client: String,
}
impl std::fmt::Debug for Flow {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Flow")
.field("account", &self.account)
.field("saml", &self.saml)
.field("expires_ms", &self.expires_ms)
.finish_non_exhaustive()
}
}
/// A login verified by the IdP, waiting for its passkey.
#[derive(Clone)]
pub struct Second {
pub account: String,
pub principal_id: String,
pub register: bool,
pub session_openssh: String,
pub session_expires_ms: i64,
pub expires_ms: i64,
}
#[derive(Default)]
struct Windows {
/// key → (minute, count)
counts: HashMap<String, (i64, u32)>,
}
/// **One listener's SSO memory.** Every map is bounded and swept on use.
#[derive(Default)]
pub struct SsoState {
egress: RwLock<Option<Arc<dyn Egress>>>,
egress_on: AtomicBool,
/// The DNS-over-HTTPS resolver a domain's proof is read from (the operator's;
/// empty: none, and a proof is refused by name, RA5-24).
dns_resolver: RwLock<String>,
allow_loopback_http: AtomicBool,
/// The operator's internal networks an IdP may live on; shared with the
/// product's resolver.
private_nets: Arc<RwLock<PrivateNets>>,
flows: Mutex<HashMap<String, Flow>>,
seconds: Mutex<HashMap<String, Second>>,
saml_seen: Mutex<HashMap<String, i64>>,
#[cfg(feature = "oidc")]
idps: Mutex<HashMap<String, crate::oidc::CachedIdp>>,
windows: Mutex<Windows>,
}
fn lock<T>(m: &Mutex<T>) -> std::sync::MutexGuard<'_, T> {
match m.lock() {
Ok(g) => g,
Err(poisoned) => poisoned.into_inner(),
}
}
impl SsoState {
/// Hand in the network (the product does, at start).
pub fn set_egress(&self, egress: Arc<dyn Egress>) {
*self.egress.write().unwrap_or_else(|p| p.into_inner()) = Some(egress);
}
/// **The operator's switch** (A34-16): no IdP is ever dialled unless the
/// operator turned SSO egress on (Gunnar: the feed's `sso_egress=on`). Off by
/// default — airgap — whatever an account admin typed.
pub fn set_egress_on(&self, on: bool) {
self.egress_on.store(on, Ordering::SeqCst);
}
/// Whether the operator turned SSO egress on.
pub fn egress_on(&self) -> bool {
self.egress_on.load(Ordering::SeqCst)
}
/// **The network**, when one was handed in AND the operator switched it on;
/// otherwise the refusal by name ([`EGRESS_OFF`], [`NO_EGRESS`]).
pub fn egress(&self) -> Result<Arc<dyn Egress>, String> {
let wired = self
.egress
.read()
.unwrap_or_else(|p| p.into_inner())
.clone()
.ok_or_else(|| NO_EGRESS.to_owned())?;
if !self.egress_on() {
return Err(EGRESS_OFF.to_owned());
}
Ok(wired)
}
/// The operator's test door: an IdP on this box's loopback, over http.
pub fn set_allow_loopback_http(&self, allow: bool) {
self.allow_loopback_http.store(allow, Ordering::SeqCst);
}
pub fn allow_loopback_http(&self) -> bool {
self.allow_loopback_http.load(Ordering::SeqCst)
}
/// The operator's DNS-over-HTTPS resolver for domain proofs (A34-5); blank
/// is none: a proof is then refused by name (`domains::NO_RESOLVER`), never
/// read from a resolver nobody named (RA5-24).
pub fn set_dns_resolver(&self, url: &str) {
*self.dns_resolver.write().unwrap_or_else(|p| p.into_inner()) = url.trim().to_owned();
}
/// The resolver a domain's proof is read from; empty when none is named.
pub fn dns_resolver(&self) -> String {
self.dns_resolver
.read()
.unwrap_or_else(|p| p.into_inner())
.clone()
}
/// The operator's internal networks an IdP may live on (estate config).
pub fn set_private_nets(&self, nets: PrivateNets) {
*self.private_nets.write().unwrap_or_else(|p| p.into_inner()) = nets;
}
pub fn private_nets(&self) -> PrivateNets {
self.private_nets
.read()
.unwrap_or_else(|p| p.into_inner())
.clone()
}
/// The same list, shared: the product's resolver reads it per lookup.
pub fn private_nets_handle(&self) -> Arc<RwLock<PrivateNets>> {
Arc::clone(&self.private_nets)
}
/// Store a begun flow under the hash of its state. Refused when the
/// listener, or this client for this account ([`MAX_FLOWS_PER_CLIENT`]), holds
/// too many.
pub fn begin(&self, state_hash: String, flow: Flow, now: i64) -> Result<(), String> {
let mut flows = lock(&self.flows);
flows.retain(|_, f| f.expires_ms > now);
if flows.len() >= MAX_FLOWS {
return Err(
"too many sign-ins are waiting for their identity providers; try again in a minute"
.to_owned(),
);
}
if flows
.values()
.filter(|f| f.account == flow.account && f.client == flow.client)
.count()
>= MAX_FLOWS_PER_CLIENT
{
return Err(format!("[sso-busy] too many sign-ins to {} from this address are waiting; try again in a minute", flow.account));
}
flows.insert(state_hash, flow);
Ok(())
}
/// **May `client` begin another sign-in to `account` this minute?** Counts
/// one ([`BEGINS_PER_CLIENT_PER_MINUTE`]).
pub fn allow_begin(&self, account: &str, client: &str, now: i64) -> bool {
self.allow(
&format!("begin\u{1f}{account}\u{1f}{}", client_key(client)),
BEGINS_PER_CLIENT_PER_MINUTE,
now,
)
}
/// Count one failed finish of `client` for `account`.
pub fn note_failure(&self, account: &str, client: &str, now: i64) {
self.allow(
&format!("fail\u{1f}{account}\u{1f}{}", client_key(client)),
u32::MAX,
now,
);
}
/// **Is `client` held back from `account` for the rest of this minute?**
/// ([`FAILURES_PER_CLIENT_PER_MINUTE`] failures). Another client never is.
pub fn failures_held(&self, account: &str, client: &str, now: i64) -> bool {
self.counted(
&format!("fail\u{1f}{account}\u{1f}{}", client_key(client)),
now,
) >= FAILURES_PER_CLIENT_PER_MINUTE
}
/// Take a flow by the hash of its state: single use, whatever happens
/// next; `None` when unknown or expired.
pub fn take_flow(&self, state_hash: &str, now: i64) -> Option<Flow> {
let mut flows = lock(&self.flows);
flows.remove(state_hash).filter(|f| f.expires_ms > now)
}
/// Park a login that owes its passkey under the hash of its ticket.
pub fn park_second(&self, ticket_hash: String, second: Second, now: i64) -> Result<(), String> {
let mut seconds = lock(&self.seconds);
seconds.retain(|_, s| s.expires_ms > now);
if seconds.len() >= MAX_FLOWS {
return Err("too many sign-ins are waiting for their passkeys".to_owned());
}
seconds.insert(ticket_hash, second);
Ok(())
}
/// Take a parked login: single use; `None` when unknown or expired.
pub fn take_second(&self, ticket_hash: &str, now: i64) -> Option<Second> {
lock(&self.seconds)
.remove(ticket_hash)
.filter(|s| s.expires_ms > now)
}
/// **A SAML assertion id, once.** False when it was seen before (a
/// replay); remembered until `until_ms`.
pub fn first_sight(&self, assertion_id: &str, until_ms: i64, now: i64) -> bool {
let mut seen = lock(&self.saml_seen);
seen.retain(|_, until| *until > now);
if seen.contains_key(assertion_id) {
return false;
}
if seen.len() >= 100_000 {
// Full of live ids: refuse rather than forget one.
return false;
}
seen.insert(assertion_id.to_owned(), until_ms.max(now + FLOW_TTL_MS));
true
}
/// An IdP's cached discovery document and keys, while fresh.
#[cfg(feature = "oidc")]
pub fn cached_idp(&self, key: &str, now: i64) -> Option<crate::oidc::CachedIdp> {
lock(&self.idps)
.get(key)
.filter(|c| now - c.fetched_ms < IDP_CACHE_MS)
.cloned()
}
/// Keep an IdP's discovery document and keys.
#[cfg(feature = "oidc")]
pub fn cache_idp(&self, key: &str, idp: crate::oidc::CachedIdp) {
let mut idps = lock(&self.idps);
if idps.len() >= 4096 {
idps.clear();
}
idps.insert(key.to_owned(), idp);
}
/// **A rate window**: count one event for `key` in the current minute;
/// false when that makes more than `limit`.
pub fn allow(&self, key: &str, limit: u32, now: i64) -> bool {
let minute = now / 60_000;
let mut w = lock(&self.windows);
if w.counts.len() > 50_000 {
w.counts.retain(|_, (m, _)| *m == minute);
}
let entry = w.counts.entry(key.to_owned()).or_insert((minute, 0));
if entry.0 != minute {
*entry = (minute, 0);
}
entry.1 = entry.1.saturating_add(1);
entry.1 <= limit
}
/// How many events `key` counted this minute (without counting one).
pub fn counted(&self, key: &str, now: i64) -> u32 {
let minute = now / 60_000;
lock(&self.windows)
.counts
.get(key)
.filter(|(m, _)| *m == minute)
.map_or(0, |(_, n)| *n)
}
}
#[cfg(test)]
mod tests {
use super::*;
pub(crate) fn flow(account: &str) -> Flow {
Flow {
account: account.into(),
saml: false,
console_origin: String::new(),
expires_ms: 1_000 + FLOW_TTL_MS,
nonce: String::new(),
verifier: String::new(),
redirect_uri: String::new(),
request_id: String::new(),
client: client_key("203.0.113.7"),
}
}
fn flow_from(account: &str, client: &str) -> Flow {
Flow {
client: client_key(client),
..flow(account)
}
}
#[test]
fn flows_are_single_use_and_expire() {
let s = SsoState::default();
s.begin("a".into(), flow("acme"), 1_000).unwrap();
assert!(s.take_flow("a", 2_000).is_some());
assert!(s.take_flow("a", 2_000).is_none(), "a flow was taken twice");
s.begin("b".into(), flow("acme"), 1_000).unwrap();
assert!(
s.take_flow("b", 1_000 + FLOW_TTL_MS + 1).is_none(),
"an expired flow was taken"
);
}
/// ★ **A few addresses cannot lock an account's sign-ins** (A34-6): open
/// flows, begins and failures are counted per (account, client); a client at
/// its limit is held back and every other client of the account is not.
#[test]
fn a_few_clients_cannot_lock_an_account_out() {
let s = SsoState::default();
for i in 0..MAX_FLOWS_PER_CLIENT {
s.begin(format!("a{i}"), flow_from("acme", "203.0.113.7"), 1_000)
.unwrap();
}
assert!(
s.begin(
"a-over".into(),
flow_from("acme", "203.0.113.7:4431"),
1_000
)
.is_err(),
"one client past its open flows"
);
assert!(
s.begin("b".into(), flow_from("acme", "198.51.100.9"), 1_000)
.is_ok(),
"another client of the account is not held"
);
assert!(
s.begin("c".into(), flow_from("globex", "203.0.113.7"), 1_000)
.is_ok(),
"the same client for another account"
);
// Thirty attacking addresses cannot fill the account for a newcomer.
for n in 0..30u8 {
for i in 0..MAX_FLOWS_PER_CLIENT {
let _ = s.begin(
format!("f{n}-{i}"),
flow_from("acme", &format!("192.0.2.{n}")),
1_000,
);
}
}
assert!(
s.begin("newcomer".into(), flow_from("acme", "100.100.1.1"), 1_000)
.is_ok(),
"an account was locked by thirty addresses"
);
for _ in 0..BEGINS_PER_CLIENT_PER_MINUTE {
assert!(s.allow_begin("acme", "203.0.113.7", 60_000));
}
assert!(
!s.allow_begin("acme", "203.0.113.7", 60_000),
"past its begins"
);
assert!(
s.allow_begin("acme", "198.51.100.9", 60_000),
"another client still begins"
);
for _ in 0..FAILURES_PER_CLIENT_PER_MINUTE {
s.note_failure("acme", "2001:db8:1:2::7", 60_000);
}
assert!(
s.failures_held("acme", "2001:db8:1:2::99", 60_000),
"the same /64 is held"
);
assert!(
!s.failures_held("acme", "198.51.100.9", 60_000),
"one client's failures held another back"
);
assert!(
!s.failures_held("acme", "2001:db8:1:2::7", 120_000),
"a new minute starts afresh"
);
}
#[test]
fn a_client_is_its_address_or_its_64() {
assert_eq!(client_key("203.0.113.7:4431"), "203.0.113.7");
assert_eq!(
client_key("[2001:db8:1:2::7]:443"),
client_key("2001:db8:1:2::99")
);
assert_ne!(client_key("2001:db8:1:2::7"), client_key("2001:db8:1:3::7"));
assert_eq!(client_key(""), "?");
}
#[test]
fn an_assertion_is_consumed_once() {
let s = SsoState::default();
assert!(s.first_sight("_a1", 10_000_000, 1_000));
assert!(
!s.first_sight("_a1", 10_000_000, 2_000),
"a replayed assertion was accepted"
);
assert!(s.first_sight("_a2", 10_000_000, 2_000));
}
#[test]
fn rate_windows_count_per_minute() {
let s = SsoState::default();
for _ in 0..3 {
assert!(s.allow("k", 3, 60_000));
}
assert!(!s.allow("k", 3, 61_000));
assert!(s.allow("k", 3, 120_000), "a new minute starts afresh");
}
struct Net;
impl Egress for Net {
fn get(&self, _: &str) -> Result<(u16, String), String> {
Ok((200, String::new()))
}
fn post_form(&self, _: &str, _: &str, _: &str) -> Result<(u16, String), String> {
Ok((200, String::new()))
}
}
/// ★ **No egress unless the operator turned it on** (A34-16): a wired
/// network is refused by name until the switch is on, and again once off.
#[test]
fn the_egress_is_the_operators_to_turn_on() {
let s = SsoState::default();
assert_eq!(
s.egress().err().as_deref(),
Some(NO_EGRESS),
"nothing wired"
);
s.set_egress(Arc::new(Net));
assert!(!s.egress_on(), "off by default");
assert_eq!(
s.egress().err().as_deref(),
Some(EGRESS_OFF),
"wired but not switched on"
);
s.set_egress_on(true);
assert!(s.egress().is_ok());
s.set_egress_on(false);
assert_eq!(
s.egress().err().as_deref(),
Some(EGRESS_OFF),
"switched off again, live"
);
assert_eq!(EGRESS_OFF, crate::Refusal::EgressOff.message());
assert_eq!(NO_EGRESS, crate::Refusal::NoEgress.message());
}
/// ★ **A restart forgets what was in flight and re-admits nothing**
/// (A34-40): a new listener knows no state, no ticket and no flow a SAML
/// response could answer, so each is refused, never accepted.
#[test]
fn a_restart_forgets_the_sign_ins_in_flight_and_admits_none() {
let before = SsoState::default();
before
.begin("state-hash".into(), flow("acme"), 1_000)
.unwrap();
let second = Second {
account: "acme".into(),
principal_id: "acme.bob".into(),
register: false,
session_openssh: String::new(),
session_expires_ms: 9_000_000,
expires_ms: 1_000 + SECOND_TTL_MS,
};
before
.park_second("ticket-hash".into(), second, 1_000)
.unwrap();
assert!(before.first_sight("_assertion", 10_000_000, 1_000));
let after = SsoState::default();
assert!(
after.take_flow("state-hash", 2_000).is_none(),
"a state from before the restart was taken"
);
assert!(
after.take_second("ticket-hash", 2_000).is_none(),
"a ticket from before the restart was taken"
);
// The replay set is empty, but a response needs a live flow to answer.
assert!(after.first_sight("_assertion", 10_000_000, 2_000));
assert!(
after.take_flow("state-hash", 2_000).is_none(),
"no flow is there for the replayed response"
);
}
#[test]
fn the_resolver_shares_the_operators_networks() {
let s = SsoState::default();
assert!(s.private_nets().is_empty());
let nets = PrivateNets::parse("10.0.2.100/32").expect("nets");
s.set_private_nets(nets.clone());
assert_eq!(
*s.private_nets_handle().read().unwrap(),
nets,
"the resolver's handle sees the list"
);
}
}