crates/monetize-server/src/control.rs
//! **Callers from OUTSIDE the estate**: the admin principals whose per-call
//! SSH signature opens monetize's control API (owner 2026-09-28: ONE shared
//! operator console administers every estate's monetize through its
//! `monetize.<apex>`; auth is SSHSIG per gRPC call, exactly gunnar's model).
//!
//! * The config's `control_principals` names each principal and its OpenSSH
//! PUBLIC key (`monetize-ui = "ssh-ed25519 AAAA… monetize-ui"`); the front
//! writes it from its install feed (`monetize_principal.<name>`). Nothing
//! secret is held here: revoking the console for this estate is removing
//! the line.
//! * [`layer`] runs before every route: a request that carries
//! `monetize-authorization` is checked in full (`nornir_auth_sshsig::check`:
//! bound to THIS method, ±5 min, a known key verifies, single use) and either
//! refused with gRPC `UNAUTHENTICATED` or marked [`ControlCaller`]. A request
//! without it goes on to the bearer, unchanged (the console IN the estate,
//! on loopback).
//! * [`interceptor`] is the one gate every service sits behind: a marked
//! caller passes, anyone else needs the bearer.
//!
//! The principals are re-read with the config (`guests::look`), so a
//! principal fed after the server started is taken in without a restart.
use std::collections::BTreeMap;
use std::sync::{Arc, RwLock};
use monetize_proto::control_auth::{sshsig, MonetizeScheme, METADATA_KEY};
use tonic::{Request, Status};
/// The request extension a verified control caller carries (its principal name).
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ControlCaller(pub String);
/// The admin principals, by fingerprint, and the replay guard.
#[derive(Debug, Default)]
pub struct ControlKeys {
keys: RwLock<BTreeMap<String, (String, String)>>,
guard: sshsig::ReplayGuard,
}
/// The prefix every control-call decision is logged under.
pub const AUDIT_TARGET: &str = "monetize::audit";
impl ControlKeys {
/// From the config's `control_principals` (name → OpenSSH public key line).
/// A line that is not a key is left out and named.
pub fn from_config(principals: &BTreeMap<String, String>) -> (ControlKeys, Vec<String>) {
let k = ControlKeys::default();
let said = k.set(principals);
(k, said)
}
/// Replace the principals; answers the lines to log (a refused line by name).
pub fn set(&self, principals: &BTreeMap<String, String>) -> Vec<String> {
let mut next = BTreeMap::new();
let mut said = Vec::new();
for (name, line) in principals {
match sshsig::fingerprint_of(line) {
Some(fp) => {
next.insert(fp, (name.clone(), line.trim().to_string()));
}
None => said.push(format!("REFUSED [control-principal-key] control_principals.{name} is not an OpenSSH public key line — that principal cannot call")),
}
}
let mut keys = self.keys.write().unwrap_or_else(|e| e.into_inner());
if *keys != next {
said.push(format!(
"control principals: {}",
if next.is_empty() {
"none".to_string()
} else {
next.values()
.map(|(n, _)| n.as_str())
.collect::<Vec<_>>()
.join(", ")
}
));
*keys = next;
}
said
}
/// The principal names, for the health line.
pub fn names(&self) -> Vec<String> {
self.keys
.read()
.unwrap_or_else(|e| e.into_inner())
.values()
.map(|(n, _)| n.clone())
.collect()
}
/// The whole check of one call; answers the principal's name.
pub fn check(
&self,
header: Option<&str>,
path: &str,
now_unix_ms: i64,
) -> Result<String, sshsig::Refusal> {
let keys = self.keys.read().unwrap_or_else(|e| e.into_inner()).clone();
let fp = sshsig::check::<MonetizeScheme>(
header,
path,
now_unix_ms,
|fp| keys.get(fp).map(|(_, l)| l.clone()),
&self.guard,
)?;
Ok(keys.get(&fp).map(|(n, _)| n.clone()).unwrap_or(fp))
}
}
fn now_ms() -> i64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as i64)
.unwrap_or(0)
}
/// **The edge**: a request with a control credential is checked here, once,
/// against the method the router is about to dispatch; the credential header
/// is removed either way so nothing downstream reads it again.
pub async fn layer(
keys: Arc<ControlKeys>,
mut req: axum::extract::Request,
next: axum::middleware::Next,
) -> axum::response::Response {
let Some(raw) = req.headers_mut().remove(METADATA_KEY) else {
return next.run(req).await;
};
let path = req.uri().path().to_string();
match keys.check(raw.to_str().ok(), &path, now_ms()) {
Ok(name) => {
eprintln!("{AUDIT_TARGET}: control call {path} by principal {name} (SSHSIG verified)");
req.extensions_mut().insert(ControlCaller(name));
next.run(req).await
}
Err(r) => {
let caller = raw
.to_str()
.ok()
.and_then(|h| sshsig::Credential::<MonetizeScheme>::parse(h).ok())
.map(|c| c.fingerprint)
.unwrap_or_else(|| "an unreadable credential".into());
eprintln!(
"{}",
refusal_line(&caller, r.reason(), &path, std::time::Instant::now())
);
// Coarse on the wire: never whether a key is known.
let why = match &r {
sshsig::Refusal::Stale { skew_ms } => format!(
"the credential is {skew_ms} ms from this server's clock; the window is ±{} ms",
sshsig::MAX_CLOCK_SKEW_MS
),
sshsig::Refusal::MethodMismatch { signed, called } => {
format!("this credential was signed for {signed} and presented at {called}")
}
sshsig::Refusal::Malformed(m) => format!("malformed credential: {m}"),
sshsig::Refusal::Replayed => {
"this credential has already been used; each one is single-use".into()
}
sshsig::Refusal::ReplayGuardFull => {
return Status::resource_exhausted(
"monetize is remembering as many single-use credentials as it will; retry",
)
.into_http()
}
_ => "the credential did not verify against any control principal of this monetize"
.into(),
};
Status::unauthenticated(why).into_http()
}
}
}
/// **One refused control call, as said** (stinkies control-refusal-spam, the
/// REAL test2 2026-10-01: the shared admin console polls an estate whose
/// monetize was not fed its key yet, and the front logged `monetize::audit:
/// control call /monetize.v1.Products/List REFUSED (not-verified)` at WARN
/// 7/min). Every refusal is still said — one line each — and the line carries
/// the caller's key fingerprint and the reason.
///
/// The FIRST refusal of a caller in [`REFUSAL_SAY_EVERY`] says `REFUSED` (the
/// front's stderr capture lifts it to WARN); the rest of the window say
/// `refused again` (INFO) with how many so far; the next window's first names
/// how many went quietly. A caller whose REASON changes is said `REFUSED` at
/// once — a new reason is news. gunnar's `gunnar_grpc::quiet` (1983c5b), for
/// monetize's own audit line.
pub fn refusal_line(caller: &str, reason: &str, path: &str, now: std::time::Instant) -> String {
use std::collections::HashMap;
use std::sync::Mutex;
/// Per caller: the window's start, how many went quietly, the last reason.
type Window = (std::time::Instant, u64, String);
static WINDOWS: Mutex<Option<HashMap<String, Window>>> = Mutex::new(None);
let mut g = WINDOWS.lock().unwrap_or_else(|p| p.into_inner());
let map = g.get_or_insert_with(HashMap::new);
let loud = match map.get_mut(caller) {
Some((since, quiet, last))
if last == reason && now.saturating_duration_since(*since) < REFUSAL_SAY_EVERY =>
{
*quiet += 1;
Err(*quiet)
}
Some(w) => {
let went = if w.2 == reason { w.1 } else { 0 };
*w = (now, 0, reason.to_owned());
Ok(went)
}
None => {
map.insert(caller.to_owned(), (now, 0, reason.to_owned()));
Ok(0)
}
};
match loud {
Ok(0) => format!("{AUDIT_TARGET}: control call {path} REFUSED ({reason}) — the caller's key {caller}"),
Ok(went) => format!("{AUDIT_TARGET}: control call {path} REFUSED ({reason}) — the caller's key {caller}; {went} more went quietly since the last said"),
Err(n) => format!(
"{AUDIT_TARGET}: control call {path} refused again ({reason}) — the caller's key {caller}; {n} so far this minute, said loudly once a minute"
),
}
}
/// One loud refusal per caller and reason per this long.
pub const REFUSAL_SAY_EVERY: std::time::Duration = std::time::Duration::from_secs(60);
/// **The one gate every service sits behind**: a caller [`layer`] verified
/// passes; anyone else presents the bearer (`nornir_service`'s rule: empty
/// token ⇒ open, a dev deployment).
pub fn interceptor(
token: Arc<String>,
) -> impl Fn(Request<()>) -> Result<Request<()>, Status> + Clone {
let bearer = nornir_service::auth_interceptor(token);
move |req: Request<()>| {
let mut req = if req.extensions().get::<ControlCaller>().is_some() {
req
} else {
bearer(req)?
};
req.extensions_mut().insert(crate::auth::Caller::Door);
Ok(req)
}
}
#[cfg(test)]
mod refusal_tests {
use super::*;
use std::time::{Duration, Instant};
/// ★ A caller retrying with a key this monetize was not fed is said at
/// WARN once a minute (the line says `REFUSED`, which the front's stderr
/// capture lifts to WARN), the rest at INFO (`refused again`) with the
/// count; a NEW reason from the same caller is said at WARN at once.
#[test]
fn a_retried_refusal_is_warn_once_a_minute_and_a_new_reason_at_once() {
let t0 = Instant::now();
let caller = "SHA256:test-a-retried-refusal";
let lines: Vec<String> = (0..14u64)
.map(|i| {
refusal_line(
caller,
"not-verified",
"/monetize.v1.Products/List",
t0 + Duration::from_secs(i * 9),
)
})
.collect();
let loud: Vec<usize> = lines
.iter()
.enumerate()
.filter(|(_, l)| l.contains("REFUSED"))
.map(|(i, _)| i)
.collect();
assert_eq!(loud, vec![0, 7], "one WARN a minute: {lines:#?}");
assert!(
lines[1].contains("refused again")
&& lines[1].contains("not-verified")
&& lines[1].contains(caller),
"{}",
lines[1]
);
assert!(
lines[7].contains("6 more went quietly"),
"the next WARN names how many went quietly: {}",
lines[7]
);
let changed = refusal_line(
caller,
"replayed",
"/monetize.v1.Products/List",
t0 + Duration::from_secs(70),
);
assert!(
changed.contains("REFUSED (replayed)"),
"a new reason is said at WARN at once: {changed}"
);
assert!(
refusal_line("SHA256:another-caller", "not-verified", "/x", t0).contains("REFUSED"),
"callers are counted apart"
);
}
}