krishvishal commented on code in PR #4222:
URL: https://github.com/apache/iggy/pull/4222#discussion_r4079663195
##########
gateways/kafka/src/server.rs:
##########
@@ -31,16 +31,49 @@ use tokio_util::task::TaskTracker;
use tracing::{debug, error, info, warn};
use tracing_appender::non_blocking::WorkerGuard;
+use crate::auth::{AuthError, SaslAuthenticator};
use crate::bridge::IggyBridge;
use crate::error::{KafkaProtocolError, Result};
use crate::protocol::api::{
- BrokerAdvertise, DEFAULT_KAFKA_PORT, GatewayState, HandleOutcome,
handle_request_bounded,
+ API_KEY_SASL_AUTHENTICATE, API_KEY_SASL_HANDSHAKE, BrokerAdvertise,
DEFAULT_KAFKA_PORT,
+ ERROR_ILLEGAL_SASL_STATE, ERROR_NONE, ERROR_SASL_AUTHENTICATION_FAILED,
+ ERROR_UNSUPPORTED_SASL_MECHANISM, ERROR_UNSUPPORTED_VERSION, GatewayState,
HandleOutcome,
+ decode_sasl_auth_bytes, decode_sasl_mechanism, encode_error_for_key,
handle_request_bounded,
+ sasl_authenticate_outcome, sasl_handshake_outcome,
};
use crate::protocol::header::{request_header_version, response_header_version};
+use crate::protocol::sasl::{
+ SASL_AUTHENTICATE_MAX_VERSION, SASL_HANDSHAKE_VERSION, SaslAction,
SaslState, parse_plain,
+};
use std::io;
const READ_CHUNK: usize = 65536;
+/// Builds the log filter, forcing the Iggy SDK quiet unless the operator
asked otherwise.
+///
+/// This is a credential-disclosure control, not noise reduction. The SDK logs
the username it
+/// signed in with at INFO on every successful login, and this gateway drives
that path once per
+/// authentication with a *Kafka client's* username, so at INFO every
principal that connects ends
+/// up in the gateway's log. Because the line fires only on success, what
leaks is precisely the
+/// set of valid accounts.
+///
+/// Appending the directive rather than only supplying a default is the point.
Reading `RUST_LOG`
+/// and using it verbatim silently drops this the moment anyone sets it,
including on the run
+/// commands this repository's own documentation gives. An explicit `iggy=`
directive still wins,
+/// so raising it deliberately for debugging remains possible.
+fn sdk_quieted_filter(rust_log: Option<&str>) -> String {
Review Comment:
`EnvFilter` matches a directive target by prefix, so `iggy=warn` also
catches this crate, `iggy_gateway_kafka`. The gateway therefore drops its own
INFO and DEBUG lines at every `RUST_LOG` value, including the SASL success and
rejection logs, so no authentication decision is recorded anywhere.
Append an `iggy_gateway_kafka=` directive after `iggy=warn`, because the
longer target wins. The unit test below asserts the built string rather than
the filter behavior, so it passes either way.
##########
gateways/kafka/src/main.rs:
##########
@@ -92,6 +116,7 @@ fn reject_unknown_kafka_env_vars() -> Result<(), String> {
if key.starts_with("IGGY_KAFKA_")
&& !KNOWN_KAFKA_ENV_VARS.contains(&key.as_str())
&& !IggyBridgeConfig::KNOWN_ENV_VARS.contains(&key.as_str())
+ && !IggyAuthenticator::KNOWN_ENV_VARS.contains(&key.as_str())
Review Comment:
`IggyAuthenticator::from_env` runs only when SASL is on, but this guard
accepts `IGGY_KAFKA_IGGY_TLS_*` either way. With SASL off an operator sets
`IGGY_KAFKA_IGGY_TLS_ENABLED=true`, sees no error and no warning, and nothing
reads it.
That is the same "believes TLS is on" mistake `from_env` refuses to start
for, reached through another door. Reject the TLS variables when SASL is off.
##########
gateways/kafka/src/server.rs:
##########
@@ -564,8 +931,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(
Review Comment:
`EnvFilter::new` is `parse_lossy`, so it drops a bad directive instead of
reporting one. The default level applies only when every directive fails, and
`iggy=warn` always parses, so one typo in `RUST_LOG` leaves every non-`iggy`
target with no output at any level.
This PR replaced `try_from_default_env().unwrap_or_else(|_|
EnvFilter::new("info"))`, which used to recover from exactly that. Use
`EnvFilter::try_new` and fall back to `sdk_quieted_filter(None)`.
##########
gateways/kafka/src/server.rs:
##########
@@ -364,19 +615,118 @@ 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(&state, req.request_api_key,
req.request_api_version, body)
- .await;
+ 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(
+ ctx: &ConnectionContext<'_>,
+ api_version: i16,
+ body: Bytes,
+) -> HandleOutcome {
+ let peer = ctx.peer;
+ let failed = || sasl_authenticate_outcome(api_version,
ERROR_SASL_AUTHENTICATION_FAILED, true);
+
+ let Some(authenticator) = ctx.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(ctx.config.pre_auth_timeout, async {
Review Comment:
The timeout frees the permit, but it does not stop the login already running
inside Iggy. `verify_login_credentials` hashes with Argon2 in a plain
synchronous function, and the dispatch drain task is detached, so a dropped
socket cancels nothing.
`max_concurrent_authentications` therefore bounds gateway-side verifications
only, not the hashing load on the shard that its own doc claims to bound. Start
the budget at permit acquisition, and soften the claim.
##########
gateways/kafka/README.md:
##########
@@ -76,6 +80,57 @@ Iggy deduplicates writes on its own partition plane, and
that does not close thi
guards the hop from the gateway to Iggy rather than the hop from the producer
to the gateway.
[docs/IDEMPOTENCE.md](docs/IDEMPOTENCE.md) has the detail and what closing it
needs.
+## Authentication ([#3549](https://github.com/apache/iggy/issues/3549))
+
+Off by default. With `IGGY_KAFKA_SASL_ENABLED=true` the gateway requires
SASL/PLAIN before it serves
+any other API, and it verifies the credentials by logging into Iggy with them.
+
+The username and password a Kafka client sends are an **Iggy** username and
password. There is no
+mapping table and no credential store in the gateway: create an Iggy user for
each Kafka principal
+and point the client at it. Credentials are verified against
`IGGY_KAFKA_IGGY_ADDR`.
+
+```bash
+IGGY_KAFKA_SASL_ENABLED=true IGGY_KAFKA_IGGY_ADDR=127.0.0.1:8090 cargo run -p
iggy-gateway-kafka
+```
+
+Transport security to Iggy is configured separately from the Kafka side,
because the two protect
+different hops:
+
+| Variable | Default | Description |
+| --- | --- | --- |
+| `IGGY_KAFKA_IGGY_TLS_ENABLED` | `false` | Encrypt the gateway's link to Iggy
(`true` or `false`, nothing else). Required if the Iggy server only accepts
TLS, otherwise every verification fails as unreachable |
Review Comment:
Only the SASL verification client reads these three variables. The bridge
builds its Iggy client with no TLS at all, and that connection carries
`IGGY_KAFKA_IGGY_PASSWORD`.
Say that the variables cover the credential check link only, and that the
bridge hop stays unencrypted. `docs/AUTHENTICATION.md:200` repeats the same
wording.
##########
gateways/kafka/src/protocol/api.rs:
##########
@@ -246,3 +282,161 @@ pub const fn advertised_min_version(api_key: i16,
firewall_min: i16) -> i16 {
firewall_min
}
}
+
+/// Advertised only when SASL is switched on, and deliberately absent from
[`SUPPORTED_RANGES`].
+///
+/// These two keys never reach [`handle_request_bounded`]: the connection loop
routes them through
+/// the SASL state machine before dispatch. Keeping them out of the firewall
table means a gateway
+/// with SASL off treats them as any other unknown key and closes, which is
what stops enabling the
Review Comment:
With SASL off the connection starts `Authenticated`, so keys 17 and 36 get
`ILLEGAL_SASL_STATE` and the socket stays open. `sasl_tests.rs:439` and
`docs/SCOPE.md` already pin that, so this comment is the outlier.
The same wrong claim sits at `handlers/api_versions.rs:76`. Correct both.
##########
gateways/kafka/docs/AUTHENTICATION.md:
##########
@@ -0,0 +1,251 @@
+# Kafka authentication and Iggy identity
+
+Status: implemented. Answers
[#3549](https://github.com/apache/iggy/issues/3549). The gateway listener
+still has no TLS, so PLAIN stays confined to a trusted network until that
lands (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
Review Comment:
No code defers to Iggy's permission checks. The verified client is shut down
as soon as the credential clears, and nothing carries the principal onto the
data plane.
This document is marked as implemented, so the present tense reads as a
claim about today. Write the paragraph as intent for later work.
##########
gateways/kafka/docs/AUTHENTICATION.md:
##########
@@ -0,0 +1,251 @@
+# Kafka authentication and Iggy identity
+
+Status: implemented. Answers
[#3549](https://github.com/apache/iggy/issues/3549). The gateway listener
+still has no TLS, so PLAIN stays confined to a trusted network until that
lands (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.
+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` | ApiVersions past the allowance | answer 34, then close
(two are allowed: the one a real broker allows, plus the KIP-511 downgrade
retry) |
+| `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.
+- **The pre-authentication allowance counts answers, not successes.** A
refusal a client may retry at a
+ lower version is one it may also repeat, and every frame resets the
pre-authentication deadline, so
+ spending the allowance only on a usable answer leaves the connection
unbounded.
+- **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
Review Comment:
No such mapping exists. `bridge/error.rs:156` picks the Kafka code from the
Iggy error rather than from the operation in flight, and neither error 30 nor
error 31 appears anywhere in the crate.
Move this paragraph under a planned heading, or drop it.
##########
gateways/kafka/tests/sasl_tests.rs:
##########
@@ -0,0 +1,974 @@
+// 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.
+
+//! SASL authentication over a real socket.
+//!
+//! Drives the whole exchange against a running `KafkaGateway` with a stub
verifier standing in
+//! for Iggy, so these cover the listener's state machine rather than Iggy's
credential checking.
+
+use std::net::SocketAddr;
+use std::sync::Arc;
+use std::time::Duration;
+
+use async_trait::async_trait;
+use bytes::{BufMut, Bytes, BytesMut};
+use tokio::io::AsyncWriteExt;
+use tokio::net::TcpStream;
+
+use iggy_gateway_kafka::GatewayConfig;
+use iggy_gateway_kafka::auth::{AuthError, SaslAuthenticator};
+use iggy_gateway_kafka::protocol::sasl::{MAX_PRE_AUTH_API_VERSIONS,
PlainCredentials};
+
+#[path = "common/codec.rs"]
+mod codec;
+#[path = "common/server.rs"]
+mod server;
+#[path = "common/tcp.rs"]
+mod tcp;
+
+use server::spawn_test_server_with_authenticator;
+use tcp::{ByteRead, build_request_frame, parse_response_payload,
read_byte_with_timeout};
+
+const API_KEY_PRODUCE: i16 = 0;
+const API_KEY_METADATA: i16 = 3;
+const API_KEY_SASL_HANDSHAKE: i16 = 17;
+const API_KEY_API_VERSIONS: i16 = 18;
+const API_KEY_SASL_AUTHENTICATE: i16 = 36;
+
+const ERROR_NONE: i16 = 0;
+const ERROR_UNSUPPORTED_SASL_MECHANISM: i16 = 33;
+const ERROR_ILLEGAL_SASL_STATE: i16 = 34;
+const ERROR_UNSUPPORTED_VERSION: i16 = 35;
+const ERROR_SASL_AUTHENTICATION_FAILED: i16 = 58;
+
+const HANDSHAKE_VERSION: i16 = 1;
+const AUTHENTICATE_VERSION: i16 = 1;
+
+/// Accepts one specific credential pair and rejects everything else.
+#[derive(Debug)]
+struct FixedCredentialAuthenticator {
+ username: &'static str,
+ password: &'static str,
+}
+
+#[async_trait]
+impl SaslAuthenticator for FixedCredentialAuthenticator {
+ async fn authenticate(&self, credentials: &PlainCredentials) -> Result<(),
AuthError> {
+ use secrecy::ExposeSecret;
+ let matches = credentials.username == self.username
+ && credentials.password.expose_secret() == self.password;
+ if matches {
+ Ok(())
+ } else {
+ Err(AuthError::Rejected)
+ }
+ }
+}
+
+fn sasl_config() -> GatewayConfig {
+ GatewayConfig {
+ sasl_enabled: true,
+ idle_timeout: Duration::from_secs(5),
+ read_timeout: Duration::from_secs(5),
+ write_timeout: Duration::from_secs(5),
+ shutdown_drain_timeout: Duration::from_secs(5),
+ ..GatewayConfig::default()
+ }
+}
+
+async fn spawn_sasl_gateway() -> SocketAddr {
+ let authenticator = Arc::new(FixedCredentialAuthenticator {
+ username: "alice",
+ password: "s3cret",
+ });
+ let (addr, shutdown) = spawn_test_server_with_authenticator(sasl_config(),
authenticator).await;
+ // Held for the whole test: dropping the sender shuts the gateway down
mid-exchange.
+ std::mem::forget(shutdown);
+ addr
+}
+
+/// `SaslHandshake` body: one legacy (non-compact) string, at both v0 and v1.
+fn handshake_body(mechanism: &str) -> Bytes {
+ let mut buf = BytesMut::new();
+ buf.put_i16(i16::try_from(mechanism.len()).expect("test mechanism fits
i16"));
+ buf.extend_from_slice(mechanism.as_bytes());
+ buf.freeze()
+}
+
+/// `SaslAuthenticate` body at v0/v1: one legacy length-prefixed bytes field.
+fn authenticate_body(auth_bytes: &[u8]) -> Bytes {
+ let mut buf = BytesMut::new();
+ buf.put_i32(i32::try_from(auth_bytes.len()).expect("test token fits i32"));
+ buf.extend_from_slice(auth_bytes);
+ buf.freeze()
+}
+
+/// `SaslAuthenticate` body at v2, where the token is a compact bytes field
and a tagged-fields
+/// byte follows it. Modern clients negotiate v2, so this framing is the one
they actually send.
+fn authenticate_body_v2(auth_bytes: &[u8]) -> Bytes {
+ let mut buf = BytesMut::new();
+ // Compact bytes: unsigned varint of len + 1. Single byte for anything
under 127.
+ let len = u8::try_from(auth_bytes.len() + 1).expect("test token is short");
+ buf.put_u8(len);
+ buf.extend_from_slice(auth_bytes);
+ buf.put_u8(0);
+ buf.freeze()
+}
+
+fn plain_token(username: &str, password: &str) -> Vec<u8> {
+ let mut token = vec![0u8];
+ token.extend_from_slice(username.as_bytes());
+ token.push(0);
+ token.extend_from_slice(password.as_bytes());
+ token
+}
+
+/// Sends one request on an existing stream and returns the decoded response
body.
+async fn send(
+ stream: &mut TcpStream,
+ api_key: i16,
+ api_version: i16,
+ correlation_id: i32,
+ body: &[u8],
+) -> Bytes {
+ let frame = build_request_frame(
+ api_key,
+ api_version,
+ correlation_id,
+ Some("sasl-test"),
+ body,
+ );
+ stream.write_all(&frame).await.expect("write request");
+ let payload = tcp::read_response_frame(stream, 8 * 1024 * 1024).await;
+ let (echoed, response) = parse_response_payload(api_key, api_version,
payload);
+ assert_eq!(echoed, correlation_id, "correlation id must be echoed");
+ response
+}
+
+/// First `i16` of a `SaslHandshake` or `SaslAuthenticate` response body is
its error code.
+fn error_code(body: &Bytes) -> i16 {
+ assert!(body.len() >= 2, "response body is too short to hold a code");
+ i16::from_be_bytes([body[0], body[1]])
+}
+
+async fn assert_closed(stream: &mut TcpStream) {
+ let read = read_byte_with_timeout(stream, Duration::from_secs(5)).await;
+ assert!(
+ matches!(read, ByteRead::Closed),
+ "connection should have been closed, got {read:?}"
+ );
+}
+
+async fn handshake_ok(stream: &mut TcpStream) {
+ let body = send(
+ stream,
+ API_KEY_SASL_HANDSHAKE,
+ HANDSHAKE_VERSION,
+ 1,
+ &handshake_body("PLAIN"),
+ )
+ .await;
+ assert_eq!(error_code(&body), ERROR_NONE, "PLAIN must be accepted");
+}
+
+#[tokio::test]
+async fn
given_valid_credentials_when_authenticating_should_serve_normal_requests() {
+ let addr = spawn_sasl_gateway().await;
+ let mut stream = TcpStream::connect(addr).await.expect("connect");
+
+ // ApiVersions is legal before the handshake, and must advertise the SASL
keys so the client
+ // knows which token framing to use.
+ let advertised = send(&mut stream, API_KEY_API_VERSIONS, 1, 1, &[]).await;
+ assert!(
+ advertised
+ .windows(2)
+ .any(|w| i16::from_be_bytes([w[0], w[1]]) ==
API_KEY_SASL_AUTHENTICATE),
+ "SaslAuthenticate must be advertised while SASL is enabled"
+ );
+
+ handshake_ok(&mut stream).await;
+
+ let token = plain_token("alice", "s3cret");
+ let body = send(
+ &mut stream,
+ API_KEY_SASL_AUTHENTICATE,
+ AUTHENTICATE_VERSION,
+ 2,
+ &authenticate_body(&token),
+ )
+ .await;
+ assert_eq!(error_code(&body), ERROR_NONE, "valid credentials accepted");
+
+ // The connection now serves ordinary traffic, and survives it.
+ let metadata = send(&mut stream, API_KEY_METADATA, 0, 3, &[0, 0, 0,
0]).await;
+ assert!(
+ !metadata.is_empty(),
+ "Metadata must answer once authenticated"
+ );
+
+ // A real Java client sends ApiVersions a second time after authenticating.
+ let second = send(&mut stream, API_KEY_API_VERSIONS, 1, 4, &[]).await;
+ assert!(!second.is_empty(), "ApiVersions must stay legal after auth");
+}
+
+#[tokio::test]
+async fn given_wrong_credentials_when_authenticating_should_fail_then_close() {
+ let addr = spawn_sasl_gateway().await;
+ let mut stream = TcpStream::connect(addr).await.expect("connect");
+ handshake_ok(&mut stream).await;
+
+ let token = plain_token("alice", "wrong-password");
+ let body = send(
+ &mut stream,
+ API_KEY_SASL_AUTHENTICATE,
+ AUTHENTICATE_VERSION,
+ 2,
+ &authenticate_body(&token),
+ )
+ .await;
+ assert_eq!(error_code(&body), ERROR_SASL_AUTHENTICATION_FAILED);
+ assert_closed(&mut stream).await;
+}
+
+#[tokio::test]
+async fn
given_a_malformed_plain_token_when_authenticating_should_fail_like_a_bad_password()
{
+ let addr = spawn_sasl_gateway().await;
+ let mut stream = TcpStream::connect(addr).await.expect("connect");
+ handshake_ok(&mut stream).await;
+
+ // Two fields instead of three. Indistinguishable from a rejection, on
purpose.
+ let body = send(
+ &mut stream,
+ API_KEY_SASL_AUTHENTICATE,
+ AUTHENTICATE_VERSION,
+ 2,
+ &authenticate_body(b"alice\0s3cret"),
+ )
+ .await;
+ assert_eq!(error_code(&body), ERROR_SASL_AUTHENTICATION_FAILED);
+ assert_closed(&mut stream).await;
+}
+
+#[tokio::test]
+async fn
given_an_unknown_mechanism_when_handshaking_should_report_the_supported_list() {
+ let addr = spawn_sasl_gateway().await;
+ let mut stream = TcpStream::connect(addr).await.expect("connect");
+
+ let body = send(
+ &mut stream,
+ API_KEY_SASL_HANDSHAKE,
+ HANDSHAKE_VERSION,
+ 1,
+ &handshake_body("SCRAM-SHA-256"),
+ )
+ .await;
+ assert_eq!(error_code(&body), ERROR_UNSUPPORTED_SASL_MECHANISM);
+ assert!(
+ body.windows(5).any(|w| w == b"PLAIN"),
+ "the refusal must still name what is supported, or the operator cannot
act on it"
+ );
+ assert_closed(&mut stream).await;
+}
+
+#[tokio::test]
+async fn
given_handshake_v0_when_offered_should_be_refused_before_any_headerless_token()
{
+ let addr = spawn_sasl_gateway().await;
+ let mut stream = TcpStream::connect(addr).await.expect("connect");
+
+ // v0 selects the pre-KIP-152 framing, where tokens arrive with no request
header at all.
+ let body = send(
+ &mut stream,
+ API_KEY_SASL_HANDSHAKE,
+ 0,
+ 1,
+ &handshake_body("PLAIN"),
+ )
+ .await;
+ assert_eq!(error_code(&body), ERROR_UNSUPPORTED_VERSION);
+ assert_closed(&mut stream).await;
+}
+
+#[tokio::test]
+async fn
given_an_unauthenticated_connection_when_a_normal_request_arrives_should_refuse_it()
{
+ let addr = spawn_sasl_gateway().await;
+ let mut stream = TcpStream::connect(addr).await.expect("connect");
+
+ // Metadata carries no top-level error field at this version, so there is
nowhere well-formed
+ // to put the code and the connection closes without a body.
+ let frame = build_request_frame(API_KEY_METADATA, 0, 1, Some("sasl-test"),
&[0, 0, 0, 0]);
+ stream.write_all(&frame).await.expect("write request");
+ assert_closed(&mut stream).await;
+}
+
+#[tokio::test]
+async fn
given_an_unauthenticated_connection_when_produce_arrives_should_close_without_answering()
{
+ let addr = spawn_sasl_gateway().await;
+ let mut stream = TcpStream::connect(addr).await.expect("connect");
+
+ // Produce never gets a body on this path: acks=0 forbids a response and
the acks value is not
+ // knowable without decoding a body this connection has not earned.
+ let mut body = BytesMut::new();
+ body.put_i16(-1);
+ body.put_i16(1);
+ body.put_i32(1000);
+ body.put_i32(0);
+ let frame = build_request_frame(API_KEY_PRODUCE, 3, 1, Some("sasl-test"),
&body.freeze());
+ stream.write_all(&frame).await.expect("write request");
+ assert_closed(&mut stream).await;
+}
+
+#[tokio::test]
+async fn given_a_token_before_a_handshake_should_be_an_illegal_state() {
+ let addr = spawn_sasl_gateway().await;
+ let mut stream = TcpStream::connect(addr).await.expect("connect");
+
+ let token = plain_token("alice", "s3cret");
+ let body = send(
+ &mut stream,
+ API_KEY_SASL_AUTHENTICATE,
+ AUTHENTICATE_VERSION,
+ 1,
+ &authenticate_body(&token),
+ )
+ .await;
+ assert_eq!(
+ error_code(&body),
+ ERROR_ILLEGAL_SASL_STATE,
+ "a token with no negotiated mechanism is an ordering violation, not a
bad password"
+ );
+ assert_closed(&mut stream).await;
+}
+
+#[tokio::test]
+async fn
given_a_second_handshake_after_authenticating_should_be_refused_without_closing()
{
+ let addr = spawn_sasl_gateway().await;
+ let mut stream = TcpStream::connect(addr).await.expect("connect");
+ handshake_ok(&mut stream).await;
+
+ let token = plain_token("alice", "s3cret");
+ let authenticated = send(
+ &mut stream,
+ API_KEY_SASL_AUTHENTICATE,
+ AUTHENTICATE_VERSION,
+ 2,
+ &authenticate_body(&token),
+ )
+ .await;
+ assert_eq!(error_code(&authenticated), ERROR_NONE);
+
+ let body = send(
+ &mut stream,
+ API_KEY_SASL_HANDSHAKE,
+ HANDSHAKE_VERSION,
+ 3,
+ &handshake_body("PLAIN"),
+ )
+ .await;
+ assert_eq!(error_code(&body), ERROR_ILLEGAL_SASL_STATE);
+
+ // Still usable: a real broker does not drop an authenticated connection
over this.
+ let metadata = send(&mut stream, API_KEY_METADATA, 0, 4, &[0, 0, 0,
0]).await;
+ assert!(!metadata.is_empty(), "connection must survive the refusal");
+}
+
+#[tokio::test]
+async fn
given_a_handshake_above_the_version_ceiling_after_authenticating_should_close()
{
+ // The keep-open promise stops where the schemas do. SaslHandshake has no
v2, so there is no
+ // body this client could parse and one shaped for v1 would be misparsed;
the version is
+ // checked before the encoder rather than left to fail inside it.
+ let addr = spawn_sasl_gateway().await;
+ let mut stream = TcpStream::connect(addr).await.expect("connect");
+ handshake_ok(&mut stream).await;
+
+ let token = plain_token("alice", "s3cret");
+ let authenticated = send(
+ &mut stream,
+ API_KEY_SASL_AUTHENTICATE,
+ AUTHENTICATE_VERSION,
+ 2,
+ &authenticate_body(&token),
+ )
+ .await;
+ assert_eq!(error_code(&authenticated), ERROR_NONE);
+
+ let frame = build_request_frame(
+ API_KEY_SASL_HANDSHAKE,
+ HANDSHAKE_VERSION + 1,
+ 3,
+ Some("sasl-test"),
+ &handshake_body("PLAIN"),
+ );
+ stream.write_all(&frame).await.expect("write request");
+ assert_closed(&mut stream).await;
+}
+
+#[tokio::test]
+async fn
given_sasl_disabled_when_a_client_connects_should_serve_without_authenticating()
{
+ // The default configuration, which is what every existing deployment runs.
+ let (addr, shutdown) = server::spawn_test_server().await;
+ std::mem::forget(shutdown);
+ let mut stream = TcpStream::connect(addr).await.expect("connect");
+
+ let metadata = send(&mut stream, API_KEY_METADATA, 0, 1, &[0, 0, 0,
0]).await;
+ assert!(!metadata.is_empty(), "Metadata must answer with SASL off");
+
+ let advertised = send(&mut stream, API_KEY_API_VERSIONS, 1, 2, &[]).await;
+ assert!(
+ !advertised
+ .windows(2)
+ .any(|w| i16::from_be_bytes([w[0], w[1]]) ==
API_KEY_SASL_HANDSHAKE),
+ "the SASL keys must not be advertised while the feature is off"
+ );
+}
+
+#[tokio::test]
+async fn
given_sasl_disabled_when_a_handshake_arrives_should_refuse_it_without_closing()
{
Review Comment:
This test asserts the real socket behavior, but two older tests still assert
the opposite. `tests/common/wire.rs:44` lists key 17 in
`OUT_OF_SCOPE_API_KEYS`, and `version_firewall_tests.rs:349` repeats it in a
literal.
Both call `handle_request` directly, below the new routing layer, so both
keep passing while describing behavior the server no longer has.
`server_e2e_tests.rs:439` slices the list to its first four entries, so nothing
catches it. Drop key 17 from both places.
##########
gateways/kafka/src/auth.rs:
##########
@@ -0,0 +1,427 @@
+// 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::bridge::config::DEFAULT_IGGY_ADDR;
+use crate::env::parse_bool;
+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;
Review Comment:
`max_retries` counts passes after the first, so `Some(1)` buys a second dial
pass plus a one second `reconnection.interval` sleep. A failed verification
holds one of only four slots for that whole time.
`Some(0)` still makes the first pass, which is all this path needs. The
Kafka client retries the handshake itself.
##########
gateways/kafka/src/server.rs:
##########
@@ -446,6 +796,23 @@ async fn dispatch_outcome(
.await?;
Ok(false)
}
+ HandleOutcome::RespondThenClose(body_response) => {
Review Comment:
Dropping the stream while unread bytes sit in the receive queue makes Linux
send RST instead of FIN. The RST discards whatever is still queued for send and
lets the peer throw away the error body this branch exists to deliver.
Call `stream.shutdown()` first, then drain input under `write_timeout`. A
real broker closes an authentication failure with `CloseMode.GRACEFUL` for this
reason.
##########
gateways/kafka/src/server.rs:
##########
@@ -321,17 +432,157 @@ 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> {
Review Comment:
`ConnectionContext` carries no cancellation token, so a credential check in
flight cannot see shutdown. `handle_connection` hands `cancel` to
`read_next_frame` alone, so a drain waits out `pre_auth_timeout` for every
connection that is verifying.
15s inside a 25s drain budget holds today, but `load_config` never compares
the two values. Select on the token alongside the verification timeout.
##########
gateways/kafka/src/server.rs:
##########
@@ -364,19 +615,118 @@ 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(&state, req.request_api_key,
req.request_api_version, body)
- .await;
+ 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(
+ ctx: &ConnectionContext<'_>,
+ api_version: i16,
+ body: Bytes,
+) -> HandleOutcome {
+ let peer = ctx.peer;
+ let failed = || sasl_authenticate_outcome(api_version,
ERROR_SASL_AUTHENTICATION_FAILED, true);
+
+ let Some(authenticator) = ctx.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(ctx.config.pre_auth_timeout, async {
+ // Acquire fails only once the semaphore is closed, which this gateway
never does.
+ let Ok(_slot) = ctx.auth_slots.acquire().await else {
+ error!(%peer, "authentication slots unavailable");
+ return None;
+ };
+ Some(authenticator.authenticate(&credentials).await)
+ })
+ .await;
+
+ let Ok(Some(result)) = verified else {
Review Comment:
There is no per-peer and no per-principal throttle on failed logins. An
attacker recycles connections freely and keeps all four verify slots busy, and
every guess still costs a full Argon2id hash inline on an Iggy shard thread.
A real login that waits past `pre_auth_timeout` is then closed with no
answer, which `sasl_tests.rs:877` already demonstrates. Add a per-peer failure
throttle on this path.
##########
gateways/kafka/docs/AUTHENTICATION.md:
##########
@@ -0,0 +1,251 @@
+# Kafka authentication and Iggy identity
+
+Status: implemented. Answers
[#3549](https://github.com/apache/iggy/issues/3549). The gateway listener
+still has no TLS, so PLAIN stays confined to a trusted network until that
lands (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.
+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` | ApiVersions past the allowance | answer 34, then close
(two are allowed: the one a real broker allows, plus the KIP-511 downgrade
retry) |
+| `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.
+- **The pre-authentication allowance counts answers, not successes.** A
refusal a client may retry at a
+ lower version is one it may also repeat, and every frame resets the
pre-authentication deadline, so
+ spending the allowance only on a usable answer leaves the connection
unbounded.
+- **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. 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
Review Comment:
No handler asks Iggy about permissions. Every handler still carries the note
that the session is kept until a handler awaits the bridge, and `auth.rs:251`
shuts the verified client down.
`README.md:128` states the real position, an admission gate rather than an
identity carried onto the data plane. Say here that authorization is not wired
yet.
--
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]
