This is an automated email from the ASF dual-hosted git repository.
numinnex pushed a commit to branch mtls_kafka_proxy
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/mtls_kafka_proxy by this push:
new 0adff2d43 address review
0adff2d43 is described below
commit 0adff2d436c2acc367e8bcdf3aba3afca9bd0ade
Author: Grzegorz Koszyk <[email protected]>
AuthorDate: Wed Oct 14 09:32:28 2026 +0200
address review
---
gateways/kafka/README.md | 4 ++-
.../kafka/src/bridge/iggy_bridge/group_offsets.rs | 1 +
gateways/kafka/src/bridge/iggy_bridge/mod.rs | 29 +++++++++++++++----
gateways/kafka/src/server.rs | 33 ++++++++++++++++------
4 files changed, 52 insertions(+), 15 deletions(-)
diff --git a/gateways/kafka/README.md b/gateways/kafka/README.md
index 6ba7425d6..e55370350 100644
--- a/gateways/kafka/README.md
+++ b/gateways/kafka/README.md
@@ -276,7 +276,9 @@ The certificate's subject alternative names must cover
every host in
`IGGY_KAFKA_ADVERTISED_LISTENERS`: a client verifies the name of each broker
address Metadata
hands it, so a host missing from the certificate fails the client's next
connection, not the
first. Client certificates are checked against the CA only; there is no CRL or
OCSP revocation
-checking, so revoke access by deleting or deactivating the Iggy user a
certificate maps to.
+checking, so revoke access by deleting or deactivating the Iggy user a
certificate maps to. That
+applies to new connections only: an open connection keeps the principal its
handshake resolved,
+and the permissions that came with it, until it closes.
```bash
kcat -b kafka.example.com:9094 -X security.protocol=SSL -X
ssl.ca.location=ca.pem -L
diff --git a/gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs
b/gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs
index 4b6e60746..b5805a4c2 100644
--- a/gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs
+++ b/gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs
@@ -462,6 +462,7 @@ mod tests {
send_slot: Arc::new(Semaphore::new(1)),
fetch_pool: Arc::new(FetchPool::new()),
probe_client: LazyClient::default(),
+ lookup_client: LazyClient::default(),
offset_pool: OffsetPool {
slots: commit_clients
.into_iter()
diff --git a/gateways/kafka/src/bridge/iggy_bridge/mod.rs
b/gateways/kafka/src/bridge/iggy_bridge/mod.rs
index b77f9e370..049a63700 100644
--- a/gateways/kafka/src/bridge/iggy_bridge/mod.rs
+++ b/gateways/kafka/src/bridge/iggy_bridge/mod.rs
@@ -25,7 +25,7 @@ use iggy::prelude::{
};
use tokio::sync::{OwnedSemaphorePermit, Semaphore, oneshot};
use tokio::time::{Instant, timeout_at};
-use tracing::info;
+use tracing::{info, warn};
use crate::auth::{AuthError, AuthenticatedPrincipal, PrincipalResolver,
resolve_principal};
use crate::bridge::config::IggyBridgeConfig;
@@ -117,8 +117,8 @@ async fn permit_by(turn: &Arc<Semaphore>, deadline:
Instant) -> Option<OwnedSema
/// Owns the connected `IggyClient`s and resolves Kafka topics against them.
///
/// One lockstep client serves Produce, Metadata and `CreateTopics` for every
Kafka connection, so
-/// those Iggy calls run one at a time. Topic probes use a client of their
own, Fetch polls one per
-/// read slot, and offset calls two per offset slot. A Produce pool is a TODO
in `docs/SCOPE.md`.
+/// those Iggy calls run one at a time. Topic probes and certificate user
lookups each use a client
+/// of their own, Fetch polls one per read slot, and offset calls two per
offset slot. A Produce pool is a TODO in `docs/SCOPE.md`.
///
/// Each client signs in at the metadata leader, and a view change or a
refused call can move it to
/// another node. So in a cluster, two clients can read replicas that are not
at the same offset.
@@ -131,6 +131,8 @@ pub struct IggyBridge {
fetch_pool: Arc<FetchPool>,
/// The client of every topic probe, Fetch's and `ListOffsets`'. See
`probe`.
probe_client: LazyClient,
+ /// The client of every certificate user lookup. See `resolve`.
+ lookup_client: LazyClient,
/// Group offset slots, each with a commit client and a read client. See
`OffsetPool`.
offset_pool: OffsetPool,
}
@@ -170,6 +172,7 @@ impl IggyBridge {
send_slot: Arc::new(Semaphore::new(1)),
fetch_pool: Arc::new(FetchPool::new()),
probe_client: LazyClient::default(),
+ lookup_client: LazyClient::default(),
offset_pool: OffsetPool::new(),
})
}
@@ -226,10 +229,12 @@ impl IggyBridge {
pub async fn close(self) -> Result<(), BridgeError> {
let fetch_clients = self.fetch_pool.close().await;
let probe_client = self.probe_client.close().await;
+ let lookup_client = self.lookup_client.close().await;
let offset_clients = self.offset_pool.close().await;
let shared_client = with_request_timeout(self.client.shutdown()).await;
fetch_clients
.and(probe_client)
+ .and(lookup_client)
.and(offset_clients)
.and(shared_client)
}
@@ -270,11 +275,25 @@ impl std::fmt::Debug for IggyBridge {
}
}
-/// The bridge's own session answers for the users client certificates name.
It runs as the
+/// A session of the bridge's own answers for the users client certificates
name. It runs as the
/// bridge's Iggy user, so that account needs `read_users` for any certificate
to be admitted.
+///
+/// Not the lockstep client: a stuck Produce send would hold every new mTLS
connection past its
+/// handshake budget, and a burst of handshakes would queue ahead of Produce.
#[async_trait]
impl PrincipalResolver for IggyBridge {
async fn resolve(&self, username: &str) -> Result<AuthenticatedPrincipal,
AuthError> {
- resolve_principal(self.client.as_ref(), username).await
+ let client = &self.lookup_client;
+ let lookup = async {
+ let connected = client.connected(&self.config).await?;
+ Ok(resolve_principal(connected.as_ref(), username).await)
+ };
+ match client.until(Instant::now() + REQUEST_TIMEOUT, lookup).await {
+ Ok(resolved) => resolved,
+ Err(error) => {
+ warn!(%error, "could not reach Iggy to look up the
certificate's user");
+ Err(AuthError::Unavailable)
+ }
+ }
}
}
diff --git a/gateways/kafka/src/server.rs b/gateways/kafka/src/server.rs
index 3610a2690..683feb4cf 100644
--- a/gateways/kafka/src/server.rs
+++ b/gateways/kafka/src/server.rs
@@ -157,6 +157,8 @@ pub struct GatewayConfig {
/// Four by default, which stays under the shard count of any node with 16
or fewer physical
/// cores. A default above that is no bound at all on the deployments most
likely to run one
/// gateway in front of one node; a larger node raises it deliberately.
+ ///
+ /// Client certificate user lookups get the same number of slots again,
separate from these.
pub max_concurrent_authentications: usize,
}
@@ -315,6 +317,9 @@ struct SharedAuth {
/// Looks up the Iggy user a client certificate names. The bridge, unless
a test swaps one in.
resolver: Option<Arc<dyn PrincipalResolver>>,
slots: Semaphore,
+ /// Certificate user lookups, kept apart from `slots` so a burst of either
kind cannot starve
+ /// the other of its budget.
+ lookup_slots: Semaphore,
failed_logins: FailedLoginThrottle,
}
@@ -328,6 +333,7 @@ impl SharedAuth {
authenticator,
resolver,
slots: Semaphore::new(max_concurrent),
+ lookup_slots: Semaphore::new(max_concurrent),
failed_logins: FailedLoginThrottle::default(),
}
}
@@ -795,6 +801,7 @@ struct ConnectionContext<'a> {
authenticator: Option<&'a dyn SaslAuthenticator>,
resolver: Option<&'a dyn PrincipalResolver>,
auth_slots: &'a Semaphore,
+ lookup_slots: &'a Semaphore,
failed_logins: &'a FailedLoginThrottle,
peer: &'a SocketAddr,
cancel: &'a CancellationToken,
@@ -962,6 +969,7 @@ where
authenticator: shared_auth.authenticator.as_deref(),
resolver: shared_auth.resolver.as_deref(),
auth_slots: &shared_auth.slots,
+ lookup_slots: &shared_auth.lookup_slots,
failed_logins: &shared_auth.failed_logins,
peer: &peer,
cancel: &cancel,
@@ -1056,11 +1064,17 @@ async fn resolve_certificate_principal(
error!(%peer, "a client certificate named a principal but nothing can
look it up");
return None;
};
- // The lookup is an Iggy call per handshake, so it shares the
authentication slots with SASL.
- let lookup = within_auth_budget(ctx, "certificate identity lookup", |slot|
async move {
- let _slot = slot;
- resolver.resolve(username).await
- });
+ // The lookup is an Iggy call per handshake, so it is bounded like a SASL
login, but in slots
+ // of its own: it costs no password hash.
+ let lookup = within_auth_budget(
+ ctx,
+ ctx.lookup_slots,
+ "certificate identity lookup",
+ |slot| async move {
+ let _slot = slot;
+ resolver.resolve(username).await
+ },
+ );
let outcome = tokio::select! {
() = ctx.cancel.cancelled() => {
debug!(%peer, "certificate identity lookup abandoned by shutdown");
@@ -1134,7 +1148,7 @@ async fn authenticate_token(
debug!(%peer, "SASL authentication abandoned by shutdown");
return (HandleOutcome::Close, None);
}
- verified = within_auth_budget(ctx, "SASL authentication", |slot| {
+ verified = within_auth_budget(ctx, ctx.auth_slots, "SASL
authentication", |slot| {
authenticator.authenticate(&credentials, slot)
}) => verified,
};
@@ -1226,8 +1240,8 @@ fn describe_acls(
}
}
-/// Waits for an authentication slot, then runs `work` holding it, each within
the
-/// pre-authentication budget. `None` means a budget ran out.
+/// Waits for one of `slots`, then runs `work` holding it, each within the
pre-authentication
+/// budget. `None` means a budget ran out.
///
/// Two budgets rather than one shared deadline. The wait needs its own bound
so a queued
/// connection cannot hold a `max_connections` permit for as long as the
backlog takes to drain,
@@ -1237,6 +1251,7 @@ fn describe_acls(
/// and hand its slot to the next one while Iggy is still working.
async fn within_auth_budget<'a, T, Work, Fut>(
ctx: &ConnectionContext<'a>,
+ slots: &'a Semaphore,
operation: &'static str,
work: Work,
) -> Option<T>
@@ -1246,7 +1261,7 @@ where
{
let peer = ctx.peer;
let budget = ctx.config.pre_auth_timeout;
- let Ok(acquired) = timeout(budget, ctx.auth_slots.acquire()).await else {
+ let Ok(acquired) = timeout(budget, slots.acquire()).await else {
warn!(%peer, operation, "no authentication slot came free within the
pre-authentication budget");
return None;
};
