This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 689436fbf feat(gateways): accept SaslHandshake v0 from older Kafka
clients (#4778)
689436fbf is described below
commit 689436fbf855baa370dcc735f1856d163bad0172
Author: Krishna Vishal <[email protected]>
AuthorDate: Sat Nov 7 16:30:36 2026 +0530
feat(gateways): accept SaslHandshake v0 from older Kafka clients (#4778)
## Which issue does this PR address?
Relates to #4528.
## Rationale
kafka-python 2.0.2 always sends `SaslHandshake` v0, and it was the
latest kafka-python release for four and a half years. Sarama sends v0
by default up to 1.40.1, and aiokafka up to 0.12 sends it with
`api_version="auto"`. Kafka 3.x brokers accept v0, so these clients work
with SASL against Kafka. With SASL on, they failed to connect to the
gateway. If v0 is not in `ApiVersions`, librdkafka gives up before its
handshake, so the gateway already listed v0 there. But the gateway
refused a v0 handshake.
After a v0 handshake, the client sends the SASL token in a bare frame
(KIP-152). The frame is a 4-byte length and the token, with no Kafka
request header. The broker answers a valid PLAIN token with an empty
token in the same kind of frame, which is four zero bytes. For any
failure, the broker closes the connection, because this exchange has no
field for an error code. kafka-python 2.0.2 and Sarama 1.40.1 wait for
these four bytes.
## What changed?
Before, the gateway answered `SaslHandshake` v0 with
`UNSUPPORTED_VERSION` (35) and closed the connection. Now it accepts v0
and reads the next frame as the bare PLAIN token, with no request header
and a cap of 4096 bytes. A valid token gets four zero bytes back, and
the connection continues as signed in. A wrong, malformed or oversized
token closes the connection without an answer.
---
core/kafka_wire/src/bounds_guard.rs | 10 +-
core/kafka_wire/src/sasl.rs | 97 +++++++++++--
gateways/kafka/docs/AUTHENTICATION.md | 15 +-
gateways/kafka/docs/MANUAL_TESTING.md | 30 ++++
gateways/kafka/src/protocol/api.rs | 18 +--
gateways/kafka/src/server.rs | 251 ++++++++++++++++++++++------------
gateways/kafka/tests/sasl_tests.rs | 113 +++++++++++++--
7 files changed, 406 insertions(+), 128 deletions(-)
diff --git a/core/kafka_wire/src/bounds_guard.rs
b/core/kafka_wire/src/bounds_guard.rs
index 3f22b95e6..479de93d9 100644
--- a/core/kafka_wire/src/bounds_guard.rs
+++ b/core/kafka_wire/src/bounds_guard.rs
@@ -1493,7 +1493,10 @@ pub fn validate_offset_delete_shape(
/// `max_frame_size`, before a header is even decoded, so an unauthenticated
peer can still make
/// the gateway hold that much. What this cap bounds is everything downstream
of the guard: what
/// reaches `parse_plain`, and what a future mechanism would carry into a
credential exchange.
-const MAX_SASL_AUTH_BYTES: usize = 4096;
+///
+/// A bare token after a `SaslHandshake` v0 gets the same cap. There the frame
reader applies it,
+/// so such a token is never buffered past it.
+pub const MAX_SASL_AUTH_BYTES: usize = 4096;
/// Mirrors the field order `DescribeConfigsRequest::decode` walks.
///
@@ -1714,9 +1717,8 @@ pub fn validate_describe_cluster_shape(version: i16,
body: &Bytes) -> Result<()>
/// `SaslHandshake` carries one non-nullable string, the mechanism name.
///
-/// The message is never flexible, so v0 and v1 share this shape, and both are
walked: the
-/// connection loop reads the mechanism name out of every handshake body
before the SASL state
-/// machine decides, and it is that decision, not this guard, that refuses v0.
+/// The message is never flexible, so v0 and v1 share this shape. The
connection loop reads the
+/// mechanism name out of every handshake body before the SASL state machine
decides.
///
/// # Errors
///
diff --git a/core/kafka_wire/src/sasl.rs b/core/kafka_wire/src/sasl.rs
index e4fa125de..bb3d1307a 100644
--- a/core/kafka_wire/src/sasl.rs
+++ b/core/kafka_wire/src/sasl.rs
@@ -146,7 +146,12 @@ pub enum SaslState {
AwaitHandshake {
api_versions_answered: u8,
},
+ /// A v1 handshake negotiated the mechanism. The token comes in a
`SaslAuthenticate` request.
AwaitToken(SaslMechanism),
+ /// A v0 handshake negotiated the mechanism. The next frame is the bare
token, with no request
+ /// header (KIP-152), so the connection loop reads it before it parses any
header, and it never
+ /// reaches [`Self::classify`].
+ AwaitRawToken(SaslMechanism),
Authenticated,
}
@@ -159,6 +164,24 @@ impl SaslState {
}
}
+ /// The state after a `SaslHandshake` at `api_version` negotiated
`mechanism`. The version
+ /// selects how the token comes (KIP-152): v0 in a bare length-prefixed
frame, v1 in a
+ /// `SaslAuthenticate` request.
+ #[must_use]
+ pub const fn awaiting_token(mechanism: SaslMechanism, api_version: i16) ->
Self {
+ if api_version == 0 {
+ Self::AwaitRawToken(mechanism)
+ } else {
+ Self::AwaitToken(mechanism)
+ }
+ }
+
+ /// Whether the next frame is a bare token rather than a request.
+ #[must_use]
+ pub const fn awaits_raw_token(self) -> bool {
+ matches!(self, Self::AwaitRawToken(_))
+ }
+
/// Records one answered pre-authentication `ApiVersions`.
///
/// Every answer counts, whatever error code it carries. Counting only
usable answers lets a
@@ -186,15 +209,13 @@ impl Default for SaslState {
pub enum SaslAction {
/// Hand the frame to the ordinary request dispatch.
Dispatch,
- /// Answer a `SaslHandshake` for this mechanism and advance to
[`SaslState::AwaitToken`].
+ /// Answer a `SaslHandshake` for this mechanism and advance to the state
that
+ /// [`SaslState::awaiting_token`] names for its version.
AcceptHandshake(SaslMechanism),
/// Answer a `SaslHandshake` with `UNSUPPORTED_SASL_MECHANISM`, then close.
RejectMechanism,
- /// Answer a `SaslHandshake` with `UNSUPPORTED_VERSION`, then close.
- ///
- /// The handshake version selects the token framing (KIP-152): at v0 the
tokens that follow
- /// arrive as bare length-prefixed frames with no request header. Refusing
v0 here is what
- /// lets the frame reader stay header-parsing-only, and it costs only
pre-1.0 clients.
+ /// A `SaslHandshake` above [`SASL_HANDSHAKE_VERSION`], or below 0. No
response schema exists
+ /// at that version, so close without an answer.
RejectHandshakeVersion,
/// Parse the frame as a `SaslAuthenticate` token and verify the
credentials it carries.
Authenticate,
@@ -221,6 +242,13 @@ pub const MAX_PRE_AUTH_API_VERSIONS: u8 = 2;
/// Highest `SaslHandshake` version accepted. See
[`SaslAction::RejectHandshakeVersion`].
pub const SASL_HANDSHAKE_VERSION: i16 = 1;
+/// What the gateway sends back for a bare token that it accepts after a v0
handshake.
+///
+/// It is the empty server token that completes PLAIN, in a length-prefixed
frame. A rejected token
+/// gets no answer, and the connection closes, as a Kafka broker does it.
kafka-python and Sarama
+/// wait for exactly these four bytes.
+pub const RAW_TOKEN_ACCEPTED: [u8; 4] = [0; 4];
+
/// Highest `SaslAuthenticate` version accepted, matching what `ApiVersions`
advertises.
///
/// Checked in [`SaslState::classify`] rather than left to the decoder.
Without it an out-of-range
@@ -261,7 +289,7 @@ impl SaslState {
SaslAction::DispatchPreAuthApiVersions
}
(Self::AwaitHandshake { .. }, API_KEY_SASL_HANDSHAKE) => {
- if api_version == SASL_HANDSHAKE_VERSION {
+ if (0..=SASL_HANDSHAKE_VERSION).contains(&api_version) {
mechanism
.and_then(|name| SaslMechanism::from_str(name).ok())
.map_or(SaslAction::RejectMechanism,
SaslAction::AcceptHandshake)
@@ -466,15 +494,64 @@ mod tests {
}
#[test]
- fn
given_await_handshake_when_handshake_v0_arrives_should_reject_the_version() {
+ fn given_await_handshake_when_handshake_v0_offers_plain_should_accept_it()
{
let state = SaslState::new();
assert_eq!(
state.classify(API_KEY_SASL_HANDSHAKE, 0, Some("PLAIN")),
- SaslAction::RejectHandshakeVersion,
- "v0 selects the headerless token framing the frame reader cannot
parse"
+ SaslAction::AcceptHandshake(SaslMechanism::Plain),
+ "Sarama up to 1.40.1, kafka-python 2.0.2 and aiokafka up to 0.12
send v0"
+ );
+ assert_eq!(
+ state.classify(API_KEY_SASL_HANDSHAKE, 0, Some("GSSAPI")),
+ SaslAction::RejectMechanism
);
}
+ #[test]
+ fn
given_a_handshake_outside_the_known_versions_should_reject_the_version() {
+ let state = SaslState::new();
+ for version in [-1, SASL_HANDSHAKE_VERSION + 1] {
+ assert_eq!(
+ state.classify(API_KEY_SASL_HANDSHAKE, version, Some("PLAIN")),
+ SaslAction::RejectHandshakeVersion,
+ "v{version} has no response schema"
+ );
+ }
+ }
+
+ #[test]
+ fn
given_a_negotiated_mechanism_should_await_the_token_framing_of_the_handshake_version()
{
+ assert_eq!(
+ SaslState::awaiting_token(SaslMechanism::Plain, 0),
+ SaslState::AwaitRawToken(SaslMechanism::Plain)
+ );
+ assert_eq!(
+ SaslState::awaiting_token(SaslMechanism::Plain, 1),
+ SaslState::AwaitToken(SaslMechanism::Plain)
+ );
+
assert!(SaslState::AwaitRawToken(SaslMechanism::Plain).awaits_raw_token());
+
assert!(!SaslState::AwaitToken(SaslMechanism::Plain).awaits_raw_token());
+ assert!(!SaslState::new().awaits_raw_token());
+ assert!(!SaslState::Authenticated.awaits_raw_token());
+ }
+
+ #[test]
+ fn
given_await_raw_token_when_a_request_is_classified_should_be_illegal_state() {
+ // The connection loop reads the bare token before it parses a header,
so a request
+ // classified in this state means that the loop skipped that step.
Fail closed.
+ let state = SaslState::AwaitRawToken(SaslMechanism::Plain);
+ for (api_key, api_version) in [
+ (API_KEY_SASL_AUTHENTICATE, 1),
+ (API_KEY_API_VERSIONS, 3),
+ (0, 9),
+ ] {
+ assert_eq!(
+ state.classify(api_key, api_version, None),
+ SaslAction::IllegalState
+ );
+ }
+ }
+
#[test]
fn given_await_handshake_when_the_mechanism_is_unknown_should_reject_it() {
let state = SaslState::new();
diff --git a/gateways/kafka/docs/AUTHENTICATION.md
b/gateways/kafka/docs/AUTHENTICATION.md
index fa0d6c1c6..07fd52b9d 100644
--- a/gateways/kafka/docs/AUTHENTICATION.md
+++ b/gateways/kafka/docs/AUTHENTICATION.md
@@ -146,23 +146,28 @@ Per connection, with SASL enabled. Only ApiVersions and
SaslHandshake are legal
| ------- | ------- | -------- |
| `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 |
+| | SaslHandshake (17) v0, supported mechanism | answer `error_code` 0, go to
`AwaitRawToken` |
+| | SaslHandshake (17) v0 or v1, unknown mechanism | answer 33 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 |
+| `AwaitRawToken` | bare PLAIN token, credentials valid | send the empty
server token in a length-prefixed frame (four zero bytes), go to
`Authenticated` |
+| | bare token rejected, malformed or longer than 4096 bytes | close without
an answer |
| `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) |
+| | SaslHandshake above v1 | close, no schema exists at that version |
| `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.
+- **Handshake v0 and v1.** The handshake version selects the token framing
(KIP-152). After v0, the next
+ frame is the bare token, with no request header. The connection loop reads
it before it parses a header,
+ and caps it at 4096 bytes. A bare token has no answer that can carry an
error, so every failure closes
+ the connection, as on a Kafka broker. kafka-python 2.0.2, Sarama up to
1.40.1 and aiokafka up to 0.12
+ send v0.
- **`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
diff --git a/gateways/kafka/docs/MANUAL_TESTING.md
b/gateways/kafka/docs/MANUAL_TESTING.md
index ae4788512..8d1c484c5 100644
--- a/gateways/kafka/docs/MANUAL_TESTING.md
+++ b/gateways/kafka/docs/MANUAL_TESTING.md
@@ -98,6 +98,8 @@ docker run --rm --network host edenhill/kcat:1.7.1 -b
127.0.0.1:9095 \
| S4 | No credentials | `docker run --rm --network host edenhill/kcat:1.7.1 -b
127.0.0.1:9095 -L` | No metadata served; client suggests SASL may be required |
Pass |
| S5 | Java client | see below | Authenticates, and lists `SaslHandshake(17):
0 to 1 [usable: 1]` and `SaslAuthenticate(36): 0 to 2 [usable: 2]` | Pass |
| S6 | Feature off | restart without `IGGY_KAFKA_SASL_ENABLED`, rerun S1 |
Client reports no broker SASL support; plaintext `-L` still served | Pass |
+| S7 | kafka-python 2.0.2 | see below | Produces 10 records and reads them
back in a group. With the password `WRONG`, it ends with `NoBrokersAvailable` |
Pass |
+| S8 | Sarama 1.40.1 | see below | Produces 10 records and reads them back.
With the password `WRONG`, it ends with `client has run out of available
brokers to talk to: EOF` | Pass |
S5 is the case that matters most, because the Java client negotiates
`SaslAuthenticate` **v2**, the
compact framing no other case here exercises, and it sends `ApiVersions` both
before and after
@@ -114,6 +116,34 @@ docker run --rm --network host -v
/tmp/client.properties:/tmp/client.properties:
--bootstrap-server 127.0.0.1:9095 --command-config /tmp/client.properties
```
+S7 and S8 send `SaslHandshake` v0 and then the token in a bare frame with no
request header
+(KIP-152). kafka-python 2.0.2 always does this, and Sarama does it by default
up to 1.40.1. Both
+cases produce, so add `IGGY_KAFKA_BRIDGE_ENABLED=true
IGGY_KAFKA_IGGY_USERNAME=iggy
+IGGY_KAFKA_IGGY_PASSWORD=iggy IGGY_KAFKA_AUTO_CREATE_TOPICS_ENABLED=true` to
the gateway in
+terminal 2. kafka-python 2.0.2 needs Python 3.11 or older:
+
+```bash
+uv venv --python 3.11 /tmp/kafka-python-2.0.2
+uv pip install --python /tmp/kafka-python-2.0.2/bin/python kafka-python==2.0.2
+/tmp/kafka-python-2.0.2/bin/python - <<'EOF'
+from kafka import KafkaConsumer, KafkaProducer
+common = dict(bootstrap_servers="127.0.0.1:9095",
security_protocol="SASL_PLAINTEXT",
+ sasl_mechanism="PLAIN", sasl_plain_username="iggy",
sasl_plain_password="iggy")
+producer = KafkaProducer(**common)
+for i in range(10):
+ producer.send("v0-kafka-python", f"kafka-python
{i}".encode()).get(timeout=10)
+producer.close()
+consumer = KafkaConsumer("v0-kafka-python", group_id="v0",
auto_offset_reset="earliest",
+ consumer_timeout_ms=15000, **common)
+print(len(list(consumer)), "records read")
+EOF
+```
+
+For S8, a Go program on `github.com/IBM/sarama v1.40.1` does the same. It sets
+`config.Net.SASL.Version = sarama.SASLHandshakeV0`, sends 10 records with a
`SyncProducer`, and
+reads them back with `ConsumePartition` from `OffsetOldest`. With
`RUST_LOG=debug`, the gateway logs
+`SASL mechanism negotiated` with `api_version=0` for each connection of both
clients.
+
### Category T — ACL view
Needs the same running stack as category S, plus two extra Iggy users so the
view has something to
diff --git a/gateways/kafka/src/protocol/api.rs
b/gateways/kafka/src/protocol/api.rs
index 4c52a94ad..69024ee90 100644
--- a/gateways/kafka/src/protocol/api.rs
+++ b/gateways/kafka/src/protocol/api.rs
@@ -698,19 +698,15 @@ pub fn supported_max_version(api_key: i16) -> Option<i16>
{
/// open, the answer a real broker gives on a PLAINTEXT listener. Keeping them
out of the firewall
/// table means dispatch never serves them on its own.
///
-/// `SaslHandshake` is pinned to v1 on both ends. v0 selects the headerless
token framing (KIP-152)
-/// that the frame reader cannot parse, so advertising it would invite exactly
the shape this
-/// gateway refuses.
+/// `SaslHandshake` v0 selects a bare token with no request header after it
(KIP-152), and v1 a
+/// token in a `SaslAuthenticate` request. The connection loop serves both.
static SASL_ADVERTISED_RANGES: &[ApiVersionRange] = &[
ApiVersionRange {
- // Advertised from v0 even though only v1 is accepted, the same split
- // [`advertised_min_version`] already applies to Produce. librdkafka
gates its whole
- // SASL-handshake feature on `SaslHandshake` v0 appearing in the
advertisement
- // (`RD_KAFKA_FEATURE_SASL_HANDSHAKE` depends on key 17 at version 0),
so advertising v1
- // alone makes it report "SASL Handshake not supported by broker" and
give up before it
- // ever sends one. It then picks v1 anyway, because `SaslAuthenticate`
being advertised is
- // what selects the KIP-152 framing. A client that really only knows
v0 still gets
- // `UNSUPPORTED_VERSION` and a closed connection from the state
machine.
+ // librdkafka gates its whole SASL-handshake feature on
`SaslHandshake` v0 appearing in the
+ // advertisement (`RD_KAFKA_FEATURE_SASL_HANDSHAKE` depends on key 17
at version 0), so
+ // advertising v1 alone makes it report "SASL Handshake not supported
by broker" and give
+ // up before it ever sends one. It then picks v1, because
`SaslAuthenticate` being
+ // advertised is what selects the KIP-152 framing.
api_key: API_KEY_SASL_HANDSHAKE,
min_version: 0,
max_version: 1,
diff --git a/gateways/kafka/src/server.rs b/gateways/kafka/src/server.rs
index b38843544..548f6bc79 100644
--- a/gateways/kafka/src/server.rs
+++ b/gateways/kafka/src/server.rs
@@ -49,17 +49,18 @@ use crate::protocol::api::{
API_KEY_DESCRIBE_ACLS, API_KEY_SASL_AUTHENTICATE, API_KEY_SASL_HANDSHAKE,
BrokerAdvertise,
Brokers, ConnectionState, DEFAULT_KAFKA_PORT, DEFAULT_NODE_ID,
ERROR_ILLEGAL_SASL_STATE,
ERROR_INVALID_REQUEST, ERROR_NONE, ERROR_SASL_AUTHENTICATION_FAILED,
- ERROR_UNKNOWN_SERVER_ERROR, ERROR_UNSUPPORTED_SASL_MECHANISM,
ERROR_UNSUPPORTED_VERSION,
- GatewayState, HandleOutcome, SASL_ADVERTISED_DESCRIBE_ACLS_VERSIONS,
decode_acl_filter,
- decode_sasl_auth_bytes, decode_sasl_mechanism, describe_acls_outcome,
encode_error_for_key,
- handle_connection_request, respond_describe_acls_error,
sasl_authenticate_outcome,
- sasl_handshake_outcome,
+ ERROR_UNKNOWN_SERVER_ERROR, ERROR_UNSUPPORTED_SASL_MECHANISM,
GatewayState, HandleOutcome,
+ SASL_ADVERTISED_DESCRIBE_ACLS_VERSIONS, decode_acl_filter,
decode_sasl_auth_bytes,
+ decode_sasl_mechanism, describe_acls_outcome, encode_error_for_key,
handle_connection_request,
+ respond_describe_acls_error, sasl_authenticate_outcome,
sasl_handshake_outcome,
};
use crate::protocol::auto_topic_creation::{self, AutoTopicCreationConfig};
+use crate::protocol::bounds_guard::MAX_SASL_AUTH_BYTES;
use crate::protocol::handlers::api_versions::Advertisement;
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,
+ RAW_TOKEN_ACCEPTED, SASL_AUTHENTICATE_MAX_VERSION, SASL_HANDSHAKE_VERSION,
SaslAction,
+ SaslState, parse_plain,
};
use crate::tls::{IdentityError, TlsConfig, TlsState};
@@ -1148,25 +1149,19 @@ async fn route_frame(
.await
}
SaslAction::AcceptHandshake(mechanism) => {
- debug!(%peer, %mechanism, "SASL mechanism negotiated");
- *sasl_state = SaslState::AwaitToken(mechanism);
- sasl_handshake_outcome(req.request_api_version, ERROR_NONE, false)
+ let api_version = req.request_api_version;
+ debug!(%peer, %mechanism, api_version, "SASL mechanism
negotiated");
+ *sasl_state = SaslState::awaiting_token(mechanism, api_version);
+ sasl_handshake_outcome(api_version, ERROR_NONE, false)
}
SaslAction::RejectMechanism => sasl_handshake_outcome(
req.request_api_version,
ERROR_UNSUPPORTED_SASL_MECHANISM,
true,
),
- // v0 is the one refusable handshake version that still has an
encodable response shape,
- // so it gets a real UNSUPPORTED_VERSION. Anything above the ceiling
has no schema at the
- // version the client asked for, and a body shaped for a different one
would be misparsed.
- SaslAction::RejectHandshakeVersion => {
- if req.request_api_version < SASL_HANDSHAKE_VERSION {
- sasl_handshake_outcome(req.request_api_version,
ERROR_UNSUPPORTED_VERSION, true)
- } else {
- HandleOutcome::Close
- }
- }
+ // No schema exists at the version the client asked for, and a body
shaped for another
+ // version would be misparsed.
+ SaslAction::RejectHandshakeVersion => HandleOutcome::Close,
SaslAction::RejectAuthenticateVersion => {
debug!(
%peer,
@@ -1288,52 +1283,34 @@ where
} else {
config.pre_auth_timeout
};
+ // A bare token gets the cap of a token, so the reader never buffers
more than one holds.
+ let frame_cap = if sasl_state.awaits_raw_token() {
+ MAX_SASL_AUTH_BYTES
+ } else {
+ config.max_frame_size
+ };
let Some(frame) =
- read_next_frame(&mut stream, &config, idle_budget, &peer,
&cancel).await?
+ read_next_frame(&mut stream, &config, frame_cap, idle_budget,
&peer, &cancel).await?
else {
return Ok(());
};
- if frame.len() < 8 {
- return Err(KafkaWireError::BufferUnderflow {
- needed: 8,
- remaining: frame.len(),
+ // After a v0 handshake, the frame is the bare token, with no request
header to parse.
+ if sasl_state.awaits_raw_token() {
+ if !authenticate_raw_token(&ctx, &mut stream, &frame, &mut
principal, &mut permissions)
+ .await?
+ {
+ return Ok(());
}
- .into());
+ sasl_state = SaslState::Authenticated;
+ continue;
}
- let api_key = i16::from_be_bytes([frame[0], frame[1]]);
- let api_version = i16::from_be_bytes([frame[2], frame[3]]);
- let req_hdr_ver = request_header_version(api_key, api_version);
- let resp_hdr_ver = response_header_version(api_key, api_version);
-
- // `request_header_version` only ever returns 1 or 2, both of which
`RequestHeader`
- // supports, so `decode` cannot fail on the version argument itself;
any error here is a
- // malformed header and closes the connection.
- let mut body = frame;
- let mut req = RequestHeader::decode(&mut body, req_hdr_ver)
- .map_err(|e| KafkaWireError::Malformed(e.to_string()))?;
-
- debug!(
- %peer,
- api_key = req.request_api_key,
- api_version = req.request_api_version,
- correlation_id = req.correlation_id,
- client_id = req.client_id.as_deref().unwrap_or(""),
- "received request"
- );
- connection_state.note_client_id(req.client_id.as_deref());
- // A group request can park for a whole rebalance timeout. `client_id`
and the tagged
- // fields are views into the frame, so keeping them would pin the
frame for that long
- // after the handler has let go of its own copy.
- req.client_id = None;
- req.unknown_tagged_fields.clear();
+ let (req, resp_hdr_ver, body) = decode_request(frame, &peer,
connection_state)?;
let Some(revision) = permissions.apply(principal.as_mut(), &peer) else
{
return Ok(());
};
let signed_in = principal.is_some();
- // `RequestHeader::decode` advances `body` past the header fields it
consumed via
- // `Buf::advance`, so `body` is already exactly the request payload.
let outcome = route_frame(&ctx, &mut sasl_state, &mut principal, &req,
body).await;
if !signed_in && principal.is_some() {
permissions.signed_in(revision);
@@ -1354,6 +1331,77 @@ where
}
}
+/// Verifies the bare token that follows a v0 handshake (KIP-152), and answers
it if it is valid.
+///
+/// `false` closes the connection. A bare token has no answer that can carry
an error, so a broker
+/// closes the connection on every failure, and the client reports the close
as one.
+async fn authenticate_raw_token<S: AsyncWrite + Unpin>(
+ ctx: &ConnectionContext<'_>,
+ stream: &mut S,
+ token: &[u8],
+ principal: &mut Option<AuthenticatedPrincipal>,
+ permissions: &mut PermissionFollow<'_>,
+) -> Result<bool> {
+ let Some(revision) = permissions.apply(principal.as_mut(), ctx.peer) else {
+ return Ok(false);
+ };
+ let signed_in = principal.is_some();
+ let TokenCheck::Accepted(authenticated) = verify_plain(ctx, token).await
else {
+ return Ok(false);
+ };
+ send_raw_token_accepted(stream, ctx.config.write_timeout).await?;
+ *principal = Some(authenticated);
+ if !signed_in {
+ permissions.signed_in(revision);
+ }
+ Ok(true)
+}
+
+/// Decodes the request header at the start of `frame`. Returns the header,
the version of the
+/// response header that answers it, and the request body after the header.
+fn decode_request(
+ frame: Bytes,
+ peer: &SocketAddr,
+ connection_state: &ConnectionState,
+) -> Result<(RequestHeader, i16, Bytes)> {
+ if frame.len() < 8 {
+ return Err(KafkaWireError::BufferUnderflow {
+ needed: 8,
+ remaining: frame.len(),
+ }
+ .into());
+ }
+ let api_key = i16::from_be_bytes([frame[0], frame[1]]);
+ let api_version = i16::from_be_bytes([frame[2], frame[3]]);
+ let req_hdr_ver = request_header_version(api_key, api_version);
+ let resp_hdr_ver = response_header_version(api_key, api_version);
+
+ // `request_header_version` only ever returns 1 or 2, both of which
`RequestHeader`
+ // supports, so `decode` cannot fail on the version argument itself; any
error here is a
+ // malformed header and closes the connection.
+ let mut body = frame;
+ let mut req = RequestHeader::decode(&mut body, req_hdr_ver)
+ .map_err(|e| KafkaWireError::Malformed(e.to_string()))?;
+
+ debug!(
+ %peer,
+ api_key = req.request_api_key,
+ api_version = req.request_api_version,
+ correlation_id = req.correlation_id,
+ client_id = req.client_id.as_deref().unwrap_or(""),
+ "received request"
+ );
+ connection_state.note_client_id(req.client_id.as_deref());
+ // A group request can park for a whole rebalance timeout. `client_id` and
the tagged
+ // fields are views into the frame, so keeping them would pin the frame
for that long
+ // after the handler has let go of its own copy.
+ req.client_id = None;
+ req.unknown_tagged_fields.clear();
+ // `RequestHeader::decode` advances `body` past the header fields it
consumed via
+ // `Buf::advance`, so `body` is already exactly the request payload.
+ Ok((req, resp_hdr_ver, body))
+}
+
/// One connection's place on the bridge's [`PermissionBoard`], which the
change log keeps.
struct PermissionFollow<'a> {
board: Option<&'a PermissionBoard>,
@@ -1483,28 +1531,57 @@ async fn authenticate_token(
None,
)
};
+ let Ok(auth_bytes) = decode_sasl_auth_bytes(api_version, body) else {
+ debug!(%peer, "malformed SaslAuthenticate token");
+ return failed();
+ };
+ match verify_plain(ctx, &auth_bytes).await {
+ TokenCheck::Accepted(authenticated) => (
+ sasl_authenticate_outcome(api_version, ERROR_NONE, false),
+ Some(authenticated),
+ ),
+ // A rejection is the client's problem and is terminal, so it earns a
parseable 58.
+ TokenCheck::Rejected => failed(),
+ // Kafka clients treat 58 as fatal and surface it to the application,
so answering with it
+ // would turn a momentary Iggy blip, an overload or a throttle into a
permanent
+ // authentication error for credentials that may be 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.
+ TokenCheck::Dropped => (HandleOutcome::Close, None),
+ }
+}
+
+/// What a PLAIN token earned.
+enum TokenCheck {
+ Accepted(AuthenticatedPrincipal),
+ /// The token is malformed, or Iggy rejected its credentials.
+ Rejected,
+ /// Nothing says that the credentials are wrong: the peer is throttled, no
slot came free, a
+ /// budget ran out, the gateway shuts down or Iggy is unreachable.
+ Dropped,
+}
+/// Verifies the PLAIN token `auth_bytes` against Iggy, within the
pre-authentication budget. Both
+/// token framings come here: a `SaslAuthenticate` request, and a bare frame
after a v0 handshake.
+async fn verify_plain(ctx: &ConnectionContext<'_>, auth_bytes: &[u8]) ->
TokenCheck {
+ let peer = ctx.peer;
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();
+ return TokenCheck::Rejected;
};
- 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 {
+ let Ok(credentials) = parse_plain(auth_bytes) else {
debug!(%peer, "malformed PLAIN initial response");
- return failed();
+ return TokenCheck::Rejected;
};
- // Checked before a slot is taken, so a throttled peer costs neither a
slot nor a hash. Closed
- // rather than answered 58 for the same reason an overload is: the client
would treat 58 as
- // fatal, and a correct password retried after the delay must still get
through.
+ // Checked before a slot is taken, so a throttled peer costs neither a
slot nor a hash. Not a
+ // rejection, for the same reason that an overload is not one: a correct
password retried
+ // after the delay must still get through.
if ctx.failed_logins.is_blocked(peer.ip()) {
debug!(%peer, "SASL authentication refused: peer is throttled after a
rejected login");
- return (HandleOutcome::Close, None);
+ return TokenCheck::Dropped;
}
// The connection loop only watches the shutdown token between frames, so
a verification in
@@ -1512,47 +1589,51 @@ async fn authenticate_token(
let verified = tokio::select! {
() = ctx.cancel.cancelled() => {
debug!(%peer, "SASL authentication abandoned by shutdown");
- return (HandleOutcome::Close, None);
+ return TokenCheck::Dropped;
}
verified = within_auth_budget(ctx, ctx.auth_slots, "SASL
authentication", |slot| {
authenticator.authenticate(&credentials, slot)
}) => verified,
};
-
+ // Overloaded or timed out. Nothing here says the credentials were wrong.
let Some(result) = verified else {
- // Overloaded or timed out. 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.
- return (HandleOutcome::Close, None);
+ return TokenCheck::Dropped;
};
match result {
Ok(authenticated) => {
debug!(%peer, "SASL authentication succeeded");
ctx.failed_logins.record_success(peer.ip());
- (
- sasl_authenticate_outcome(api_version, ERROR_NONE, false),
- Some(authenticated),
- )
+ TokenCheck::Accepted(authenticated)
}
- // A rejection is the client's problem and is terminal, so it earns a
parseable 58.
Err(AuthError::Rejected) => {
debug!(%peer, "SASL authentication rejected");
ctx.failed_logins.record_rejection(peer.ip());
- failed()
+ TokenCheck::Rejected
}
- // 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, None)
+ TokenCheck::Dropped
}
}
}
+/// Answers an accepted bare token with [`RAW_TOKEN_ACCEPTED`].
+async fn send_raw_token_accepted<S: AsyncWrite + Unpin>(
+ stream: &mut S,
+ write_timeout: Duration,
+) -> Result<()> {
+ // The flush matters on TLS, for the same reason as in `send_response`.
+ let write = async {
+ stream.write_all(&RAW_TOKEN_ACCEPTED).await?;
+ stream.flush().await
+ };
+ timeout(write_timeout, write)
+ .await
+ .map_err(|_| io::Error::new(io::ErrorKind::TimedOut, "write
timeout"))??;
+ Ok(())
+}
+
/// Answers `DescribeAcls` from the permissions captured when this connection
authenticated.
///
/// Only reachable on a listener that names principals: elsewhere the key is
not advertised and
@@ -1685,10 +1766,12 @@ fn illegal_state_outcome(
}
}
-/// Returns `Ok(None)` when shutdown cancellation wins; `Ok(Some(frame))` on a
full frame.
+/// Returns `Ok(None)` when shutdown cancellation wins; `Ok(Some(frame))` on a
full frame of at
+/// most `max_frame_size` bytes.
async fn read_next_frame<S: AsyncRead + Unpin>(
stream: &mut S,
config: &GatewayConfig,
+ max_frame_size: usize,
idle_timeout: Duration,
peer: &SocketAddr,
cancel: &CancellationToken,
@@ -1700,7 +1783,7 @@ async fn read_next_frame<S: AsyncRead + Unpin>(
}
result = read_frame(
stream,
- config.max_frame_size,
+ max_frame_size,
idle_timeout,
config.read_timeout,
) => match result {
diff --git a/gateways/kafka/tests/sasl_tests.rs
b/gateways/kafka/tests/sasl_tests.rs
index 831fc54fa..717e7568e 100644
--- a/gateways/kafka/tests/sasl_tests.rs
+++ b/gateways/kafka/tests/sasl_tests.rs
@@ -35,6 +35,7 @@ use iggy_gateway_kafka::GatewayConfig;
use iggy_gateway_kafka::auth::{AuthError, AuthenticatedPrincipal,
SaslAuthenticator};
use iggy_gateway_kafka::listener::SecurityProtocol;
use iggy_gateway_kafka::protocol::acl::PrincipalPermissions;
+use iggy_gateway_kafka::protocol::bounds_guard::MAX_SASL_AUTH_BYTES;
use iggy_gateway_kafka::protocol::sasl::{MAX_PRE_AUTH_API_VERSIONS,
PlainCredentials};
#[path = "common/codec.rs"]
@@ -56,7 +57,6 @@ 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 ERROR_UNKNOWN_SERVER_ERROR: i16 = -1;
@@ -434,21 +434,110 @@ async fn
given_an_unknown_mechanism_when_handshaking_should_report_the_supported
assert_closed(&mut stream).await;
}
+async fn handshake_v0_ok(stream: &mut TcpStream) {
+ let body = send(
+ stream,
+ API_KEY_SASL_HANDSHAKE,
+ 0,
+ 1,
+ &handshake_body("PLAIN"),
+ )
+ .await;
+ assert_eq!(
+ error_code(&body),
+ ERROR_NONE,
+ "PLAIN must be accepted at v0"
+ );
+}
+
+/// Sends `token` as kafka-python 2.0.2 and Sarama up to 1.40.1 send it after
a v0 handshake
+/// (KIP-152): in a bare length-prefixed frame, with no request header.
+async fn send_raw_token(stream: &mut TcpStream, token: &[u8]) {
+ let mut frame = BytesMut::new();
+ frame.put_i32(i32::try_from(token.len()).expect("test token fits i32"));
+ frame.extend_from_slice(token);
+ stream
+ .write_all(&frame)
+ .await
+ .expect("write the bare token");
+}
+
#[tokio::test]
-async fn
given_handshake_v0_when_offered_should_be_refused_before_any_headerless_token()
{
+async fn
given_handshake_v0_when_a_valid_bare_token_follows_should_answer_it_and_serve_requests()
{
+ let addr = spawn_sasl_gateway().await;
+ let mut stream = TcpStream::connect(addr).await.expect("connect");
+ handshake_v0_ok(&mut stream).await;
+
+ send_raw_token(&mut stream, &plain_token("alice", "s3cret")).await;
+ // The empty server token of PLAIN, in a length-prefixed frame.
kafka-python compares these
+ // four bytes, and Sarama reads exactly four.
+ let mut answer = [0xff_u8; 4];
+ tokio::time::timeout(Duration::from_secs(5), stream.read_exact(&mut
answer))
+ .await
+ .expect("the gateway must answer the bare token")
+ .expect("read the answer to the bare token");
+ assert_eq!(answer, [0, 0, 0, 0]);
+
+ // Requests with headers follow, as on any connection that authenticated.
+ let metadata = send(&mut stream, API_KEY_METADATA, 0, 2, &[0, 0, 0,
0]).await;
+ assert!(
+ !metadata.is_empty(),
+ "Metadata must answer once authenticated"
+ );
+}
+
+#[tokio::test]
+async fn
given_handshake_v0_when_the_bare_token_is_wrong_should_close_without_an_answer()
{
+ // A bare token has no answer that can carry an error code. A broker
closes the connection,
+ // and kafka-python and Sarama report the close as the failure.
+ let addr = spawn_sasl_gateway().await;
+ let mut stream = TcpStream::connect(addr).await.expect("connect");
+ handshake_v0_ok(&mut stream).await;
+
+ send_raw_token(&mut stream, &plain_token("alice", "wrong-password")).await;
+ assert_closed(&mut stream).await;
+}
+
+#[tokio::test]
+async fn
given_handshake_v0_when_the_bare_token_is_oversized_should_close_before_its_body()
{
+ // The token cap applies to the read itself, so an unauthenticated peer
cannot make the
+ // gateway wait for, or buffer, a frame up to `max_frame_size`.
+ let addr = spawn_sasl_gateway().await;
+ let mut stream = TcpStream::connect(addr).await.expect("connect");
+ handshake_v0_ok(&mut stream).await;
+
+ let oversized = i32::try_from(MAX_SASL_AUTH_BYTES + 1).expect("the cap
fits i32");
+ stream
+ .write_all(&oversized.to_be_bytes())
+ .await
+ .expect("write the length of the bare token");
+ // Only the length went out. A gateway that waited for the body would hold
the connection
+ // until its 5 s read timeout.
+ let read = read_byte_with_timeout(&mut stream,
Duration::from_secs(2)).await;
+ assert!(
+ matches!(read, ByteRead::Closed),
+ "the gateway must close at the length, got {read:?}"
+ );
+}
+
+#[tokio::test]
+async fn
given_handshake_v0_with_an_unknown_mechanism_should_report_the_supported_list()
{
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"),
+ &handshake_body("SCRAM-SHA-256"),
)
.await;
- assert_eq!(error_code(&body), ERROR_UNSUPPORTED_VERSION);
+ assert_eq!(error_code(&body), ERROR_UNSUPPORTED_SASL_MECHANISM);
+ assert!(
+ body.windows(5).any(|w| w == b"PLAIN"),
+ "the v0 refusal must name what is supported, as the v1 one does"
+ );
assert_closed(&mut stream).await;
}
@@ -673,11 +762,10 @@ fn parse_api_versions(body: &Bytes) -> Vec<(i16, i16,
i16)> {
#[tokio::test]
async fn
given_sasl_enabled_when_advertising_should_offer_handshake_from_version_zero() {
- // Regression. Advertising SaslHandshake as v1-only is what a version
firewall pinned to v1
- // suggests, and it breaks librdkafka outright: its SASL-handshake feature
detection depends on
- // key 17 appearing at version 0, so it reports "SASL Handshake not
supported by broker" and
- // never sends one. Advertising from 0 while still refusing a v0 handshake
at runtime is the
- // same split Produce already uses. Caught by a real client, not by a stub.
+ // Regression. Advertising SaslHandshake as v1-only breaks librdkafka
outright: its
+ // SASL-handshake feature detection depends on key 17 appearing at version
0, so it reports
+ // "SASL Handshake not supported by broker" and never sends one. Caught by
a real client, not
+ // by a stub.
let addr = spawn_sasl_gateway().await;
let mut stream = TcpStream::connect(addr).await.expect("connect");
@@ -692,10 +780,7 @@ async fn
given_sasl_enabled_when_advertising_should_offer_handshake_from_version
handshake.1, 0,
"librdkafka needs SaslHandshake advertised from v0 or it sees no SASL
at all"
);
- assert_eq!(
- handshake.2, 1,
- "v0 framing is refused, so v1 is the ceiling"
- );
+ assert_eq!(handshake.2, 1, "v1 is the highest SaslHandshake version");
let authenticate = rows
.iter()
