This is an automated email from the ASF dual-hosted git repository.
numinnex pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 3c935261e feat(server-ng): verify trusted-issuer a2a JWT and expose
SDK refresh (#3626)
3c935261e is described below
commit 3c935261ea71515ce68349e1c111fdd9a0876128
Author: Hubert Gruszecki <[email protected]>
AuthorDate: Fri Jul 10 13:17:34 2026 +0200
feat(server-ng): verify trusted-issuer a2a JWT and expose SDK refresh
(#3626)
The HTTP listener accepted only self-issued HS256 tokens and
silently ignored [[http.jwt.trusted_issuers]]. Port the legacy
JWKS client (RS256/EC, kid rotation) and verify external a2a
tokens against trusted issuers, remapping sub to the configured
user. Refresh of trusted-issuer tokens is rejected (they have
their own lifecycle) and a trusted issuer mapped to root now
fails at startup.
Harden the legacy server's JWKS client with the same pre-auth
amplification guard (per-issuer single-flight plus a 10s
rate-limit / negative-cache window) so an unknown-key-id flood
cannot fan out to one outbound fetch per request. Both shipped
configs document the opt-in trusted_issuers block; legacy no
longer ships an active example issuer.
Also expose the SDK's dead HTTP refresh_access_token, fixing a
read-guard self-deadlock in it, and cover both paths with
integration tests.
---
Cargo.lock | 2 +
.../src/harness/orchestrator/harness.rs | 13 +
core/integration/tests/sdk/http_refresh.rs | 63 +++++
core/integration/tests/sdk/mod.rs | 2 +
core/integration/tests/server/a2a_jwt/jwt_tests.rs | 195 ++++++++++++-
core/integration/tests/server/mod.rs | 5 +-
core/sdk/src/http/http_client.rs | 67 +++--
core/sdk/src/http/http_transport.rs | 6 +-
core/server-ng/Cargo.toml | 1 +
core/server-ng/LISTENER_SUPPORT_PLAN.md | 286 -------------------
core/server-ng/config.toml | 23 +-
core/server-ng/src/http.rs | 9 +-
core/server-ng/src/http/extractor.rs | 44 +--
core/server-ng/src/http/handlers.rs | 16 +-
.../src/http/jwt => server-ng/src/http}/jwks.rs | 150 ++++++----
core/server-ng/src/http/jwt.rs | 310 +++++++++++++++++++--
core/server/Cargo.toml | 1 +
core/server/config.toml | 23 +-
core/server/src/http/jwt/jwks.rs | 86 ++++--
19 files changed, 842 insertions(+), 460 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index d678d7145..e5ed364e9 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -11739,6 +11739,7 @@ dependencies = [
"sysinfo 0.39.5",
"tempfile",
"thiserror 2.0.18",
+ "tokio",
"toml 1.1.2+spec-1.1.0",
"tower-http 0.7.0",
"tracing",
@@ -11808,6 +11809,7 @@ dependencies = [
"secrecy",
"send_wrapper",
"serde",
+ "serde_json",
"server_common",
"shard",
"shard_allocator",
diff --git a/core/integration/src/harness/orchestrator/harness.rs
b/core/integration/src/harness/orchestrator/harness.rs
index ece57d83c..31320235d 100644
--- a/core/integration/src/harness/orchestrator/harness.rs
+++ b/core/integration/src/harness/orchestrator/harness.rs
@@ -364,6 +364,19 @@ impl TestHarness {
&self.servers
}
+ /// Number of requests the trusted-issuer JWKS mock has served, or 0 when
no
+ /// JWKS mock is configured. Lets a test bound the server's outbound JWKS
+ /// fetches (e.g. assert an unknown-`kid` flood does not amplify).
+ pub async fn jwks_request_count(&self) -> usize {
+ match &self.jwks_server {
+ Some(server) => server
+ .received_requests()
+ .await
+ .map_or(0, |reqs| reqs.len()),
+ None => 0,
+ }
+ }
+
/// Get the number of server nodes (1 for single server, N for cluster).
pub fn cluster_size(&self) -> usize {
self.servers.len()
diff --git a/core/integration/tests/sdk/http_refresh.rs
b/core/integration/tests/sdk/http_refresh.rs
new file mode 100644
index 000000000..98236fb38
--- /dev/null
+++ b/core/integration/tests/sdk/http_refresh.rs
@@ -0,0 +1,63 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! End-to-end coverage for the SDK HTTP client's `refresh_access_token`
+//! against a live server-ng listener: the reissued token must replace the one
+//! the client holds and keep it authenticated.
+
+use iggy::http::http_client::HttpClient;
+use iggy::prelude::*;
+use integration::iggy_harness;
+
+#[iggy_harness]
+async fn
given_logged_in_http_client_when_refreshing_should_swap_to_a_working_token(
+ harness: &TestHarness,
+) {
+ let addr = harness
+ .server()
+ .http_addr()
+ .expect("HTTP transport not configured on test server");
+ let client = HttpClient::new(&format!("http://{addr}";)).expect("build SDK
HTTP client");
+
+ let logged_in = client
+ .login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD)
+ .await
+ .expect("login as root");
+ let login_token = logged_in.access_token.expect("login returns a token");
+
+ let refreshed = client
+ .refresh_access_token()
+ .await
+ .expect("refresh the access token");
+ let refreshed_token = refreshed.access_token.expect("refresh returns a
token");
+
+ assert_eq!(
+ refreshed.user_id, logged_in.user_id,
+ "refresh must re-issue for the same user"
+ );
+ assert_ne!(
+ refreshed_token.token, login_token.token,
+ "refresh must mint a token distinct from the one it replaced"
+ );
+
+ // The client swapped its stored bearer to the reissued token; an authed
+ // call carries that stored token, so success proves the swap landed.
+ client
+ .get_stats()
+ .await
+ .expect("authed call after refresh must use the reissued token");
+}
diff --git a/core/integration/tests/sdk/mod.rs
b/core/integration/tests/sdk/mod.rs
index 70b102c46..0cd484a18 100644
--- a/core/integration/tests/sdk/mod.rs
+++ b/core/integration/tests/sdk/mod.rs
@@ -17,6 +17,8 @@
mod hello_world;
#[cfg(feature = "vsr")]
+mod http_refresh;
+#[cfg(feature = "vsr")]
mod mcp_parity;
#[cfg(feature = "vsr")]
mod messages;
diff --git a/core/integration/tests/server/a2a_jwt/jwt_tests.rs
b/core/integration/tests/server/a2a_jwt/jwt_tests.rs
index 767372cb0..4d0b2b194 100644
--- a/core/integration/tests/server/a2a_jwt/jwt_tests.rs
+++ b/core/integration/tests/server/a2a_jwt/jwt_tests.rs
@@ -15,7 +15,12 @@
// specific language governing permissions and limitations
// under the License.
-use iggy::prelude::{GlobalPermissions, IggyClientBuilder, Permissions,
UserStatus};
+use std::sync::Arc;
+
+use iggy::http::http_client::HttpClient;
+use iggy::prelude::{
+ GlobalPermissions, HttpClientConfig, IggyClientBuilder, Permissions,
UserStatus,
+};
use iggy_common::{StreamClient, UserClient};
use integration::iggy_harness;
use jsonwebtoken::{Algorithm, EncodingKey, Header, encode};
@@ -169,6 +174,52 @@ fn create_unknown_issuer_jwt() -> String {
encode(&header, &claims, &encoding_key).unwrap()
}
+/// Craft an algorithm-confusion token: it claims the trusted issuer and
carries
+/// the real JWKS `kid`, but is HS256 (symmetric) and signed with an
+/// attacker-controlled secret the server never shares. The issuer publishes
only
+/// an RSA key, so verifying an HMAC signature against it must fail closed
rather
+/// than treat the RSA public-key bytes as an HMAC secret.
+fn create_algorithm_confusion_jwt() -> String {
+ let now = now_timestamp();
+ let claims = TestClaims {
+ jti: uuid::Uuid::now_v7().to_string(),
+ iss: TEST_ISSUER.to_string(),
+ aud: Audience::from(TEST_AUDIENCE),
+ sub: "external-a2a-user-123".to_string(),
+ exp: now + 3600,
+ iat: now,
+ nbf: now,
+ };
+
+ let mut header = Header::new(Algorithm::HS256);
+ header.kid = Some(TEST_KEY_ID.to_string());
+ let encoding_key =
EncodingKey::from_secret(b"attacker-controlled-hmac-secret");
+
+ encode(&header, &claims, &encoding_key).unwrap()
+}
+
+/// Creates a valid-shape RS256 token stamped with an arbitrary `kid`. Used to
+/// drive the unknown-key-id path: the issuer is trusted but no JWKS entry
+/// matches, so verification reaches the fetch/rate-limit path and then fails.
+fn create_jwt_with_kid(kid: &str) -> String {
+ let now = now_timestamp();
+ let claims = TestClaims {
+ jti: uuid::Uuid::now_v7().to_string(),
+ iss: TEST_ISSUER.to_string(),
+ aud: Audience::from(TEST_AUDIENCE),
+ sub: "external-a2a-user-123".to_string(),
+ exp: now + 3600,
+ iat: now,
+ nbf: now,
+ };
+
+ let mut header = Header::new(Algorithm::RS256);
+ header.kid = Some(kid.to_string());
+ let encoding_key = EncodingKey::from_rsa_pem(TEST_PRIVATE_KEY).unwrap();
+
+ encode(&header, &claims, &encoding_key).unwrap()
+}
+
/// Create an IggyClient with the provided JWT token
async fn create_client_with_jwt(http_addr: &str, token: String) ->
iggy::prelude::IggyClient {
IggyClientBuilder::new()
@@ -291,6 +342,107 @@ async fn test_a2a_jwt_unknown_issuer(harness:
&TestHarness) {
);
}
+/// Test that an algorithm-confusion token is rejected: HS256-signed but
claiming
+/// the trusted issuer and the real JWKS `kid`. The issuer's key set holds
only an
+/// RSA key, so an HMAC signature must never verify against it - proving end
to end
+/// that the RSA public key can never be misused as an HMAC secret.
+#[iggy_harness(
+ server(config_path = "tests/server/a2a_jwt/config.toml"),
+ jwks_server(store_path =
"tests/server/a2a_jwt/wiremock/__files/jwks.json"),
+ seed = seed_a2a_user
+)]
+async fn test_a2a_jwt_algorithm_confusion(harness: &TestHarness) {
+ let server = harness
+ .all_servers()
+ .first()
+ .expect("server should be available");
+ let http_addr = server
+ .http_addr()
+ .expect("http address should be available");
+
+ // First prove the JWKS path resolves this issuer and kid: a valid RS256
token
+ // carrying the same kid must authenticate, warming the RSA key into cache.
+ // Only then can the rejection below be attributed to the HMAC-vs-RSA
+ // algorithm check rather than an incidental JWKS-fetch failure, which
would
+ // also surface as Unauthenticated and pass the assertion for the wrong
reason.
+ let valid_client = create_client_with_jwt(&http_addr.to_string(),
create_valid_jwt(3600)).await;
+ assert!(
+ valid_client.get_streams().await.is_ok(),
+ "valid RS256 token must authenticate (JWKS reachable, kid resolvable)"
+ );
+
+ let token = create_algorithm_confusion_jwt();
+ let client = create_client_with_jwt(&http_addr.to_string(), token).await;
+
+ // get_streams() should fail with Unauthenticated error
+ let result = client.get_streams().await;
+ assert!(
+ result.is_err(),
+ "algorithm-confusion token must be rejected, got {:?}",
+ result
+ );
+ let err = result.unwrap_err();
+ assert_eq!(
+ err.as_code(),
+ iggy::prelude::IggyError::Unauthenticated.as_code(),
+ "Expected Unauthenticated error, got {:?}",
+ err
+ );
+}
+
+/// Test that a valid trusted-issuer (A2A) token cannot be refreshed. The same
+/// token authenticates reads (see `test_a2a_jwt_valid_token`), but exchanging
it
+/// for a self-issued token would mint a bearer outliving the external grant
the
+/// issuer controls, so `POST /users/refresh-token` must reject it and issue
+/// nothing.
+#[iggy_harness(
+ server(config_path = "tests/server/a2a_jwt/config.toml"),
+ jwks_server(store_path =
"tests/server/a2a_jwt/wiremock/__files/jwks.json"),
+ seed = seed_a2a_user
+)]
+async fn test_a2a_jwt_refresh_rejected(harness: &TestHarness) {
+ let server = harness
+ .all_servers()
+ .first()
+ .expect("server should be available");
+ let http_addr = server
+ .http_addr()
+ .expect("http address should be available");
+
+ // Pre-set the A2A token as the client's bearer. `HttpClient`'s retry
+ // middleware bridges the HTTP-listener warmup the readiness gate does not.
+ let client = HttpClient::create(Arc::new(HttpClientConfig {
+ api_url: format!("http://{}";, http_addr),
+ jwt: Some(create_valid_jwt(3600)),
+ ..HttpClientConfig::default()
+ }))
+ .expect("failed to build HTTP client");
+
+ // Prove the token authenticates a normal request first: a successful read
+ // runs the same trusted-issuer JWKS verify, so the refresh rejection
below is
+ // the trusted-issuer refusal and not an incidental JWKS-fetch failure
(both
+ // collapse to Unauthenticated on the wire).
+ assert!(
+ client.get_streams().await.is_ok(),
+ "A2A token must authenticate a read before refresh is attempted"
+ );
+
+ // An `Ok` would mean a self-issued token was handed back - the regression.
+ let result = client.refresh_access_token().await;
+ assert!(
+ result.is_err(),
+ "trusted-issuer token must not be refreshable, got {:?}",
+ result
+ );
+ let err = result.unwrap_err();
+ assert_eq!(
+ err.as_code(),
+ iggy::prelude::IggyError::Unauthenticated.as_code(),
+ "Expected Unauthenticated error, got {:?}",
+ err
+ );
+}
+
/// Test that missing JWT token results in authentication failure
#[iggy_harness(
server(config_path = "tests/server/a2a_jwt/config.toml"),
@@ -327,3 +479,44 @@ async fn test_a2a_jwt_missing_token(harness: &TestHarness)
{
err
);
}
+
+/// A token that names the trusted issuer but carries an unknown `kid` reaches
the
+/// JWKS fetch path pre-signature. Replaying it must not amplify into one
upstream
+/// fetch per request: the per-issuer rate-limit / negative-cache window
collapses
+/// an unknown-key-id flood to at most one outbound JWKS request.
+#[iggy_harness(
+ server(config_path = "tests/server/a2a_jwt/config.toml"),
+ jwks_server(store_path = "tests/server/a2a_jwt/wiremock/__files/jwks.json")
+)]
+async fn test_a2a_jwt_unknown_kid_does_not_amplify_jwks_fetch(harness:
&TestHarness) {
+ let server = harness
+ .all_servers()
+ .first()
+ .expect("server should be available");
+ let http_addr = server
+ .http_addr()
+ .expect("http address should be available");
+
+ // The same unknown kid replayed: each request is a cache miss that
classifies
+ // the trusted issuer and would fetch its JWKS on every request if
unbounded.
+ let token = create_jwt_with_kid("novel-unknown-kid");
+ for _ in 0..8 {
+ let client = create_client_with_jwt(&http_addr.to_string(),
token.clone()).await;
+ let result = client.get_streams().await;
+ assert!(
+ result.is_err(),
+ "unknown-kid token must be rejected, got {:?}",
+ result
+ );
+ }
+
+ // Exactly one fetch: the first miss reads the issuer's key set, and every
+ // replay inside the refresh window is then served as a known-absent kid
+ // without touching the network. Without the fix this would be one fetch
per
+ // request. The `== 1` also proves the path was exercised (not a vacuous
0).
+ let fetches = harness.jwks_request_count().await;
+ assert_eq!(
+ fetches, 1,
+ "unknown-kid replay must collapse to a single JWKS fetch, got
{fetches}"
+ );
+}
diff --git a/core/integration/tests/server/mod.rs
b/core/integration/tests/server/mod.rs
index 90e2e58bf..4e88fb969 100644
--- a/core/integration/tests/server/mod.rs
+++ b/core/integration/tests/server/mod.rs
@@ -15,9 +15,8 @@
// specific language governing permissions and limitations
// under the License.
-// a2a_jwt exercises trusted-issuer (JWKS) tokens; server-ng's HTTP JWT
-// verifier has no trusted-issuer path.
-#[cfg(not(feature = "vsr"))]
+// a2a_jwt exercises trusted-issuer (JWKS) tokens; both the legacy verifier and
+// server-ng's ported trusted-issuer path verify them.
mod a2a_jwt;
mod cg;
// Flush (FLUSH_UNSAVED_BUFFER) has no server-ng primitive; it must deny typed.
diff --git a/core/sdk/src/http/http_client.rs b/core/sdk/src/http/http_client.rs
index f04c355f0..5cd4bb413 100644
--- a/core/sdk/src/http/http_client.rs
+++ b/core/sdk/src/http/http_client.rs
@@ -233,30 +233,6 @@ impl HttpTransport for HttpClient {
!token.is_empty()
}
- /// Refresh the access token using the current access token.
- // TODO(hubcio): method `refresh_access_token` is never used
- async fn _refresh_access_token(&self) -> Result<(), IggyError> {
- let token = self.access_token.read().await;
- if token.is_empty() {
- return Err(IggyError::AccessTokenMissing);
- }
-
- let command = _RefreshToken {
- token: token.to_owned(),
- };
- let response = self.post("/users/refresh-token", &command).await?;
- let identity_info: IdentityInfo = response
- .json()
- .await
- .map_err(|_| IggyError::InvalidJsonResponse)?;
- if identity_info.access_token.is_none() {
- return Err(IggyError::JwtMissing);
- }
-
- self.set_token_from_identity(&identity_info).await?;
- Ok(())
- }
-
/// Set the access token.
async fn set_access_token(&self, token: Option<String>) {
let mut current_token = self.access_token.write().await;
@@ -321,6 +297,46 @@ impl HttpClient {
))
}
+ /// Present the stored access token to `POST /users/refresh-token`, then
+ /// swap it for the reissued one. Returns the new identity so the caller
can
+ /// schedule the next refresh from `IdentityInfo.access_token.expiry`
+ /// (unix seconds). Scheduling is the caller's job: no auto-refresh or
+ /// retry-on-401 happens anywhere in the request path.
+ ///
+ /// Server semantics differ and the caller must account for it:
+ /// - Legacy server: one-shot. The presented token is revoked as it is
+ /// consumed, so a concurrent in-flight request still carrying the old
+ /// token may fail with 401.
+ /// - server-ng: stateless. The old token stays valid until its natural
+ /// expiry; refreshing never revokes it.
+ pub async fn refresh_access_token(&self) -> Result<IdentityInfo,
IggyError> {
+ // Release the read guard before `set_token_from_identity` takes the
+ // write guard on the same lock, otherwise the reissue self-deadlocks.
+ let current_token = {
+ let token = self.access_token.read().await;
+ if token.is_empty() {
+ return Err(IggyError::AccessTokenMissing);
+ }
+ token.to_owned()
+ };
+
+ let response = self
+ .post(
+ "/users/refresh-token",
+ &RefreshToken {
+ token: current_token,
+ },
+ )
+ .await?;
+ let identity_info: IdentityInfo = response
+ .json()
+ .await
+ .map_err(|_| IggyError::InvalidJsonResponse)?;
+
+ self.set_token_from_identity(&identity_info).await?;
+ Ok(identity_info)
+ }
+
async fn handle_response(response: Response) -> Result<Response,
IggyError> {
let status = response.status();
match status.is_success() {
@@ -357,12 +373,11 @@ impl HttpClient {
}
#[derive(Debug, Serialize)]
-struct _RefreshToken {
+struct RefreshToken {
token: String,
}
/// Unit tests for HttpClient.
-/// Currently only tests for "from_connection_string()" are implemented.
/// TODO: Add complete unit tests for HttpClient.
#[cfg(test)]
mod tests {
diff --git a/core/sdk/src/http/http_transport.rs
b/core/sdk/src/http/http_transport.rs
index b651009d6..428acda38 100644
--- a/core/sdk/src/http/http_transport.rs
+++ b/core/sdk/src/http/http_transport.rs
@@ -72,13 +72,9 @@ pub trait HttpTransport {
/// Returns true if the client is authenticated.
async fn is_authenticated(&self) -> bool;
- /// Refresh the access token using the provided refresh token.
- //method `refresh_access_token` is never used
- async fn _refresh_access_token(&self) -> Result<(), IggyError>;
-
/// Set the access token.
async fn set_access_token(&self, token: Option<String>);
- /// Set the access token and refresh token from the provided identity.
+ /// Set the access token from the provided identity.
async fn set_token_from_identity(&self, identity: &IdentityInfo) ->
Result<(), IggyError>;
}
diff --git a/core/server-ng/Cargo.toml b/core/server-ng/Cargo.toml
index 9869a1b96..26f397d00 100644
--- a/core/server-ng/Cargo.toml
+++ b/core/server-ng/Cargo.toml
@@ -146,6 +146,7 @@ rustls-pemfile = { workspace = true }
secrecy = { workspace = true }
send_wrapper = { workspace = true }
serde = { workspace = true }
+serde_json = { workspace = true }
server_common = { workspace = true }
shard = { workspace = true }
shard_allocator = { workspace = true }
diff --git a/core/server-ng/LISTENER_SUPPORT_PLAN.md
b/core/server-ng/LISTENER_SUPPORT_PLAN.md
deleted file mode 100644
index e2c4210e0..000000000
--- a/core/server-ng/LISTENER_SUPPORT_PLAN.md
+++ /dev/null
@@ -1,286 +0,0 @@
-# `server-ng` Listener Support Plan
-
-## Scope
-
-Add `server-ng` bootstrap support for the `message_bus` client listener planes
that were added upstream:
-
-- QUIC
-- WebSocket
-- TCP-TLS
-
-`HTTP` remains out of scope for this change.
-
-This plan is based on the current rebased tree, not on the intended end state.
-
-## What Exists Today
-
-### `server-ng` is still effectively TCP-only
-
-In [`src/bootstrap.rs`](./src/bootstrap.rs), `RunServerNg::run()` builds only:
-
-- `AcceptedReplicaFn`
-- `AcceptedClientFn`
-
-and then calls `start_tcp_runtime(...)`.
-
-The runtime split is:
-
-- `start_single_node_tcp_runtime(...)`
-- `start_cluster_tcp_runtime(...)`
-
-Both branches only wire plain TCP listeners today.
-
-### The new `message_bus` entry point already supports the extra planes
-
-[`core/message_bus/src/replica/io.rs`](../message_bus/src/replica/io.rs)
exposes:
-
-- `replica_io::start_on_shard_zero(...)`
-
-That function can bind:
-
-- replica TCP
-- client TCP
-- WS
-- QUIC
-- TCP-TLS
-- WSS
-
-and returns `BoundPlanes { replica, client, ws, quic, tcp_tls, wss }`.
-
-`server-ng` is still calling `start_on_shard_zero_default(...)`, which
hardcodes all optional listener planes to `None`.
-
-### `server-ng` is not yet using the `ServerNgConfig` / `message_bus` config
path
-
-This is the most important bootstrap mismatch in the current tree:
-
-- `load_config()` still loads `configs::server::ServerConfig`
-- `bootstrap()` still constructs the bus with `IggyMessageBus::new(SHARD_ID)`
-
-That means:
-
-- the `[message_bus]` config section is not being consumed
-- bus WS tuning is ignored
-- bus reconnect / handshake / queue tunables are still defaulted
-- QUIC tuning from `ServerNgConfig` is not flowing into the bus constructor
-
-Because `IggyMessageBus::with_config(...)` expects
`configs::server_ng::ServerNgConfig`, listener support should be implemented
together with config wiring.
-
-## Recommended Implementation Shape
-
-### 1. Move `server-ng` bootstrap to `ServerNgConfig`
-
-Update `core/server-ng` to load and pass around
`configs::server_ng::ServerNgConfig` instead of the legacy `ServerConfig`.
-
-Why this should happen first:
-
-- `message_bus` production constructor is
`IggyMessageBus::with_config(shard_id, cfg)`
-- the QUIC listener needs bus-side QUIC tuning
-- the WS listener needs bus-side WS handshake config
-- the repo already has a dedicated `server_ng_config` schema for exactly this
wiring
-
-Expected touch points:
-
-- [`src/bootstrap.rs`](./src/bootstrap.rs)
-- [`src/config_writer.rs`](./src/config_writer.rs)
-- any `server-ng` call sites currently typed to `ServerConfig`
-
-### 2. Replace `IggyMessageBus::new(...)` with
`IggyMessageBus::with_config(...)`
-
-In `bootstrap()`, build the bus from the validated `ServerNgConfig`.
-
-This makes the listener work use the correct runtime tunables immediately:
-
-- `message_bus.handshake_grace`
-- `message_bus.close_grace`
-- `message_bus.close_peer_timeout`
-- `message_bus.reconnect_period`
-- `message_bus.ws_*`
-- `quic.*` transport tuning
-
-### 3. Generalize the TCP-only bootstrap helpers into transport bootstrap
-
-The current names and signatures are too narrow:
-
-- `resolve_tcp_topology(...)`
-- `start_tcp_runtime(...)`
-- `start_cluster_tcp_runtime(...)`
-- `start_single_node_tcp_runtime(...)`
-
-Recommended refactor:
-
-- keep `TcpTopology` for replica/client TCP addressing if convenient
-- add a listener-settings helper that derives all optional listen addresses
from config
-- rename runtime helpers to reflect multi-transport startup rather than
TCP-only startup
-
-At minimum, bootstrap needs to derive:
-
-- replica TCP address
-- client TCP address
-- optional WS listen address from `[websocket]`
-- optional QUIC listen address from `[quic]`
-- optional TCP-TLS listen address
-
-Important detail:
-
-- TCP-TLS does not have its own address section
-- it is derived from `[tcp].address` when `[tcp.tls].enabled = true`
-
-Same pattern exists for WSS via `[websocket]` + `[websocket.tls]`, even if WSS
is not part of the first implementation slice.
-
-### 4. In cluster mode, switch from `start_on_shard_zero_default(...)` to
`start_on_shard_zero(...)`
-
-This is the main listener wiring change.
-
-Instead of:
-
-- `replica_io::start_on_shard_zero_default(...)`
-
-call:
-
-- `replica_io::start_on_shard_zero(...)`
-
-and populate the optional planes based on config:
-
-- `ws_listen_addr`
-- `quic_listen_addr`
-- `tcp_tls_listen_addr`
-- optionally `wss_listen_addr`
-
-Also provide the matching accept callbacks:
-
-- `AcceptedWsClientFn`
-- `AcceptedQuicClientFn`
-- `AcceptedTlsClientFn`
-- optionally `AcceptedWssClientFn`
-
-### 5. Add transport-specific accepted-client closures in `server-ng`
-
-`server-ng` already has:
-
-- `make_local_client_accept_fn(...)`
-
-Add transport-specific variants that mint `client_id` the same way and then
install through the correct `message_bus::installer` entry point.
-
-Expected closures:
-
-- plain TCP: `installer::install_client_tcp(...)` or existing wrapper path
-- WS: reuse the pre-upgrade local path, most likely by duping the accepted fd
and calling `ConnectionInstaller::install_client_ws_fd(...)`
-- QUIC: `installer::install_client_quic(...)`
-- TCP-TLS: `installer::install_client_tcp_tls(...)`
-
-Important nuance for WS:
-
-- upstream `message_bus` docs describe WS as a pre-upgrade TCP accept path
that can be fd-shipped to another shard
-- `server-ng` is single-shard today, so it can terminate WS locally on shard 0
-- the accepted callback receives raw `TcpStream`, so
`installer::install_client_ws(...)` is not directly callable there because it
expects a post-upgrade `WebSocketStream`
-- the clean reuse path is to keep using the existing pre-upgrade installer
flow (`install_client_ws_fd(...)`) locally on shard 0 rather than inventing a
second WS handshake path in `server-ng`
-
-Important nuance for QUIC / TCP-TLS:
-
-- these are shard-0 terminal by design in `message_bus`
-- do not try to thread them through a cross-shard setup path
-
-### 6. Add credential-loading helpers for TCP-TLS and QUIC
-
-Bootstrap needs to construct:
-
-- `message_bus::TlsServerCredentials`
-- `message_bus::replica_io::QuicServerCredentials`
-
-Reuse existing primitives where possible:
-
-- `message_bus::transports::tls::load_pem(...)`
-- `message_bus::transports::tls::self_signed_for_loopback()`
-- `message_bus::transports::tls::install_default_crypto_provider()`
-
-For QUIC, mirror the existing legacy-server behavior:
-
-- self-signed when configured and files are absent
-- otherwise load cert/key from configured paths
-
-This logic should live in `server-ng`, not be duplicated inside `message_bus`.
-
-### 7. Single-node mode should also use the same transport bootstrap path
-
-Right now single-node startup bypasses `replica_io` entirely and binds plain
TCP directly via `client_listener::bind(...)`.
-
-That should be replaced with one transport-aware path so cluster and
non-cluster mode do not diverge on feature support.
-
-Recommended shape:
-
-- cluster mode: bind replica + client planes through
`replica_io::start_on_shard_zero(...)`
-- single-node mode: either
- - reuse `start_on_shard_zero(...)` with a degenerate one-node topology, or
- - add a small `client_planes_only(...)` helper if the replica-plane
assumptions make that cleaner
-
-The important part is to avoid maintaining separate TCP-only and
multi-transport startup logic.
-
-## `current_config.toml` Follow-Up
-
-[`src/config_writer.rs`](./src/config_writer.rs) currently writes only:
-
-- `tcp.address`
-- `cluster.nodes[*].ports.tcp`
-- `cluster.nodes[*].ports.tcp_replica`
-
-That is insufficient once new listeners are enabled.
-
-Extend it to write the effective bound addresses for:
-
-- `quic.address`
-- `websocket.address`
-
-and update cluster node ports when relevant:
-
-- `ports.quic`
-- `ports.websocket`
-
-For TCP-TLS there is no separate top-level address field today, so there may
be nothing new to serialize beyond the shared TCP address unless the config
model changes.
-
-## Error Handling Changes
-
-[`src/server_error.rs`](./src/server_error.rs) still exposes:
-
-- `StartTcpListeners`
-
-That name is too narrow after this change. Rename or replace it with something
transport-neutral, for example:
-
-- `StartListeners`
-- `StartTransportListeners`
-
-Also expect new bootstrap errors for:
-
-- TLS credential loading
-- QUIC credential loading
-- invalid derived listener configuration
-
-## Proposed Execution Order
-
-1. Switch `server-ng` from `ServerConfig` to `ServerNgConfig`.
-2. Construct the bus with `IggyMessageBus::with_config(...)`.
-3. Add credential-loading helpers for TLS-family and QUIC listeners.
-4. Introduce transport-specific accepted-client closures.
-5. Replace `start_on_shard_zero_default(...)` with `start_on_shard_zero(...)`
in cluster mode.
-6. Unify single-node startup with the same multi-transport bootstrap path.
-7. Extend `write_current_config(...)` to include QUIC / WS bound addresses.
-8. Rename the TCP-specific listener-startup error variant and add any new
bootstrap errors.
-
-## Test Plan
-
-### Compile and unit-level checks
-
-- `cargo check -p server-ng`
-- `cargo check -p message_bus`
-- `cargo check -p integration`
-
-## First Implementation Slice
-
-To keep the change set controlled, the first slice should be:
-
-1. move to `ServerNgConfig`
-2. wire `IggyMessageBus::with_config(...)`
-3. add QUIC / WS / TCP-TLS listener startup in cluster mode
-4. mirror the same startup in single-node mode
-5. update `current_config.toml` for QUIC and WS
-
-WSS can be added immediately after if desired, because the bootstrap pattern
is the same once the transport-aware startup path exists.
diff --git a/core/server-ng/config.toml b/core/server-ng/config.toml
index 92638aa18..bd54c661d 100644
--- a/core/server-ng/config.toml
+++ b/core/server-ng/config.toml
@@ -126,11 +126,24 @@ decoding_secret = ""
# `false` means the secret is in plain text.
use_base64_secret = false
-# Trusted issuers for A2A (Application-to-Application) authentication
-[[http.jwt.trusted_issuers]]
-issuer = "test-issuer"
-jwks_url = "http://127.0.0.1:8081/.well-known/jwks.json";
-audience = "iggy.apache.org"
+# Trusted issuers for A2A (Application-to-Application) authentication. Opt-in:
+# with none configured the listener accepts only self-issued HS256 tokens.
+# `issuer`, `audience` and `jwks_url` are required per entry; `user_id` is
+# optional but defaults to 0 (root), which is rejected - set it to the non-zero
+# iggy user every token from that issuer is remapped onto.
+#
+# Operational note: enabling an issuer opens an outbound JWKS fetch that is
+# reachable before a token's signature is verified - a token naming this issuer
+# with an unknown key id can trigger a fetch to `jwks_url`. The target is fixed
+# (not attacker-chosen); concurrent misses coalesce onto one fetch and repeats
+# are rate-limited to at most one outbound request per issuer per short window,
+# so an unknown-key-id flood cannot amplify. The same window bounds how
quickly a
+# freshly rotated signing key is picked up.
+# [[http.jwt.trusted_issuers]]
+# issuer = "test-issuer"
+# jwks_url = "http://127.0.0.1:8081/.well-known/jwks.json";
+# audience = "iggy.apache.org"
+# user_id = 1
# Metrics configuration for HTTP.
[http.metrics]
diff --git a/core/server-ng/src/http.rs b/core/server-ng/src/http.rs
index 9ebace6e4..2593c4dad 100644
--- a/core/server-ng/src/http.rs
+++ b/core/server-ng/src/http.rs
@@ -17,15 +17,16 @@
//! Shard-0 HTTP/REST listener. This root binds the listener and assembles the
//! router; the rest is split across submodules: the `state` bridge and axum
-//! `State`, the bearer `extractor` and `jwt` issuer, the route `handlers`, the
-//! `reads` gates, the `submit` write paths, `wire` request mapping,
-//! partition-write `admission`, committed-reply `reply` decoding, the
rejection
-//! `error` types, and per-credential `session` state.
+//! `State`, the bearer `extractor`, the `jwt` issuer and its `jwks` resolver,
+//! the route `handlers`, the `reads` gates, the `submit` write paths, `wire`
+//! request mapping, partition-write `admission`, committed-reply `reply`
+//! decoding, the rejection `error` types, and per-credential `session` state.
mod admission;
mod error;
mod extractor;
mod handlers;
+mod jwks;
mod jwt;
mod reads;
mod reply;
diff --git a/core/server-ng/src/http/extractor.rs
b/core/server-ng/src/http/extractor.rs
index 2287ef012..bcd523658 100644
--- a/core/server-ng/src/http/extractor.rs
+++ b/core/server-ng/src/http/extractor.rs
@@ -61,12 +61,16 @@ impl FromRequestParts<HttpState> for Authenticated {
) -> Result<Self, Self::Rejection> {
let bearer = bearer_token(parts)?;
- let (key, user_id, expiry) = resolve_credential(state, bearer)?;
-
- // `resolve_session` is `Rc`-based and `!Send`, yet axum requires this
- // extractor future to be `Send`. `SendWrapper` bridges the gap: sound
- // because compio pins the future to shard 0's thread - the only thread
- // that ever touches the session table (mirrors legacy
`HttpSafeShard`).
+ // Both `resolve_credential` (its JWT verify may await a `!Send` JWKS
+ // fetch through cyper) and `resolve_session` (`Rc`-based, `!Send`)
must
+ // run, yet axum requires this extractor future to be `Send`.
+ // `SendWrapper` bridges each: sound because compio pins the future to
+ // shard 0's thread - the only thread the JWKS client and session table
+ // ever run on (mirrors legacy `HttpSafeShard`). Neither future holds a
+ // `RefCell` borrow or `DashMap` guard across the `.await` (each such
+ // critical section is synchronous), so a cooperatively-scheduled
sibling
+ // task on this thread never observes a borrowed session table.
+ let (key, user_id, expiry) =
SendWrapper::new(resolve_credential(state, bearer)).await?;
let session = SendWrapper::new(state.resolve_session(key, user_id,
expiry)).await?;
Ok(Self {
session: SendWrapper::new(session),
@@ -74,8 +78,9 @@ impl FromRequestParts<HttpState> for Authenticated {
}
}
-/// Read-only caller identity for a protected read route: the authenticated
-/// user id and nothing else.
+/// Read-only caller identity for a protected read route: the authenticated
user
+/// id plus the request path/query used for follower redirects, and no minted
VSR
+/// session.
///
/// Unlike [`Authenticated`], it verifies the bearer WITHOUT minting or
/// Registering a VSR session. Reads are served from the local metadata STM and
@@ -103,10 +108,13 @@ impl FromRequestParts<HttpState> for Identity {
// Verify only. The session key and expiry `resolve_credential` also
// returns feed the write path's session table; a read discards them.
- // No `.await` and no session borrow here, so the extractor future
needs
- // no `SendWrapper` bridge (contrast [`Authenticated`], which awaits
the
- // `!Send` `resolve_session`).
- let (_key, user_id, _expiry) = resolve_credential(state, bearer)?;
+ // The verify is `!Send` (a trusted-issuer JWT may await a JWKS fetch),
+ // so bridge it with `SendWrapper` - sound only because compio pins
this
+ // future to shard 0's single thread, the only thread the JWKS client
+ // ever runs on (mirrors legacy `HttpSafeShard`). It holds no `RefCell`
+ // borrow or `DashMap` guard across the `.await`, so a sibling task
+ // scheduled on this thread meanwhile never observes a borrowed cell.
+ let (_key, user_id, _expiry) =
SendWrapper::new(resolve_credential(state, bearer)).await?;
let path_and_query = parts
.uri
.path_and_query()
@@ -137,10 +145,14 @@ fn bearer_token(parts: &Parts) -> Result<&str, IggyError>
{
/// the bearer is the documented fallback, keyed by the same 256-bit BLAKE3
/// digest (`PersonalAccessToken::hash_token`) Iggy already indexes PATs by, so
/// the key is stable and collision-free while the raw secret never enters the
-/// table. Both checks are local and synchronous, so no borrow is held across
an
-/// await here.
-fn resolve_credential(state: &HttpState, bearer: &str) -> Result<(String, u32,
u64), AuthError> {
- if let Ok(claims) = state.jwt.decode(bearer)
+/// table. The JWT verify is `async` and `!Send` (a trusted-issuer token may
+/// fetch the issuer's JWKS on a cache miss), so both call sites drive it
inside
+/// a `SendWrapper`; the PAT check is local and synchronous.
+async fn resolve_credential(
+ state: &HttpState,
+ bearer: &str,
+) -> Result<(String, u32, u64), AuthError> {
+ if let Ok(claims) = state.jwt.decode(bearer).await
&& let Ok(user_id) = claims.sub.parse::<u32>()
{
return Ok((
diff --git a/core/server-ng/src/http/handlers.rs
b/core/server-ng/src/http/handlers.rs
index 6b6a0f2b7..a796ff7c2 100644
--- a/core/server-ng/src/http/handlers.rs
+++ b/core/server-ng/src/http/handlers.rs
@@ -220,7 +220,10 @@ pub(in crate::http) async fn refresh_token(
if command.token.is_empty() {
return Err(IggyError::Unauthenticated.into());
}
- let claims = state.jwt.decode(&command.token)?;
+ // Refresh rejects trusted-issuer tokens (they keep their own lifecycle);
+ // `decode_for_refresh` is `!Send` (a trusted-issuer token may await a JWKS
+ // fetch), so bridge it like every other shard-0 path (see
[`logout_user`]).
+ let claims =
SendWrapper::new(state.jwt.decode_for_refresh(&command.token)).await?;
let user_id = claims
.sub
.parse::<u32>()
@@ -945,7 +948,7 @@ pub(in crate::http) async fn delete_partitions(
/// the path.
///
/// `DeleteSegments` is not itself a consensus op: [`submit_write`] carries it
-/// through [`submit_gated`], which resolves it to the `TruncatePartition` that
+/// through `submit_gated`, which resolves it to the `TruncatePartition` that
/// commits the trim. RBAC (`delete_segments`) is enforced in-apply on that
/// truncate, like the sibling topic writes.
pub(in crate::http) async fn delete_segments(
@@ -1316,7 +1319,8 @@ pub(in crate::http) async fn delete_cg(
///
/// The plaintext password rides the JSON body; [`submit_write`] hashes it on
/// shard 0 before the request enters consensus (see
-/// [`maybe_rewrite_user_password_request`]), so no plaintext is ever
replicated.
+/// [`crate::users::maybe_rewrite_user_password_request`]), so no plaintext is
+/// ever replicated.
pub(in crate::http) async fn create_user(
State(state): State<HttpState>,
identity: Authenticated,
@@ -1397,8 +1401,8 @@ pub(in crate::http) async fn delete_user(
/// `PUT /users/{user_id}/password`: change a user's password. Returns 204.
///
/// Both passwords ride the JSON body in plaintext. On shard 0, before the op
-/// enters consensus, [`maybe_rewrite_user_password_request`] hashes the new
-/// password and strips the current one (so neither plaintext is ever
+/// enters consensus, [`crate::users::maybe_rewrite_user_password_request`]
hashes
+/// the new password and strips the current one (so neither plaintext is ever
/// replicated), and verifies `current_password` against the target's stored
/// hash. A wrong current password is not denied pre-consensus: the op still
/// commits, carrying an empty new-password hash the replicated apply turns
into
@@ -1483,7 +1487,7 @@ pub(in crate::http) async fn get_pats(
/// legacy server returns, with HTTP 200.
///
/// The raw token is non-deterministic and secret, so it must never enter
-/// consensus: [`rewrite_pat_request_for_user`] (invoked inside
+/// consensus: [`crate::pat::rewrite_pat_request_for_user`] (invoked inside
/// [`submit_committed`]) mints it on shard 0 and replicates only its hash, so
a
/// successful committed reply body is empty. [`build_raw_pat_reply`] then
splices
/// the raw secret back into that reply locally, using the confirmed commit
diff --git a/core/server/src/http/jwt/jwks.rs b/core/server-ng/src/http/jwks.rs
similarity index 62%
copy from core/server/src/http/jwt/jwks.rs
copy to core/server-ng/src/http/jwks.rs
index 591d0a707..d3de9931f 100644
--- a/core/server/src/http/jwt/jwks.rs
+++ b/core/server-ng/src/http/jwks.rs
@@ -15,18 +15,37 @@
// specific language governing permissions and limitations
// under the License.
+//! JWKS key resolver for trusted-issuer (A2A) JWT verification.
+//!
+//! Ported from `server::http::jwt::jwks`. Keys are fetched lazily per
+//! `{issuer, kid}` and cached in a `DashMap`; a refresh re-reads the issuer's
+//! key set and evicts cached kids no longer present (rotation cleanup).
+
+use std::collections::HashSet;
+use std::hash::Hash;
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+
use dashmap::DashMap;
use iggy_common::IggyError;
use jsonwebtoken::DecodingKey;
use serde::Deserialize;
-use std::hash::Hash;
-use strum::{Display, EnumString};
+use strum::EnumString;
+use tokio::sync::Mutex;
+
+/// Minimum wall-clock gap between outbound JWKS fetches for one trusted
issuer.
+/// Inside this window a cache miss is authoritative: the issuer's key set was
+/// just read, so an absent `kid` is genuinely absent and no fetch is issued.
+/// This bounds a pre-auth caller replaying unknown `kid`s to at most one
+/// outbound request per issuer per window; the trade is that a freshly rotated
+/// key is only discoverable once the window elapses.
+const JWKS_REFRESH_MIN_INTERVAL: Duration = Duration::from_secs(10);
thread_local! {
- // cyper 0.9's `Client` is `!Send`/`!Sync` (`Rc`-backed) and `new()`
- // now returns a `Result`, so it can no longer be a global `OnceLock`.
- // compio is thread-per-core; keep one client per thread. The `Rc`
- // inner makes cloning cheap, so callers take an owned handle.
+ // cyper's `Client` is `!Send`/`!Sync` (`Rc`-backed) and `new()` returns a
+ // `Result`, so it cannot be a global `OnceLock`. compio is
thread-per-core;
+ // keep one client per thread. The `Rc` inner makes cloning cheap, so
callers
+ // take an owned handle.
static HTTP_CLIENT: cyper::Client =
cyper::Client::new().expect("failed to build cyper HTTP client for
JWKS");
}
@@ -36,22 +55,18 @@ fn get_http_client() -> cyper::Client {
}
/// JWK key type enumeration
-#[derive(Debug, Clone, Copy, Display, EnumString, Deserialize, PartialEq, Eq)]
-#[strum(serialize_all = "UPPERCASE")]
+#[derive(Debug, Clone, Copy, Deserialize)]
#[serde(rename_all = "UPPERCASE")]
enum JwkKeyType {
/// RSA key type
- #[strum(serialize = "RSA")]
Rsa,
/// EC (Elliptic Curve) key type
- #[strum(serialize = "EC")]
Ec,
}
/// EC curve type enumeration
-#[derive(Debug, Clone, Copy, Display, EnumString, Deserialize, PartialEq, Eq)]
+#[derive(Debug, Clone, Copy, EnumString)]
#[strum(serialize_all = "UPPERCASE")]
-#[serde(rename_all = "UPPERCASE")]
enum EcCurve {
/// P-256 curve
#[strum(serialize = "P-256")]
@@ -86,56 +101,75 @@ struct CacheKey {
kid: String,
}
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Default)]
pub struct JwksClient {
cache: DashMap<CacheKey, DecodingKey>,
-}
-
-impl Default for JwksClient {
- fn default() -> Self {
- Self {
- cache: DashMap::new(),
- }
- }
+ /// Per-issuer single-flight and rate-limit guard. The async mutex
serialises
+ /// refresh attempts for one issuer so a burst of concurrent misses
collapses
+ /// onto a single fetch; the inner instant is when that issuer was last
+ /// fetched and gates [`JWKS_REFRESH_MIN_INTERVAL`]. Keyed only by issuer
+ /// (operator-configured, bounded), never by the attacker-controlled `kid`.
+ refresh_guards: DashMap<String, Arc<Mutex<Option<Instant>>>>,
}
impl JwksClient {
- pub async fn get_key(&self, issuer: &str, jwks_url: &str, kid: &str) ->
Option<DecodingKey> {
+ /// Resolve the decoding key for `{issuer, kid}`, fetching and caching the
+ /// issuer's JWKS on a cache miss. Returns the rich [`IggyError`] built
during
+ /// the fetch so the caller can log why a trusted-issuer token could not be
+ /// verified instead of collapsing every failure into an opaque miss.
+ ///
+ /// A cache miss is served under a per-issuer guard: concurrent misses
fetch
+ /// once, and within [`JWKS_REFRESH_MIN_INTERVAL`] of the last fetch a
miss is
+ /// treated as a known-absent `kid` and rejected without an outbound
request.
+ /// This keeps an unauthenticated caller replaying unknown `kid`s from
+ /// amplifying into unbounded fetches against the issuer's JWKS endpoint.
+ // The per-issuer guard is deliberately held across the JWKS fetch: that
hold
+ // is what serialises concurrent misses onto a single outbound request.
Drop-
+ // tightening would release it before the await and defeat the
single-flight.
+ #[allow(clippy::significant_drop_tightening)]
+ pub async fn get_key(
+ &self,
+ issuer: &str,
+ jwks_url: &str,
+ kid: &str,
+ ) -> Result<DecodingKey, IggyError> {
let cache_key = CacheKey {
issuer: issuer.to_string(),
kid: kid.to_string(),
};
- // try to get from cache first
+ // Positive-cache fast path: no lock, no fetch.
if let Some(key) = self.cache.get(&cache_key) {
- return Some(key.clone());
+ return Ok(key.clone());
}
- // fetch and cache if not found
- if let Ok(key) = self.fetch_and_cache_key(issuer, jwks_url, kid).await
{
- return Some(key);
- }
+ // Take the per-issuer guard so concurrent misses serialise onto one
+ // fetch. Clone the Arc out and drop the DashMap entry lock before the
+ // await, so no shard lock is held across the network I/O.
+ let entry = self.refresh_guards.entry(issuer.to_string()).or_default();
+ let guard = entry.value().clone();
+ drop(entry);
+ let mut last_fetch = guard.lock().await;
- None
- }
+ // A prior holder of the guard may have populated our kid while we
waited.
+ if let Some(key) = self.cache.get(&cache_key) {
+ return Ok(key.clone());
+ }
- async fn fetch_and_cache_key(
- &self,
- issuer: &str,
- jwks_url: &str,
- kid: &str,
- ) -> Result<DecodingKey, IggyError> {
- if let Err(e) = self.refresh_keys(issuer, jwks_url).await {
- return Err(IggyError::CannotFetchJwks(format!(
- "Failed to refresh keys: {}",
- e
- )));
+ // Inside the refresh window the last fetch's key set still stands, so
a
+ // miss here means the kid is genuinely absent: reject without touching
+ // the network. One per-issuer timestamp both negative-caches unknown
kids
+ // and rate-limits outbound fetches, with no attacker-keyed state.
+ if let Some(fetched_at) = *last_fetch
+ && fetched_at.elapsed() < JWKS_REFRESH_MIN_INTERVAL
+ {
+ return Err(IggyError::InvalidAccessToken);
}
- let cache_key = CacheKey {
- issuer: issuer.to_string(),
- kid: kid.to_string(),
- };
+ // Stale or first contact: fetch. Record the attempt up front so a
failing
+ // issuer is rate-limited too, not re-hit on every miss.
+ *last_fetch = Some(Instant::now());
+ self.refresh_keys(issuer, jwks_url).await?;
self.cache
.get(&cache_key)
@@ -144,29 +178,28 @@ impl JwksClient {
}
async fn refresh_keys(&self, issuer: &str, jwks_url: &str) -> Result<(),
IggyError> {
- // The cyper client is `!Send` since 0.9; callers reached from axum
- // middleware wrap this future in `SendWrapper` (see
- // `http::jwt::middleware::jwt_auth`), so it's free to await cyper
- // directly here.
+ // The cyper client is `!Send`; callers reached from the axum extractor
+ // wrap this future in `SendWrapper` (see `http::extractor`), so it is
+ // free to await cyper directly here.
let client = get_http_client();
let request = client
.get(jwks_url)
- .map_err(|e| IggyError::CannotFetchJwks(format!("Failed to build
request: {}", e)))?
+ .map_err(|e| IggyError::CannotFetchJwks(format!("Failed to build
request: {e}")))?
.build();
let response = client
.execute(request)
.await
- .map_err(|e| IggyError::CannotFetchJwks(format!("HTTP request
failed: {}", e)))?;
+ .map_err(|e| IggyError::CannotFetchJwks(format!("HTTP request
failed: {e}")))?;
let body = response.text().await.map_err(|e| {
- IggyError::CannotFetchJwks(format!("Failed to read response body:
{}", e))
+ IggyError::CannotFetchJwks(format!("Failed to read response body:
{e}"))
})?;
let jwks: JwkSet = serde_json::from_str(&body)
- .map_err(|e| IggyError::CannotFetchJwks(format!("Failed to parse
JWKS: {}", e)))?;
+ .map_err(|e| IggyError::CannotFetchJwks(format!("Failed to parse
JWKS: {e}")))?;
// Collect all current kids from the JWKS response
- let current_kids: std::collections::HashSet<String> =
+ let current_kids: HashSet<String> =
jwks.keys.iter().filter_map(|key| key.kid.clone()).collect();
// Remove cached keys for this issuer that are no longer in the JWKS
response
@@ -190,7 +223,7 @@ impl JwksClient {
JwkKeyType::Rsa => {
if let (Some(n), Some(e)) = (key.n.as_deref(),
key.e.as_deref()) {
DecodingKey::from_rsa_components(n, e).map_err(|e|
{
- IggyError::CannotFetchJwks(format!("Invalid
RSA key: {}", e))
+ IggyError::CannotFetchJwks(format!("Invalid
RSA key: {e}"))
})?
} else {
continue;
@@ -202,7 +235,7 @@ impl JwksClient {
{
if let Ok(_curve) = crv_str.parse::<EcCurve>() {
DecodingKey::from_ec_components(x,
y).map_err(|e| {
-
IggyError::CannotFetchJwks(format!("Invalid EC key: {}", e))
+
IggyError::CannotFetchJwks(format!("Invalid EC key: {e}"))
})?
} else {
continue;
@@ -293,10 +326,9 @@ mod tests {
};
let decoding_key = create_test_decoding_key();
- client.cache.insert(cache_key.clone(), decoding_key.clone());
+ client.cache.insert(cache_key.clone(), decoding_key);
- let cached = client.cache.get(&cache_key);
- assert!(cached.is_some());
+ assert!(client.cache.get(&cache_key).is_some());
}
#[test]
diff --git a/core/server-ng/src/http/jwt.rs b/core/server-ng/src/http/jwt.rs
index 5548963cf..91f1830bc 100644
--- a/core/server-ng/src/http/jwt.rs
+++ b/core/server-ng/src/http/jwt.rs
@@ -18,24 +18,31 @@
//! Minimal JWT issuer/verifier for the shard-0 HTTP listener.
//!
//! Ported from the legacy `server::http::jwt::jwt_manager::JwtManager`,
reduced
-//! to the issue + verify half: no revoked-token persistence, no JWKS /
-//! trusted-issuer path. The refresh-token route re-issues by composing
`decode`
-//! with `generate` at the handler; without a revocation list that refresh is
-//! stateless, so the token it was minted from stays valid until its own `exp`.
-//! The claim set (`JwtClaims`, including the `jti`) and the `IggyError ->
HTTP`
-//! grading are reused verbatim so issued tokens and error bodies stay
identical
-//! to the legacy server.
+//! to the issue + verify half: no revoked-token persistence. Self-issued HS256
+//! is the common path; with `[[http.jwt.trusted_issuers]]` configured,
`decode`
+//! also verifies external RS256/EC tokens against the issuer's JWKS
+//! ([`super::jwks`]) and remaps their subject onto the configured iggy user.
+//! The refresh-token route re-issues by composing `decode_for_refresh` (verify
+//! plus a refusal of trusted-issuer tokens) with `generate` at the handler;
+//! without a revocation list that refresh is stateless, so the token it was
+//! minted from stays valid until its own `exp`. The claim set
+//! (`JwtClaims`, including the `jti`) and the `IggyError -> HTTP` grading are
+//! reused verbatim so issued tokens and error bodies stay identical to the
+//! legacy server.
+use std::collections::HashMap;
use std::ops::Range;
-use configs::http::HttpJwtConfig;
+use configs::http::{HttpJwtConfig, TrustedIssuerConfig};
use iggy_common::{IggyDuration, IggyError, IggyExpiry, IggyTimestamp, UserId};
use jsonwebtoken::{Algorithm, DecodingKey, EncodingKey, Header, Validation,
decode, encode};
use serde::{Deserialize, Deserializer, Serialize, Serializer};
use server_common::crypto;
-use tracing::warn;
+use tracing::{debug, error, warn};
use uuid::Uuid;
+use super::jwks::JwksClient;
+
/// Length window for the random secret minted when no secret is configured.
/// Matches the legacy server so both behave identically on an empty secret.
const GENERATED_SECRET_LEN: Range<usize> = 32..64;
@@ -54,12 +61,18 @@ pub struct JwtManager {
encoding_key: EncodingKey,
decoding_key: DecodingKey,
validation: Validation,
+ jwks_client: JwksClient,
+ /// Trusted external issuers keyed by normalized issuer URL. Empty unless
+ /// `[[http.jwt.trusted_issuers]]` is configured, in which case `decode`
+ /// takes the JWKS verification path for tokens from these issuers.
+ trusted_issuers: HashMap<String, TrustedIssuerConfig>,
}
impl JwtManager {
/// Build the manager from the `[http.jwt]` config, normalizing an
/// unconfigured secret the same way the legacy server does (mint a random
- /// ephemeral secret, or mirror whichever half is set).
+ /// ephemeral secret, or mirror whichever half is set) and loading any
+ /// configured trusted issuers (keyed by normalized issuer URL).
///
/// # Errors
///
@@ -79,6 +92,42 @@ impl JwtManager {
// `jsonwebtoken` leaves `validate_nbf` off by default, so this is
// explicit; the claim is always present, so it only ever tightens.
validation.validate_nbf = true;
+ // Validate every trusted issuer at startup so a broken config fails
loud
+ // instead of silently rejecting every A2A token at first use: an empty
+ // issuer or `jwks_url` can never verify, and mapping onto the root
user
+ // (id 0) would resolve every external subject to root (`decode` guards
+ // the root case again at verify time as defense in depth).
+ for issuer_config in config.trusted_issuers.iter().flatten() {
+ if issuer_config.issuer.is_empty() {
+ error!("trusted issuer configured with an empty issuer URL");
+ return Err(IggyError::InvalidConfiguration);
+ }
+ if issuer_config.jwks_url.is_empty() {
+ error!(
+ issuer = %issuer_config.issuer,
+ "trusted issuer configured with an empty jwks_url"
+ );
+ return Err(IggyError::InvalidConfiguration);
+ }
+ if issuer_config.user_id == 0 {
+ error!(
+ issuer = %issuer_config.issuer,
+ "trusted issuer user_id is missing or 0 (must be a
non-zero iggy user id)"
+ );
+ return Err(IggyError::InvalidConfiguration);
+ }
+ }
+ let trusted_issuers = config
+ .trusted_issuers
+ .iter()
+ .flatten()
+ .map(|issuer_config| {
+ (
+ normalize_issuer_url(&issuer_config.issuer),
+ issuer_config.clone(),
+ )
+ })
+ .collect();
Ok(Self {
algorithm,
issuer: config.issuer,
@@ -88,6 +137,8 @@ impl JwtManager {
encoding_key,
decoding_key,
validation,
+ jwks_client: JwksClient::default(),
+ trusted_issuers,
})
}
@@ -128,17 +179,154 @@ impl JwtManager {
/// Verify a token's signature and registered claims, returning its claim
/// set (the `jti` is what a session table keys on).
///
+ /// With no trusted issuers this is the self-issued HS256 verify and the
+ /// returned future resolves without awaiting. When the token's issuer
+ /// matches a configured trusted issuer, verification instead uses the
+ /// issuer's JWKS key (awaiting the fetch on a cache miss) and the returned
+ /// `sub` is the configured iggy user id, not the external subject.
+ ///
/// # Errors
///
/// Returns [`IggyError::Unauthenticated`] if the token is malformed,
/// expired, or fails issuer/audience/signature validation.
- pub fn decode(&self, token: &str) -> Result<JwtClaims, IggyError> {
+ pub async fn decode(&self, token: &str) -> Result<JwtClaims, IggyError> {
+ // With no trusted issuers this is the original self-issued verifier,
+ // resolving without ever awaiting.
+ if self.trusted_issuers.is_empty() {
+ return self.decode_self_issued(token);
+ }
+
+ // Try the self-issued verify first. A trusted-issuer token is signed
by
+ // the issuer's key, which the self-issued verify rejects at the
algorithm
+ // check (when self-issued is HS*, the shipped default) or at signature
+ // verification (if an operator configures self-issued RS*/ES*) -
either
+ // way it can never accept an external token; a valid self-issued token
+ // returns here and the classification below only runs once it has
failed.
+ if let Ok(claims) = self.decode_self_issued(token) {
+ return Ok(claims);
+ }
+
+ // Self-issued verify failed. Classify the token by its (untrusted)
issuer
+ // claim; if it names a trusted issuer, verify it against that issuer's
+ // JWKS key. Everything read here is re-checked under that key below.
+ let Ok(insecure) =
jsonwebtoken::dangerous::insecure_decode::<JwtClaims>(token) else {
+ debug!("token is not self-issued and cannot be parsed to classify
its issuer");
+ return Err(IggyError::Unauthenticated);
+ };
+ let normalized_iss = normalize_issuer_url(&insecure.claims.iss);
+ let Some(config) = self.trusted_issuers.get(&normalized_iss) else {
+ debug!(
+ issuer = %insecure.claims.iss,
+ "token issuer matches no trusted issuer and failed self-issued
verify"
+ );
+ return Err(IggyError::Unauthenticated);
+ };
+
+ // A trusted external subject must never resolve to the root user.
+ if config.user_id == 0 {
+ error!(
+ issuer = %config.issuer,
+ "trusted-issuer token cannot map to root user (user_id = 0)"
+ );
+ return Err(IggyError::Unauthenticated);
+ }
+
+ // `insecure.header` is parsed from the same header segment
`decode_header`
+ // reads, so reuse it for `alg` and `kid` instead of parsing twice.
+ let Some(kid) = insecure.header.kid.as_deref() else {
+ debug!(
+ issuer = %config.issuer,
+ "trusted-issuer token has no `kid` header for JWKS key lookup"
+ );
+ return Err(IggyError::Unauthenticated);
+ };
+ let decoding_key = match self
+ .jwks_client
+ .get_key(&config.issuer, &config.jwks_url, kid)
+ .await
+ {
+ Ok(key) => key,
+ Err(error) => {
+ // Fail closed but loud: a broken `jwks_url` or a rotated-away
+ // `kid` otherwise surfaces only as an opaque 401.
+ warn!(
+ %error,
+ issuer = %config.issuer,
+ kid,
+ "cannot resolve JWKS key for trusted-issuer token"
+ );
+ return Err(IggyError::Unauthenticated);
+ }
+ };
+
+ // The algorithm is taken from the token header (e.g. RS256), not the
+ // self-issued HS256; issuer and audience are pinned to the config, and
+ // the timing checks mirror the self-issued path (enforce `nbf`, honor
+ // the configured clock skew).
+ let mut validation = Validation::new(insecure.header.alg);
+ validation.set_issuer(std::slice::from_ref(&config.issuer));
+ validation.set_audience(std::slice::from_ref(&config.audience));
+ validation.validate_nbf = true;
+ validation.leeway = self.validation.leeway;
+ let mut claims = decode::<JwtClaims>(token, &decoding_key, &validation)
+ .map(|data| data.claims)
+ .map_err(|error| {
+ error!(%error, "trusted-issuer JWT verification failed");
+ IggyError::Unauthenticated
+ })?;
+
+ // Remap the external subject onto the fixed configured iggy user; the
+ // raw external `sub` is discarded (no auto-provisioning).
+ claims.sub = config.user_id.to_string();
+ Ok(claims)
+ }
+
+ /// Verify a token for the refresh route, additionally refusing any token
+ /// minted by a trusted external issuer: those carry their own lifecycle
(the
+ /// issuer controls their `exp`), so exchanging one for a self-issued token
+ /// would mint a bearer that outlives the external grant. Mirrors the
legacy
+ /// `JwtManager::refresh_token` reject.
+ ///
+ /// # Errors
+ ///
+ /// Returns [`IggyError::InvalidAccessToken`] for a trusted-issuer token,
or
+ /// whatever [`Self::decode`] returns for an otherwise invalid one.
+ pub async fn decode_for_refresh(&self, token: &str) -> Result<JwtClaims,
IggyError> {
+ let claims = self.decode(token).await?;
+ if self
+ .trusted_issuers
+ .contains_key(&normalize_issuer_url(&claims.iss))
+ {
+ error!(issuer = %claims.iss, "refusing to refresh a trusted-issuer
token");
+ return Err(IggyError::InvalidAccessToken);
+ }
+ Ok(claims)
+ }
+
+ /// Self-issued HS256 verify against the configured secret and validation.
+ /// The synchronous common path shared by the fast path and by every
+ /// trusted-issuer classification miss.
+ fn decode_self_issued(&self, token: &str) -> Result<JwtClaims, IggyError> {
decode::<JwtClaims>(token, &self.decoding_key, &self.validation)
.map(|data| data.claims)
.map_err(|_| IggyError::Unauthenticated)
}
}
+/// Normalize an issuer URL by lowercasing scheme and host while preserving
path
+/// case, so trusted-issuer lookup is case-insensitive on the parts that are
+/// (`https://Example.COM/PATH` -> `https://example.com/PATH`).
+fn normalize_issuer_url(url: &str) -> String {
+ match url.split_once("://") {
+ Some((scheme, rest)) => {
+ let scheme = scheme.to_lowercase();
+ let (host, path) = rest.find('/').map_or((rest, ""), |idx|
rest.split_at(idx));
+ format!("{}://{}{}", scheme, host.to_lowercase(), path)
+ }
+ None => url.trim_end_matches('/').to_lowercase(),
+ }
+}
+
/// Fill in an unconfigured JWT secret exactly like the legacy HTTP server:
/// both empty -> one random ephemeral secret; one empty -> mirror the other;
/// both set but different under an HMAC algorithm -> warn (they must match).
@@ -283,25 +471,113 @@ mod tests {
}
}
- #[test]
- fn token_presented_before_its_nbf_is_rejected() {
+ // The default config has no trusted issuers, so `decode` takes the
+ // synchronous self-issued path; any executor can drive the ready future.
+ #[tokio::test]
+ async fn token_presented_before_its_nbf_is_rejected() {
// not_before well past the leeway window, so the freshly issued token
// is not yet valid and decode must reject it.
let manager = JwtManager::build(&config("3600 s", "5
s")).expect("builds");
let token = manager.generate(7).expect("issues");
assert!(
- manager.decode(&token.access_token).is_err(),
+ manager.decode(&token.access_token).await.is_err(),
"a token presented before its nbf must be rejected"
);
}
- #[test]
- fn token_at_its_nbf_is_accepted() {
+ #[tokio::test]
+ async fn token_at_its_nbf_is_accepted() {
// Default not_before (0s) => nbf == iat, so the token is valid now and
// enabling nbf validation does not reject a normally issued token.
let manager = JwtManager::build(&config("0 s", "5
s")).expect("builds");
let token = manager.generate(7).expect("issues");
- let claims = manager.decode(&token.access_token).expect("valid now");
+ let claims = manager
+ .decode(&token.access_token)
+ .await
+ .expect("valid now");
assert_eq!(claims.sub, "7");
}
+
+ #[test]
+ fn build_rejects_trusted_issuer_mapping_to_root() {
+ let jwt = HttpJwtConfig {
+ trusted_issuers: Some(vec![TrustedIssuerConfig {
+ issuer: "https://external.example".to_string(),
+ audience: "iggy".to_string(),
+ jwks_url:
"https://external.example/.well-known/jwks.json".to_string(),
+ user_id: 0,
+ }]),
+ ..HttpJwtConfig::default()
+ };
+ match JwtManager::build(&jwt) {
+ Err(IggyError::InvalidConfiguration) => {}
+ Err(other) => panic!("expected InvalidConfiguration, got
{other:?}"),
+ Ok(_) => panic!("build must reject a trusted issuer mapping to the
root user"),
+ }
+ }
+
+ #[test]
+ fn build_rejects_trusted_issuer_with_empty_issuer() {
+ let jwt = HttpJwtConfig {
+ trusted_issuers: Some(vec![TrustedIssuerConfig {
+ issuer: String::new(),
+ audience: "iggy".to_string(),
+ jwks_url:
"https://external.example/.well-known/jwks.json".to_string(),
+ user_id: 1,
+ }]),
+ ..HttpJwtConfig::default()
+ };
+ match JwtManager::build(&jwt) {
+ Err(IggyError::InvalidConfiguration) => {}
+ Err(other) => panic!("expected InvalidConfiguration, got
{other:?}"),
+ Ok(_) => panic!("build must reject a trusted issuer with an empty
issuer"),
+ }
+ }
+
+ #[test]
+ fn build_rejects_trusted_issuer_with_empty_jwks_url() {
+ let jwt = HttpJwtConfig {
+ trusted_issuers: Some(vec![TrustedIssuerConfig {
+ issuer: "https://external.example".to_string(),
+ audience: "iggy".to_string(),
+ jwks_url: String::new(),
+ user_id: 1,
+ }]),
+ ..HttpJwtConfig::default()
+ };
+ match JwtManager::build(&jwt) {
+ Err(IggyError::InvalidConfiguration) => {}
+ Err(other) => panic!("expected InvalidConfiguration, got
{other:?}"),
+ Ok(_) => panic!("build must reject a trusted issuer with an empty
jwks_url"),
+ }
+ }
+
+ #[test]
+ fn normalize_issuer_url_lowercases_scheme_and_host_preserving_path() {
+ assert_eq!(
+ normalize_issuer_url("HTTPS://Example.COM/PATH"),
+ "https://example.com/PATH";
+ );
+ }
+
+ #[test]
+ fn normalize_issuer_url_no_path() {
+ assert_eq!(
+ normalize_issuer_url("HTTPS://Example.COM"),
+ "https://example.com";
+ );
+ }
+
+ #[test]
+ fn normalize_issuer_url_no_scheme_trims_trailing_slash() {
+ assert_eq!(normalize_issuer_url("Example.COM/"), "example.com");
+ }
+
+ #[test]
+ fn normalize_issuer_url_already_normalized_is_stable() {
+ assert_eq!(
+ normalize_issuer_url("https://example.com/path";),
+ "https://example.com/path";
+ );
+ }
}
diff --git a/core/server/Cargo.toml b/core/server/Cargo.toml
index cb8f78640..037bd30b9 100644
--- a/core/server/Cargo.toml
+++ b/core/server/Cargo.toml
@@ -89,6 +89,7 @@ strum = { workspace = true }
sysinfo = { workspace = true }
tempfile = { workspace = true }
thiserror = { workspace = true }
+tokio = { workspace = true }
toml = { workspace = true }
tower-http = { workspace = true }
tracing = { workspace = true }
diff --git a/core/server/config.toml b/core/server/config.toml
index 2add7ca24..b1609e7a3 100644
--- a/core/server/config.toml
+++ b/core/server/config.toml
@@ -126,11 +126,24 @@ decoding_secret = ""
# `false` means the secret is in plain text.
use_base64_secret = false
-# Trusted issuers for A2A (Application-to-Application) authentication
-[[http.jwt.trusted_issuers]]
-issuer = "test-issuer"
-jwks_url = "http://127.0.0.1:8081/.well-known/jwks.json";
-audience = "iggy.apache.org"
+# Trusted issuers for A2A (Application-to-Application) authentication. Opt-in:
+# with none configured the listener accepts only self-issued HS256 tokens.
+# `issuer`, `audience` and `jwks_url` are required per entry; `user_id` is
+# optional but defaults to 0 (root), which is rejected - set it to the non-zero
+# iggy user every token from that issuer is remapped onto.
+#
+# Operational note: enabling an issuer opens an outbound JWKS fetch that is
+# reachable before a token's signature is verified - a token naming this issuer
+# with an unknown key id can trigger a fetch to `jwks_url`. The target is fixed
+# (not attacker-chosen); concurrent misses coalesce onto one fetch and repeats
+# are rate-limited to at most one outbound request per issuer per short window,
+# so an unknown-key-id flood cannot amplify. The same window bounds how
quickly a
+# freshly rotated signing key is picked up.
+# [[http.jwt.trusted_issuers]]
+# issuer = "test-issuer"
+# jwks_url = "http://127.0.0.1:8081/.well-known/jwks.json";
+# audience = "iggy.apache.org"
+# user_id = 1
# Metrics configuration for HTTP.
[http.metrics]
diff --git a/core/server/src/http/jwt/jwks.rs b/core/server/src/http/jwt/jwks.rs
index 591d0a707..0932e5ed4 100644
--- a/core/server/src/http/jwt/jwks.rs
+++ b/core/server/src/http/jwt/jwks.rs
@@ -15,12 +15,24 @@
// specific language governing permissions and limitations
// under the License.
+use std::hash::Hash;
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+
use dashmap::DashMap;
use iggy_common::IggyError;
use jsonwebtoken::DecodingKey;
use serde::Deserialize;
-use std::hash::Hash;
use strum::{Display, EnumString};
+use tokio::sync::Mutex;
+
+/// Minimum wall-clock gap between outbound JWKS fetches for one trusted
issuer.
+/// Inside this window a cache miss is authoritative: the issuer's key set was
+/// just read, so an absent `kid` is genuinely absent and no fetch is issued.
+/// This bounds a pre-auth caller replaying unknown `kid`s to at most one
+/// outbound request per issuer per window; the trade is that a freshly rotated
+/// key is only discoverable once the window elapses.
+const JWKS_REFRESH_MIN_INTERVAL: Duration = Duration::from_secs(10);
thread_local! {
// cyper 0.9's `Client` is `!Send`/`!Sync` (`Rc`-backed) and `new()`
@@ -89,58 +101,78 @@ struct CacheKey {
#[derive(Debug, Clone)]
pub struct JwksClient {
cache: DashMap<CacheKey, DecodingKey>,
+ /// Per-issuer single-flight and rate-limit guard. The async mutex
serialises
+ /// refresh attempts for one issuer so a burst of concurrent misses
collapses
+ /// onto a single fetch; the inner instant is when that issuer was last
+ /// fetched and gates [`JWKS_REFRESH_MIN_INTERVAL`]. Keyed only by issuer
+ /// (operator-configured, bounded), never by the attacker-controlled `kid`.
+ refresh_guards: DashMap<String, Arc<Mutex<Option<Instant>>>>,
}
impl Default for JwksClient {
fn default() -> Self {
Self {
cache: DashMap::new(),
+ refresh_guards: DashMap::new(),
}
}
}
impl JwksClient {
+ /// Resolve the decoding key for `{issuer, kid}`, fetching and caching the
+ /// issuer's JWKS on a cache miss.
+ ///
+ /// A cache miss is served under a per-issuer guard: concurrent misses
fetch
+ /// once, and within [`JWKS_REFRESH_MIN_INTERVAL`] of the last fetch a
miss is
+ /// treated as a known-absent `kid` and rejected without an outbound
request.
+ /// This keeps an unauthenticated caller replaying unknown `kid`s from
+ /// amplifying into unbounded fetches against the issuer's JWKS endpoint.
+ // The per-issuer guard is deliberately held across the JWKS fetch: that
hold
+ // is what serialises concurrent misses onto a single outbound request.
Drop-
+ // tightening would release it before the await and defeat the
single-flight.
+ #[allow(clippy::significant_drop_tightening)]
pub async fn get_key(&self, issuer: &str, jwks_url: &str, kid: &str) ->
Option<DecodingKey> {
let cache_key = CacheKey {
issuer: issuer.to_string(),
kid: kid.to_string(),
};
- // try to get from cache first
+ // Positive-cache fast path: no lock, no fetch.
if let Some(key) = self.cache.get(&cache_key) {
return Some(key.clone());
}
- // fetch and cache if not found
- if let Ok(key) = self.fetch_and_cache_key(issuer, jwks_url, kid).await
{
- return Some(key);
- }
+ // Take the per-issuer guard so concurrent misses serialise onto one
+ // fetch. Clone the Arc out and drop the DashMap entry lock before the
+ // await, so no shard lock is held across the network I/O.
+ let entry = self.refresh_guards.entry(issuer.to_string()).or_default();
+ let guard = entry.value().clone();
+ drop(entry);
+ let mut last_fetch = guard.lock().await;
- None
- }
+ // A prior holder of the guard may have populated our kid while we
waited.
+ if let Some(key) = self.cache.get(&cache_key) {
+ return Some(key.clone());
+ }
- async fn fetch_and_cache_key(
- &self,
- issuer: &str,
- jwks_url: &str,
- kid: &str,
- ) -> Result<DecodingKey, IggyError> {
- if let Err(e) = self.refresh_keys(issuer, jwks_url).await {
- return Err(IggyError::CannotFetchJwks(format!(
- "Failed to refresh keys: {}",
- e
- )));
+ // Inside the refresh window the last fetch's key set still stands, so
a
+ // miss here means the kid is genuinely absent: reject without touching
+ // the network. One per-issuer timestamp both negative-caches unknown
kids
+ // and rate-limits outbound fetches, with no attacker-keyed state.
+ if let Some(fetched_at) = *last_fetch
+ && fetched_at.elapsed() < JWKS_REFRESH_MIN_INTERVAL
+ {
+ return None;
}
- let cache_key = CacheKey {
- issuer: issuer.to_string(),
- kid: kid.to_string(),
- };
+ // Stale or first contact: fetch. Record the attempt up front so a
failing
+ // issuer is rate-limited too, not re-hit on every miss.
+ *last_fetch = Some(Instant::now());
+ if self.refresh_keys(issuer, jwks_url).await.is_err() {
+ return None;
+ }
- self.cache
- .get(&cache_key)
- .map(|entry| entry.clone())
- .ok_or(IggyError::InvalidAccessToken)
+ self.cache.get(&cache_key).map(|entry| entry.clone())
}
async fn refresh_keys(&self, issuer: &str, jwks_url: &str) -> Result<(),
IggyError> {
