Watch
0
0
Fork
You've already forked hyperhive
0
hyperhive/swarm-nats-auth/src/respond.rs
atlas 98c873965b swarm-nats-auth: a review classifier hears try-outs and answers by reply only
swarm-queue-client gains review_classifier: the try-out subject, the
request and reply that travel on it, and the deadlines both ends share.
The controller's reader grant may publish the subject; a client named
with --review-classifier-client may subscribe to it and its inbox, and
answers under NATS allow_responses (resp in the user JWT) within the
run deadline. Every other client id is still denied.
2026-10-12 13:11:21 +02:00

494 lines
19 KiB
Rust

//! Minting the reply the server expects on `msg.reply`.
//!
//! Both JWTs are hand-built because neither is expressible through
//! `nats-jwt`: its `Claims` has no `aud`, which the `authorization_response`
//! wrapper needs (the asking server's id, so a captured reply cannot be
//! replayed elsewhere in the cluster) and the user token needs (the account
//! name). Its `IntoNatsClaims` also returns a closed
//! `NatsClaims { User, Account }`.
//!
//! ⛔ **`issuer_account` must be ABSENT from the user token.** It is an
//! operator-mode field, and `nix/host-modules/swarm-nats.nix` renders
//! *server-config* mode, where its mere presence makes the server refuse the
//! client — `Error non operator mode account "AUTH": attempted to use
//! issuer_account`, with the responder having answered `granted=true`.
//! `nats_jwt::Token::new_user` always sets it, which is why reaching for that
//! constructor again would reintroduce the bug.
//!
//! Signing follows `nats-jwt`'s own `sign()`: serialise the claims with `jti`
//! **empty**, sha256 that, `BASE32HEX_NOPAD` the digest into `jti`,
//! re-serialise, sign `"<b64url header>.<b64url claims>"` with the account
//! key. ⚠️ `BASE32HEX`, not `BASE32` — one word, and the only symptom is a
//! token the server rejects without saying why.
//!
//! `nats-jwt` is therefore a **dev-dependency, not a runtime one**: it is this
//! module's test oracle: `tests::hand_built_matches_the_reference` builds a
//! user token both ways and requires byte equality, so the encoder is checked
//! against the reference on the one shape that crate does model.
use data_encoding::{BASE32HEX_NOPAD, BASE64URL_NOPAD};
use nkeys::KeyPair;
use serde::Serialize;
use sha2::{Digest, Sha256};
/// Header every NATS JWT carries. Byte-identical to `nats-jwt`'s, and the
/// equality test would catch it drifting.
const JWT_HEADER: &str = r#"{"typ":"JWT","alg":"ed25519-nkey"}"#;
/// A NATS JWT claim set, with `aud` — which the callout response needs and
/// `nats-jwt`'s own `Claims` lacks.
#[derive(Serialize)]
struct Claims<T: Serialize> {
iat: i64,
iss: String,
jti: String,
sub: String,
name: String,
#[serde(skip_serializing_if = "Option::is_none")]
aud: Option<String>,
nats: T,
}
/// The `nats` object of an `authorization_response`.
///
/// Exactly one of `jwt` / `error` is set. They are `Option`s rather than an
/// enum so the serialised shape matches the wire format directly, and the
/// constructors below are the only way to build one — a caller cannot
/// accidentally produce a response that carries both, which a server would be
/// free to read either way.
#[derive(Serialize)]
struct AuthResponse {
#[serde(skip_serializing_if = "Option::is_none")]
jwt: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
error: Option<String>,
#[serde(rename = "type")]
kind: &'static str,
version: i64,
}
/// Sign a claim set the way `nats-jwt` does. `now` is injected so the test
/// can pin it; production passes the wall clock.
fn sign<T: Serialize>(mut claims: Claims<T>, key: &KeyPair) -> String {
claims.jti = String::new();
let unhashed = serde_json::to_string(&claims).expect("claims serialisation cannot fail");
let digest = Sha256::digest(unhashed.as_bytes());
claims.jti = BASE32HEX_NOPAD.encode(&digest);
let body = serde_json::to_string(&claims).expect("claims serialisation cannot fail");
let half = format!(
"{}.{}",
BASE64URL_NOPAD.encode(JWT_HEADER.as_bytes()),
BASE64URL_NOPAD.encode(body.as_bytes())
);
let sig = key
.sign(half.as_bytes())
.expect("ed25519 signing cannot fail");
format!("{half}.{}", BASE64URL_NOPAD.encode(&sig))
}
/// Wrap a decision in an `authorization_response` addressed to the server
/// that asked.
///
/// A denial is a *signed response carrying an error*, never silence: the
/// server cannot tell an absent responder from a refusing one, so staying
/// quiet turns every rejection into a timeout and hides an outage behind
/// what looks like normal denials.
fn response(
now: i64,
issuer: &KeyPair,
server_id: &str,
user_nkey: &str,
verdict: Result<String, String>,
) -> String {
let (jwt, error) = match verdict {
Ok(user_jwt) => (Some(user_jwt), None),
Err(reason) => (None, Some(reason)),
};
sign(
Claims {
iat: now,
iss: issuer.public_key(),
jti: String::new(),
sub: user_nkey.to_owned(),
name: user_nkey.to_owned(),
aud: Some(server_id.to_owned()),
nats: AuthResponse {
jwt,
error,
kind: "authorization_response",
version: 2,
},
},
issuer,
)
}
/// Current unix seconds, as the JWT `iat`.
fn now_secs() -> i64 {
i64::try_from(
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.expect("system time is after the unix epoch")
.as_secs(),
)
.expect("seconds since the epoch fit in an i64")
}
/// The `nats` object of a **user** JWT, server-config mode.
///
/// Field-for-field what `nats-jwt` emits **minus `issuer_account`**, whose
/// presence a non-operator server rejects outright (see the module docs). The
/// account is named by the enclosing claims' `aud` instead.
///
/// The `-1`s are NATS' "unlimited": no cap on subscriptions, payload size or
/// total data. Limits belong on the account, which the module owns; putting
/// them here would mean every issued token silently disagreed with the
/// deployed policy the moment that policy changed.
#[derive(Serialize)]
struct UserNats {
#[serde(rename = "type")]
kind: &'static str,
/// Subjects this client may publish to.
///
/// The server enforces what is minted here — measured, because a
/// non-operator server validates minted claims against its own config
/// rather than trusting them, and this responder's own history includes a
/// field the server rejected while the responder said `granted=true`.
#[serde(rename = "pub")]
publish: Permission,
/// Subjects this client may subscribe to; absent means every subject.
#[serde(rename = "sub", skip_serializing_if = "Option::is_none")]
subscribe: Option<Permission>,
/// The reply allowance; absent means none.
#[serde(rename = "resp", skip_serializing_if = "Option::is_none")]
responses: Option<ResponsePermission>,
subs: i64,
data: i64,
payload: i64,
bearer_token: bool,
version: i64,
}
/// A NATS permission block.
///
/// Only `allow` is modelled. NATS also takes `deny`, and a struct that has it
/// is a struct someone will use: an allow-list plus a deny-list has two places
/// deciding the same question, and the interaction between them is a thing to
/// remember rather than to read.
#[derive(Serialize)]
struct Permission {
allow: Vec<String>,
}
/// NATS' `allow_responses`: the client may publish `max` messages to the
/// reply subject of each request it receives, within `ttl` nanoseconds (Go's
/// `time.Duration` in `jwt/v2`'s `ResponsePermission`).
#[derive(Serialize)]
struct ResponsePermission {
max: i64,
ttl: i64,
}
impl ResponsePermission {
/// One answer per request, within `window`.
fn one_within(window: std::time::Duration) -> Self {
Self {
max: 1,
ttl: i64::try_from(window.as_nanos()).unwrap_or(i64::MAX),
}
}
}
/// Mint the user JWT an admitted client presents.
///
/// `account` is the account **name** from the server's `accounts` block (the
/// module's `clientAccount`), not a public key — in config mode the server
/// resolves `aud` against its own config rather than against a key.
fn user_jwt(
now: i64,
issuer: &KeyPair,
account: &str,
user_nkey: &str,
permissions: &crate::policy::Permissions,
) -> String {
sign(
Claims {
iat: now,
iss: issuer.public_key(),
jti: String::new(),
sub: user_nkey.to_owned(),
name: user_nkey.to_owned(),
aud: Some(account.to_owned()),
nats: UserNats {
kind: "user",
publish: Permission {
allow: permissions.publish.clone(),
},
subscribe: permissions
.subscribe
.clone()
.map(|allow| Permission { allow }),
responses: permissions.responses.map(ResponsePermission::one_within),
subs: -1,
data: -1,
payload: -1,
bearer_token: false,
version: 2,
},
},
issuer,
)
}
/// Grant: mint a user JWT placing the client in `account`, scoped to
/// `permissions`, and wrap it.
///
/// Taking the permissions by value rather than defaulting them is the point:
/// there is no way to call this and get an unscoped grant by omission, so a
/// future caller cannot forget the argument that makes the scoping real.
pub fn grant(
issuer: &KeyPair,
account: &str,
server_id: &str,
user_nkey: &str,
permissions: &crate::policy::Permissions,
) -> String {
let now = now_secs();
let jwt = user_jwt(now, issuer, account, user_nkey, permissions);
response(now, issuer, server_id, user_nkey, Ok(jwt))
}
/// Deny, with a reason the server can log.
///
/// The reason is deliberately coarse (`"unauthorized"`), not a description of
/// *why*: this string reaches an unauthenticated peer, and a precise one turns
/// the auth path into an oracle for which tokens exist.
pub fn deny(issuer: &KeyPair, server_id: &str, user_nkey: &str) -> String {
response(
now_secs(),
issuer,
server_id,
user_nkey,
Err("unauthorized".to_owned()),
)
}
#[cfg(test)]
mod tests {
use super::*;
/// Mirror of what `nats-jwt` puts in a user token, **in its field order**.
///
/// Deliberately *not* [`super::UserNats`]: this one carries
/// `issuer_account`, because the reference does. Production must not, and
/// keeping the two structs separate is what lets one test pin the encoder
/// against `nats-jwt` while another asserts production diverges from it in
/// exactly one field.
///
/// 🩸 The first version of the equality test round-tripped `nats` through
/// `serde_json::Value` and failed — not because the encoder was wrong, but
/// because `Value` is a `BTreeMap`, so re-serialising sorted the keys and
/// changed the bytes. The gate's *method* was the defect. Worth keeping the
/// scar: `jti` is a hash over serialised claims, so anything that reorders
/// fields between hashing and signing is a live hazard here, not merely a
/// test artefact.
#[derive(serde::Serialize)]
struct ReferenceUserNats {
#[serde(rename = "type")]
kind: &'static str,
issuer_account: String,
subs: i64,
data: i64,
payload: i64,
bearer_token: bool,
version: i64,
}
/// The gate that makes the hand-built encoder trustworthy: build a *user*
/// token with `nats-jwt` and with this module's `sign`, and require the
/// bytes to be identical.
///
/// It works because `nats-jwt`'s `sign()` derives `iat` from the wall
/// clock and everything else from its inputs, so signing the same claims
/// in the same second must produce the same string. If this module's
/// header, field order, `jti` alphabet or signing input ever drifts from
/// the reference, this fails — including the `BASE32HEX`-vs-`BASE32`
/// distinction, which is otherwise invisible until a server refuses a
/// token.
#[test]
fn hand_built_matches_the_reference() {
let account = KeyPair::new_account();
let user = KeyPair::new_user();
let reference =
nats_jwt::Token::new_user(account.public_key(), user.public_key()).sign(&account);
// Mirror what `nats-jwt` puts in a user token **as a struct**, in its
// field order.
//
// 🩸 The first version of this test round-tripped `nats` through
// `serde_json::Value` and failed — not because the encoder was wrong,
// but because `Value` is a `BTreeMap`, so re-serialising sorted the
// keys and changed the bytes. The gate's *method* was the defect, and
// it is worth keeping the scar: `jti` is a hash over serialised
// claims, so anything that reorders fields between hashing and
// signing is a live hazard here, not just a test artefact.
let decoded = decode_claims(&reference);
let mine = sign(
Claims {
iat: decoded["iat"].as_i64().expect("iat"),
iss: account.public_key(),
jti: String::new(),
sub: user.public_key(),
name: user.public_key(),
aud: None,
nats: ReferenceUserNats {
kind: "user",
issuer_account: account.public_key(),
subs: -1,
data: -1,
payload: -1,
bearer_token: false,
version: 2,
},
},
&account,
);
assert_eq!(mine, reference, "hand-built encoder drifted from nats-jwt");
}
#[test]
fn a_denial_is_signed_and_carries_no_jwt() {
let account = KeyPair::new_account();
let token = deny(&account, "NSERVER", "UCLIENT");
let claims = decode_claims(&token);
assert_eq!(claims["nats"]["type"], "authorization_response");
assert_eq!(claims["nats"]["error"], "unauthorized");
assert!(
claims["nats"]["jwt"].is_null(),
"a denial must not carry a user jwt"
);
// `aud` binds the reply to the asking server; without it a captured
// response is replayable at any other server in the cluster.
assert_eq!(claims["aud"], "NSERVER");
}
/// The permissions a test grant carries. Any non-empty set will do for
/// the wrapper-shape assertions; `policy.rs` owns what the real ones are.
fn perms() -> crate::policy::Permissions {
crate::policy::Permissions {
publish: vec!["$KV.hive-status.alpha".to_owned()],
subscribe: None,
responses: None,
}
}
#[test]
fn the_issued_user_jwt_carries_the_permissions_it_was_given() {
// The scoping is only real if it survives into the minted token. A
// policy that computes the right subjects and a grant that drops them
// look identical from every test that stops at the policy.
let account = KeyPair::new_account();
let wrapper = decode_claims(&grant(&account, "APP", "NSERVER", "UCLIENT", &perms()));
let user = decode_claims(wrapper["nats"]["jwt"].as_str().expect("a user jwt"));
assert_eq!(user["nats"]["pub"]["allow"][0], "$KV.hive-status.alpha");
// An empty or absent allow-list is NATS' "everything": the one shape
// that turns this whole change into a no-op while every other
// assertion still passes.
assert!(
user["nats"]["pub"]["allow"]
.as_array()
.is_some_and(|a| !a.is_empty()),
"an empty pub.allow is an unscoped grant"
);
}
#[test]
fn the_issued_user_jwt_carries_a_subscribe_allow_list_only_when_given_one() {
let account = KeyPair::new_account();
let scoped = crate::policy::Permissions {
subscribe: Some(vec!["_INBOX.agent.atlas.>".to_owned()]),
..perms()
};
let wrapper = decode_claims(&grant(&account, "APP", "NSERVER", "UCLIENT", &scoped));
let user = decode_claims(wrapper["nats"]["jwt"].as_str().expect("a user jwt"));
assert_eq!(
user["nats"]["sub"]["allow"],
serde_json::json!(["_INBOX.agent.atlas.>"])
);
let wrapper = decode_claims(&grant(&account, "APP", "NSERVER", "UCLIENT", &perms()));
let user = decode_claims(wrapper["nats"]["jwt"].as_str().expect("a user jwt"));
assert!(user["nats"]["sub"].is_null(), "{}", user["nats"]);
}
#[test]
fn the_issued_user_jwt_carries_a_reply_allowance_only_when_given_one() {
let account = KeyPair::new_account();
let answering = crate::policy::Permissions {
responses: Some(std::time::Duration::from_mins(31)),
..perms()
};
let wrapper = decode_claims(&grant(&account, "APP", "NSERVER", "UCLIENT", &answering));
let user = decode_claims(wrapper["nats"]["jwt"].as_str().expect("a user jwt"));
// The names and units nats-server's `jwt/v2` reads: `max` messages,
// `ttl` as a Go duration in nanoseconds.
assert_eq!(
user["nats"]["resp"],
serde_json::json!({ "max": 1, "ttl": 31 * 60 * 1_000_000_000_i64 })
);
let wrapper = decode_claims(&grant(&account, "APP", "NSERVER", "UCLIENT", &perms()));
let user = decode_claims(wrapper["nats"]["jwt"].as_str().expect("a user jwt"));
assert!(user["nats"]["resp"].is_null(), "{}", user["nats"]);
}
#[test]
fn a_grant_carries_a_user_jwt_and_no_error() {
let account = KeyPair::new_account();
let claims = decode_claims(&grant(&account, "APP", "NSERVER", "UCLIENT", &perms()));
assert!(claims["nats"]["error"].is_null());
assert!(
claims["nats"]["jwt"]
.as_str()
.is_some_and(|j| j.contains('.'))
);
}
/// 🩸 THE REGRESSION TEST FOR THE BUG THE UNIT TESTS COULD NOT SEE.
///
/// The first version of this module minted the user JWT with
/// `nats_jwt::Token::new_user`, which always sets `issuer_account`. Every
/// test passed, the responder logged `granted=true`, and the server still
/// refused every client: `Error non operator mode account "AUTH":
/// attempted to use issuer_account`.
///
/// The lesson is in what this asserts — an **absence**. The suite was full
/// of assertions about fields that had to be *present*, and a field that
/// must not exist is invisible to all of them. It took running a real
/// server to find, so the cheap check goes here to keep it found.
#[test]
fn the_issued_user_jwt_names_its_account_by_aud_and_sets_no_issuer_account() {
let account = KeyPair::new_account();
let wrapper = decode_claims(&grant(&account, "APP", "NSERVER", "UCLIENT", &perms()));
let user = decode_claims(wrapper["nats"]["jwt"].as_str().expect("a user jwt"));
// In server-config mode the account is named by `aud`...
assert_eq!(user["aud"], "APP", "the user jwt must name its account");
// ...and `issuer_account` is an operator-mode field whose mere
// presence makes the server reject the client.
assert!(
user["nats"]["issuer_account"].is_null(),
"issuer_account is fatal in non-operator mode"
);
assert_eq!(user["nats"]["type"], "user");
assert_eq!(user["sub"], "UCLIENT");
assert_eq!(user["iss"], account.public_key());
}
fn decode_claims(jwt: &str) -> serde_json::Value {
let body = jwt.split('.').nth(1).expect("claims segment");
serde_json::from_slice(&BASE64URL_NOPAD.decode(body.as_bytes()).expect("base64url"))
.expect("claims json")
}
}