An agent's queue credential could subscribe to every subject in the swarm account, so any agent could read every other agent's terminal, turn state, operator input and request replies. Agents are the least trusted clients on the queue (mara's ruling on #4901), so their grants now carry a subscribe allow-list: - an agent's own credential (the harness and its subagent daemon): `_INBOX.agent.<agent>.>` and `$SWARM.input.<agent>` - the hive's shared agent client: `_INBOX.client.<client-id>.>` Hive and swarm-controller grants are unchanged (no `sub` claim, so still every subject). The clients connect with the matching inbox prefix, so the replies they rely on still arrive: the icon write's JetStream ack and the schedule DIRECT.GET. `connect_with_token` takes the agent name and sets the prefix itself; the hive-shared fallback uses the new `connect_with_inbox_prefix`. The harness no longer subscribes to its input subject when it fell back to the hive's shared client: that identity names no agent, so it cannot be granted one agent's input without all of them. It logs that and skips, the same as the schedule reader on that path. Refs #4901
452 lines
18 KiB
Rust
452 lines
18 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>,
|
|
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>,
|
|
}
|
|
|
|
/// 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,
|
|
publish: Vec<String>,
|
|
subscribe: Option<Vec<String>>,
|
|
) -> 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: publish },
|
|
subscribe: subscribe.map(|allow| Permission { allow }),
|
|
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.publish.clone(),
|
|
permissions.subscribe.clone(),
|
|
);
|
|
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,
|
|
}
|
|
}
|
|
|
|
#[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 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")
|
|
}
|
|
}
|