hubcio commented on code in PR #4222:
URL: https://github.com/apache/iggy/pull/4222#discussion_r4061941930
##########
gateways/kafka/src/server.rs:
##########
@@ -307,17 +411,168 @@ fn enable_tcp_keepalive(stream: &TcpStream) ->
std::io::Result<()> {
Ok(())
}
+/// Per-connection borrows that every frame's routing needs and none of it
changes.
+///
+/// Bundled rather than passed one by one: the routing decision already takes
the mutable SASL
+/// state, the header and the body, and four more parameters would put it over
the argument limit
+/// for no gain in clarity.
+struct ConnectionContext<'a> {
+ config: &'a GatewayConfig,
+ broker: &'a BrokerAdvertise,
+ authenticator: Option<&'a dyn SaslAuthenticator>,
+ auth_slots: &'a Semaphore,
+ peer: &'a SocketAddr,
+}
+
+/// Decides what one decoded frame earns, advancing `sasl_state` when it
authenticates.
+///
+/// Split out of [`handle_connection`] so the loop stays about framing and
this stays about the
+/// SASL state machine.
+async fn route_frame(
+ ctx: &ConnectionContext<'_>,
+ sasl_state: &mut SaslState,
+ req: &RequestHeader,
+ body: Bytes,
+) -> HandleOutcome {
+ let peer = ctx.peer;
+
+ // The mechanism name lives in the handshake body, and the state machine
needs it to decide.
+ // Decoding it here keeps `classify` free of wire concerns; a body that
will not decode yields
+ // `None`, which `classify` already treats as an unsupported mechanism.
+ let mechanism = if req.request_api_key == API_KEY_SASL_HANDSHAKE {
+ decode_sasl_mechanism(req.request_api_version, body.clone()).ok()
+ } else {
+ None
+ };
+
+ let action = sasl_state.classify(
+ req.request_api_key,
+ req.request_api_version,
+ mechanism.as_deref(),
+ );
+ match action {
+ SaslAction::Dispatch => handle_request_bounded(
+ req.request_api_key,
+ req.request_api_version,
+ body,
+ ctx.broker,
+ ctx.config.max_frame_size,
+ ctx.config.sasl_enabled,
+ ),
+ SaslAction::DispatchFirstApiVersions => {
+ let outcome = handle_request_bounded(
+ req.request_api_key,
+ req.request_api_version,
+ body,
+ ctx.broker,
+ ctx.config.max_frame_size,
+ ctx.config.sasl_enabled,
+ );
+ // Spend the allowance only on an answer the client can use. A
refusal it is entitled
+ // to retry at a lower version, which is what the KIP-511
downgrade path does, must not
+ // consume the single attempt and strand a conformant client on
the retry.
+ if answered_successfully(&outcome) {
Review Comment:
warning: the pre-auth ApiVersions allowance is spent only on a success
answer, so a peer that repeats a rejected version holds a max_connections
permit forever. count every answered frame against it, and keep the KIP-511
downgrade retry working.
##########
gateways/kafka/src/server.rs:
##########
@@ -554,8 +927,9 @@ pub async fn read_frame(
/// Returns the [`WorkerGuard`]; it must be held for the lifetime of `main`
(dropping it stops the
/// worker thread and any buffered-but-unflushed log lines are lost) - see
`main.rs`.
pub fn init_tracing() -> WorkerGuard {
- let filter = tracing_subscriber::EnvFilter::try_from_default_env()
- .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info"));
+ let filter = tracing_subscriber::EnvFilter::new(sdk_quieted_filter(
+ std::env::var("RUST_LOG").ok().as_deref(),
Review Comment:
nit: the doc at line 1305 still names `EnvFilter::try_from_default_env`, but
this line reads RUST_LOG with `std::env::var`. update the doc.
##########
gateways/kafka/src/main.rs:
##########
@@ -164,6 +195,19 @@ fn load_config() -> Result<GatewayConfig, String> {
config.write_timeout =
Duration::from_secs(parse_positive("IGGY_KAFKA_WRITE_TIMEOUT_SECS", &raw)?);
}
+ if let Some(raw) = env_var("IGGY_KAFKA_SASL_ENABLED") {
+ config.sasl_enabled = parse_bool("IGGY_KAFKA_SASL_ENABLED", &raw)?;
+ }
+ if let Some(raw) = env_var("IGGY_KAFKA_PRE_AUTH_TIMEOUT_SECS") {
+ config.pre_auth_timeout =
+
Duration::from_secs(parse_positive("IGGY_KAFKA_PRE_AUTH_TIMEOUT_SECS", &raw)?);
+ }
+ if let Some(raw) = env_var("IGGY_KAFKA_MAX_CONCURRENT_AUTHENTICATIONS") {
Review Comment:
warning: this only rejects zero, so a value above Semaphore::MAX_PERMITS
panics Semaphore::new at startup. add the guard that IGGY_KAFKA_MAX_CONNECTIONS
already gets.
##########
gateways/kafka/docs/SCOPE.md:
##########
@@ -142,7 +142,15 @@ Offset persistence design
([#3540](https://github.com/apache/iggy/issues/3540)):
InitProducerId and idempotent producers
([#3545](https://github.com/apache/iggy/issues/3545)):
[`IDEMPOTENCE.md`](IDEMPOTENCE.md).
-- [ ] SASL (17, 36) if required by deployment
+Authentication design ([#3549](https://github.com/apache/iggy/issues/3549)):
+[`AUTHENTICATION.md`](AUTHENTICATION.md).
+
+- [x] SASL/PLAIN (17, 36), opt-in via `IGGY_KAFKA_SASL_ENABLED`, credentials
verified against Iggy.
+ Kept out of `SUPPORTED_RANGES` on purpose: the connection loop routes
both keys through the
+ SASL state machine before dispatch, so a gateway with the feature off
refuses them like any
Review Comment:
nit: a sasl-off gateway answers both keys with error 34 and keeps the
connection rather than closing like an unlisted key. describe the real shape,
as AUTHENTICATION.md does.
##########
gateways/kafka/docs/AUTHENTICATION.md:
##########
@@ -0,0 +1,250 @@
+# Kafka authentication and Iggy identity
+
+Status: proposed. Answers [#3549](https://github.com/apache/iggy/issues/3549)
and needs a TLS listener
+first (see [Transport security](#transport-security)). The mechanism choice
below is forced by how Iggy
+stores credentials, not preferred.
+
+## Decision
+
+Four parts.
+
+**Mechanism: SASL/PLAIN only.** SCRAM cannot be implemented against Iggy's
credential store at all, and
+nothing else in Kafka's mechanism set maps without new server-side storage.
+
+**Credentials pass straight through.** The username and password a Kafka
client sends in its PLAIN payload
+are an Iggy username and password. The gateway forwards them to a login and
keeps no credential store and
+no mapping table of its own. An Iggy account is created for each Kafka
principal, and the Iggy server stays
+the only place a credential is defined or verified.
+
+**Identity: one Iggy user per Kafka principal.** Not one shared account for
everyone. Credentials are
+verified on every connection, against Iggy, and that verification is
deliberately not cached. See
+[Verification is per
connection](#verification-is-per-connection-and-cannot-be-cached).
+
+**Authorization stays in Iggy.** The gateway does not implement its own ACL
model. It authenticates a Kafka
+client into an Iggy identity and lets the server's existing permission checks
decide what that identity can do.
+
+## Why SCRAM is out
+
+Iggy stores exactly one credential per user, an Argon2id PHC string
(`core/metadata/src/stm/user.rs:57`,
+hashed at `core/server_common/src/crypto.rs:30`, verified at
`core/server/src/dispatch/session_ops.rs:126`).
+
+SCRAM-SHA-256 and SCRAM-SHA-512 require the server to hold `StoredKey` and
`ServerKey`, both derived from
+`PBKDF2(password, salt, iterations)`, and to send the salt and iteration count
to the client in the
+server-first message. None of that is derivable from an Argon2 hash, and the
stored hash is never exposed
+through any API. There is no second credential column and no per-mechanism
credential table anywhere in
+the tree.
+
+Supporting SCRAM therefore means a new replicated credential table plus
derivation at user-creation time,
+which is a change to Iggy's own wire format and state machine. That is a
product decision and does not
+belong to this gateway.
+
+SASL/OAUTHBEARER is the one other candidate, because a bearer token maps onto
an Iggy personal access
+token, which is verified by a BLAKE3 hash lookup
(`core/common/src/utils/hash.rs:25`) rather than Argon2.
+It is out of scope here and worth revisiting, since it avoids putting a
password on the wire.
+
+## Why not one Iggy session per Kafka connection
+
+The obvious reading of "one-to-one user mapping" is one authenticated Iggy
connection behind every Kafka
+connection. The cost rules it out.
+
+| Per login | Cost |
+| ----------- | ------ |
+| Argon2id verify, run inline on the shard thread | 19 MiB, t=2, p=1 |
+| Replicated VSR `Register` | one consensus commit |
+| Leader settlement round trip, plus a second login if it redirects | one or
two round trips |
+| Session slots, cluster-wide across all transports | 8192, oldest committed
evicted past it |
+
+The Argon2 verify runs on the shard's compio runtime, which has no blocking
pool. A burst of Kafka
+connections therefore stalls that shard's entire request loop, including
produce and fetch traffic that has
+nothing to do with authentication. The 8192 ceiling
(`core/consensus/src/impls.rs:178`) is cluster-wide, so
+connection churn on the Kafka side turns directly into session-table pressure
and eviction on the Iggy side.
+
+## Why not a shared service account
+
+The other option in the issue is authenticating every Kafka client against one
`kafka-bridge` account. It
+removes the cost above and creates a worse problem: every Kafka client would
reach Iggy as the same
+privileged principal, so Iggy's own permission model could no longer
distinguish them. The gateway would
+have to carry an ACL model of its own, and become the only thing standing
between any authenticated Kafka
+client and every stream the bridge account can reach.
+
+That is a larger security surface than the one it removes.
+
+## Verification is per connection, and cannot be cached
+
+A Kafka principal is the SASL `authcid`. Every SASL exchange verifies the
presented credentials by
+logging into Iggy with them, and that verification happens on every connection.
+
+The gateway reaches Iggy over the binary TCP transport only. Nothing uses the
HTTP API, so there is no way
+to check a password without establishing a full session. First contact for a
connection therefore costs one
+Argon2 verify, run inline on the shard thread, and one replicated `Register`.
+
+An earlier revision of this document proposed caching the authenticated client
per principal and attaching
+later connections to it. That is wrong, and implementing it surfaced why. A
cache keyed on the username
+alone lets a second connection present *any* password for a principal already
seen, which is an
+authentication bypass rather than an optimisation. Keying it on the credential
instead means holding
+something password-equivalent in gateway memory for every principal seen.
Neither is acceptable, so
+verification stays on the connection path.
+
+Two consequences worth naming rather than discovering:
+
+- **Login rate tracks connection rate.** A client that reconnects frequently
drives an Argon2 verify per
+ reconnect, on a shard runtime with no blocking pool, where it competes with
ordinary produce and fetch
+ traffic on that shard. Measured on debug builds, a single login takes about
10 ms and 32 concurrent
+ logins take about 75 ms each, with throughput flattening near 410 per
second. The shape is the point,
+ latency climbing while throughput plateaus, not the absolute numbers.
`MANUAL_TESTING.md` has the table.
+- **Nothing holds the verified session.** The connection that proved a
credential drops it immediately,
+ because no handler consumes an Iggy session yet. When Produce and Fetch need
one
+ ([#3535](https://github.com/apache/iggy/issues/3535),
+ [#3536](https://github.com/apache/iggy/issues/3536)), the question of how a
principal's data client is
+ pooled becomes live, and it is a separate question from how its credentials
are verified. Pooling clients
+ is gateway-side code over the existing public SDK and needs no SDK change.
+
+If either consequence starts to hurt, the lever is below, and it is a smaller
change than a credential
+cache because it moves cost off the server without weakening what the gateway
checks.
+
+### Personal access tokens, if login cost becomes a problem
+
+Not part of the baseline. The gateway logs in with the username and password
the Kafka client presented,
+and that is the whole credential path. This section records why a token is the
lever to reach for if the
+login rate turns out to hurt, and what it costs to pull.
+
+Argon2 is the expensive half of a login and the only half that runs inline on
a shard thread with no
+blocking pool. A token login replaces it with a hash lookup. The replicated
`Register` is unchanged either
+way, so tokens buy server CPU, not consensus, and only under churn: a cached
client that stays connected
+pays neither.
+
+Minting is self-scoped, so a token can only be created for a principal while
authenticated as that
+principal. First contact therefore always pays the password path regardless,
and a token would pay off only
+afterwards, on internal SDK reconnects, pool growth and cache repopulation.
Credentials also cannot be
+swapped on a live client, so a token would apply to the next client built for
that principal, never the one
+that minted it.
+
+What makes it a real cost rather than a free optimisation: the raw token is
returned exactly once and is
+never retrievable, so a restarted gateway cannot recover what it minted and
must delete before creating
+again, handling `PersonalAccessTokenAlreadyExists` (51). Names have to be
instance-scoped or two gateways
+fight over one token. A user holds at most `max_tokens_per_user` tokens, 100
by default
+(`core/server/config.toml:305`), and tokens leaked by instances that never
cleaned up count against that.
+Token expiry has to outlive the cache TTL with margin. The gateway would also
be holding bearer credentials
+carrying the full rights of their users with no scope narrowing, in memory,
for every principal it has seen.
+
+Measure the login rate before taking any of that on.
+
+## Handshake
+
+Per connection, with SASL enabled. Only ApiVersions and SaslHandshake are
legal before authentication.
+
+| State | Input | Action |
+| ------- | ------- | -------- |
+| `AwaitHandshake` | ApiVersions (18) | answer, stay |
+| | SaslHandshake (17) v1, supported mechanism | answer `error_code` 0, go to
`AwaitToken` |
+| | SaslHandshake (17) v1, unknown mechanism | answer 33 with the mechanism
list, then close |
+| | SaslHandshake (17) v0 | answer 35 with the mechanism list, then close |
+| | anything else | answer 34 shaped for that API, then close |
+| `AwaitToken` | SaslAuthenticate (36), credentials valid | answer 0, empty
`auth_bytes`, go to `Authenticated` |
+| | SaslAuthenticate (36), credentials rejected | answer 58 with a generic
message, then close |
+| | anything else | answer 34, then close |
+| `Authenticated` | any supported API, including ApiVersions again | serve |
+| `AwaitHandshake` | a second ApiVersions | answer 34, then close (one is
allowed, as on a real broker) |
+| `AwaitToken` | SaslAuthenticate above the advertised ceiling | close, no
schema exists at that version |
+| any | Metadata or Produce while unauthenticated | close with no body (no
error field, and acks=0 forbids one) |
+| | SaslHandshake or SaslAuthenticate | answer 34, keep the connection |
+
+Notes that decide the implementation.
+
+- **Handshake v1 only.** The handshake version selects the token framing
(KIP-152). Refusing v0 means the
+ legacy bare-token mode is never entered, so the frame reader stays
header-parsing-only. Only pre-1.0
+ Kafka clients are excluded.
+- **`session_lifetime_ms` is 0.** No KIP-368 re-authentication, which keeps
mid-connection identity change
+ out of scope. That matters beyond convenience, see the re-authentication
note below.
+- **ApiVersions is answered twice.** The Java client sends it once before the
handshake and again after
+ authenticating, so it must stay legal in both states.
+- **Produce with `acks=0` stays silent.** Answering an unauthenticated
fire-and-forget produce desyncs the
+ client's correlation stream, so that case closes without writing. The
existing rationale in
+ `protocol/api.rs` applies unchanged.
+- **Pre-authentication deadline.** An unauthenticated connection currently
holds a `max_connections` permit
+ for up to the idle timeout, ten minutes by default. Authentication needs its
own, much shorter deadline.
+
+## Error mapping
+
+At authentication time a rejected credential becomes
`SASL_AUTHENTICATION_FAILED` (58) with a generic
+message. An Iggy that cannot be reached, or an overloaded gateway, closes the
connection without a body
+instead: Kafka clients treat 58 as fatal and raise it to the application, so
borrowing it for a transient
+condition turns a blip into a permanent failure for credentials that were
always correct. A close reads as
+a transport failure, which is retriable, and still says nothing about whether
the account exists. Iggy's login path already runs a dummy hash for unknown
users to avoid a user-enumeration oracle,
+so the gateway must not reintroduce one by distinguishing unknown user from
wrong password in the message
+or by returning early.
+
+After authentication, Iggy reports exactly one permission-denied code,
`IggyError::Unauthorized` (41,
+`core/common/src/error/iggy_error.rs:97`), alongside `Unauthenticated` (40).
Kafka distinguishes
+`TOPIC_AUTHORIZATION_FAILED` (29), `GROUP_AUTHORIZATION_FAILED` (30) and
`CLUSTER_AUTHORIZATION_FAILED`
+(31). The gateway therefore picks the Kafka code from the operation it was
performing, not from the Iggy
+error, because the Iggy error cannot tell them apart.
+
+Iggy's data-plane permission checks read the local shard's view, so a
permission revocation is visible on
+the control plane immediately and on the data plane only after that shard
applies it. Combined with the
+principal cache TTL above, revocation is eventually consistent by two
mechanisms rather than one. Say so
+in the README rather than leaving an operator to discover it.
+
+## Transport security
+
+SASL/PLAIN puts the password in the clear on the wire, and so does Iggy's own
login. Both hops need TLS.
+
+The gateway listener has no TLS at all today, which makes a TLS listener a
prerequisite rather than a
+follow-up. The Iggy side already supports it on both ends (client at
+`core/common/src/types/configuration/tcp_config/tcp_client_config.rs:30`,
server at
+`core/configs/src/server_config/tcp.rs:31`), shipped disabled.
+
+The two hops are configured independently. `IGGY_KAFKA_IGGY_TLS_ENABLED` and
its companions encrypt
+the gateway's link to Iggy and are what make the gateway usable at all against
a TLS-only Iggy
+server, where every verification would otherwise fail as unreachable. The
Kafka-side listener is the
+half that is still missing.
+
+Two limits worth recording. There is no mTLS anywhere in the tree, every
rustls config uses
+`with_no_client_auth()`, so certificate-based Kafka client authentication
cannot map to an Iggy identity
+without an external terminator. And an Iggy password is capped at 100 bytes
with a `u8` length prefix on
+the wire (`core/common/src/http/users/defaults.rs:20`), so a Kafka client with
a longer PLAIN secret is
+rejected.
+
+## Re-authentication
+
+Iggy's only correct re-authentication is logout followed by login, which drops
and re-mints the VSR session.
+Sending a login on a still-bound connection takes a replay branch that
verifies the new credentials and
+returns the new user id to the caller while leaving the server bound to the
previous user
+(`core/server/src/dispatch/session_ops.rs:232`). Building Kafka
re-authentication on that would silently
+cross identities.
+
+Returning `session_lifetime_ms` of 0 avoids the question entirely for now.
Whoever implements KIP-368 later
+has to resolve it first.
+
+## Out of scope
+
+- SCRAM-SHA-256 and SCRAM-SHA-512, blocked on credential storage that does not
exist.
+- SASL/OAUTHBEARER and GSSAPI.
+- mTLS and certificate-based identity.
+- Mapping Kafka ACL administration APIs onto Iggy permissions. Authorization
is enforced, but the
+ `DescribeAcls` and `CreateAcls` API keys stay unimplemented.
+- KIP-368 re-authentication.
+
+## Open questions
+
+1. **Who creates the Iggy accounts?** Each Kafka principal needs an Iggy user,
since its credentials are
+ the ones the Kafka client sends. An operator can create them with the
existing user API
+ (`core/common/src/traits/user_client.rs:38`), or the gateway can provision
on first SASL login, which
+ needs the `manage_users` permission and means the gateway holds a
credential that can create users.
+ Auto-provisioning also cannot work here: the gateway only ever sees a
password it cannot validate
+ against anything until a user already exists, so provisioning would accept
any credential as a new
+ account. Default: an operator creates the accounts, and the gateway only
consumes them.
+2. **What is the principal cache TTL?** It bounds how long a revoked Iggy user
keeps working through the
Review Comment:
warning: this sets a 5 minute default for a cache that lines 81-86 call an
authentication bypass, and line 185 treats that same TTL as live. drop the
question and both references.
##########
gateways/kafka/src/server.rs:
##########
@@ -350,23 +605,124 @@ async fn handle_connection(
// `RequestHeader::decode` advances `body` past the header fields it
consumed via
// `Buf::advance`, so `body` is already exactly the request payload.
- let outcome = handle_request_bounded(
- req.request_api_key,
- req.request_api_version,
- body,
- &broker,
- config.max_frame_size,
- );
+ let outcome = route_frame(&ctx, &mut sasl_state, &req, body).await;
if dispatch_outcome(&mut stream, &peer, &config, &req, resp_hdr_ver,
outcome).await? {
return Ok(());
}
}
}
+/// Verifies a `SaslAuthenticate` token and answers it.
+///
+/// Every failure answers with the same code and the same message. A malformed
token, an unknown
+/// user, a wrong password and an unreachable Iggy are indistinguishable to
the client on purpose;
+/// the gateway's own log carries the difference.
+async fn authenticate_token(
Review Comment:
simplification: six parameters, four of which come straight from `ctx`. pass
`ctx` with `api_version` and `body` instead.
##########
gateways/kafka/src/server.rs:
##########
@@ -350,23 +605,124 @@ async fn handle_connection(
// `RequestHeader::decode` advances `body` past the header fields it
consumed via
// `Buf::advance`, so `body` is already exactly the request payload.
- let outcome = handle_request_bounded(
- req.request_api_key,
- req.request_api_version,
- body,
- &broker,
- config.max_frame_size,
- );
+ let outcome = route_frame(&ctx, &mut sasl_state, &req, body).await;
if dispatch_outcome(&mut stream, &peer, &config, &req, resp_hdr_ver,
outcome).await? {
return Ok(());
}
}
}
+/// Verifies a `SaslAuthenticate` token and answers it.
+///
+/// Every failure answers with the same code and the same message. A malformed
token, an unknown
+/// user, a wrong password and an unreachable Iggy are indistinguishable to
the client on purpose;
+/// the gateway's own log carries the difference.
+async fn authenticate_token(
+ authenticator: Option<&dyn SaslAuthenticator>,
+ auth_slots: &Semaphore,
+ budget: Duration,
+ api_version: i16,
+ body: Bytes,
+ peer: &SocketAddr,
+) -> HandleOutcome {
+ let failed = || sasl_authenticate_outcome(api_version,
ERROR_SASL_AUTHENTICATION_FAILED, true);
+
+ let Some(authenticator) = authenticator else {
+ // Unreachable: `run` refuses to start in this combination. Fail
closed anyway, since the
+ // alternative is admitting an unauthenticated connection.
+ error!(%peer, "SASL is enabled but no authenticator is configured");
+ return failed();
+ };
+ let Ok(auth_bytes) = decode_sasl_auth_bytes(api_version, body) else {
+ debug!(%peer, "malformed SaslAuthenticate token");
+ return failed();
+ };
+ let Ok(credentials) = parse_plain(&auth_bytes) else {
+ debug!(%peer, "malformed PLAIN initial response");
+ return failed();
+ };
+
+ // The wait for a slot and the verification itself share one deadline, and
it is the same
+ // pre-authentication budget every other unauthenticated read gets.
Without it the queue is the
+ // bound: an unauthenticated connection would sit in `acquire()` for as
long as the backlog
+ // takes to drain, holding a `max_connections` permit the whole time,
which is precisely the
+ // invariant `pre_auth_timeout` is documented to enforce.
+ let verified = tokio::time::timeout(budget, async {
+ // Acquire fails only once the semaphore is closed, which this gateway
never does.
+ let Ok(_slot) = auth_slots.acquire().await else {
+ error!(%peer, "authentication slots unavailable");
+ return None;
+ };
+ Some(authenticator.authenticate(&credentials).await)
+ })
+ .await;
+
+ let Ok(Some(result)) = verified else {
+ // Overloaded or shutting down. Close rather than answer 58: a Kafka
client treats that
+ // code as fatal and surfaces it to the application, and nothing here
says the credentials
+ // were wrong. A close reads as a transport failure, which is
retriable.
+ warn!(%peer, "authentication did not complete within the
pre-authentication budget");
+ return HandleOutcome::Close;
+ };
+
+ match result {
+ Ok(()) => {
+ debug!(%peer, "SASL authentication succeeded");
+ sasl_authenticate_outcome(api_version, ERROR_NONE, false)
+ }
+ // A rejection is the client's problem and is terminal, so it earns a
parseable 58.
+ Err(AuthError::Rejected) => {
+ debug!(%peer, "SASL authentication rejected");
+ failed()
+ }
+ // An outage is not. Kafka clients treat 58 as fatal and surface it to
the application, so
+ // answering with it would turn a momentary Iggy blip into a permanent
authentication error
+ // for credentials that were always correct. Closing without a body
reads as a transport
+ // failure instead, which is retriable, and still tells the client
nothing about whether
+ // the account exists.
+ Err(AuthError::Unavailable) => {
+ warn!(%peer, "SASL authentication could not be completed; Iggy is
unreachable");
+ HandleOutcome::Close
+ }
+ }
+}
+
+/// Whether an `ApiVersions` outcome carries `error_code` 0, meaning the
client got a usable
+/// answer rather than one it is expected to retry at a lower version.
+fn answered_successfully(outcome: &HandleOutcome) -> bool {
+ let (HandleOutcome::Respond(body) | HandleOutcome::RespondThenClose(body))
= outcome else {
+ return false;
+ };
+ // `ApiVersions` puts its error code in the first two bytes of the body at
every version.
+ body.len() >= 2 && i16::from_be_bytes([body[0], body[1]]) == ERROR_NONE
+}
+
+/// Answers a request that is well-formed but not legal in this connection's
SASL state.
+///
+/// `keep_open` marks the one case a real broker does not treat as fatal: a
SASL request arriving
+/// on a connection that already authenticated.
+fn illegal_state_outcome(
+ api_key: i16,
+ api_version: i16,
+ keep_open: bool,
+ sasl_enabled: bool,
+) -> HandleOutcome {
+ match api_key {
Review Comment:
nit: an authenticated client that sends SaslHandshake v2 gets a closed
connection instead of the error 34 this path promises. move the version check
ahead of the encoders.
##########
gateways/kafka/src/protocol/api.rs:
##########
@@ -108,7 +134,7 @@ impl HandleOutcome {
#[must_use]
pub fn expect_response(self, msg: &str) -> Bytes {
match self {
- Self::Respond(body) => body,
+ Self::Respond(body) | Self::RespondThenClose(body) => body,
Review Comment:
nit: the summary at line 129 promises a panic for anything but
`Self::Respond`, but this arm also returns the body of
`Self::RespondThenClose`. name both variants in the doc.
##########
gateways/kafka/src/server.rs:
##########
@@ -76,6 +133,9 @@ impl Default for GatewayConfig {
read_timeout: Duration::from_secs(15),
write_timeout: Duration::from_secs(10),
shutdown_drain_timeout: Duration::from_secs(25),
+ sasl_enabled: false,
+ pre_auth_timeout: Duration::from_secs(15),
+ max_concurrent_authentications: 16,
Review Comment:
warning: 16 concurrent authentications by default is above the shard count
of any node with 16 or fewer physical cores. lower it to 4 and let large nodes
raise it.
##########
gateways/kafka/docs/AUTHENTICATION.md:
##########
@@ -0,0 +1,250 @@
+# Kafka authentication and Iggy identity
+
+Status: proposed. Answers [#3549](https://github.com/apache/iggy/issues/3549)
and needs a TLS listener
Review Comment:
nit: this ships sasl, so the "proposed" status is stale. mark the document
implemented, since the TLS prerequisite still holds.
##########
gateways/kafka/src/auth.rs:
##########
@@ -0,0 +1,352 @@
+// 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.
+
+//! Turning SASL credentials into a verified Iggy identity.
+//!
+//! The gateway keeps no credential store and no user mapping. A Kafka
principal's username and
+//! password are an Iggy username and password, so authenticating is
forwarding them to an Iggy
+//! login and seeing whether it succeeds. `docs/AUTHENTICATION.md` has the
reasoning.
+
+use std::time::Duration;
+
+use async_trait::async_trait;
+use iggy::prelude::{AutoLogin, Client, Credentials, IggyClientBuilder,
IggyError};
+use tracing::{debug, warn};
+
+use crate::protocol::sasl::PlainCredentials;
+
+/// Bound on one credential verification.
+///
+/// Covers the dial and the login. Teardown has its own, much smaller budget
+/// ([`TEARDOWN_TIMEOUT`]) so that one attempt cannot hold an authentication
permit for twice this
+/// long. The caller bounds the whole thing again from outside, against its
pre-authentication
+/// budget, because a permit wait is not covered here at all.
+///
+/// A verification that has not answered inside this is indistinguishable,
from the Kafka client's
+/// side, from one that failed, and the client is holding a connection open
waiting for it. Shorter
+/// than the bridge's own 15s request budget because a login is a bounded
handshake against a
+/// server that is either reachable or not, not an arbitrary data operation.
+///
+/// "Bounded" is not "one round trip". `establish_session` also performs a
cluster-metadata lookup
+/// for leader settlement, and on a redirect it reconnects and logs in a
second time, so a single
+/// verification can cost two logins against a clustered deployment. The
measurements in
+/// `docs/MANUAL_TESTING.md` were taken against one node and therefore never
exercise that path.
+const VERIFY_TIMEOUT: Duration = Duration::from_secs(10);
+
+/// Retries the dial makes before giving up, not the SDK's unlimited default.
+///
+/// A Kafka client retries the whole authentication itself, so an unbounded
inner loop would only
+/// hide the failure underneath one the client cannot see.
+const VERIFY_RECONNECTION_RETRIES: u32 = 1;
+
+/// Budget for tearing the verification client down again.
+///
+/// Deliberately far shorter than [`VERIFY_TIMEOUT`]. Teardown happens while
the caller still holds
+/// an authentication permit, so giving it the full verify budget would let
one attempt occupy a
+/// slot for twice as long as the doc on [`VERIFY_TIMEOUT`] claims the whole
operation can take.
+/// Nothing is lost by cutting it short: `Drop` aborts the heartbeat task
regardless.
+const TEARDOWN_TIMEOUT: Duration = Duration::from_secs(1);
+
+/// Why a SASL exchange did not produce a verified identity.
+///
+/// Both variants reach the client as the same `SASL_AUTHENTICATION_FAILED`
with the same generic
+/// message. They are distinct here only so the gateway's own log can tell an
operator whether
+/// their Iggy server is unreachable or their user typed the wrong password.
+#[derive(Debug)]
+pub enum AuthError {
+ /// Iggy rejected the credentials.
+ Rejected,
+ /// Iggy could not be reached, or did not answer inside the verification
timeout.
+ Unavailable,
+}
+
+/// Verifies Kafka-supplied credentials.
+///
+/// A trait rather than a concrete type so the protocol tests can drive the
whole SASL exchange
+/// over a socket without an Iggy server behind it.
+#[async_trait]
+pub trait SaslAuthenticator: Send + Sync + std::fmt::Debug {
+ /// Returns `Ok(())` when `credentials` name a real, active Iggy user.
+ ///
+ /// # Errors
+ ///
+ /// Returns [`AuthError::Rejected`] when Iggy refuses the credentials and
+ /// [`AuthError::Unavailable`] when it cannot be asked.
+ async fn authenticate(&self, credentials: &PlainCredentials) -> Result<(),
AuthError>;
+}
+
+/// How the verifier reaches Iggy.
+///
+/// Separate from the gateway's own listener security. A deployment can
terminate TLS on the Kafka
+/// side and still speak plain TCP to a co-located Iggy, or the reverse, and
the two are configured
+/// independently because they protect different hops.
+#[derive(Debug, Clone, Default)]
+pub struct IggyTls {
+ pub enabled: bool,
+ /// Name checked against the server certificate. Empty means derive it
from the address.
+ pub domain: String,
+ /// PEM roots to trust. Unset uses the SDK's bundled roots, not the system
trust store.
+ pub ca_file: Option<String>,
+}
+
+/// Verifies credentials by logging into a real Iggy server with them.
+///
+/// Every verification opens its own connection, logs in, and shuts down
again. That is one Argon2
+/// verify and one replicated `Register` per authenticated Kafka connection,
which is the cost
+/// `docs/AUTHENTICATION.md` describes and does not hide.
+///
+/// It is also why there is no credential cache here. Caching a verification
keyed on the username
+/// alone would let a second connection present any password for a principal
already seen, which
+/// is an authentication bypass rather than an optimisation. Caching it keyed
on the credential
+/// means storing something password-equivalent in gateway memory. Neither is
worth doing before
+/// there is a handler whose throughput the login cost actually limits, and
today Produce and Fetch
+/// are still stubs, so no verified session has a consumer to be held for.
+#[derive(Debug)]
+pub struct IggyAuthenticator {
+ address: String,
+ tls: IggyTls,
+}
+
+/// Iggy address used when `IGGY_KAFKA_IGGY_ADDR` is unset.
+///
+/// Matches `bridge::config`'s own default. The two read the same variable for
the same purpose, so
+/// they must not disagree about what it falls back to.
+const DEFAULT_IGGY_ADDR: &str = "127.0.0.1:8090";
+
+impl IggyAuthenticator {
+ /// The complete set of `IGGY_KAFKA_*` vars this type reads, for `main`'s
unknown-var guard.
+ ///
+ /// `IGGY_KAFKA_IGGY_ADDR` is listed here even though `bridge::config`
also carries it. This
+ /// type reads it directly, and relying on the bridge's list to cover it
is the cross-list
+ /// coupling the guard exists to avoid: a bridge rename would then break
SASL startup on a
+ /// variable SASL reads itself.
+ pub const KNOWN_ENV_VARS: &'static [&'static str] = &[
+ "IGGY_KAFKA_IGGY_ADDR",
+ "IGGY_KAFKA_IGGY_TLS_ENABLED",
+ "IGGY_KAFKA_IGGY_TLS_DOMAIN",
+ "IGGY_KAFKA_IGGY_TLS_CA_FILE",
+ ];
+
+ #[must_use]
+ pub const fn new(address: String) -> Self {
+ Self {
+ address,
+ tls: IggyTls {
+ enabled: false,
+ domain: String::new(),
+ ca_file: None,
+ },
+ }
+ }
+
+ #[must_use]
+ pub fn with_tls(mut self, tls: IggyTls) -> Self {
+ self.tls = tls;
+ self
+ }
+
+ #[must_use]
+ pub const fn is_tls_enabled(&self) -> bool {
+ self.tls.enabled
+ }
+
+ /// Reads the Iggy address and transport security from the environment.
+ ///
+ /// No credentials of its own: every verification uses the credentials the
Kafka client
+ /// presented, which is the whole point of forwarding them rather than
mapping them.
+ ///
+ /// # Errors
+ ///
+ /// Returns a message naming the offending variable when
`IGGY_KAFKA_IGGY_TLS_ENABLED` is set
+ /// to anything but `true` or `false`. Defaulting a mistyped security
switch to off is how a
+ /// deployment ends up sending passwords in the clear while believing it
does not.
+ pub fn from_env() -> Result<Self, String> {
+ let address =
+ std::env::var("IGGY_KAFKA_IGGY_ADDR").unwrap_or_else(|_|
DEFAULT_IGGY_ADDR.to_string());
+ let enabled = match
std::env::var("IGGY_KAFKA_IGGY_TLS_ENABLED").as_deref() {
+ Ok("true") => true,
Review Comment:
nit: this repeats the true/false parse from `parse_bool` in main.rs, and
nothing tests `from_env` or the TLS guard below. move the helper into the
library and reuse it here.
##########
gateways/kafka/src/server.rs:
##########
@@ -185,6 +264,25 @@ impl KafkaGateway {
listener: TcpListener,
mut shutdown: broadcast::Receiver<()>,
) -> Result<()> {
+ if !self.config.sasl_enabled && self.authenticator.is_some() {
+ // The mirror of the guard below, and the quieter mistake: a
verifier attached while the
+ // flag is off means every connection is served unauthenticated,
with nothing in the log
+ // to say so. Refusing to start is the only way that failure is
visible.
+ return Err(KafkaProtocolError::InvalidConfig(
+ "an authenticator is configured but SASL is disabled; every
connection would be \
+ served unauthenticated. Set IGGY_KAFKA_SASL_ENABLED=true, or
remove the \
+ authenticator"
+ .into(),
+ ));
+ }
+ if self.config.sasl_enabled && self.authenticator.is_none() {
+ return Err(KafkaProtocolError::InvalidConfig(
+ "SASL is enabled but no authenticator is configured; every
client would be \
+ rejected. Set IGGY_KAFKA_IGGY_ADDR and the Iggy credentials,
or unset \
Review Comment:
nit: the gateway reads no credential variable, so telling operators to set
"the Iggy credentials" sends them looking for a setting that does not exist.
drop the clause or name the real variable.
##########
gateways/kafka/tests/common/server.rs:
##########
@@ -38,6 +41,7 @@ pub async fn spawn_test_server() -> (SocketAddr,
broadcast::Sender<()>) {
read_timeout: Duration::from_secs(5),
write_timeout: Duration::from_secs(5),
shutdown_drain_timeout: Duration::from_secs(5),
+ ..GatewayConfig::default()
Review Comment:
simplification: the default supplies five of these fields, and
`spawn_test_server_with_config` overwrites `bind_addr` anyway. drop the dead
fields, also at lines 91, 123, 154, 306, 503 of
`tests/listener_robustness_tests.rs`.
##########
gateways/kafka/src/auth.rs:
##########
@@ -0,0 +1,352 @@
+// 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.
+
+//! Turning SASL credentials into a verified Iggy identity.
+//!
+//! The gateway keeps no credential store and no user mapping. A Kafka
principal's username and
+//! password are an Iggy username and password, so authenticating is
forwarding them to an Iggy
+//! login and seeing whether it succeeds. `docs/AUTHENTICATION.md` has the
reasoning.
+
+use std::time::Duration;
+
+use async_trait::async_trait;
+use iggy::prelude::{AutoLogin, Client, Credentials, IggyClientBuilder,
IggyError};
+use tracing::{debug, warn};
+
+use crate::protocol::sasl::PlainCredentials;
+
+/// Bound on one credential verification.
+///
+/// Covers the dial and the login. Teardown has its own, much smaller budget
+/// ([`TEARDOWN_TIMEOUT`]) so that one attempt cannot hold an authentication
permit for twice this
+/// long. The caller bounds the whole thing again from outside, against its
pre-authentication
+/// budget, because a permit wait is not covered here at all.
+///
+/// A verification that has not answered inside this is indistinguishable,
from the Kafka client's
+/// side, from one that failed, and the client is holding a connection open
waiting for it. Shorter
+/// than the bridge's own 15s request budget because a login is a bounded
handshake against a
+/// server that is either reachable or not, not an arbitrary data operation.
+///
+/// "Bounded" is not "one round trip". `establish_session` also performs a
cluster-metadata lookup
+/// for leader settlement, and on a redirect it reconnects and logs in a
second time, so a single
+/// verification can cost two logins against a clustered deployment. The
measurements in
+/// `docs/MANUAL_TESTING.md` were taken against one node and therefore never
exercise that path.
+const VERIFY_TIMEOUT: Duration = Duration::from_secs(10);
+
+/// Retries the dial makes before giving up, not the SDK's unlimited default.
+///
+/// A Kafka client retries the whole authentication itself, so an unbounded
inner loop would only
+/// hide the failure underneath one the client cannot see.
+const VERIFY_RECONNECTION_RETRIES: u32 = 1;
+
+/// Budget for tearing the verification client down again.
+///
+/// Deliberately far shorter than [`VERIFY_TIMEOUT`]. Teardown happens while
the caller still holds
+/// an authentication permit, so giving it the full verify budget would let
one attempt occupy a
+/// slot for twice as long as the doc on [`VERIFY_TIMEOUT`] claims the whole
operation can take.
+/// Nothing is lost by cutting it short: `Drop` aborts the heartbeat task
regardless.
+const TEARDOWN_TIMEOUT: Duration = Duration::from_secs(1);
+
+/// Why a SASL exchange did not produce a verified identity.
+///
+/// Both variants reach the client as the same `SASL_AUTHENTICATION_FAILED`
with the same generic
+/// message. They are distinct here only so the gateway's own log can tell an
operator whether
+/// their Iggy server is unreachable or their user typed the wrong password.
+#[derive(Debug)]
+pub enum AuthError {
+ /// Iggy rejected the credentials.
+ Rejected,
+ /// Iggy could not be reached, or did not answer inside the verification
timeout.
+ Unavailable,
+}
+
+/// Verifies Kafka-supplied credentials.
+///
+/// A trait rather than a concrete type so the protocol tests can drive the
whole SASL exchange
+/// over a socket without an Iggy server behind it.
+#[async_trait]
+pub trait SaslAuthenticator: Send + Sync + std::fmt::Debug {
+ /// Returns `Ok(())` when `credentials` name a real, active Iggy user.
+ ///
+ /// # Errors
+ ///
+ /// Returns [`AuthError::Rejected`] when Iggy refuses the credentials and
+ /// [`AuthError::Unavailable`] when it cannot be asked.
+ async fn authenticate(&self, credentials: &PlainCredentials) -> Result<(),
AuthError>;
+}
+
+/// How the verifier reaches Iggy.
+///
+/// Separate from the gateway's own listener security. A deployment can
terminate TLS on the Kafka
+/// side and still speak plain TCP to a co-located Iggy, or the reverse, and
the two are configured
+/// independently because they protect different hops.
+#[derive(Debug, Clone, Default)]
+pub struct IggyTls {
+ pub enabled: bool,
+ /// Name checked against the server certificate. Empty means derive it
from the address.
+ pub domain: String,
+ /// PEM roots to trust. Unset uses the SDK's bundled roots, not the system
trust store.
+ pub ca_file: Option<String>,
+}
+
+/// Verifies credentials by logging into a real Iggy server with them.
+///
+/// Every verification opens its own connection, logs in, and shuts down
again. That is one Argon2
+/// verify and one replicated `Register` per authenticated Kafka connection,
which is the cost
+/// `docs/AUTHENTICATION.md` describes and does not hide.
+///
+/// It is also why there is no credential cache here. Caching a verification
keyed on the username
+/// alone would let a second connection present any password for a principal
already seen, which
+/// is an authentication bypass rather than an optimisation. Caching it keyed
on the credential
+/// means storing something password-equivalent in gateway memory. Neither is
worth doing before
+/// there is a handler whose throughput the login cost actually limits, and
today Produce and Fetch
+/// are still stubs, so no verified session has a consumer to be held for.
+#[derive(Debug)]
+pub struct IggyAuthenticator {
+ address: String,
+ tls: IggyTls,
+}
+
+/// Iggy address used when `IGGY_KAFKA_IGGY_ADDR` is unset.
+///
+/// Matches `bridge::config`'s own default. The two read the same variable for
the same purpose, so
+/// they must not disagree about what it falls back to.
+const DEFAULT_IGGY_ADDR: &str = "127.0.0.1:8090";
Review Comment:
simplification: this duplicates `DEFAULT_IGGY_ADDR` from `bridge/config.rs`
and nothing enforces that the two agree. make the bridge constant `pub(crate)`
and use it here.
##########
gateways/kafka/src/protocol/api.rs:
##########
@@ -123,6 +149,15 @@ impl HandleOutcome {
pub const fn is_close(&self) -> bool {
matches!(self, Self::Close)
}
+
+ /// Whether applying this outcome ends the connection, whether or not it
writes first.
+ ///
+ /// Distinct from [`Self::is_close`], which asks only whether the
connection ends *without* a
+ /// response. Callers deciding whether to keep reading want this one.
+ #[must_use]
+ pub const fn closes_connection(&self) -> bool {
Review Comment:
simplification: nothing calls this - 0 call sites in the repo, and the close
flag is computed inline in `dispatch_outcome`. delete it and its doc.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
