ryerraguntla commented on code in PR #4245:
URL: https://github.com/apache/iggy/pull/4245#discussion_r4082239453
##########
gateways/kafka/docs/TEST_SUITE.md:
##########
@@ -73,6 +74,27 @@ is test-only primitive encode/decode scaffolding for
hand-building legacy/advers
---
+## Real-client end-to-end suite
+
+`kafka_client_e2e_tests.rs` needs two things the rest of the suite does not:
Docker, and an
+already-built `iggy-server` in the same target directory. Missing either makes
it skip with a
+printed reason rather than fail, which is what lets `cargo test -p
iggy-gateway-kafka` stay usable
+without either.
+
+```bash
+cargo build --bin iggy-server
+KAFKA_E2E_REQUIRED=1 cargo test -p iggy-gateway-kafka --test
kafka_client_e2e_tests
+```
+
+`KAFKA_E2E_REQUIRED=1` turns a skip into a failure, mirroring
`KAFKA_FIXTURES_REQUIRED`, so a CI
+job that means to run these cannot report a pass over zero assertions. Set it
there.
+
Review Comment:
Claims kafka_bridge group. Nextest is kafka_client_e2e (nextest.toml:67).
Fix: rewrite.
##########
gateways/kafka/src/server.rs:
##########
@@ -682,7 +704,57 @@ async fn authenticate_token(
// the account exists.
Err(AuthError::Unavailable) => {
warn!(%peer, "SASL authentication could not be completed; Iggy is
unreachable");
- HandleOutcome::Close
+ (HandleOutcome::Close, None)
+ }
+ }
+}
+
+/// Answers `DescribeAcls` from the permissions captured when this connection
authenticated.
+///
+/// Only reachable on an authenticated connection: the SASL gate refuses every
key but one before a
+/// principal exists, and the caller additionally gates this on the feature
being on, so a gateway
+/// with SASL off never routes here. The `None` arm is a fail-closed guard,
not a reachable path.
+fn describe_acls(
+ principal: Option<&AuthenticatedPrincipal>,
+ api_version: i16,
+ body: Bytes,
+ peer: &SocketAddr,
+) -> HandleOutcome {
+ let Some(principal) = principal else {
+ // `error!` rather than `debug!` on purpose, unlike every other
refusal here: reaching this
+ // means the routing guards above disagree with each other, which is a
gateway fault and not
+ // something a client can provoke.
+ error!(%peer, "DescribeAcls reached a connection with no authenticated
principal");
+ // Not `encode_error_for_key`: key 29 is absent from
`SUPPORTED_RANGES`, so that helper
+ // always returns `Close` here and the code would read as if it
answers when it cannot.
+ return respond_describe_acls_error(api_version,
ERROR_ILLEGAL_SASL_STATE, true);
+ };
+ // The firewall table cannot cover this key: it is kept out of
`SUPPORTED_RANGES` on purpose,
+ // so the advertised range would otherwise be enforced only by whatever
`kafka_protocol`'s
+ // schema happens to accept. A crate bump adding v4 would start answering
v4 while ApiVersions
+ // still says 3, which is exactly what the sibling SASL keys pin
explicitly against.
+ if !SASL_ADVERTISED_DESCRIBE_ACLS_VERSIONS.contains(&api_version) {
+ debug!(%peer, api_version, "DescribeAcls version outside the
advertised range");
+ return HandleOutcome::Close;
+ }
+ if !principal.permissions_known {
+ // The permission read failed after a successful login, so this
connection holds no real
+ // answer. Reporting the empty fallback would tell an operator the
principal has no access,
+ // which is a different statement from "we could not find out".
+ warn!(%peer, "DescribeAcls asked on a connection whose permissions
were never read");
+ return respond_describe_acls_error(api_version,
ERROR_UNKNOWN_SERVER_ERROR, false);
Review Comment:
permissions_known=false → -1 keep-open. Fakes always known: true
(sasl_tests.rs:84). Empty-0 vs -1 can swap unseen. Java fails immediately, no
hang. **Fix**: stub known=false; assert -1 + stay-open.
##########
gateways/kafka/src/protocol/acl.rs:
##########
@@ -0,0 +1,522 @@
+// 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.
+
+//! Rendering an Iggy principal's permissions as Kafka ACL bindings.
+//!
+//! Pure and synchronous, and deliberately free of any Iggy type: the
permissions arrive as
+//! [`PrincipalPermissions`], which `auth` fills in. That keeps the protocol
layer independent of
+//! the SDK and makes the mapping unit-testable without a server.
+//!
+//! See `docs/ACL_MAPPING.md` for the decisions this implements, including
what is not mapped.
+
+/// Kafka resource types (`org.apache.kafka.common.resource.ResourceType`).
`kafka_protocol` carries
+/// these as bare `i8` with no enum, so the values live here.
+pub mod resource_type {
+ pub const ANY: i8 = 1;
+ pub const TOPIC: i8 = 2;
+ pub const GROUP: i8 = 3;
+ pub const CLUSTER: i8 = 4;
+}
+
+/// Kafka pattern types (`org.apache.kafka.common.resource.PatternType`).
+pub mod pattern_type {
+ pub const ANY: i8 = 1;
+ /// Kafka's own "match anything of this type" lookup, which a filter may
ask for.
+ pub const MATCH: i8 = 2;
+ pub const LITERAL: i8 = 3;
+}
+
+/// Kafka ACL operations (`org.apache.kafka.common.acl.AclOperation`).
+pub mod operation {
+ pub const ANY: i8 = 1;
+ pub const ALL: i8 = 2;
+ pub const READ: i8 = 3;
+ pub const WRITE: i8 = 4;
+ pub const CREATE: i8 = 5;
+ pub const DELETE: i8 = 6;
+ pub const ALTER: i8 = 7;
+ pub const DESCRIBE: i8 = 8;
+}
+
+/// Kafka ACL permission types
(`org.apache.kafka.common.acl.AclPermissionType`).
+pub mod permission_type {
+ pub const ANY: i8 = 1;
+ pub const ALLOW: i8 = 3;
+}
+
+/// Name Kafka gives the cluster resource. There is exactly one, and it is
always called this.
+pub const CLUSTER_NAME: &str = "kafka-cluster";
+
+/// How Kafka spells "every resource of this type": a literal pattern named
`*`.
+pub const WILDCARD: &str = "*";
+
+/// Host scope on every binding this gateway renders. Iggy has no host-scoped
permissions.
+pub const ANY_HOST: &str = "*";
+
+/// The subset of an Iggy principal's global permissions that has a Kafka
meaning.
+///
+/// Stream-level flags are folded into the topic ones by the caller: Kafka has
no resource above a
+/// topic, and every Kafka topic lives inside one Iggy stream, so a stream
grant is in practice a
+/// grant over the topics a Kafka client can reach.
+///
+/// The boolean count mirrors Iggy's own `GlobalPermissions`, which is a flat
set of independent
+/// grants. Collapsing them into a bitfield would hide which grant is which at
every call site for
+/// no gain, so the lint is allowed here the way it is elsewhere in this
repository.
+#[allow(clippy::struct_excessive_bools)]
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+pub struct PrincipalPermissions {
+ /// True when the principal holds either of Iggy's server flags.
`manage_servers` is not
+ /// carried separately: Iggy reads it in exactly one rule, as an alias for
this one, so it
+ /// gates no mutation anywhere and renders nothing of its own.
`docs/ACL_MAPPING.md` has the
+ /// argument, which is the same one that keeps `manage_users` out of the
table.
+ pub read_servers: bool,
+ pub read_topics: bool,
+ pub manage_topics: bool,
+ pub poll_messages: bool,
+ pub send_messages: bool,
+}
+
+/// One rendered binding. Principal, host and permission type are constant for
every binding this
+/// gateway produces, so they are not carried here.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct AclBinding {
+ pub resource_type: i8,
+ pub resource_name: &'static str,
+ pub operation: i8,
+}
+
+impl AclBinding {
+ const fn new(resource_type: i8, resource_name: &'static str, operation:
i8) -> Self {
+ Self {
+ resource_type,
+ resource_name,
+ operation,
+ }
+ }
+}
+
+/// Renders a principal's permissions as Kafka ACL bindings.
+///
+/// Only global permissions are rendered, as wildcard bindings. Iggy keys its
per-stream and
+/// per-topic permissions by numeric id, and the topic mapping is one-way, so
a named Kafka binding
+/// cannot be reconstructed from them without a reverse index that does not
exist. The result
+/// under-reports rather than over-reports, which is the safe direction for an
authorization view.
+#[must_use]
+pub fn bindings_for(permissions: &PrincipalPermissions) -> Vec<AclBinding> {
+ let mut bindings = Vec::new();
+
+ if permissions.read_servers {
+ bindings.push(AclBinding::new(
+ resource_type::CLUSTER,
+ CLUSTER_NAME,
+ operation::DESCRIBE,
+ ));
+ }
+ if permissions.read_topics {
+ bindings.push(AclBinding::new(
+ resource_type::TOPIC,
+ WILDCARD,
+ operation::DESCRIBE,
+ ));
+ }
+ if permissions.manage_topics {
+ for op in [operation::CREATE, operation::DELETE, operation::ALTER] {
+ bindings.push(AclBinding::new(resource_type::TOPIC, WILDCARD, op));
+ }
+ }
+ if permissions.poll_messages {
+ bindings.push(AclBinding::new(
+ resource_type::TOPIC,
+ WILDCARD,
+ operation::READ,
+ ));
+ }
+ // Derived, not stored: Iggy has no group-level permission. Group
*membership* operations
+ // (create, delete, get, join, leave) route through
`Permissioner::get_topic`
+ // (`permissioner_rules/consumer_groups.rs`), which admits on the read and
manage flags and
+ // never consults `poll_messages`. Deriving this from polling instead
granted a group to
+ // principals Iggy denies, which is the over-report the design commits
against.
+ //
+ // Not every group-shaped operation goes that way: offset commit and fetch
route through
+ // `poll_messages` (`permissioner_rules/consumer_offsets.rs`), while Kafka
gates them on this
+ // same GROUP READ. A principal with polling but no read grant is
therefore shown no group
+ // binding while Iggy would let it commit an offset. That under-reports,
which is the safe
+ // direction, and the alternative over-reports membership.
+ if permissions.read_topics {
+ bindings.push(AclBinding::new(
+ resource_type::GROUP,
+ WILDCARD,
+ operation::READ,
+ ));
+ }
+ if permissions.send_messages {
+ bindings.push(AclBinding::new(
+ resource_type::TOPIC,
+ WILDCARD,
+ operation::WRITE,
+ ));
+ }
+
+ bindings
+}
+
+/// The filter carried by a `DescribeAcls` request, already decoded.
+#[derive(Debug, Clone)]
+pub struct AclFilter {
+ pub resource_type: i8,
+ pub resource_name: Option<String>,
+ pub pattern_type: i8,
+ pub principal: Option<String>,
+ pub host: Option<String>,
+ pub operation: i8,
+ pub permission_type: i8,
+}
+
+impl AclFilter {
+ /// Whether this filter selects `binding`, belonging to `principal`.
+ ///
+ /// Kafka's `ANY` sentinel matches everything, and an absent string field
is the same as `ANY`.
+ /// A principal filter naming anyone else matches nothing: the gateway
holds no administrative
+ /// credentials and can only ever read the caller's own record, so
claiming an empty result for
+ /// another user is the only honest answer it can give.
+ #[must_use]
+ pub fn matches(&self, binding: &AclBinding, principal: &str) -> bool {
+ if self.resource_type != resource_type::ANY && self.resource_type !=
binding.resource_type {
+ return false;
+ }
+ // PREFIXED selects nothing, because nothing here is prefix-scoped.
+ if !matches!(
+ self.pattern_type,
+ pattern_type::ANY | pattern_type::MATCH | pattern_type::LITERAL
+ ) {
+ return false;
+ }
+ if !self.matches_resource_name(binding.resource_name) {
+ return false;
+ }
+ if !matches_name(self.principal.as_deref(),
&format!("User:{principal}")) {
Review Comment:
acl.rs:211 / README.md:136 — kafka-acls.sh --list uses ANY, prints caller
bindings as cluster catalog. Other User: empty 0, not 31. Documented self-only;
operator still reads as full dump. **Fix**: README: --list = self-snapshot, not
broker dump.
##########
gateways/kafka/tests/kafka_client_e2e_tests.rs:
##########
@@ -0,0 +1,648 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! End-to-end tests driving **real Kafka clients** against the gateway.
+//!
+//! Every other suite in this crate hand-builds wire frames, which cannot
catch a client
+//! compatibility problem by construction. One already slipped through that
way: advertising
+//! `SaslHandshake` from v1 rather than v0 passed every hand-built test and
made librdkafka report
+//! "SASL Handshake not supported by broker" before it sent anything. These
tests exist so that
+//! class of bug fails in CI instead of during manual testing.
+//!
+//! The stack is real on all three sides: a spawned `iggy-server` process, the
gateway in-process
+//! with a real `IggyAuthenticator`, and a client from a container. They
automate categories S and T
+//! of `docs/MANUAL_TESTING.md`.
+//!
+//! Prerequisites are Docker and an already-built `iggy-server`. Missing
either skips, the way the
+//! wire-fixture suites do, unless `KAFKA_E2E_REQUIRED=1` is set, which turns
a skip into a failure
+//! so a broken CI step cannot leave these silently green.
+
+use std::net::SocketAddr;
+use std::path::PathBuf;
+use std::process::{Child, Command, Stdio};
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+
+use iggy_gateway_kafka::GatewayConfig;
+use iggy_gateway_kafka::auth::IggyAuthenticator;
+
+#[path = "common/server.rs"]
+mod server;
+
+use server::spawn_test_server_with_authenticator;
+
+const KCAT_IMAGE: &str = "edenhill/kcat:1.7.1";
+const KAFKA_IMAGE: &str = "apache/kafka:3.9.0";
+const ROOT_USER: &str = "iggy";
+const ROOT_PASSWORD: &str = "iggy";
+/// Password for every non-root principal these tests create.
+const USER_PASSWORD: &str = "s3cretpass";
+
+/// Budget for `iggy-server` to start listening. Generous: it is a cold
process start, and a debug
+/// build on a loaded machine is not quick.
+const SERVER_READY_TIMEOUT: Duration = Duration::from_secs(45);
+
+/// Reports why the suite cannot run, and whether that is fatal.
+///
+/// Returns `true` when the caller should skip. `KAFKA_E2E_REQUIRED=1` makes
it panic instead, so a
+/// CI job that means to run these fails loudly rather than reporting a pass
over zero assertions.
+fn skip(reason: &str) -> bool {
+ assert!(
+ std::env::var("KAFKA_E2E_REQUIRED").as_deref() != Ok("1"),
+ "KAFKA_E2E_REQUIRED=1 but the suite cannot run: {reason}"
+ );
+ eprintln!("skipping real-client end-to-end test: {reason}");
+ true
+}
+
+fn docker_missing() -> bool {
+ let available = Command::new("docker")
+ .arg("info")
+ .stdout(Stdio::null())
+ .stderr(Stdio::null())
+ .status()
+ .is_ok_and(|status| status.success());
+ if available {
+ false
+ } else {
+ skip("docker is unavailable")
+ }
+}
+
+/// Locates the already-built `iggy-server` alongside this test binary. Does
not build it.
+fn iggy_server_binary() -> Option<PathBuf> {
+ let mut dir = std::env::current_exe().ok()?;
+ // .../target/<profile>/deps/<test binary> -> .../target/<profile>
+ dir.pop();
+ dir.pop();
+ let candidate = dir.join(format!("iggy-server{}",
std::env::consts::EXE_SUFFIX));
+ candidate.is_file().then_some(candidate)
+}
+
+/// Outcome of waiting for the spawned server to come up.
+enum Ready {
+ Listening,
+ /// The process exited before it bound, carrying its status.
+ Exited(String),
+ TimedOut,
+}
+
+/// A spawned `iggy-server`, killed on drop.
+///
+/// The environment recipe mirrors `bridge_iggy_integration_tests.rs`, which
explains each setting:
+/// the other listeners are off so concurrently spawned servers do not fight
over their fixed
+/// ports, and the shard pool is capped so a spawned server does not size
itself to the whole
+/// machine against unrelated packages' tests in the same run.
+struct TestServer {
+ child: Child,
+ address: String,
+ http_address: String,
+ _data_dir: tempfile::TempDir,
+}
+
+impl TestServer {
+ fn spawn() -> Result<Self, String> {
+ let Some(binary) = iggy_server_binary() else {
+ return Err(
+ "iggy-server is not built; run `cargo build --bin iggy-server`
first".to_string(),
+ );
+ };
+ let data_dir = tempfile::tempdir().expect("create server data dir");
+ let port = free_port();
+ let address = format!("127.0.0.1:{port}");
+ let http_address = format!("127.0.0.1:{}", free_port());
+
+ let child = Command::new(binary)
+ .arg("--fresh")
+ .env("IGGY_PATH", data_dir.path().display().to_string())
+ .env("IGGY_TCP_ADDRESS", &address)
+ // Left on, unlike the sibling suite: the three-principal
procedure creates its
+ // non-root users over the HTTP API, which is the only
administrative surface reachable
+ // from a test without pulling in the SDK.
+ .env("IGGY_HTTP_ENABLED", "true")
+ .env("IGGY_HTTP_ADDRESS", &http_address)
+ .env("IGGY_QUIC_ENABLED", "false")
+ .env("IGGY_WEBSOCKET_ENABLED", "false")
+ .env("IGGY_SHARDING_PIN_CORES", "false")
+ .env("IGGY_SHARDING_CPU_ALLOCATION", "0..4")
+ .env("IGGY_ROOT_USERNAME", ROOT_USER)
+ .env("IGGY_ROOT_PASSWORD", ROOT_PASSWORD)
+ .stdout(Stdio::null())
+ .stderr(Stdio::null())
+ .spawn()
+ .expect("spawn iggy-server");
+
+ let mut server = Self {
+ child,
+ address,
+ http_address,
+ _data_dir: data_dir,
+ };
+ match server.wait_ready() {
+ Ready::Listening => Ok(server),
+ Ready::Exited(status) => Err(format!(
+ "iggy-server exited during startup with {status}; its output
is suppressed, so \
+ rerun it by hand with the same environment to see why"
+ )),
+ Ready::TimedOut => Err(format!(
+ "iggy-server never bound {} within {SERVER_READY_TIMEOUT:?}",
+ server.address
+ )),
+ }
+ }
+
+ fn wait_ready(&mut self) -> Ready {
+ let deadline = Instant::now() + SERVER_READY_TIMEOUT;
+ while Instant::now() < deadline {
+ // Both listeners, not just the data one: the server binds HTTP
after TCP, so probing
+ // TCP alone declares readiness while the provisioning calls that
follow would still
+ // be refused.
+ if std::net::TcpStream::connect(&self.address).is_ok()
+ && std::net::TcpStream::connect(&self.http_address).is_ok()
+ {
+ return Ready::Listening;
+ }
+ // Without this a server that dies at boot burns the whole budget
and is then reported
+ // as a timeout, which sends the reader looking at the wrong thing
entirely.
+ if let Ok(Some(status)) = self.child.try_wait() {
+ return Ready::Exited(status.to_string());
+ }
+ std::thread::sleep(Duration::from_millis(200));
+ }
+ Ready::TimedOut
+ }
+}
+
+impl Drop for TestServer {
+ fn drop(&mut self) {
+ let _ = self.child.kill();
+ let _ = self.child.wait();
+ }
+}
+
+/// Binds an ephemeral port and releases it, so the server can take it.
+///
+/// A window exists between release and rebind. Acceptable here because this
suite is serialized
+/// into its own nextest group, so nothing else in it is drawing ports
concurrently.
+fn free_port() -> u16 {
+ let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind
ephemeral port");
+ listener.local_addr().expect("local addr").port()
+}
+
+/// Starts the gateway with SASL on, verifying credentials against
`iggy_address`.
+async fn spawn_gateway(iggy_address: &str) -> SocketAddr {
+ let config = GatewayConfig {
+ sasl_enabled: true,
+ ..GatewayConfig::default()
+ };
+ let authenticator =
Arc::new(IggyAuthenticator::new(iggy_address.to_string()));
+ let (addr, shutdown) = spawn_test_server_with_authenticator(config,
authenticator).await;
+ // Held for the test's lifetime: dropping the sender shuts the gateway
down mid-exchange.
+ std::mem::forget(shutdown);
+ addr
+}
+
+/// Runs a container against the host network and returns its combined output.
+///
+/// `--network host` is what lets a containerised client reach a gateway bound
to the host's
+/// loopback. It is Linux-specific, which matches where CI runs.
+///
+/// This blocks the calling thread for the life of the container, which is why
every test here uses
+/// a multi-threaded runtime: the gateway runs as a spawned task, and on the
single-threaded runtime
+/// `#[tokio::test]` gives by default, this call would starve it and nothing
would ever listen.
+fn run_client(image: &str, args: &[&str], mounts: &[(&str, &str)]) ->
ClientRun {
+ let mut command = Command::new("docker");
+ command.args(["run", "--rm", "--network", "host"]);
+ for (host, guest) in mounts {
+ command.args(["-v", &format!("{host}:{guest}:ro")]);
+ }
+ command.arg(image).args(args);
+ let output = command.output().expect("run client container");
+ ClientRun {
+ succeeded: output.status.success(),
+ text: format!(
+ "{}{}",
+ String::from_utf8_lossy(&output.stdout),
+ String::from_utf8_lossy(&output.stderr)
+ ),
+ }
+}
+
+/// What a client container produced, and whether it ran at all.
+///
+/// The exit status is carried deliberately. Every negative assertion in this
suite is of the form
+/// "the output does not contain X", and a container that never started
produces output satisfying
+/// all of them. Without this a wrong image tag, a malformed config or a
Docker failure reads as a
+/// passing test.
+struct ClientRun {
+ succeeded: bool,
+ text: String,
+}
+
+impl ClientRun {
+ /// Asserts the client itself ran, then hands back its output for content
assertions.
+ fn expect_ran(self, what: &str) -> String {
+ assert!(
+ self.succeeded,
+ "{what}: the client container failed to run, so any assertion
about its output would \
+ be meaningless. Output: {}",
+ self.text
+ );
+ self.text
+ }
+
+ /// For the cases where the client is *expected* to fail, so only the
output matters.
+ fn text(self) -> String {
+ self.text
+ }
+}
+
+fn kcat(addr: SocketAddr, username: &str, password: &str, mechanism: &str) ->
ClientRun {
+ run_client(
+ KCAT_IMAGE,
+ &[
+ "-b",
+ &addr.to_string(),
+ "-X",
+ "security.protocol=SASL_PLAINTEXT",
+ "-X",
+ &format!("sasl.mechanisms={mechanism}"),
+ "-X",
+ &format!("sasl.username={username}"),
+ "-X",
+ &format!("sasl.password={password}"),
+ "-L",
+ ],
+ &[],
+ )
+}
+
+/// Creates an Iggy user over the HTTP API with the given global permissions.
+///
+/// The spec's three-principal procedure needs non-root accounts, and root
holds every flag, so a
+/// suite that only ever authenticates as root checks neither the inheritance
projection nor the
+/// empty view against a real client.
+fn create_user(http: &str, token: &str, username: &str, permissions: &str) {
+ let body = format!(
+
r#"{{"username":"{username}","password":"{USER_PASSWORD}","status":"active","permissions":{permissions}}}"#
+ );
+ let status = Command::new("curl")
+ .args([
+ "-s",
+ "-o",
+ "/dev/null",
+ "-w",
+ "%{http_code}",
+ "-X",
+ "POST",
+ &format!("http://{http}/users";),
+ "-H",
+ &format!("Authorization: Bearer {token}"),
+ "-H",
+ "Content-Type: application/json",
+ "-d",
+ &body,
+ ])
+ .output()
+ .expect("create user");
+ let code = String::from_utf8_lossy(&status.stdout).to_string();
+ assert!(
+ code.starts_with('2'),
+ "creating {username} returned HTTP {code}"
+ );
+}
+
+/// Logs in as root over HTTP and returns the bearer token.
+fn root_token(http: &str) -> String {
+ let output = Command::new("curl")
+ .args([
+ "-s",
+ "-X",
+ "POST",
+ &format!("http://{http}/users/login";),
+ "-H",
+ "Content-Type: application/json",
+ "-d",
+
&format!(r#"{{"username":"{ROOT_USER}","password":"{ROOT_PASSWORD}"}}"#),
+ ])
+ .output()
+ .expect("root login");
+ let body = String::from_utf8_lossy(&output.stdout);
+ // Avoids a JSON dependency for one field: the token is the value after
this key.
+ let key = "\"token\":\"";
+ let start = body.find(key).expect("login response carries a token") +
key.len();
+ let end = start + body[start..].find('"').expect("token is terminated");
+ body[start..end].to_string()
+}
+
+/// Writes a JAAS client config for the Java tools and returns its path, kept
alive by the handle.
+fn java_client_config(username: &str, password: &str) -> (tempfile::TempDir,
PathBuf) {
Review Comment:
no request.timeout.ms; list_acls ×5. Hang ≈ 300s kill. **Fix**: 10s in
command-config.
##########
gateways/kafka/src/protocol/acl.rs:
##########
@@ -0,0 +1,522 @@
+// 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.
+
+//! Rendering an Iggy principal's permissions as Kafka ACL bindings.
+//!
+//! Pure and synchronous, and deliberately free of any Iggy type: the
permissions arrive as
+//! [`PrincipalPermissions`], which `auth` fills in. That keeps the protocol
layer independent of
+//! the SDK and makes the mapping unit-testable without a server.
+//!
+//! See `docs/ACL_MAPPING.md` for the decisions this implements, including
what is not mapped.
+
+/// Kafka resource types (`org.apache.kafka.common.resource.ResourceType`).
`kafka_protocol` carries
+/// these as bare `i8` with no enum, so the values live here.
+pub mod resource_type {
+ pub const ANY: i8 = 1;
+ pub const TOPIC: i8 = 2;
+ pub const GROUP: i8 = 3;
+ pub const CLUSTER: i8 = 4;
+}
+
+/// Kafka pattern types (`org.apache.kafka.common.resource.PatternType`).
+pub mod pattern_type {
+ pub const ANY: i8 = 1;
+ /// Kafka's own "match anything of this type" lookup, which a filter may
ask for.
+ pub const MATCH: i8 = 2;
+ pub const LITERAL: i8 = 3;
+}
+
+/// Kafka ACL operations (`org.apache.kafka.common.acl.AclOperation`).
+pub mod operation {
+ pub const ANY: i8 = 1;
+ pub const ALL: i8 = 2;
+ pub const READ: i8 = 3;
+ pub const WRITE: i8 = 4;
+ pub const CREATE: i8 = 5;
+ pub const DELETE: i8 = 6;
+ pub const ALTER: i8 = 7;
+ pub const DESCRIBE: i8 = 8;
+}
+
+/// Kafka ACL permission types
(`org.apache.kafka.common.acl.AclPermissionType`).
+pub mod permission_type {
+ pub const ANY: i8 = 1;
+ pub const ALLOW: i8 = 3;
+}
+
+/// Name Kafka gives the cluster resource. There is exactly one, and it is
always called this.
+pub const CLUSTER_NAME: &str = "kafka-cluster";
+
+/// How Kafka spells "every resource of this type": a literal pattern named
`*`.
+pub const WILDCARD: &str = "*";
+
+/// Host scope on every binding this gateway renders. Iggy has no host-scoped
permissions.
+pub const ANY_HOST: &str = "*";
+
+/// The subset of an Iggy principal's global permissions that has a Kafka
meaning.
+///
+/// Stream-level flags are folded into the topic ones by the caller: Kafka has
no resource above a
+/// topic, and every Kafka topic lives inside one Iggy stream, so a stream
grant is in practice a
+/// grant over the topics a Kafka client can reach.
+///
+/// The boolean count mirrors Iggy's own `GlobalPermissions`, which is a flat
set of independent
+/// grants. Collapsing them into a bitfield would hide which grant is which at
every call site for
+/// no gain, so the lint is allowed here the way it is elsewhere in this
repository.
+#[allow(clippy::struct_excessive_bools)]
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+pub struct PrincipalPermissions {
Review Comment:
PrincipalPermissions global bools. From drops Permissions.streams.
DescribeAcls under-report OK. Later Produce off this snapshot over-denies
per-topic grants. Not authorizer today. **Fix**: store full Permissions or
never authz off this type.
##########
gateways/kafka/src/auth.rs:
##########
@@ -265,7 +316,92 @@ impl SaslAuthenticator for IggyAuthenticator {
}
}
- outcome
+ outcome.map(|(permissions, permissions_known)| AuthenticatedPrincipal {
+ username: credentials.username.clone(),
+ permissions,
+ permissions_known,
+ })
+ }
+}
+
+/// Reads the just-authenticated user's own record and projects its global
permissions.
+///
+/// A missing record or absent permissions both yield an empty set rather than
an error: the
+/// credentials were already accepted, so refusing the connection here would
reject a valid login
+/// over an authorization view it never asked for.
+async fn fetch_permissions(
+ client: &impl Client,
+ username: &str,
+) -> Result<(PrincipalPermissions, bool), AuthError> {
+ // Unreachable in practice: a username that reached a successful login is
already inside
+ // Identifier's own length bounds. Propagating rather than degrading is
still the wrong shape
+ // for this function, so it degrades like every other failure below.
+ let Ok(identifier) = Identifier::named(username) else {
+ warn!("authenticated, but the principal's name is not a valid Iggy
identifier");
+ return Ok((PrincipalPermissions::default(), false));
+ };
+ // Deliberately not propagated as a failure. The credentials were already
accepted by the login
+ // above, so turning a stumble on this second round trip into a rejection
would answer a correct
+ // password with `SASL_AUTHENTICATION_FAILED`, which a Kafka client treats
as fatal and raises
+ // to the application. Losing the ACL view is the lesser harm, and it
degrades to an empty one.
+ let fetched = tokio::time::timeout(PERMISSION_READ_TIMEOUT,
client.get_user(&identifier)).await;
Review Comment:
1s get_user timeout cancels wait only. SDK tokio::spawn holds stream lock
until RESPONSE_READ_TIMEOUT 30s. Slot free ~12s. Logged-in Iggy TCP zombie. New
success-path leak after login already proved. **Fix**: drop slot after login;
abort/join spawn or skip shutdown wait.
##########
gateways/kafka/tests/kafka_client_e2e_tests.rs:
##########
@@ -0,0 +1,648 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! End-to-end tests driving **real Kafka clients** against the gateway.
+//!
+//! Every other suite in this crate hand-builds wire frames, which cannot
catch a client
+//! compatibility problem by construction. One already slipped through that
way: advertising
+//! `SaslHandshake` from v1 rather than v0 passed every hand-built test and
made librdkafka report
+//! "SASL Handshake not supported by broker" before it sent anything. These
tests exist so that
+//! class of bug fails in CI instead of during manual testing.
+//!
+//! The stack is real on all three sides: a spawned `iggy-server` process, the
gateway in-process
+//! with a real `IggyAuthenticator`, and a client from a container. They
automate categories S and T
+//! of `docs/MANUAL_TESTING.md`.
+//!
+//! Prerequisites are Docker and an already-built `iggy-server`. Missing
either skips, the way the
+//! wire-fixture suites do, unless `KAFKA_E2E_REQUIRED=1` is set, which turns
a skip into a failure
+//! so a broken CI step cannot leave these silently green.
+
+use std::net::SocketAddr;
+use std::path::PathBuf;
+use std::process::{Child, Command, Stdio};
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+
+use iggy_gateway_kafka::GatewayConfig;
+use iggy_gateway_kafka::auth::IggyAuthenticator;
+
+#[path = "common/server.rs"]
+mod server;
+
+use server::spawn_test_server_with_authenticator;
+
+const KCAT_IMAGE: &str = "edenhill/kcat:1.7.1";
+const KAFKA_IMAGE: &str = "apache/kafka:3.9.0";
+const ROOT_USER: &str = "iggy";
+const ROOT_PASSWORD: &str = "iggy";
+/// Password for every non-root principal these tests create.
+const USER_PASSWORD: &str = "s3cretpass";
+
+/// Budget for `iggy-server` to start listening. Generous: it is a cold
process start, and a debug
+/// build on a loaded machine is not quick.
+const SERVER_READY_TIMEOUT: Duration = Duration::from_secs(45);
+
+/// Reports why the suite cannot run, and whether that is fatal.
+///
+/// Returns `true` when the caller should skip. `KAFKA_E2E_REQUIRED=1` makes
it panic instead, so a
+/// CI job that means to run these fails loudly rather than reporting a pass
over zero assertions.
+fn skip(reason: &str) -> bool {
+ assert!(
+ std::env::var("KAFKA_E2E_REQUIRED").as_deref() != Ok("1"),
+ "KAFKA_E2E_REQUIRED=1 but the suite cannot run: {reason}"
+ );
+ eprintln!("skipping real-client end-to-end test: {reason}");
+ true
+}
+
+fn docker_missing() -> bool {
+ let available = Command::new("docker")
+ .arg("info")
+ .stdout(Stdio::null())
+ .stderr(Stdio::null())
+ .status()
+ .is_ok_and(|status| status.success());
+ if available {
+ false
+ } else {
+ skip("docker is unavailable")
+ }
+}
+
+/// Locates the already-built `iggy-server` alongside this test binary. Does
not build it.
+fn iggy_server_binary() -> Option<PathBuf> {
+ let mut dir = std::env::current_exe().ok()?;
+ // .../target/<profile>/deps/<test binary> -> .../target/<profile>
+ dir.pop();
+ dir.pop();
+ let candidate = dir.join(format!("iggy-server{}",
std::env::consts::EXE_SUFFIX));
+ candidate.is_file().then_some(candidate)
+}
+
+/// Outcome of waiting for the spawned server to come up.
+enum Ready {
+ Listening,
+ /// The process exited before it bound, carrying its status.
+ Exited(String),
+ TimedOut,
+}
+
+/// A spawned `iggy-server`, killed on drop.
+///
+/// The environment recipe mirrors `bridge_iggy_integration_tests.rs`, which
explains each setting:
+/// the other listeners are off so concurrently spawned servers do not fight
over their fixed
+/// ports, and the shard pool is capped so a spawned server does not size
itself to the whole
+/// machine against unrelated packages' tests in the same run.
+struct TestServer {
+ child: Child,
+ address: String,
+ http_address: String,
+ _data_dir: tempfile::TempDir,
+}
+
+impl TestServer {
+ fn spawn() -> Result<Self, String> {
+ let Some(binary) = iggy_server_binary() else {
+ return Err(
+ "iggy-server is not built; run `cargo build --bin iggy-server`
first".to_string(),
+ );
+ };
+ let data_dir = tempfile::tempdir().expect("create server data dir");
+ let port = free_port();
+ let address = format!("127.0.0.1:{port}");
+ let http_address = format!("127.0.0.1:{}", free_port());
+
+ let child = Command::new(binary)
+ .arg("--fresh")
+ .env("IGGY_PATH", data_dir.path().display().to_string())
+ .env("IGGY_TCP_ADDRESS", &address)
+ // Left on, unlike the sibling suite: the three-principal
procedure creates its
+ // non-root users over the HTTP API, which is the only
administrative surface reachable
+ // from a test without pulling in the SDK.
+ .env("IGGY_HTTP_ENABLED", "true")
+ .env("IGGY_HTTP_ADDRESS", &http_address)
+ .env("IGGY_QUIC_ENABLED", "false")
+ .env("IGGY_WEBSOCKET_ENABLED", "false")
+ .env("IGGY_SHARDING_PIN_CORES", "false")
+ .env("IGGY_SHARDING_CPU_ALLOCATION", "0..4")
+ .env("IGGY_ROOT_USERNAME", ROOT_USER)
+ .env("IGGY_ROOT_PASSWORD", ROOT_PASSWORD)
+ .stdout(Stdio::null())
+ .stderr(Stdio::null())
+ .spawn()
+ .expect("spawn iggy-server");
+
+ let mut server = Self {
+ child,
+ address,
+ http_address,
+ _data_dir: data_dir,
+ };
+ match server.wait_ready() {
+ Ready::Listening => Ok(server),
+ Ready::Exited(status) => Err(format!(
+ "iggy-server exited during startup with {status}; its output
is suppressed, so \
+ rerun it by hand with the same environment to see why"
+ )),
+ Ready::TimedOut => Err(format!(
+ "iggy-server never bound {} within {SERVER_READY_TIMEOUT:?}",
+ server.address
+ )),
+ }
+ }
+
+ fn wait_ready(&mut self) -> Ready {
+ let deadline = Instant::now() + SERVER_READY_TIMEOUT;
+ while Instant::now() < deadline {
+ // Both listeners, not just the data one: the server binds HTTP
after TCP, so probing
+ // TCP alone declares readiness while the provisioning calls that
follow would still
+ // be refused.
+ if std::net::TcpStream::connect(&self.address).is_ok()
+ && std::net::TcpStream::connect(&self.http_address).is_ok()
+ {
+ return Ready::Listening;
+ }
+ // Without this a server that dies at boot burns the whole budget
and is then reported
+ // as a timeout, which sends the reader looking at the wrong thing
entirely.
+ if let Ok(Some(status)) = self.child.try_wait() {
+ return Ready::Exited(status.to_string());
+ }
+ std::thread::sleep(Duration::from_millis(200));
+ }
+ Ready::TimedOut
+ }
+}
+
+impl Drop for TestServer {
+ fn drop(&mut self) {
+ let _ = self.child.kill();
+ let _ = self.child.wait();
+ }
+}
+
+/// Binds an ephemeral port and releases it, so the server can take it.
+///
+/// A window exists between release and rebind. Acceptable here because this
suite is serialized
+/// into its own nextest group, so nothing else in it is drawing ports
concurrently.
+fn free_port() -> u16 {
+ let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind
ephemeral port");
+ listener.local_addr().expect("local addr").port()
+}
+
+/// Starts the gateway with SASL on, verifying credentials against
`iggy_address`.
+async fn spawn_gateway(iggy_address: &str) -> SocketAddr {
+ let config = GatewayConfig {
+ sasl_enabled: true,
+ ..GatewayConfig::default()
+ };
+ let authenticator =
Arc::new(IggyAuthenticator::new(iggy_address.to_string()));
+ let (addr, shutdown) = spawn_test_server_with_authenticator(config,
authenticator).await;
+ // Held for the test's lifetime: dropping the sender shuts the gateway
down mid-exchange.
+ std::mem::forget(shutdown);
+ addr
+}
+
+/// Runs a container against the host network and returns its combined output.
+///
+/// `--network host` is what lets a containerised client reach a gateway bound
to the host's
+/// loopback. It is Linux-specific, which matches where CI runs.
+///
+/// This blocks the calling thread for the life of the container, which is why
every test here uses
+/// a multi-threaded runtime: the gateway runs as a spawned task, and on the
single-threaded runtime
+/// `#[tokio::test]` gives by default, this call would starve it and nothing
would ever listen.
+fn run_client(image: &str, args: &[&str], mounts: &[(&str, &str)]) ->
ClientRun {
Review Comment:
docker run no timeout, no --pull never. Pre-pull fail then ~400MB inside
nextest 60s×5. **Fix**: wrap timeout + --pull never.
##########
gateways/kafka/tests/kafka_client_e2e_tests.rs:
##########
@@ -0,0 +1,648 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! End-to-end tests driving **real Kafka clients** against the gateway.
+//!
+//! Every other suite in this crate hand-builds wire frames, which cannot
catch a client
+//! compatibility problem by construction. One already slipped through that
way: advertising
+//! `SaslHandshake` from v1 rather than v0 passed every hand-built test and
made librdkafka report
+//! "SASL Handshake not supported by broker" before it sent anything. These
tests exist so that
+//! class of bug fails in CI instead of during manual testing.
+//!
+//! The stack is real on all three sides: a spawned `iggy-server` process, the
gateway in-process
+//! with a real `IggyAuthenticator`, and a client from a container. They
automate categories S and T
+//! of `docs/MANUAL_TESTING.md`.
+//!
+//! Prerequisites are Docker and an already-built `iggy-server`. Missing
either skips, the way the
+//! wire-fixture suites do, unless `KAFKA_E2E_REQUIRED=1` is set, which turns
a skip into a failure
+//! so a broken CI step cannot leave these silently green.
+
+use std::net::SocketAddr;
+use std::path::PathBuf;
+use std::process::{Child, Command, Stdio};
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+
+use iggy_gateway_kafka::GatewayConfig;
+use iggy_gateway_kafka::auth::IggyAuthenticator;
+
+#[path = "common/server.rs"]
+mod server;
+
+use server::spawn_test_server_with_authenticator;
+
+const KCAT_IMAGE: &str = "edenhill/kcat:1.7.1";
+const KAFKA_IMAGE: &str = "apache/kafka:3.9.0";
+const ROOT_USER: &str = "iggy";
+const ROOT_PASSWORD: &str = "iggy";
+/// Password for every non-root principal these tests create.
+const USER_PASSWORD: &str = "s3cretpass";
+
+/// Budget for `iggy-server` to start listening. Generous: it is a cold
process start, and a debug
+/// build on a loaded machine is not quick.
+const SERVER_READY_TIMEOUT: Duration = Duration::from_secs(45);
+
+/// Reports why the suite cannot run, and whether that is fatal.
+///
+/// Returns `true` when the caller should skip. `KAFKA_E2E_REQUIRED=1` makes
it panic instead, so a
+/// CI job that means to run these fails loudly rather than reporting a pass
over zero assertions.
+fn skip(reason: &str) -> bool {
+ assert!(
+ std::env::var("KAFKA_E2E_REQUIRED").as_deref() != Ok("1"),
+ "KAFKA_E2E_REQUIRED=1 but the suite cannot run: {reason}"
+ );
+ eprintln!("skipping real-client end-to-end test: {reason}");
+ true
+}
+
+fn docker_missing() -> bool {
+ let available = Command::new("docker")
+ .arg("info")
+ .stdout(Stdio::null())
+ .stderr(Stdio::null())
+ .status()
+ .is_ok_and(|status| status.success());
+ if available {
+ false
+ } else {
+ skip("docker is unavailable")
+ }
+}
+
+/// Locates the already-built `iggy-server` alongside this test binary. Does
not build it.
+fn iggy_server_binary() -> Option<PathBuf> {
+ let mut dir = std::env::current_exe().ok()?;
+ // .../target/<profile>/deps/<test binary> -> .../target/<profile>
+ dir.pop();
+ dir.pop();
+ let candidate = dir.join(format!("iggy-server{}",
std::env::consts::EXE_SUFFIX));
+ candidate.is_file().then_some(candidate)
+}
+
+/// Outcome of waiting for the spawned server to come up.
+enum Ready {
+ Listening,
+ /// The process exited before it bound, carrying its status.
+ Exited(String),
+ TimedOut,
+}
+
+/// A spawned `iggy-server`, killed on drop.
+///
+/// The environment recipe mirrors `bridge_iggy_integration_tests.rs`, which
explains each setting:
+/// the other listeners are off so concurrently spawned servers do not fight
over their fixed
+/// ports, and the shard pool is capped so a spawned server does not size
itself to the whole
+/// machine against unrelated packages' tests in the same run.
+struct TestServer {
+ child: Child,
+ address: String,
+ http_address: String,
+ _data_dir: tempfile::TempDir,
+}
+
+impl TestServer {
+ fn spawn() -> Result<Self, String> {
+ let Some(binary) = iggy_server_binary() else {
+ return Err(
+ "iggy-server is not built; run `cargo build --bin iggy-server`
first".to_string(),
+ );
+ };
+ let data_dir = tempfile::tempdir().expect("create server data dir");
+ let port = free_port();
+ let address = format!("127.0.0.1:{port}");
+ let http_address = format!("127.0.0.1:{}", free_port());
+
+ let child = Command::new(binary)
+ .arg("--fresh")
+ .env("IGGY_PATH", data_dir.path().display().to_string())
+ .env("IGGY_TCP_ADDRESS", &address)
+ // Left on, unlike the sibling suite: the three-principal
procedure creates its
+ // non-root users over the HTTP API, which is the only
administrative surface reachable
+ // from a test without pulling in the SDK.
+ .env("IGGY_HTTP_ENABLED", "true")
+ .env("IGGY_HTTP_ADDRESS", &http_address)
+ .env("IGGY_QUIC_ENABLED", "false")
+ .env("IGGY_WEBSOCKET_ENABLED", "false")
+ .env("IGGY_SHARDING_PIN_CORES", "false")
+ .env("IGGY_SHARDING_CPU_ALLOCATION", "0..4")
+ .env("IGGY_ROOT_USERNAME", ROOT_USER)
+ .env("IGGY_ROOT_PASSWORD", ROOT_PASSWORD)
+ .stdout(Stdio::null())
+ .stderr(Stdio::null())
+ .spawn()
+ .expect("spawn iggy-server");
+
+ let mut server = Self {
+ child,
+ address,
+ http_address,
+ _data_dir: data_dir,
+ };
+ match server.wait_ready() {
+ Ready::Listening => Ok(server),
+ Ready::Exited(status) => Err(format!(
+ "iggy-server exited during startup with {status}; its output
is suppressed, so \
+ rerun it by hand with the same environment to see why"
+ )),
+ Ready::TimedOut => Err(format!(
+ "iggy-server never bound {} within {SERVER_READY_TIMEOUT:?}",
+ server.address
+ )),
+ }
+ }
+
+ fn wait_ready(&mut self) -> Ready {
+ let deadline = Instant::now() + SERVER_READY_TIMEOUT;
+ while Instant::now() < deadline {
+ // Both listeners, not just the data one: the server binds HTTP
after TCP, so probing
+ // TCP alone declares readiness while the provisioning calls that
follow would still
+ // be refused.
+ if std::net::TcpStream::connect(&self.address).is_ok()
+ && std::net::TcpStream::connect(&self.http_address).is_ok()
+ {
+ return Ready::Listening;
+ }
+ // Without this a server that dies at boot burns the whole budget
and is then reported
+ // as a timeout, which sends the reader looking at the wrong thing
entirely.
+ if let Ok(Some(status)) = self.child.try_wait() {
+ return Ready::Exited(status.to_string());
+ }
+ std::thread::sleep(Duration::from_millis(200));
+ }
+ Ready::TimedOut
+ }
+}
+
+impl Drop for TestServer {
+ fn drop(&mut self) {
+ let _ = self.child.kill();
+ let _ = self.child.wait();
+ }
+}
+
+/// Binds an ephemeral port and releases it, so the server can take it.
+///
+/// A window exists between release and rebind. Acceptable here because this
suite is serialized
+/// into its own nextest group, so nothing else in it is drawing ports
concurrently.
+fn free_port() -> u16 {
+ let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind
ephemeral port");
+ listener.local_addr().expect("local addr").port()
+}
+
+/// Starts the gateway with SASL on, verifying credentials against
`iggy_address`.
+async fn spawn_gateway(iggy_address: &str) -> SocketAddr {
+ let config = GatewayConfig {
+ sasl_enabled: true,
+ ..GatewayConfig::default()
+ };
+ let authenticator =
Arc::new(IggyAuthenticator::new(iggy_address.to_string()));
+ let (addr, shutdown) = spawn_test_server_with_authenticator(config,
authenticator).await;
+ // Held for the test's lifetime: dropping the sender shuts the gateway
down mid-exchange.
+ std::mem::forget(shutdown);
+ addr
+}
+
+/// Runs a container against the host network and returns its combined output.
+///
+/// `--network host` is what lets a containerised client reach a gateway bound
to the host's
+/// loopback. It is Linux-specific, which matches where CI runs.
+///
+/// This blocks the calling thread for the life of the container, which is why
every test here uses
+/// a multi-threaded runtime: the gateway runs as a spawned task, and on the
single-threaded runtime
+/// `#[tokio::test]` gives by default, this call would starve it and nothing
would ever listen.
+fn run_client(image: &str, args: &[&str], mounts: &[(&str, &str)]) ->
ClientRun {
+ let mut command = Command::new("docker");
+ command.args(["run", "--rm", "--network", "host"]);
+ for (host, guest) in mounts {
+ command.args(["-v", &format!("{host}:{guest}:ro")]);
+ }
+ command.arg(image).args(args);
+ let output = command.output().expect("run client container");
+ ClientRun {
+ succeeded: output.status.success(),
+ text: format!(
+ "{}{}",
+ String::from_utf8_lossy(&output.stdout),
+ String::from_utf8_lossy(&output.stderr)
+ ),
+ }
+}
+
+/// What a client container produced, and whether it ran at all.
+///
+/// The exit status is carried deliberately. Every negative assertion in this
suite is of the form
+/// "the output does not contain X", and a container that never started
produces output satisfying
+/// all of them. Without this a wrong image tag, a malformed config or a
Docker failure reads as a
+/// passing test.
+struct ClientRun {
+ succeeded: bool,
+ text: String,
+}
+
+impl ClientRun {
+ /// Asserts the client itself ran, then hands back its output for content
assertions.
+ fn expect_ran(self, what: &str) -> String {
+ assert!(
+ self.succeeded,
+ "{what}: the client container failed to run, so any assertion
about its output would \
+ be meaningless. Output: {}",
+ self.text
+ );
+ self.text
+ }
+
+ /// For the cases where the client is *expected* to fail, so only the
output matters.
+ fn text(self) -> String {
+ self.text
+ }
+}
+
+fn kcat(addr: SocketAddr, username: &str, password: &str, mechanism: &str) ->
ClientRun {
+ run_client(
+ KCAT_IMAGE,
+ &[
+ "-b",
+ &addr.to_string(),
+ "-X",
+ "security.protocol=SASL_PLAINTEXT",
+ "-X",
+ &format!("sasl.mechanisms={mechanism}"),
+ "-X",
+ &format!("sasl.username={username}"),
+ "-X",
+ &format!("sasl.password={password}"),
+ "-L",
+ ],
+ &[],
+ )
+}
+
+/// Creates an Iggy user over the HTTP API with the given global permissions.
+///
+/// The spec's three-principal procedure needs non-root accounts, and root
holds every flag, so a
+/// suite that only ever authenticates as root checks neither the inheritance
projection nor the
+/// empty view against a real client.
+fn create_user(http: &str, token: &str, username: &str, permissions: &str) {
+ let body = format!(
+
r#"{{"username":"{username}","password":"{USER_PASSWORD}","status":"active","permissions":{permissions}}}"#
+ );
+ let status = Command::new("curl")
+ .args([
+ "-s",
+ "-o",
+ "/dev/null",
+ "-w",
+ "%{http_code}",
+ "-X",
+ "POST",
+ &format!("http://{http}/users";),
+ "-H",
+ &format!("Authorization: Bearer {token}"),
+ "-H",
+ "Content-Type: application/json",
+ "-d",
+ &body,
+ ])
+ .output()
+ .expect("create user");
+ let code = String::from_utf8_lossy(&status.stdout).to_string();
+ assert!(
+ code.starts_with('2'),
+ "creating {username} returned HTTP {code}"
+ );
+}
+
+/// Logs in as root over HTTP and returns the bearer token.
+fn root_token(http: &str) -> String {
+ let output = Command::new("curl")
+ .args([
+ "-s",
+ "-X",
+ "POST",
+ &format!("http://{http}/users/login";),
+ "-H",
+ "Content-Type: application/json",
+ "-d",
+
&format!(r#"{{"username":"{ROOT_USER}","password":"{ROOT_PASSWORD}"}}"#),
+ ])
+ .output()
+ .expect("root login");
+ let body = String::from_utf8_lossy(&output.stdout);
+ // Avoids a JSON dependency for one field: the token is the value after
this key.
+ let key = "\"token\":\"";
+ let start = body.find(key).expect("login response carries a token") +
key.len();
+ let end = start + body[start..].find('"').expect("token is terminated");
+ body[start..end].to_string()
+}
+
+/// Writes a JAAS client config for the Java tools and returns its path, kept
alive by the handle.
+fn java_client_config(username: &str, password: &str) -> (tempfile::TempDir,
PathBuf) {
+ let dir = tempfile::tempdir().expect("create config dir");
+ let path = dir.path().join("client.properties");
+ std::fs::write(
+ &path,
+ format!(
+ "security.protocol=SASL_PLAINTEXT\n\
+ sasl.mechanism=PLAIN\n\
+
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule
required \
+ username=\"{username}\" password=\"{password}\";\n"
+ ),
+ )
+ .expect("write client config");
+ (dir, path)
+}
+
+/// Brings up a server and a gateway, or reports why it could not.
+async fn stack() -> Option<(TestServer, SocketAddr)> {
+ if docker_missing() {
+ return None;
+ }
+ let server = match TestServer::spawn() {
+ Ok(server) => server,
+ Err(reason) => {
+ skip(&reason);
+ return None;
+ }
+ };
+ let gateway = spawn_gateway(&server.address).await;
+ Some((server, gateway))
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn
given_valid_credentials_when_a_real_client_connects_should_serve_metadata() {
+ let Some((_server, gateway)) = stack().await else {
+ return;
+ };
+ let output = kcat(gateway, ROOT_USER, ROOT_PASSWORD,
"PLAIN").expect_ran("valid credentials");
+ assert!(
+ output.contains("Metadata for all topics"),
+ "librdkafka must authenticate and receive metadata, got: {output}"
+ );
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn
given_a_wrong_password_when_a_real_client_connects_should_report_an_auth_failure()
{
+ let Some((_server, gateway)) = stack().await else {
+ return;
+ };
+ // kcat exits non-zero here by design, so only its output is asserted.
+ let output = kcat(gateway, ROOT_USER, "definitely-not-the-password",
"PLAIN").text();
+ assert!(
+ output.contains("Authentication failed"),
+ "a rejected credential must reach the client as an auth failure, got:
{output}"
+ );
+ assert!(
+ !output.contains("Metadata for all topics"),
+ "nothing may be served to a rejected client"
+ );
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn given_an_unsupported_mechanism_should_name_the_supported_one_back() {
+ let Some((_server, gateway)) = stack().await else {
+ return;
+ };
+ let output = kcat(gateway, ROOT_USER, ROOT_PASSWORD,
"SCRAM-SHA-256").text();
+ assert!(
+ output.contains("PLAIN"),
+ "the refusal must name what is supported or an operator cannot act on
it, got: {output}"
+ );
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn given_no_credentials_when_a_real_client_connects_should_be_refused() {
+ let Some((_server, gateway)) = stack().await else {
+ return;
+ };
+ // kcat exits non-zero here by design, so only its output is asserted, and
the assertion has
+ // to be a positive one: a container that never started also fails to
print the metadata this
+ // test forbids, which would pass over nothing. librdkafka only reports
this particular
+ // diagnosis after it connected and the broker then dropped it before
authenticating, so it
+ // stands in for the refusal itself.
+ let output = run_client(KCAT_IMAGE, &["-b", &gateway.to_string(), "-L"],
&[]).text();
+ assert!(
+ output.contains("broker might require SASL authentication"),
+ "an unauthenticated client must be disconnected by the broker, got:
{output}"
+ );
+ assert!(
+ !output.contains("Metadata for all topics"),
+ "an unauthenticated client must not be served, got: {output}"
+ );
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn given_the_java_client_should_authenticate_and_list_the_sasl_apis() {
+ // The Java client is the one that negotiates `SaslAuthenticate` v2 and
sends `ApiVersions`
+ // both before and after authenticating, none of which kcat exercises
identically.
+ let Some((_server, gateway)) = stack().await else {
+ return;
+ };
+ let (_dir, config) = java_client_config(ROOT_USER, ROOT_PASSWORD);
+ let output = run_client(
+ KAFKA_IMAGE,
+ &[
+ "/opt/kafka/bin/kafka-broker-api-versions.sh",
+ "--bootstrap-server",
+ &gateway.to_string(),
+ "--command-config",
+ "/tmp/client.properties",
+ ],
+ &[(
+ config.to_str().expect("config path is utf-8"),
+ "/tmp/client.properties",
+ )],
+ )
+ .expect_ran("java client api-versions");
+ assert!(
+ output.contains("SaslHandshake(17)"),
+ "the Java client must authenticate and read the advertisement, got:
{output}"
+ );
+ assert!(
+ output.contains("SaslAuthenticate(36)"),
+ "both SASL keys must be advertised while the feature is on, got:
{output}"
+ );
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn
given_distinct_principals_when_listing_acls_should_describe_each_differently() {
+ // The spec's category T procedure. Root alone proves nothing about the
mapping, because it
+ // holds every flag: a fixed set would satisfy it. Each of the others is
chosen to fail
+ // differently if the projection is wrong. The consume-only principal
shows the projection
+ // distinguishes principals and applies Iggy's inheritance; the poll-only
one is the single
+ // case that separates the group binding's real source from
`poll_messages`; the produce-only
+ // one shows a write grant drags in neither a read nor a group; and the
ungranted one shows an
+ // empty view is a successful answer rather than a failure.
+ let Some((server, gateway)) = stack().await else {
+ return;
+ };
+ let token = root_token(&server.http_address);
+ create_user(
+ &server.http_address,
+ &token,
+ "consumer-only",
+
r#"{"global":{"manage_servers":false,"read_servers":false,"manage_users":false,
+
"read_users":false,"manage_streams":false,"read_streams":false,"manage_topics":false,
+
"read_topics":true,"poll_messages":false,"send_messages":false},"streams":null}"#,
+ );
+ create_user(&server.http_address, &token, "no-grants", "null");
+ // Discriminates the group-derivation fix. `poll_messages` alone leaves
the projected topic
+ // read false, so Iggy would refuse this principal a consumer group, yet
the old derivation
+ // rendered one from polling. Reverting that fix makes this principal's
listing grow a GROUP
+ // section, which the assertion below catches.
+ create_user(
+ &server.http_address,
+ &token,
+ "poller-only",
+
r#"{"global":{"manage_servers":false,"read_servers":false,"manage_users":false,
+
"read_users":false,"manage_streams":false,"read_streams":false,"manage_topics":false,
+
"read_topics":false,"poll_messages":true,"send_messages":false},"streams":null}"#,
+ );
+ create_user(
+ &server.http_address,
+ &token,
+ "producer-only",
+
r#"{"global":{"manage_servers":false,"read_servers":false,"manage_users":false,
+
"read_users":false,"manage_streams":false,"read_streams":false,"manage_topics":false,
+
"read_topics":false,"poll_messages":false,"send_messages":true},"streams":null}"#,
+ );
+
+ let root = list_acls(gateway, ROOT_USER, ROOT_PASSWORD);
Review Comment:
Root pin TOPIC WRITE only. CLUSTER DESCRIBE present / ALTER absent
unenforced vs MANUAL T1. **Fix**: grants(root, CLUSTER, DESCRIBE) &&
!grants(..., ALTER).
--
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]
