You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by kr...@apache.org on 2026/09/28 10:27:03 UTC
(iggy) branch master updated: feat(gateways): describe Iggy permissions as Kafka ACLs (#4245)
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 8152a6074 feat(gateways): describe Iggy permissions as Kafka ACLs (#4245)
8152a6074 is described below
commit 8152a607410b686f37ca5224e18b4e2eabecbd60
Author: Grzegorz Koszyk <11...@users.noreply.github.com>
AuthorDate: Mon Sep 28 12:26:52 2026 +0200
feat(gateways): describe Iggy permissions as Kafka ACLs (#4245)
---
.config/nextest.toml | 12 +
.github/actions/rust/pre-merge/action.yml | 23 +
gateways/kafka/README.md | 24 +-
gateways/kafka/docs/ACL_MAPPING.md | 195 +++++++
gateways/kafka/docs/MANUAL_TESTING.md | 52 ++
gateways/kafka/docs/SCOPE.md | 7 +-
gateways/kafka/docs/TEST_SUITE.md | 29 +
gateways/kafka/src/auth.rs | 247 ++++++++-
gateways/kafka/src/protocol/acl.rs | 600 ++++++++++++++++++++
gateways/kafka/src/protocol/api.rs | 85 ++-
gateways/kafka/src/protocol/bounds_guard.rs | 77 +++
gateways/kafka/src/protocol/mod.rs | 1 +
gateways/kafka/src/server.rs | 117 +++-
gateways/kafka/tests/kafka_client_e2e_tests.rs | 725 +++++++++++++++++++++++++
gateways/kafka/tests/sasl_tests.rs | 624 ++++++++++++++++++++-
15 files changed, 2762 insertions(+), 56 deletions(-)
diff --git a/.config/nextest.toml b/.config/nextest.toml
index c8428a4e4..4cb32eb3c 100644
--- a/.config/nextest.toml
+++ b/.config/nextest.toml
@@ -65,6 +65,18 @@ filter = '''
'''
test-group = "kafka_bridge"
+# The real-client suite spawns its own server too, so it needs serializing for the same reason, but
+# in its own group rather than queued behind the bridge tests: sharing one group would put six
+# container-driven tests behind twenty-one unrelated ones, serially, for no isolation gained. Both
+# groups cap their servers' shard pools themselves, so running the two alongside each other is
+# bounded.
+[test-groups.kafka_client_e2e]
+max-threads = 1
+
+[[profile.default.overrides]]
+filter = 'binary_id(iggy-gateway-kafka::kafka_client_e2e_tests)'
+test-group = "kafka_client_e2e"
+
[profile.default]
slow-timeout = { period = "60s", terminate-after = 5 }
diff --git a/.github/actions/rust/pre-merge/action.yml b/.github/actions/rust/pre-merge/action.yml
index b9dee0eed..79cb1797a 100644
--- a/.github/actions/rust/pre-merge/action.yml
+++ b/.github/actions/rust/pre-merge/action.yml
@@ -396,6 +396,29 @@ runs:
# Fixtures are expected to exist from here on; a missing one now means
# generation silently failed, not a legitimate local-dev skip.
export KAFKA_FIXTURES_REQUIRED=1
+ # Same reasoning for the real-client suite: Docker and a built iggy-server are both
+ # present in this job, so a skip here means the prerequisite broke rather than a
+ # legitimate local-dev skip, and an unarmed guard would report a pass over zero
+ # assertions.
+ export KAFKA_E2E_REQUIRED=1
+ # Pull the client images here rather than letting the first test pull them. A cold
+ # runner fetches ~400 MB for the Kafka distribution, and inside a test that competes
+ # with nextest's per-test kill budget, so a slow registry surfaces as a test timeout
+ # naming the feature under test instead of naming the pull. The suite runs its
+ # containers with --pull never, so a failed pull here makes it skip, which the guard
+ # above turns into a job failure that names the missing image.
+ # Read the tags out of the suite itself; hardcoding them here lets a bump in the test
+ # desync silently, which pre-pulls the superseded image and puts the real one back
+ # inside the budget this block exists to avoid.
+ E2E_IMAGES=$(grep -oP '^const (KCAT|KAFKA)_IMAGE: &str = "\K[^"]+' \
+ gateways/kafka/tests/kafka_client_e2e_tests.rs)
+ if [[ -z "$E2E_IMAGES" ]]; then
+ echo "::warning::could not read client image tags from kafka_client_e2e_tests.rs; skipping pre-pull"
+ fi
+ for image in $E2E_IMAGES; do
+ timeout 300 docker pull -q "$image" \
+ || echo "::warning::docker pull $image failed; the real-client suite will fail naming it"
+ done
fi
# Start D-Bus and unlock keyring right before test execution to avoid
diff --git a/gateways/kafka/README.md b/gateways/kafka/README.md
index 3bd591eb7..6d7b632e2 100644
--- a/gateways/kafka/README.md
+++ b/gateways/kafka/README.md
@@ -86,6 +86,7 @@ See [docs/SCOPE.md](docs/SCOPE.md) for [#3421](https://github.com/apache/iggy/is
- [docs/IDEMPOTENCE.md](docs/IDEMPOTENCE.md) — InitProducerId, and why delivery is at-least-once
- [docs/OFFSET_STORAGE.md](docs/OFFSET_STORAGE.md) — where Kafka consumer group offsets live
- [docs/AUTHENTICATION.md](docs/AUTHENTICATION.md) — how a Kafka client authenticates, and why PLAIN only
+- [docs/ACL_MAPPING.md](docs/ACL_MAPPING.md) — how Iggy permissions are described as Kafka ACLs
### Delivery guarantees
@@ -145,12 +146,25 @@ Four things to know before switching it on:
would let a second connection present any password. Connection churn is therefore server load.
`IGGY_KAFKA_MAX_CONCURRENT_AUTHENTICATIONS` bounds the checks the gateway runs at once, and a peer
whose login was rejected is refused for a delay that doubles per rejection, from 0.5s up to 30s.
-- **Authentication only, for now.** The gateway verifies the credentials and then drops the
- session, because no handler consumes one yet. Iggy's permissions will decide what a principal can
- do once Produce and Fetch are wired to it
+- **Authentication and an ACL view, not enforcement.** The gateway verifies the credentials and
+ can describe what Iggy grants the principal, but nothing gates an operation yet. Iggy's
+ permissions decide that once Produce and Fetch are wired to it
([#3535](https://github.com/apache/iggy/issues/3535),
- [#3536](https://github.com/apache/iggy/issues/3536)); until then this is an admission gate, not
- an identity carried onto the data plane. Do not read it as per-topic authorization yet.
+ [#3536](https://github.com/apache/iggy/issues/3536)).
+
+### ACLs
+
+`DescribeAcls` renders the authenticated principal's Iggy permissions as Kafka ACL bindings, so
+`kafka-acls.sh --list` works against the gateway. It is read only: `CreateAcls` and `DeleteAcls`
+are not implemented and not advertised.
+
+`--list` here is a snapshot of the caller's own grants, not the broker-wide dump it is against a
+Kafka cluster. A principal sees its own permissions and nobody else's, because the gateway holds no
+administrative credentials, so filtering on another `User:` returns an empty listing whatever that
+user holds, and root's listing is root's grants rather than a catalog of every binding. Only global permissions are rendered, as wildcard bindings, and the view is a snapshot
+taken when the connection authenticated, so a permission changed afterwards is invisible until the
+client reconnects. [docs/ACL_MAPPING.md](docs/ACL_MAPPING.md) has the mapping table and what is
+deliberately left out.
Full reasoning, including what was rejected and why, is in
[docs/AUTHENTICATION.md](docs/AUTHENTICATION.md).
diff --git a/gateways/kafka/docs/ACL_MAPPING.md b/gateways/kafka/docs/ACL_MAPPING.md
new file mode 100644
index 000000000..463158f7e
--- /dev/null
+++ b/gateways/kafka/docs/ACL_MAPPING.md
@@ -0,0 +1,195 @@
+# Kafka ACLs and Iggy permissions
+
+Status: proposed. Answers the ACL half of [#3549](https://github.com/apache/iggy/issues/3549).
+Depends on SASL, because every answer here is about the principal that authenticated.
+
+## Decision
+
+**Read only.** `DescribeAcls` (29) is implemented. `CreateAcls` (30) and `DeleteAcls` (31) are not,
+and are not advertised, so they are refused the way every other unlisted key is.
+
+**A principal can see its own permissions and nothing else.** The gateway holds no administrative
+credentials, by design: a Kafka client's credentials are the only ones it ever has. Reading another
+user's record needs a permission the gateway cannot supply, so a filter naming a different
+principal returns an empty result rather than an error.
+
+**Permissions are read once, at authentication.** The login that verifies the credentials also
+fetches the principal's own record, and the connection keeps the result. No second session, no
+stored password, one extra round trip on a path that already does one.
+
+**Global permissions only, as wildcard bindings.** Iggy's per-stream and per-topic permissions are
+not rendered. See [What is not mapped](#what-is-not-mapped).
+
+## Why this is worth doing before the data plane exists
+
+Authorization cannot be *enforced* yet. Produce and Fetch are stubs
+([#3535](https://github.com/apache/iggy/issues/3535),
+[#3536](https://github.com/apache/iggy/issues/3536)), so there is no operation to gate, and the
+verified identity is currently discarded.
+
+`DescribeAcls` is the one authorization surface that needs neither. It is a read: take the
+authenticated principal, take the permissions Iggy already holds for it, and answer in Kafka's
+vocabulary. That makes it verifiable end to end today with a real admin client, which nothing else
+in the authorization story is. It also gives the discarded identity its first consumer, which is
+what turns carrying it onto the connection from speculative plumbing into something with a caller.
+
+## The mapping
+
+Iggy grants permissions to a user. Kafka describes them as bindings of
+`(resource, principal, host, operation, permission type)`. Three of those five are constant here.
+
+| Kafka field | Value | Why |
+| ------------- | ------- | ----- |
+| principal | `User:<username>` | Kafka's own convention for a SASL principal |
+| host | `*` | Iggy has no host-scoped permissions |
+| permission type | `ALLOW` | Iggy has no deny rules, only grants |
+
+The rest come from Iggy's global permissions, **after applying Iggy's own inheritance**. Its
+enforcement is hierarchical, not flag for flag: polling is granted by any of the four read or manage
+flags on topics or streams, appending by either manage flag, and reading server state by either
+server flag. A literal copy of the flags therefore describes a principal that cannot do things Iggy
+will in fact let it do, which is the same falsehood the derived group binding below exists to avoid.
+The rules live in `core/metadata/src/permissioner/permissioner_rules/`.
+
+| Iggy global permission | Kafka resource | Operation |
+| ------------------------ | ---------------- | ----------- |
+| `read_servers` or `manage_servers` | `CLUSTER` `kafka-cluster` | `DESCRIBE` |
+| any read or manage flag on topics or streams | `TOPIC` `*` | `DESCRIBE` |
+| `manage_topics` or `manage_streams` | `TOPIC` `*` | `CREATE`, `DELETE`, `ALTER` |
+| `poll_messages`, or any read or manage flag | `TOPIC` `*` | `READ` |
+| `send_messages`, `manage_topics` or `manage_streams` | `TOPIC` `*` | `WRITE` |
+| same as the topic `DESCRIBE` row | `GROUP` `*` | `READ` |
+
+Three notes on that table.
+
+**Stream permissions fold into topic ones.** Iggy puts topics inside streams, and Kafka has no
+resource above a topic. The topic mapping places every Kafka topic inside one Iggy stream, so a
+stream-level grant is in practice a grant over the topics a Kafka client can reach. Rendering it as
+a topic binding says the true thing in Kafka's vocabulary; inventing a resource type for it would
+not.
+
+**A consumer group grant is derived, not stored.** Iggy has no group-level permission. A Kafka
+consumer needs `READ` on its group as well as on the topic, so a principal Iggy would admit to a
+group is shown as having it. It is derived from the topic read grant, not from `poll_messages`,
+because group *membership* operations route through Iggy's own `get_topic` rule, which admits on the
+read and manage flags and never consults polling. Deriving it from polling granted a group to
+principals Iggy denies.
+
+Offset commit and fetch are the exception: Iggy routes those through `poll_messages`, while Kafka
+gates them on this same `GROUP READ`. A principal holding polling but no read grant is therefore
+shown no group binding even though Iggy would let it commit an offset. That under-reports, which is
+the safe direction, where deriving from polling would over-report membership.
+
+**`manage_servers` renders nothing of its own.** Iggy reads it in exactly one rule, as an alias for
+`read_servers`, so it gates no mutation anywhere. Rendering `CLUSTER ALTER` from it would advertise
+an ability with nowhere to be used, and the write ACL APIs that operation authorizes are not even
+advertised. This is the same argument that keeps `manage_users` out of the table.
+
+**`manage_users` and `read_users` are not rendered.** Kafka's `USER` resource covers credential and
+delegation-token administration, which this gateway does not expose at all. Mapping onto it would
+claim an ability that has nowhere to be used.
+
+Wildcards use `LITERAL` with the name `*`, which is how Kafka itself represents "every resource of
+this type". `PREFIXED` is never emitted, because nothing in Iggy's model is prefix-scoped.
+
+## Filters
+
+`DescribeAcls` carries a filter. Each field is matched, with Kafka's `ANY` sentinel matching
+everything:
+
+- **resource type** filters the table above. `ANY` returns all of it.
+- **resource name** matches the rendered name, so `*` matches the wildcard bindings.
+- **principal** matches the authenticated principal. Naming anyone else returns empty, per the
+ decision above.
+- **host** matches `*`.
+- **pattern type** selects nothing when it is `PREFIXED` or unknown, because nothing rendered is
+ prefix-scoped.
+- **operation** and **permission type** filter the rendered rows. Only `ANY` is a wildcard here:
+ Kafka's `ALL` is a concrete operation, so a filter naming it selects nothing, because nothing
+ rendered is an `ALL` binding.
+
+`MATCH` widens the name comparison rather than narrowing it. A named filter under `MATCH` also
+selects a literal pattern named `*`, which is how `kafka-acls.sh --topic orders
+--resource-pattern-type match` asks what affects one topic. Since every binding here is a wildcard,
+exact comparison alone would answer nothing to the one query that should find them all.
+
+An empty result is `error_code` 0 with no resources, not an error. Kafka draws a firm line between
+"no bindings match" and "the request failed", and an admin tool prints them very differently.
+
+## What is not mapped
+
+**Per-stream and per-topic permissions.** Iggy keys them by numeric slab id, not by name, so
+rendering a Kafka topic name means resolving every id through a lookup, one call per resource, on
+what should be a cheap read. Worse, the resolution can be ambiguous in the direction we need it:
+the topic mapping is one-way, from Kafka name to Iggy stream and topic, and an override means an
+Iggy topic name does not identify the Kafka name it came from. The same limitation is already
+recorded against `OffsetFetch` with a null topic list. Closing it needs a reverse index, which
+belongs with whichever change first needs one.
+
+The consequence is honest but worth stating: a principal whose access is granted per topic rather
+than globally is described as having no topic bindings. That under-reports rather than
+over-reports, which is the safe direction for an authorization view.
+
+**Writes.** `CreateAcls` and `DeleteAcls` would mean updating another user's permissions, which
+needs `manage_users` on the *Kafka client's own* Iggy user. Coherent, but it turns a read-only
+surface into one that mutates accounts, and it cannot be verified the way a read can. Left out
+deliberately rather than half-built.
+
+**Enforcement.** Nothing here gates an operation. This describes what Iggy would allow; Iggy is
+still the thing that decides, once there is an operation for it to decide about.
+
+## One divergence from a real broker
+
+A real broker gates `DescribeAcls` on `CLUSTER DESCRIBE`. This gateway answers any authenticated
+principal, including one the same response describes as lacking that permission.
+
+That is deliberate rather than an oversight. The only thing a principal can learn here is its own
+access, which it can discover anyway by attempting an operation, and enforcing the gate would mainly
+hide a principal's own permissions from itself. Recorded so the difference is a decision rather than
+a surprise.
+
+## Staleness
+
+The permissions a connection reports are the ones its principal had when it authenticated. A
+permission changed afterwards is not visible until the client reconnects.
+
+This is deliberate. The alternative is either holding an Iggy session open per connection, which
+the authentication design rejects on cost, or keeping the password to log in again, which it
+rejects outright. It also matches what Iggy already does on its own data plane, where a revocation
+becomes visible only once the owning shard applies it.
+
+Say so in the README rather than leaving an operator to discover that an ACL view can lag.
+
+## Testing
+
+The state machine and the mapping are pure, so they unit test directly: a permission set in, a set
+of bindings out, including the empty and wildcard cases.
+
+End to end, `kafka-acls.sh --list` from the Kafka distribution image, against a gateway with SASL
+enabled and a real Iggy server behind it. This is the part that matters, because it is the first
+authorization behaviour that can be checked against a real client rather than a stub. It runs five
+principals, chosen so that each one fails differently if the mapping is wrong:
+
+| Principal | Holds | What it pins down |
+| ----------- | ------- | ------------------- |
+| root | everything | every row of the table renders |
+| `consumer-only` | `read_topics` | the derived group binding appears |
+| `poller-only` | `poll_messages` | and *only* here: a topic read with no group binding. Deriving the group from polling instead makes this the one listing that grows a GROUP section |
+| `producer-only` | `send_messages` | a write with no read and no group |
+| `no-grants` | nothing | an empty list is a success, not an error |
+
+Each assertion is paired: the binding that must be present and the one that must be absent. A
+principal asserted only on absence would also be satisfied by a client container that never
+started, which is how an earlier revision of this suite passed over nothing.
+
+`kafka_client_e2e_tests.rs` skips itself when Docker is missing, so a local run without it is green
+rather than broken. CI sets `KAFKA_E2E_REQUIRED=1`, which turns that skip into a failure, because
+there the missing prerequisite is a broken job rather than a developer without Docker.
+
+## References
+
+- Authentication: [`AUTHENTICATION.md`](AUTHENTICATION.md)
+- Record mapping: [`BRIDGE_MAPPING.md`](BRIDGE_MAPPING.md)
+- Scope and phases: [`SCOPE.md`](SCOPE.md)
+- Iggy permissions: `core/common/src/types/permissions/`
+- Self-read exemption: `core/server/src/dispatch/authz.rs`
diff --git a/gateways/kafka/docs/MANUAL_TESTING.md b/gateways/kafka/docs/MANUAL_TESTING.md
index 152485676..94cb2326d 100644
--- a/gateways/kafka/docs/MANUAL_TESTING.md
+++ b/gateways/kafka/docs/MANUAL_TESTING.md
@@ -113,6 +113,58 @@ docker run --rm --network host -v /tmp/client.properties:/tmp/client.properties:
--bootstrap-server 127.0.0.1:9095 --command-config /tmp/client.properties
```
+### Category T — ACL view
+
+Needs the same running stack as category S, plus two extra Iggy users so the view has something to
+distinguish. Create them over the HTTP API, logging in as root first:
+
+```bash
+TOKEN=$(curl -s -X POST http://127.0.0.1:3000/users/login \
+ -H 'Content-Type: application/json' \
+ -d '{"username":"iggy","password":"iggy"}' | jq -r .access_token.token)
+
+curl -s -X POST http://127.0.0.1:3000/users -H "Authorization: Bearer $TOKEN" \
+ -H 'Content-Type: application/json' -d '{"username":"consumer-only","password":"s3cretpass",
+ "status":"active","permissions":{"global":{"read_topics":true,"poll_messages":true,
+ "manage_servers":false,"read_servers":false,"manage_users":false,"read_users":false,
+ "manage_streams":false,"read_streams":false,"manage_topics":false,"send_messages":false},
+ "streams":null}}'
+
+curl -s -X POST http://127.0.0.1:3000/users -H "Authorization: Bearer $TOKEN" \
+ -H 'Content-Type: application/json' \
+ -d '{"username":"no-grants","password":"s3cretpass","status":"active","permissions":null}'
+```
+
+Then list each principal's ACLs with the real admin client, swapping the username and password in
+`client.properties`:
+
+```bash
+docker run --rm --network host -v /tmp/client.properties:/tmp/client.properties:ro \
+ apache/kafka:3.9.0 /opt/kafka/bin/kafka-acls.sh \
+ --bootstrap-server 127.0.0.1:9095 --command-config /tmp/client.properties --list
+```
+
+| ID | Principal | Pass criteria | Last run |
+| ---- | ----------- | --------------- | ---------- |
+| T1 | `iggy` (root) | `CLUSTER kafka-cluster` with DESCRIBE and no other cluster operation; `TOPIC *` with all six operations; `GROUP *` with READ | Pass |
+| T2 | `consumer-only` | `TOPIC *` with DESCRIBE and READ only, `GROUP *` with READ, **no** WRITE and **no** CLUSTER | Pass |
+| T3 | `no-grants` | No output at all, and no error | Pass |
+
+T2 is the one that matters. It proves the mapping distinguishes principals rather than echoing a
+fixed set, and that the derived group binding appears for a principal Iggy would admit to a group.
+T3 proves an empty view is a successful answer rather than a failure, which is a distinction an
+admin tool prints very differently.
+
+T1 asserts the absence of `CLUSTER ALTER` rather than its presence. Iggy reads `manage_servers` in
+one rule, as an alias for `read_servers`, so rendering an alter grant from it would advertise an
+ability with nowhere to be used; [`ACL_MAPPING.md`](ACL_MAPPING.md) carries the argument.
+
+This procedure is now automated, in `gateways/kafka/tests/kafka_client_e2e_tests.rs`, driving the
+same `kafka-acls.sh` image against the same stack. The automated version runs two principals this
+table does not: one holding only `poll_messages` and one holding only `send_messages`, which are
+what separate the group binding's real source from polling. Run the manual procedure when changing
+the mapping by hand; otherwise the test is the faster check and the one CI enforces.
+
#### Login cost
Every authenticated connection costs one Iggy login. Measured against a debug build of both
diff --git a/gateways/kafka/docs/SCOPE.md b/gateways/kafka/docs/SCOPE.md
index 7a3a4ae26..a9773f6b3 100644
--- a/gateways/kafka/docs/SCOPE.md
+++ b/gateways/kafka/docs/SCOPE.md
@@ -77,7 +77,9 @@ All API keys not listed above close the connection (see Governance model above)
| 10 | FindCoordinator | Consumer group — later issue |
| 11–16 | JoinGroup, Heartbeat, LeaveGroup, SyncGroup, DescribeGroups, ListGroups | Consumer group — later issue |
| 17 | SaslHandshake | Implemented behind `IGGY_KAFKA_SASL_ENABLED`, advertised only while it is on ([`AUTHENTICATION.md`](AUTHENTICATION.md)) |
-| 20+ | DeleteTopics, InitProducerId, transactions, ACLs, etc. | Later issues |
+| 29 | DescribeAcls | Implemented behind `IGGY_KAFKA_SASL_ENABLED`, advertised only while it is on ([`ACL_MAPPING.md`](ACL_MAPPING.md)) |
+| 36 | SaslAuthenticate | Implemented behind `IGGY_KAFKA_SASL_ENABLED`, advertised only while it is on ([`AUTHENTICATION.md`](AUTHENTICATION.md)) |
+| 20+ | DeleteTopics, InitProducerId, transactions, etc. | Later issues |
Full reference for future phases: [`kafka_api_keys_reference.md`](kafka_api_keys_reference.md).
@@ -283,6 +285,9 @@ Authentication design ([#3549](https://github.com/apache/iggy/issues/3549)):
connection rather than closing it as an unlisted key would. Enabling it later therefore cannot
silently widen what an unauthenticated client may send. SCRAM is ruled out by Iggy's
credential storage, not deferred
+- [x] `DescribeAcls` (29), rendering the authenticated principal's Iggy permissions as Kafka ACL
+ bindings ([`ACL_MAPPING.md`](ACL_MAPPING.md)). Read only: `CreateAcls` (30) and `DeleteAcls`
+ (31) are not implemented and not advertised
- [ ] TLS on the gateway listener, a prerequisite for using PLAIN outside a trusted network
- [ ] Tune `max_frame_size` per workload (Kafka defaults: ~1 MiB produce, ~50 MiB fetch; current default 8 MiB)
- [ ] Target **~15–20 API keys** total for a functional bridge — not all 74+ admin keys
diff --git a/gateways/kafka/docs/TEST_SUITE.md b/gateways/kafka/docs/TEST_SUITE.md
index 101932120..da897e3d0 100644
--- a/gateways/kafka/docs/TEST_SUITE.md
+++ b/gateways/kafka/docs/TEST_SUITE.md
@@ -64,6 +64,7 @@ file under `tests/` anymore.
| [`server_e2e_tests.rs`](../tests/server_e2e_tests.rs) | Full `KafkaGateway` TCP round-trips | Partial |
| [`listener_robustness_tests.rs`](../tests/listener_robustness_tests.rs) | TCP listener robustness — framing, pipelining, concurrency, connection limits | No |
| [`sasl_tests.rs`](../tests/sasl_tests.rs) | SASL/PLAIN over a socket — full handshake, every refusal path, and the disabled default. Drives a stub verifier implementing `SaslAuthenticator`, so no Iggy server is needed | No |
+| [`kafka_client_e2e_tests.rs`](../tests/kafka_client_e2e_tests.rs) | **Real Kafka clients** against the whole stack: a spawned `iggy-server`, the gateway in-process with a real authenticator, and kcat / the Java tools from containers. The only suite that can catch a client-compatibility bug, since every other one hand-builds frames | No, but needs Docker and a built `iggy-server` |
| [`bridge_iggy_integration_tests.rs`](../tests/bridge_iggy_integration_tests.rs) | `IggyBridge` against a real, spawned `iggy-server` — provisioning idempotency, high watermark, credential/connection edge cases | No (needs the `iggy-server` binary - see Prerequisites) |
| [`produce_real_bridge_tests.rs`](../tests/produce_real_bridge_tests.rs) | Produce (key 0) through the whole handler against a real, spawned `iggy-server` — records go in as Kafka wire bytes and come back through the Iggy SDK, plus one error code per partition | No (needs the `iggy-server` binary - see Prerequisites) |
@@ -74,6 +75,34 @@ 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.
+
+Client containers run with `--pull never`, so pull the two images first. Without them the suite
+skips and names the missing one.
+
+```bash
+cargo build --bin iggy-server
+docker pull edenhill/kcat:1.7.1
+docker pull apache/kafka:3.9.0
+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.
+
+The suite runs in its own `kafka_client_e2e` nextest group, capped at one thread, so its spawned
+servers are serialized against each other. It is kept apart from the `kafka_bridge` group so the
+container-driven tests do not queue behind the bridge tests, and both groups cap their servers'
+shard pools, so the two can run alongside each other.
+
+It automates categories S and T of [`MANUAL_TESTING.md`](MANUAL_TESTING.md). Those procedures stay,
+because they cover cases a test does not assert, but the load-bearing ones now run in CI.
+
## Adding new tests
1. **New API key or version range** — update `SUPPORTED_RANGES` in `api.rs`, `SCOPE.md`, and add
diff --git a/gateways/kafka/src/auth.rs b/gateways/kafka/src/auth.rs
index 5a45c2995..704c1218a 100644
--- a/gateways/kafka/src/auth.rs
+++ b/gateways/kafka/src/auth.rs
@@ -23,23 +23,28 @@
use std::collections::HashMap;
use std::net::{IpAddr, Ipv6Addr};
-use std::sync::{Mutex, PoisonError};
+use std::sync::{Arc, Mutex, PoisonError};
use std::time::{Duration, Instant};
use async_trait::async_trait;
-use iggy::prelude::{AutoLogin, Client, Credentials, IggyClientBuilder, IggyError};
+use iggy::prelude::{
+ AutoLogin, Client, Credentials, Identifier, IggyError, Permissions, TcpClient,
+ TcpClientConfigBuilder,
+};
+use tokio::sync::SemaphorePermit;
use tracing::{debug, warn};
use crate::bridge::config::DEFAULT_IGGY_ADDR;
use crate::env::parse_bool;
+use crate::protocol::acl::PrincipalPermissions;
use crate::protocol::sasl::PlainCredentials;
/// Bound on one credential verification.
///
-/// Covers the dial and the login. Teardown has its own, much smaller budget
-/// ([`TEARDOWN_TIMEOUT`]) so that one attempt cannot hold an authentication permit for twice this
-/// long. The caller bounds the whole thing again from outside, against its pre-authentication
-/// budget, because a permit wait is not covered here at all.
+/// Covers the dial and the login only. The teardown has its own, much smaller budget
+/// ([`TEARDOWN_TIMEOUT`]). The permission read has none of its own: it is the only request on its
+/// connection, and the caller's pre-authentication budget bounds it from outside along with
+/// everything else, including a permit wait that is not covered here at all.
///
/// A verification that has not answered inside this is indistinguishable, from the Kafka client's
/// side, from one that failed, and the client is holding a connection open waiting for it. Shorter
@@ -62,10 +67,10 @@ const VERIFY_RECONNECTION_RETRIES: u32 = 0;
/// Budget for tearing the verification client down again.
///
-/// Deliberately far shorter than [`VERIFY_TIMEOUT`]. Teardown happens while the caller still holds
-/// an authentication permit, so giving it the full verify budget would let one attempt occupy a
-/// slot for twice as long as the doc on [`VERIFY_TIMEOUT`] claims the whole operation can take.
-/// Nothing is lost by cutting it short: `Drop` aborts the heartbeat task regardless.
+/// Deliberately far shorter than [`VERIFY_TIMEOUT`]. The authentication slot is already released by
+/// the time teardown runs, but every second here still comes out of the caller's own
+/// pre-authentication budget. Nothing is lost by cutting it short: dropping the client closes the
+/// socket regardless.
const TEARDOWN_TIMEOUT: Duration = Duration::from_secs(1);
/// Delay imposed on a peer after its first rejected login, doubled on every further rejection.
@@ -188,19 +193,51 @@ pub enum AuthError {
Unavailable,
}
+/// Who authenticated, and what Iggy already allows them to do.
+///
+/// The permissions are a snapshot taken during the login that verified the credentials, not a live
+/// view. Refreshing them would mean either holding an Iggy session open per connection, which the
+/// authentication design rejects on cost, or keeping the password, which it rejects outright. A
+/// permission changed mid-connection is therefore invisible until the client reconnects, the same
+/// way Iggy's own data plane lags a revocation until the owning shard applies it.
+#[derive(Debug, Clone)]
+pub struct AuthenticatedPrincipal {
+ pub username: String,
+ pub permissions: PrincipalPermissions,
+ /// Whether [`Self::permissions`] is a real answer or a fallback.
+ ///
+ /// A read that failed after a successful login degrades to an empty set, which is
+ /// indistinguishable on the wire from a principal that genuinely holds nothing. Silently
+ /// reporting "no access" for "we could not tell" is the wrong answer to give an operator
+ /// debugging access, so the two are kept apart here and answered differently.
+ ///
+ /// This matters more once Produce and Fetch consume the snapshot: nothing may authorize off a
+ /// value that was never read.
+ pub permissions_known: bool,
+}
+
/// Verifies Kafka-supplied credentials.
///
/// A trait rather than a concrete type so the protocol tests can drive the whole SASL exchange
/// over a socket without an Iggy server behind it.
#[async_trait]
pub trait SaslAuthenticator: Send + Sync + std::fmt::Debug {
- /// Returns `Ok(())` when `credentials` name a real, active Iggy user.
+ /// Returns the principal when `credentials` name a real, active Iggy user.
+ ///
+ /// `slot` is the caller's authentication slot, which bounds how many logins run against Iggy at
+ /// once. An implementation should release it as soon as the credentials are checked: whatever
+ /// follows the login costs Iggy no password verify, and holding the slot across it makes other
+ /// connections queue behind work the slot does not exist to bound.
///
/// # Errors
///
/// Returns [`AuthError::Rejected`] when Iggy refuses the credentials and
/// [`AuthError::Unavailable`] when it cannot be asked.
- async fn authenticate(&self, credentials: &PlainCredentials) -> Result<(), AuthError>;
+ async fn authenticate(
+ &self,
+ credentials: &PlainCredentials,
+ slot: SemaphorePermit<'_>,
+ ) -> Result<AuthenticatedPrincipal, AuthError>;
}
/// How the verifier reaches Iggy.
@@ -324,13 +361,19 @@ impl std::fmt::Display for IggyAuthenticator {
#[async_trait]
impl SaslAuthenticator for IggyAuthenticator {
- async fn authenticate(&self, credentials: &PlainCredentials) -> Result<(), AuthError> {
+ async fn authenticate(
+ &self,
+ credentials: &PlainCredentials,
+ slot: SemaphorePermit<'_>,
+ ) -> Result<AuthenticatedPrincipal, AuthError> {
let auto_login = AutoLogin::Enabled(Credentials::UsernamePassword(
credentials.username.clone(),
credentials.password.clone(),
));
- let mut builder = IggyClientBuilder::new()
- .with_tcp()
+ // A bare transport client, not an `IggyClient`. The wrapper starts a heartbeat the moment it
+ // connects, and on a lockstep connection that ping queues the permission read behind it.
+ // Nothing here needs a heartbeat anyway: the client lives for one login and one read.
+ let mut builder = TcpClientConfigBuilder::new()
.with_server_address(self.address.clone())
.with_auto_sign_in(auto_login)
.with_reconnection_max_retries(Some(VERIFY_RECONNECTION_RETRIES));
@@ -343,23 +386,33 @@ impl SaslAuthenticator for IggyAuthenticator {
builder = builder.with_tls_ca_file(ca_file.clone());
}
}
- let client = builder.build().map_err(|error| classify(&error))?;
+ let client = builder
+ .build()
+ .and_then(|config| TcpClient::create(Arc::new(config)))
+ .map_err(|error| classify(&error))?;
let connected = tokio::time::timeout(VERIFY_TIMEOUT, client.connect()).await;
+ // The password verify the slot bounds is done. Holding it across the permission read and
+ // teardown would keep a slot busy for two more round trips per login.
+ drop(slot);
let outcome = match connected {
Err(_elapsed) => Err(AuthError::Unavailable),
Ok(Err(error)) => Err(classify(&error)),
- Ok(Ok(())) => Ok(()),
+ // The login already proved the credentials. Reading the principal's own record on the
+ // same session is the one extra round trip that lets `DescribeAcls` answer later
+ // without a second login or a stored password. A user may always read itself, with no
+ // permission required (`dispatch/authz.rs` exempts a self-targeted read), so this
+ // cannot fail for want of a grant.
+ Ok(Ok(())) => fetch_permissions(&client, &credentials.username).await,
};
- // Shut down on both paths, and not `disconnect`: only `shutdown` stops the heartbeat task,
- // which would otherwise keep pinging, observe the dropped transport, and reconnect using
- // the very credentials this call was only meant to check.
+ // Shut down rather than drop, so the socket closes cleanly and Iggy releases the session
+ // now instead of noticing the dead peer later.
//
// Both failure shapes are reported, and separately: the outer `Elapsed` and the inner
// `IggyError` mean different things, and folding them together would log "timed out" for a
- // shutdown that failed immediately. Neither leaks the client, since `Drop` aborts the
- // heartbeat regardless, but an invisible failure here would hide a real one.
+ // shutdown that failed immediately. Neither leaks the client, since dropping it closes the
+ // socket regardless, but an invisible failure here would hide a real one.
match tokio::time::timeout(TEARDOWN_TIMEOUT, client.shutdown()).await {
Ok(Ok(())) => {}
Ok(Err(error)) => {
@@ -370,7 +423,101 @@ impl SaslAuthenticator for IggyAuthenticator {
}
}
- outcome
+ outcome.map(|(permissions, read)| AuthenticatedPrincipal {
+ username: credentials.username.clone(),
+ permissions,
+ permissions_known: read == PermissionRead::Resolved,
+ })
+ }
+}
+
+/// How the permission read after a login ended.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum PermissionRead {
+ /// The record resolved, so the permissions are a real answer, possibly an empty one.
+ Resolved,
+ /// The read answered but gave no usable record.
+ Failed,
+}
+
+/// 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, PermissionRead), 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(), PermissionRead::Failed));
+ };
+ // 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.
+ //
+ // Not bounded here either. The SDK runs the exchange on a task of its own that holds the
+ // connection until the server answers or its own response deadline passes, so cancelling this
+ // future would stop only the waiting, not the read, and leave a live session behind the slot
+ // this login already gave back. The caller's pre-authentication budget still bounds the wait.
+ let fetched = client.get_user(&identifier).await;
+ let mut read = PermissionRead::Resolved;
+ let user = match fetched {
+ Ok(None) => {
+ read = PermissionRead::Failed;
+ // Distinct from an error: the login succeeded, so the account exists. A record that
+ // resolves to nothing here means the read raced a deletion, and silently reporting an
+ // empty ACL view for it would look identical to a principal with no grants.
+ warn!("authenticated, but the principal's own record resolved to nothing");
+ None
+ }
+ Ok(user) => user,
+ Err(error) => {
+ read = PermissionRead::Failed;
+ warn!(%error, "authenticated, but could not read the principal's permissions");
+ None
+ }
+ };
+
+ // A record that resolved but carries no permissions is a real answer: the principal holds
+ // nothing. Only a read that did not resolve is unknown.
+ Ok((
+ user.and_then(|user| user.permissions)
+ .as_ref()
+ .map(PrincipalPermissions::from)
+ .unwrap_or_default(),
+ read,
+ ))
+}
+
+/// Projects Iggy's global permissions onto the subset that has a Kafka meaning.
+///
+/// Stream flags fold into the topic ones because Kafka has no resource above a topic and every
+/// Kafka topic lives inside an Iggy stream, so a stream grant is in practice a grant over the
+/// topics a Kafka client can reach. `docs/ACL_MAPPING.md` has the full table.
+impl From<&Permissions> for PrincipalPermissions {
+ fn from(permissions: &Permissions) -> Self {
+ let global = &permissions.global;
+ // Iggy's enforcement is hierarchical, so a flag-for-flag copy describes a principal that
+ // cannot do things Iggy will in fact let it do. The rules are in
+ // `core/metadata/src/permissioner/permissioner_rules/`: polling is granted by any of the
+ // four read/manage flags on topics or streams, appending by either manage flag, and
+ // reading server state by either server flag. Mirroring that here is the difference
+ // between describing a consumer and describing a principal that appears unable to consume.
+ let manages = global.manage_topics || global.manage_streams;
+ let reads = global.read_topics || global.read_streams;
+ Self {
+ read_servers: global.read_servers || global.manage_servers,
+ read_topics: reads || manages,
+ manage_topics: manages,
+ poll_messages: global.poll_messages || reads || manages,
+ send_messages: global.send_messages || manages,
+ }
}
}
@@ -400,6 +547,7 @@ fn classify(error: &IggyError) -> AuthError {
mod tests {
use secrecy::SecretString;
use serial_test::serial;
+ use tokio::sync::Semaphore;
use super::*;
@@ -422,6 +570,51 @@ mod tests {
}
}
+ /// Iggy grants polling to anyone holding any of the four read/manage flags on topics or
+ /// streams, not only to the explicit `poll_messages` flag. Copying flags one for one described
+ /// such a principal as unable to consume, which is the exact falsehood the derived group
+ /// binding exists to avoid.
+ #[test]
+ fn given_only_read_topics_when_projected_should_still_be_able_to_poll() {
+ let mut permissions = Permissions::default();
+ permissions.global.read_topics = true;
+ let projected = PrincipalPermissions::from(&permissions);
+ assert!(
+ projected.poll_messages,
+ "read_topics grants polling in Iggy"
+ );
+ assert!(projected.read_topics);
+ assert!(!projected.send_messages, "it does not grant appending");
+ }
+
+ #[test]
+ fn given_only_manage_streams_when_projected_should_grant_both_directions() {
+ // Managing streams grants appending and polling, and implies the topic-level reads.
+ let mut permissions = Permissions::default();
+ permissions.global.manage_streams = true;
+ let projected = PrincipalPermissions::from(&permissions);
+ assert!(projected.send_messages);
+ assert!(projected.poll_messages);
+ assert!(projected.manage_topics);
+ }
+
+ #[test]
+ fn given_only_manage_servers_when_projected_should_also_allow_describing_the_cluster() {
+ let mut permissions = Permissions::default();
+ permissions.global.manage_servers = true;
+ let projected = PrincipalPermissions::from(&permissions);
+ assert!(
+ projected.read_servers,
+ "managing implies reading, as the permissioner's own rule does"
+ );
+ }
+
+ #[test]
+ fn given_no_grants_when_projected_should_stay_empty() {
+ let projected = PrincipalPermissions::from(&Permissions::default());
+ assert_eq!(projected, PrincipalPermissions::default());
+ }
+
#[test]
fn given_a_credential_rejection_when_classified_should_not_look_like_an_outage() {
assert!(matches!(
@@ -597,7 +790,13 @@ mod tests {
// Port 1 on loopback refuses immediately, so this exercises the failure path without
// waiting out VERIFY_TIMEOUT.
let authenticator = IggyAuthenticator::new("127.0.0.1:1".to_string());
- let result = authenticator.authenticate(&credentials()).await;
+ let slots = Semaphore::new(1);
+ let result = authenticator
+ .authenticate(
+ &credentials(),
+ slots.acquire().await.expect("an open semaphore"),
+ )
+ .await;
assert!(
matches!(result, Err(AuthError::Unavailable)),
"an unreachable server must never be reported as a credential rejection: {result:?}"
diff --git a/gateways/kafka/src/protocol/acl.rs b/gateways/kafka/src/protocol/acl.rs
new file mode 100644
index 000000000..a421ed202
--- /dev/null
+++ b/gateways/kafka/src/protocol/acl.rs
@@ -0,0 +1,600 @@
+// 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.
+
+use bytes::Bytes;
+use kafka_protocol::messages::DescribeAclsResponse;
+use kafka_protocol::messages::describe_acls_response::{AclDescription, DescribeAclsResource};
+use kafka_protocol::protocol::StrBytes;
+
+use crate::error::Result;
+use crate::protocol::api::ERROR_NONE;
+use crate::protocol::handlers::encode_message;
+
+/// 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.
+///
+/// A description, never an authorization input. It holds only global flags, so a principal whose
+/// grants are per-stream or per-topic reads here as holding nothing, and an authorizer built on it
+/// would deny what Iggy allows. Enforcement belongs to Iggy's own `Permissions`, evaluated by the
+/// server.
+///
+/// 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}")) {
+ return false;
+ }
+ if !matches_name(self.host.as_deref(), ANY_HOST) {
+ return false;
+ }
+ // Only `ANY` is a wildcard. `AccessControlEntryFilter.matches` compares everything else by
+ // equality, so `ALL` selects bindings whose operation is literally `ALL`. Nothing here
+ // renders one, so a filter asking for it is correctly empty. Treating it as a wildcard
+ // would return every binding, which over-reports a principal's access.
+ if self.operation != operation::ANY && self.operation != binding.operation {
+ return false;
+ }
+ if self.permission_type != permission_type::ANY
+ && self.permission_type != permission_type::ALLOW
+ {
+ return false;
+ }
+ true
+ }
+}
+
+/// An absent filter field means `ANY`, so it matches. Present fields compare exactly.
+fn matches_name(filter: Option<&str>, value: &str) -> bool {
+ filter.is_none_or(|wanted| wanted == value)
+}
+
+impl AclFilter {
+ /// Name matching, which `MATCH` widens.
+ ///
+ /// `ResourcePatternFilter.matches` gives `MATCH` a second branch: a named filter also selects a
+ /// literal pattern named `*`. That is how `kafka-acls.sh --topic foo --resource-pattern-type
+ /// match` asks "what affects topic foo", and since every binding here is a wildcard, exact
+ /// comparison alone would answer nothing to the one query that should find them all.
+ fn matches_resource_name(&self, name: &str) -> bool {
+ let Some(wanted) = self.resource_name.as_deref() else {
+ return true;
+ };
+ if wanted == name {
+ return true;
+ }
+ self.pattern_type == pattern_type::MATCH && name == WILDCARD
+ }
+}
+
+/// `DescribeAcls` response, grouping the selected bindings by resource.
+///
+/// Kafka nests ACLs under the resource they apply to, so bindings that share a resource type and
+/// name become one entry with several operations. An empty result is `error_code` 0 with no
+/// resources, never an error: a real broker distinguishes "nothing matched" from "the request
+/// failed", and an admin tool prints them very differently.
+///
+/// # Errors
+///
+/// Returns an error when `kafka_protocol` cannot encode the response at `version`.
+pub fn encode_describe_acls_response(
+ version: i16,
+ principal: &str,
+ bindings: &[AclBinding],
+) -> Result<Bytes> {
+ let mut grouped: Vec<DescribeAclsResource> = Vec::new();
+ for binding in bindings {
+ let description = AclDescription::default()
+ .with_principal(StrBytes::from_string(format!("User:{principal}")))
+ .with_host(StrBytes::from_static_str(ANY_HOST))
+ .with_operation(binding.operation)
+ .with_permission_type(permission_type::ALLOW);
+
+ if let Some(resource) = grouped.iter_mut().find(|resource| {
+ resource.resource_type == binding.resource_type
+ && resource.resource_name.as_str() == binding.resource_name
+ }) {
+ resource.acls.push(description);
+ } else {
+ grouped.push(
+ DescribeAclsResource::default()
+ .with_resource_type(binding.resource_type)
+ .with_resource_name(StrBytes::from_static_str(binding.resource_name))
+ .with_pattern_type(pattern_type::LITERAL)
+ .with_acls(vec![description]),
+ );
+ }
+ }
+
+ // `error_message` is explicitly null, not left at the type's default: that default is
+ // `Some("")`, and Java substitutes its own text for an error only when the field is null, so a
+ // defaulted empty string reaches an operator as a failure with no stated cause.
+ let resp = DescribeAclsResponse::default()
+ .with_error_code(ERROR_NONE)
+ .with_error_message(None)
+ .with_resources(grouped);
+ encode_message(&resp, version, 256)
+}
+
+/// Well-formed `DescribeAcls` response carrying only an error.
+///
+/// # Errors
+///
+/// Returns an error when `kafka_protocol` cannot encode the response at `version`.
+pub fn encode_describe_acls_error_response(version: i16, error_code: i16) -> Result<Bytes> {
+ // Null rather than the type's default `Some("")`, so the client substitutes the standard text
+ // for the code instead of printing an empty reason.
+ let resp = DescribeAclsResponse::default()
+ .with_error_code(error_code)
+ .with_error_message(None)
+ .with_resources(Vec::new());
+ encode_message(&resp, version, 64)
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ fn all_permissions() -> PrincipalPermissions {
+ PrincipalPermissions {
+ read_servers: true,
+ read_topics: true,
+ manage_topics: true,
+ poll_messages: true,
+ send_messages: true,
+ }
+ }
+
+ fn any_filter() -> AclFilter {
+ AclFilter {
+ resource_type: resource_type::ANY,
+ resource_name: None,
+ pattern_type: pattern_type::ANY,
+ principal: None,
+ host: None,
+ operation: operation::ANY,
+ permission_type: permission_type::ANY,
+ }
+ }
+
+ #[test]
+ fn given_no_permissions_when_rendered_should_produce_no_bindings() {
+ assert!(bindings_for(&PrincipalPermissions::default()).is_empty());
+ }
+
+ #[test]
+ fn given_only_poll_when_rendered_should_grant_the_topic_read_but_no_group() {
+ // Consumer-group operations route through Iggy's topic rule, which never consults
+ // `poll_messages`. Deriving the group binding from polling advertised access Iggy denies.
+ let permissions = PrincipalPermissions {
+ poll_messages: true,
+ ..PrincipalPermissions::default()
+ };
+ let bindings = bindings_for(&permissions);
+ assert_eq!(
+ bindings,
+ vec![AclBinding::new(
+ resource_type::TOPIC,
+ WILDCARD,
+ operation::READ
+ )]
+ );
+ }
+
+ #[test]
+ fn given_a_topic_read_grant_should_render_the_derived_group_binding() {
+ // The grant Iggy actually admits consumer-group operations on.
+ let permissions = PrincipalPermissions {
+ read_topics: true,
+ ..PrincipalPermissions::default()
+ };
+ let bindings = bindings_for(&permissions);
+ assert!(bindings.contains(&AclBinding::new(
+ resource_type::GROUP,
+ WILDCARD,
+ operation::READ
+ )));
+ }
+
+ #[test]
+ fn given_only_send_when_rendered_should_grant_write_and_no_group() {
+ let permissions = PrincipalPermissions {
+ send_messages: true,
+ ..PrincipalPermissions::default()
+ };
+ let bindings = bindings_for(&permissions);
+ assert_eq!(
+ bindings,
+ vec![AclBinding::new(
+ resource_type::TOPIC,
+ WILDCARD,
+ operation::WRITE
+ )],
+ "a producer has no group to read"
+ );
+ }
+
+ #[test]
+ fn given_manage_topics_when_rendered_should_grant_the_three_admin_operations() {
+ let permissions = PrincipalPermissions {
+ manage_topics: true,
+ ..PrincipalPermissions::default()
+ };
+ let operations: Vec<i8> = bindings_for(&permissions)
+ .iter()
+ .map(|binding| binding.operation)
+ .collect();
+ assert_eq!(
+ operations,
+ vec![operation::CREATE, operation::DELETE, operation::ALTER]
+ );
+ }
+
+ #[test]
+ fn given_server_permissions_when_rendered_should_scope_them_to_the_cluster() {
+ let permissions = PrincipalPermissions {
+ read_servers: true,
+ ..PrincipalPermissions::default()
+ };
+ let bindings = bindings_for(&permissions);
+ assert!(
+ bindings
+ .iter()
+ .all(|binding| binding.resource_type == resource_type::CLUSTER
+ && binding.resource_name == CLUSTER_NAME)
+ );
+ }
+
+ #[test]
+ fn given_an_any_filter_when_matching_should_select_every_binding() {
+ let filter = any_filter();
+ let bindings = bindings_for(&all_permissions());
+ assert!(
+ bindings
+ .iter()
+ .all(|binding| filter.matches(binding, "alice"))
+ );
+ }
+
+ #[test]
+ fn given_a_resource_type_filter_when_matching_should_select_only_that_type() {
+ let filter = AclFilter {
+ resource_type: resource_type::GROUP,
+ ..any_filter()
+ };
+ let selected: Vec<AclBinding> = bindings_for(&all_permissions())
+ .into_iter()
+ .filter(|binding| filter.matches(binding, "alice"))
+ .collect();
+ assert_eq!(
+ selected,
+ vec![AclBinding::new(
+ resource_type::GROUP,
+ WILDCARD,
+ operation::READ
+ )]
+ );
+ }
+
+ #[test]
+ fn given_a_filter_naming_another_principal_should_select_nothing() {
+ // The gateway can only ever read the caller's own record, so anything else is empty.
+ let filter = AclFilter {
+ principal: Some("User:someone-else".to_string()),
+ ..any_filter()
+ };
+ let bindings = bindings_for(&all_permissions());
+ assert!(
+ !bindings
+ .iter()
+ .any(|binding| filter.matches(binding, "alice"))
+ );
+ }
+
+ #[test]
+ fn given_a_filter_naming_the_caller_should_select_their_bindings() {
+ let filter = AclFilter {
+ principal: Some("User:alice".to_string()),
+ ..any_filter()
+ };
+ let bindings = bindings_for(&all_permissions());
+ assert!(
+ bindings
+ .iter()
+ .all(|binding| filter.matches(binding, "alice"))
+ );
+ }
+
+ #[test]
+ fn given_a_deny_filter_when_matching_should_select_nothing() {
+ // Iggy has no deny rules, so a filter asking for them is correctly empty rather than an
+ // error.
+ let filter = AclFilter {
+ permission_type: 2,
+ ..any_filter()
+ };
+ let bindings = bindings_for(&all_permissions());
+ assert!(!bindings.iter().any(|b| filter.matches(b, "alice")));
+ }
+
+ #[test]
+ fn given_a_prefixed_pattern_filter_should_select_nothing() {
+ // Nothing here is prefix-scoped, so this is empty rather than a wrong match.
+ let filter = AclFilter {
+ pattern_type: 4,
+ ..any_filter()
+ };
+ let bindings = bindings_for(&all_permissions());
+ assert!(!bindings.iter().any(|b| filter.matches(b, "alice")));
+ }
+
+ #[test]
+ fn given_an_operation_filter_of_all_should_select_nothing() {
+ // `ALL` is a concrete operation in Kafka, not a filter wildcard: only `ANY` is one, and
+ // `AccessControlEntryFilter.matches` compares everything else by equality. Nothing here
+ // renders an `ALL` binding, so this is correctly empty. Treating it as a wildcard returned
+ // every binding, which over-reports a principal's access.
+ let filter = AclFilter {
+ operation: operation::ALL,
+ ..any_filter()
+ };
+ let bindings = bindings_for(&all_permissions());
+ assert!(!bindings.iter().any(|b| filter.matches(b, "alice")));
+ }
+
+ #[test]
+ fn given_a_match_pattern_filter_with_a_name_should_select_the_wildcard_bindings() {
+ // `kafka-acls.sh --topic orders --resource-pattern-type match` asks "what affects orders".
+ // Every binding here is a wildcard, so exact comparison alone answered nothing to the one
+ // query that should find them all.
+ let filter = AclFilter {
+ resource_type: resource_type::TOPIC,
+ resource_name: Some("orders".to_string()),
+ pattern_type: pattern_type::MATCH,
+ ..any_filter()
+ };
+ let selected: Vec<AclBinding> = bindings_for(&all_permissions())
+ .into_iter()
+ .filter(|binding| filter.matches(binding, "alice"))
+ .collect();
+ assert!(
+ !selected.is_empty(),
+ "a MATCH filter must find the wildcard bindings that cover the named topic"
+ );
+ assert!(
+ selected
+ .iter()
+ .all(|binding| binding.resource_type == resource_type::TOPIC)
+ );
+ }
+
+ #[test]
+ fn given_a_literal_pattern_filter_with_a_name_should_not_select_the_wildcard() {
+ // The widening belongs to MATCH alone. A LITERAL filter means the name exactly.
+ let filter = AclFilter {
+ resource_name: Some("orders".to_string()),
+ pattern_type: pattern_type::LITERAL,
+ ..any_filter()
+ };
+ let bindings = bindings_for(&all_permissions());
+ assert!(!bindings.iter().any(|b| filter.matches(b, "alice")));
+ }
+
+ #[test]
+ fn given_a_resource_name_filter_should_distinguish_wildcard_from_cluster() {
+ let filter = AclFilter {
+ resource_name: Some(CLUSTER_NAME.to_string()),
+ ..any_filter()
+ };
+ let selected: Vec<AclBinding> = bindings_for(&all_permissions())
+ .into_iter()
+ .filter(|binding| filter.matches(binding, "alice"))
+ .collect();
+ assert!(
+ selected
+ .iter()
+ .all(|binding| binding.resource_type == resource_type::CLUSTER)
+ );
+ assert!(!selected.is_empty());
+ }
+}
diff --git a/gateways/kafka/src/protocol/api.rs b/gateways/kafka/src/protocol/api.rs
index b92274f92..0d9217e81 100644
--- a/gateways/kafka/src/protocol/api.rs
+++ b/gateways/kafka/src/protocol/api.rs
@@ -19,16 +19,23 @@ use std::sync::Arc;
use bytes::Bytes;
use kafka_protocol::error::ResponseError;
-use kafka_protocol::messages::{SaslAuthenticateRequest, SaslHandshakeRequest};
+use kafka_protocol::messages::{
+ DescribeAclsRequest, SaslAuthenticateRequest, SaslHandshakeRequest,
+};
use tokio::sync::Semaphore;
use crate::bridge::IggyBridge;
use crate::error::Result;
+use crate::protocol::acl::{
+ self, AclBinding, AclFilter, PrincipalPermissions, encode_describe_acls_error_response,
+ encode_describe_acls_response,
+};
use crate::protocol::bounds_guard::{
- validate_sasl_authenticate_shape, validate_sasl_handshake_shape,
+ validate_describe_acls_shape, validate_sasl_authenticate_shape, validate_sasl_handshake_shape,
};
use crate::protocol::handlers::{
api_versions, create_topics, decode_guarded, dispatch, fetch, list_offsets, metadata, produce,
+ respond_or_close,
};
use crate::protocol::sasl::{
SaslMechanism, encode_sasl_authenticate_response, encode_sasl_handshake_response,
@@ -41,6 +48,7 @@ pub const API_KEY_METADATA: i16 = 3;
pub const API_KEY_SASL_HANDSHAKE: i16 = 17;
pub const API_KEY_API_VERSIONS: i16 = 18;
pub const API_KEY_CREATE_TOPICS: i16 = 19;
+pub const API_KEY_DESCRIBE_ACLS: i16 = 29;
pub const API_KEY_SASL_AUTHENTICATE: i16 = 36;
pub const DEFAULT_KAFKA_PORT: u16 = 9093;
@@ -355,6 +363,14 @@ static SASL_ADVERTISED_RANGES: &[ApiVersionRange] = &[
min_version: 0,
max_version: 2,
},
+ // Grouped with the SASL keys rather than with `SUPPORTED_RANGES` because it answers about the
+ // authenticated principal. With SASL off there is no principal, so there is nothing it could
+ // truthfully describe, and advertising it would invite a question with no answer.
+ ApiVersionRange {
+ api_key: API_KEY_DESCRIBE_ACLS,
+ min_version: 1,
+ max_version: 3,
+ },
];
/// Sent with every `SASL_AUTHENTICATION_FAILED`, whatever the real cause.
@@ -365,6 +381,71 @@ static SASL_ADVERTISED_RANGES: &[ApiVersionRange] = &[
/// gateway's own log is where the difference is recorded.
pub const SASL_AUTH_FAILED_MESSAGE: &str = "Authentication failed";
+/// Decodes a `DescribeAcls` filter.
+///
+/// # Errors
+///
+/// Returns an error when the body is not a well-formed `DescribeAcls` request at `api_version`.
+pub fn decode_acl_filter(api_version: i16, body: Bytes) -> Result<AclFilter> {
+ let req = decode_guarded::<DescribeAclsRequest>(api_version, body, |v, b| {
+ validate_describe_acls_shape(v, b)
+ })?;
+ Ok(AclFilter {
+ resource_type: req.resource_type_filter,
+ resource_name: req.resource_name_filter.map(|name| name.to_string()),
+ pattern_type: req.pattern_type_filter,
+ principal: req.principal_filter.map(|name| name.to_string()),
+ host: req.host_filter.map(|name| name.to_string()),
+ operation: req.operation,
+ permission_type: req.permission_type,
+ })
+}
+
+/// Answers `DescribeAcls` for `principal`, selecting from what Iggy already grants them.
+#[must_use]
+pub fn describe_acls_outcome(
+ api_version: i16,
+ principal: &str,
+ permissions: &PrincipalPermissions,
+ filter: &AclFilter,
+) -> HandleOutcome {
+ let selected: Vec<AclBinding> = acl::bindings_for(permissions)
+ .into_iter()
+ .filter(|binding| filter.matches(binding, principal))
+ .collect();
+ respond_or_close(
+ encode_describe_acls_response(api_version, principal, &selected),
+ "DescribeAcls",
+ )
+}
+
+/// Versions of `DescribeAcls` this gateway answers, and the range `ApiVersions` advertises.
+///
+/// Public so the connection loop can enforce it. The key is deliberately absent from this module's
+/// `SUPPORTED_RANGES` firewall table, which is what makes this the only bound there is.
+pub const SASL_ADVERTISED_DESCRIBE_ACLS_VERSIONS: std::ops::RangeInclusive<i16> = 1..=3;
+
+/// `DescribeAcls` answer carrying only an error code.
+///
+/// `close` marks the refusals that must end the connection. A caller that keeps it open is saying
+/// the client may usefully send something else on it, which is true of a malformed filter and not
+/// of a state violation.
+#[must_use]
+pub fn respond_describe_acls_error(
+ api_version: i16,
+ error_code: i16,
+ close: bool,
+) -> HandleOutcome {
+ match encode_describe_acls_error_response(api_version, error_code) {
+ Ok(body) if close => HandleOutcome::RespondThenClose(body),
+ Ok(body) => HandleOutcome::Respond(body),
+ Err(error) => {
+ tracing::warn!(%error, "failed to encode DescribeAcls error; closing connection");
+ HandleOutcome::Close
+ }
+ }
+}
+
/// Reads the mechanism name out of a `SaslHandshake` body without consuming the caller's copy.
///
/// # Errors
diff --git a/gateways/kafka/src/protocol/bounds_guard.rs b/gateways/kafka/src/protocol/bounds_guard.rs
index d0c2cd4e5..cd80d5321 100644
--- a/gateways/kafka/src/protocol/bounds_guard.rs
+++ b/gateways/kafka/src/protocol/bounds_guard.rs
@@ -743,6 +743,39 @@ pub fn validate_api_versions_shape(version: i16, body: &Bytes) -> Result<()> {
/// reaches `parse_plain`, and what a future mechanism would carry into a credential exchange.
const MAX_SASL_AUTH_BYTES: usize = 4096;
+/// `DescribeAcls` carries a fixed-shape filter: four enums and three nullable strings.
+///
+/// No arrays and nothing echoed into the response, so there is no amplification to project. The
+/// walk exists to reject a truncated filter before `kafka_protocol` reads past the frame.
+///
+/// # Errors
+///
+/// Returns an error when a declared string length does not fit the remaining frame.
+pub fn validate_describe_acls_shape(version: i16, body: &Bytes) -> Result<()> {
+ let mut c = ShapeCursor::new(body.clone(), usize::MAX);
+ let flexible = version >= 2;
+ c.read_i8()?;
+ if flexible {
+ c.compact_string(true)?;
+ } else {
+ c.legacy_string(true)?;
+ }
+ c.read_i8()?;
+ if flexible {
+ c.compact_string(true)?;
+ c.compact_string(true)?;
+ } else {
+ c.legacy_string(true)?;
+ c.legacy_string(true)?;
+ }
+ c.read_i8()?;
+ c.read_i8()?;
+ if flexible {
+ c.tagged_fields()?;
+ }
+ Ok(())
+}
+
/// `SaslHandshake` carries one non-nullable string, the mechanism name.
///
/// `_version` is unused: the message is never flexible, so v0 and v1 share this shape, and the
@@ -791,6 +824,50 @@ mod tests {
const TEST_MAX_FRAME_SIZE: usize = 8 * 1024 * 1024;
+ #[test]
+ fn describe_acls_v1_with_all_strings_present_accepted() {
+ // Both wire fixtures null every string, so without this the walk's string readers ran in
+ // no test, which is exactly where a byte-count desync hides.
+ let body = Bytes::from_static(&[
+ 0x02, // resource_type_filter
+ 0x00, 0x01, b'x', // resource_name_filter
+ 0x03, // pattern_type_filter
+ 0x00, 0x04, b'U', b's', b'e', b'r', // principal_filter
+ 0x00, 0x01, b'*', // host_filter
+ 0x03, // operation
+ 0x03, // permission_type
+ ]);
+ assert!(validate_describe_acls_shape(1, &body).is_ok());
+ }
+
+ #[test]
+ fn describe_acls_v3_compact_strings_accepted() {
+ let body = Bytes::from_static(&[
+ 0x02, // resource_type_filter
+ 0x02, b'x', // compact name, len + 1
+ 0x03, // pattern_type_filter
+ 0x05, b'U', b's', b'e', b'r', // compact principal
+ 0x02, b'*', // compact host
+ 0x03, // operation
+ 0x03, // permission_type
+ 0x00, // tagged fields
+ ]);
+ assert!(validate_describe_acls_shape(3, &body).is_ok());
+ }
+
+ #[test]
+ fn describe_acls_declared_string_past_the_frame_rejected() {
+ let body = Bytes::from_static(&[0x02, 0x00, 0x40, b'x', b'y']);
+ assert!(validate_describe_acls_shape(1, &body).is_err());
+ }
+
+ #[test]
+ fn describe_acls_truncated_after_the_filter_rejected() {
+ // Every string null, then nothing where the two trailing enums belong.
+ let body = Bytes::from_static(&[0x02, 0xFF, 0xFF, 0x03, 0xFF, 0xFF, 0xFF, 0xFF]);
+ assert!(validate_describe_acls_shape(1, &body).is_err());
+ }
+
#[test]
fn sasl_handshake_well_formed_mechanism_accepted() {
let body = Bytes::from_static(&[0x00, 0x05, b'P', b'L', b'A', b'I', b'N']);
diff --git a/gateways/kafka/src/protocol/mod.rs b/gateways/kafka/src/protocol/mod.rs
index 127c761ff..34d8ed498 100644
--- a/gateways/kafka/src/protocol/mod.rs
+++ b/gateways/kafka/src/protocol/mod.rs
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+pub mod acl;
pub mod api;
pub mod bounds_guard;
pub mod handlers;
diff --git a/gateways/kafka/src/server.rs b/gateways/kafka/src/server.rs
index 8859f0f80..2e6bb6686 100644
--- a/gateways/kafka/src/server.rs
+++ b/gateways/kafka/src/server.rs
@@ -33,14 +33,16 @@ use tracing_appender::non_blocking::WorkerGuard;
use tracing_subscriber::EnvFilter;
use tracing_subscriber::filter::LevelFilter;
-use crate::auth::{AuthError, FailedLoginThrottle, SaslAuthenticator};
+use crate::auth::{AuthError, AuthenticatedPrincipal, FailedLoginThrottle, SaslAuthenticator};
use crate::bridge::IggyBridge;
use crate::error::{KafkaProtocolError, Result};
use crate::protocol::api::{
- API_KEY_SASL_AUTHENTICATE, API_KEY_SASL_HANDSHAKE, BrokerAdvertise, DEFAULT_KAFKA_PORT,
- ERROR_ILLEGAL_SASL_STATE, ERROR_NONE, ERROR_SASL_AUTHENTICATION_FAILED,
- ERROR_UNSUPPORTED_SASL_MECHANISM, ERROR_UNSUPPORTED_VERSION, GatewayState, HandleOutcome,
- decode_sasl_auth_bytes, decode_sasl_mechanism, encode_error_for_key, handle_request_bounded,
+ API_KEY_DESCRIBE_ACLS, API_KEY_SASL_AUTHENTICATE, API_KEY_SASL_HANDSHAKE, BrokerAdvertise,
+ DEFAULT_KAFKA_PORT, 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_request_bounded, respond_describe_acls_error,
sasl_authenticate_outcome, sasl_handshake_outcome,
};
use crate::protocol::header::{request_header_version, response_header_version};
@@ -489,6 +491,7 @@ struct ConnectionContext<'a> {
async fn route_frame(
ctx: &ConnectionContext<'_>,
sasl_state: &mut SaslState,
+ principal: &mut Option<AuthenticatedPrincipal>,
req: &RequestHeader,
body: Bytes,
) -> HandleOutcome {
@@ -509,6 +512,15 @@ async fn route_frame(
mechanism.as_deref(),
);
match action {
+ // Answered here rather than in the stateless dispatch, because it describes the principal
+ // this connection authenticated as, which only the connection knows. Gated on the feature
+ // as well as the key: with SASL off there is no principal, and the key is not advertised,
+ // so it falls through to ordinary dispatch and is refused as the unlisted key it is.
+ SaslAction::Dispatch
+ if ctx.config.sasl_enabled && req.request_api_key == API_KEY_DESCRIBE_ACLS =>
+ {
+ describe_acls(principal.as_ref(), req.request_api_version, body, peer)
+ }
SaslAction::Dispatch => {
handle_request_bounded(
ctx.state,
@@ -562,11 +574,12 @@ async fn route_frame(
HandleOutcome::Close
}
SaslAction::Authenticate => {
- let outcome = authenticate_token(ctx, req.request_api_version, body).await;
+ let (outcome, verified) = authenticate_token(ctx, req.request_api_version, body).await;
// Only a plain `Respond` is success: every refusal path answers with
// `RespondThenClose`, so the state advances on exactly the accepting branch.
if matches!(outcome, HandleOutcome::Respond(_)) {
*sasl_state = SaslState::Authenticated;
+ *principal = verified;
}
outcome
}
@@ -603,6 +616,7 @@ async fn handle_connection(
} else {
SaslState::Authenticated
};
+ let mut principal: Option<AuthenticatedPrincipal> = None;
let ctx = ConnectionContext {
config: &config,
state: &state,
@@ -654,7 +668,7 @@ async fn handle_connection(
// `RequestHeader::decode` advances `body` past the header fields it consumed via
// `Buf::advance`, so `body` is already exactly the request payload.
- let outcome = route_frame(&ctx, &mut sasl_state, &req, body).await;
+ let outcome = route_frame(&ctx, &mut sasl_state, &mut principal, &req, body).await;
if dispatch_outcome(
&mut stream,
&peer,
@@ -680,9 +694,14 @@ async fn authenticate_token(
ctx: &ConnectionContext<'_>,
api_version: i16,
body: Bytes,
-) -> HandleOutcome {
+) -> (HandleOutcome, Option<AuthenticatedPrincipal>) {
let peer = ctx.peer;
- let failed = || sasl_authenticate_outcome(api_version, ERROR_SASL_AUTHENTICATION_FAILED, true);
+ let failed = || {
+ (
+ sasl_authenticate_outcome(api_version, ERROR_SASL_AUTHENTICATION_FAILED, true),
+ None,
+ )
+ };
let Some(authenticator) = ctx.authenticator else {
// Unreachable: `run` refuses to start in this combination. Fail closed anyway, since the
@@ -704,7 +723,7 @@ async fn authenticate_token(
// fatal, and 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;
+ return (HandleOutcome::Close, None);
}
// The connection loop only watches the shutdown token between frames, so a verification in
@@ -712,7 +731,7 @@ async fn authenticate_token(
let verified = tokio::select! {
() = ctx.cancel.cancelled() => {
debug!(%peer, "SASL authentication abandoned by shutdown");
- return HandleOutcome::Close;
+ return (HandleOutcome::Close, None);
}
verified = verify_within_budget(ctx, authenticator, &credentials) => verified,
};
@@ -721,14 +740,17 @@ async fn authenticate_token(
// 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;
+ return (HandleOutcome::Close, None);
};
match result {
- Ok(()) => {
+ Ok(authenticated) => {
debug!(%peer, "SASL authentication succeeded");
ctx.failed_logins.record_success(peer.ip());
- sasl_authenticate_outcome(api_version, ERROR_NONE, false)
+ (
+ 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.
Err(AuthError::Rejected) => {
@@ -743,7 +765,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);
+ }
+ match decode_acl_filter(api_version, body) {
+ Ok(filter) => describe_acls_outcome(
+ api_version,
+ &principal.username,
+ &principal.permissions,
+ &filter,
+ ),
+ Err(error) => {
+ debug!(%peer, %error, "failed to decode DescribeAcls filter");
+ // Kept open: a client that sent one bad filter may send a good one.
+ respond_describe_acls_error(api_version, ERROR_INVALID_REQUEST, false)
}
}
}
@@ -760,7 +832,7 @@ async fn verify_within_budget(
ctx: &ConnectionContext<'_>,
authenticator: &dyn SaslAuthenticator,
credentials: &PlainCredentials,
-) -> Option<std::result::Result<(), AuthError>> {
+) -> Option<std::result::Result<AuthenticatedPrincipal, AuthError>> {
let peer = ctx.peer;
let budget = ctx.config.pre_auth_timeout;
let Ok(acquired) = timeout(budget, ctx.auth_slots.acquire()).await else {
@@ -768,11 +840,11 @@ async fn verify_within_budget(
return None;
};
// Acquire fails only once the semaphore is closed, which this gateway never does.
- let Ok(_slot) = acquired else {
+ let Ok(slot) = acquired else {
error!(%peer, "authentication slots unavailable");
return None;
};
- let Ok(result) = timeout(budget, authenticator.authenticate(credentials)).await else {
+ let Ok(result) = timeout(budget, authenticator.authenticate(credentials, slot)).await else {
warn!(%peer, "authentication did not complete within the pre-authentication budget");
return None;
};
@@ -801,7 +873,16 @@ fn illegal_state_outcome(
API_KEY_SASL_AUTHENTICATE if (0..=SASL_AUTHENTICATE_MAX_VERSION).contains(&api_version) => {
sasl_authenticate_outcome(api_version, ERROR_ILLEGAL_SASL_STATE, !keep_open)
}
+ // `encode_error_for_key` consults the firewall table, and this key is deliberately absent
+ // from it, so routing through there would answer an unauthenticated client with a bodyless
+ // close while the unreachable guard above is the one that speaks. Answer it directly.
+ API_KEY_DESCRIBE_ACLS
+ if sasl_enabled && SASL_ADVERTISED_DESCRIBE_ACLS_VERSIONS.contains(&api_version) =>
+ {
+ respond_describe_acls_error(api_version, ERROR_ILLEGAL_SASL_STATE, !keep_open)
+ }
API_KEY_SASL_HANDSHAKE | API_KEY_SASL_AUTHENTICATE => HandleOutcome::Close,
+ API_KEY_DESCRIBE_ACLS if sasl_enabled => HandleOutcome::Close,
_ => encode_error_for_key(api_key, api_version, ERROR_ILLEGAL_SASL_STATE, sasl_enabled),
}
}
diff --git a/gateways/kafka/tests/kafka_client_e2e_tests.rs b/gateways/kafka/tests/kafka_client_e2e_tests.rs
new file mode 100644
index 000000000..641c86706
--- /dev/null
+++ b/gateways/kafka/tests/kafka_client_e2e_tests.rs
@@ -0,0 +1,725 @@
+// 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);
+
+/// Wall-clock cap on one client container. The ACL test runs five in a row inside nextest's 300s
+/// kill budget, so a single wedged client must fail on its own terms, well before that budget
+/// kills the test and hides which step hung.
+const CLIENT_RUN_TIMEOUT: &str = "45s";
+
+/// Per-request and per-call budget for the Java admin tools. Their defaults (30s and 60s) let one
+/// unanswered request eat most of `CLIENT_RUN_TIMEOUT` retrying.
+const JAVA_CLIENT_TIMEOUT_MS: u32 = 10_000;
+
+/// 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")
+ }
+}
+
+/// Whether `image` is already in the local Docker store.
+///
+/// Containers run with `--pull never`, so a missing image would otherwise surface as a client that
+/// failed to start, reported against whichever feature that test happened to cover.
+/// Whether this host can run `run_client` at all.
+///
+/// `run_client` depends on two things outside Docker: `--network host` reaching the host's
+/// loopback, which only Linux does, and GNU `timeout` with `--kill-after`, which macOS lacks. A
+/// host without either would get past `stack` and then panic inside `expect_ran` or the spawn
+/// itself, reading as a gateway failure rather than a missing prerequisite.
+fn host_unsupported() -> bool {
+ if !cfg!(target_os = "linux") {
+ return skip("`docker run --network host` reaches the host loopback only on Linux");
+ }
+ let gnu_timeout = Command::new("timeout")
+ .args(["--kill-after=1s", "1s", "true"])
+ .stdout(Stdio::null())
+ .stderr(Stdio::null())
+ .status()
+ .is_ok_and(|status| status.success());
+ if gnu_timeout {
+ false
+ } else {
+ skip("GNU `timeout` with `--kill-after` is unavailable")
+ }
+}
+
+fn image_present(image: &str) -> bool {
+ Command::new("docker")
+ .args(["image", "inspect", image])
+ .stdout(Stdio::null())
+ .stderr(Stdio::null())
+ .status()
+ .is_ok_and(|status| status.success())
+}
+
+/// 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.
+///
+/// `--pull never` keeps a registry download out of the test's time budget: images are pulled up
+/// front, and `stack` skips when they are absent. `timeout` bounds the run itself, and a run it
+/// cuts short exits non-zero, so `expect_ran` reports it rather than asserting over partial output.
+fn run_client(image: &str, args: &[&str], mounts: &[(&str, &str)]) -> ClientRun {
+ let mut command = Command::new("timeout");
+ command.args(["--kill-after=10s", CLIENT_RUN_TIMEOUT]);
+ command.args([
+ "docker",
+ "run",
+ "--rm",
+ "--pull",
+ "never",
+ "--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\
+ request.timeout.ms={JAVA_CLIENT_TIMEOUT_MS}\n\
+ default.api.timeout.ms={JAVA_CLIENT_TIMEOUT_MS}\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 host_unsupported() || docker_missing() {
+ return None;
+ }
+ if let Some(image) = [KCAT_IMAGE, KAFKA_IMAGE]
+ .into_iter()
+ .find(|image| !image_present(image))
+ {
+ skip(&format!(
+ "client image {image} is not pulled; run `docker pull {image}`"
+ ));
+ 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);
+ assert!(
+ root.contains(&format!("principal=User:{ROOT_USER}")),
+ "every rendered binding names the authenticated principal, got: {root}"
+ );
+ assert!(
+ grants(&root, "TOPIC", "WRITE"),
+ "root holds every permission, so it must be described as able to write, got: {root}"
+ );
+ assert!(
+ grants(&root, "CLUSTER", "DESCRIBE"),
+ "root holds the server flags, so the cluster must render as describable, got: {root}"
+ );
+ assert!(
+ !grants(&root, "CLUSTER", "ALTER"),
+ "no Iggy flag gates a cluster mutation, so even root must not be shown one, got: {root}"
+ );
+
+ let consumer = list_acls(gateway, "consumer-only", USER_PASSWORD);
+ // Paired assertions: which operation sits under which resource is the whole claim.
+ assert!(
+ grants(&consumer, "TOPIC", "READ"),
+ "read_topics grants polling in Iggy, so a topic read must render, got: {consumer}"
+ );
+ assert!(
+ grants(&consumer, "TOPIC", "DESCRIBE"),
+ "a read grant must also describe, got: {consumer}"
+ );
+ assert!(
+ grants(&consumer, "GROUP", "READ"),
+ "a principal Iggy admits to consumer groups needs the derived group binding, got: \
+ {consumer}"
+ );
+ assert!(
+ !grants(&consumer, "TOPIC", "WRITE"),
+ "it may not write, and describing it as able to would over-report, got: {consumer}"
+ );
+ assert!(
+ !consumer.contains("resourceType=CLUSTER"),
+ "it holds no server permission, got: {consumer}"
+ );
+
+ let poller = list_acls(gateway, "poller-only", USER_PASSWORD);
+ assert!(
+ grants(&poller, "TOPIC", "READ"),
+ "poll_messages must still render a topic read, got: {poller}"
+ );
+ assert!(
+ !poller.contains("resourceType=GROUP"),
+ "polling alone grants no topic read, so Iggy would refuse this principal a consumer \
+ group; rendering one over-reports, got: {poller}"
+ );
+
+ let producer = list_acls(gateway, "producer-only", USER_PASSWORD);
+ assert!(
+ grants(&producer, "TOPIC", "WRITE"),
+ "send_messages must render a topic write, got: {producer}"
+ );
+ assert!(
+ !producer.contains("resourceType=GROUP"),
+ "a principal with no topic read grant must get no group binding, because Iggy would \
+ refuse it one, got: {producer}"
+ );
+
+ let none = list_acls(gateway, "no-grants", USER_PASSWORD);
+ assert!(
+ !none.contains("principal=User:"),
+ "a principal with no grants must be described with no bindings, got: {none}"
+ );
+ assert!(
+ !none.to_lowercase().contains("error"),
+ "an empty view is a successful answer, not a failure, got: {none}"
+ );
+}
+
+/// Splits `kafka-acls.sh --list` output into `(resource_type, operations)` per resource section.
+///
+/// Independent substring scans over the whole blob cannot tell which resource an operation belongs
+/// to, so `READ on TOPIC` and `DESCRIBE on GROUP` satisfy an assertion meant to prove the reverse.
+fn acl_sections(listing: &str) -> Vec<(String, Vec<String>)> {
+ let mut sections: Vec<(String, Vec<String>)> = Vec::new();
+ for line in listing.lines() {
+ if let Some(rest) = line.split_once("resourceType=") {
+ let resource = rest
+ .1
+ .split([',', ')'])
+ .next()
+ .unwrap_or_default()
+ .trim()
+ .to_string();
+ sections.push((resource, Vec::new()));
+ } else if let Some(rest) = line.split_once("operation=") {
+ let operation = rest
+ .1
+ .split([',', ')'])
+ .next()
+ .unwrap_or_default()
+ .trim()
+ .to_string();
+ if let Some(current) = sections.last_mut() {
+ current.1.push(operation);
+ }
+ }
+ }
+ sections
+}
+
+/// Whether `listing` grants `operation` on `resource`, with the pairing actually checked.
+fn grants(listing: &str, resource: &str, operation: &str) -> bool {
+ acl_sections(listing)
+ .iter()
+ .any(|(kind, operations)| kind == resource && operations.iter().any(|op| op == operation))
+}
+
+/// Lists a principal's ACLs with the real Kafka admin client.
+fn list_acls(gateway: SocketAddr, username: &str, password: &str) -> String {
+ let (_dir, config) = java_client_config(username, password);
+ run_client(
+ KAFKA_IMAGE,
+ &[
+ "/opt/kafka/bin/kafka-acls.sh",
+ "--bootstrap-server",
+ &gateway.to_string(),
+ "--command-config",
+ "/tmp/client.properties",
+ "--list",
+ ],
+ &[(
+ config.to_str().expect("config path is utf-8"),
+ "/tmp/client.properties",
+ )],
+ )
+ .expect_ran(&format!("listing ACLs as {username}"))
+}
diff --git a/gateways/kafka/tests/sasl_tests.rs b/gateways/kafka/tests/sasl_tests.rs
index 903ece36d..2f6d5e3f8 100644
--- a/gateways/kafka/tests/sasl_tests.rs
+++ b/gateways/kafka/tests/sasl_tests.rs
@@ -22,15 +22,18 @@
use std::net::SocketAddr;
use std::sync::Arc;
+use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;
use async_trait::async_trait;
use bytes::{BufMut, Bytes, BytesMut};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpStream;
+use tokio::sync::SemaphorePermit;
use iggy_gateway_kafka::GatewayConfig;
-use iggy_gateway_kafka::auth::{AuthError, SaslAuthenticator};
+use iggy_gateway_kafka::auth::{AuthError, AuthenticatedPrincipal, SaslAuthenticator};
+use iggy_gateway_kafka::protocol::acl::PrincipalPermissions;
use iggy_gateway_kafka::protocol::sasl::{MAX_PRE_AUTH_API_VERSIONS, PlainCredentials};
#[path = "common/codec.rs"]
@@ -54,6 +57,7 @@ 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;
const HANDSHAKE_VERSION: i16 = 1;
const AUTHENTICATE_VERSION: i16 = 1;
@@ -63,16 +67,25 @@ const AUTHENTICATE_VERSION: i16 = 1;
struct FixedCredentialAuthenticator {
username: &'static str,
password: &'static str,
+ permissions: PrincipalPermissions,
}
#[async_trait]
impl SaslAuthenticator for FixedCredentialAuthenticator {
- async fn authenticate(&self, credentials: &PlainCredentials) -> Result<(), AuthError> {
+ async fn authenticate(
+ &self,
+ credentials: &PlainCredentials,
+ _slot: SemaphorePermit<'_>,
+ ) -> Result<AuthenticatedPrincipal, AuthError> {
use secrecy::ExposeSecret;
let matches = credentials.username == self.username
&& credentials.password.expose_secret() == self.password;
if matches {
- Ok(())
+ Ok(AuthenticatedPrincipal {
+ username: credentials.username.clone(),
+ permissions: self.permissions,
+ permissions_known: true,
+ })
} else {
Err(AuthError::Rejected)
}
@@ -94,6 +107,7 @@ async fn spawn_sasl_gateway() -> SocketAddr {
let authenticator = Arc::new(FixedCredentialAuthenticator {
username: "alice",
password: "s3cret",
+ permissions: PrincipalPermissions::default(),
});
let (addr, shutdown) = spawn_test_server_with_authenticator(sasl_config(), authenticator).await;
// Held for the whole test: dropping the sender shuts the gateway down mid-exchange.
@@ -159,6 +173,102 @@ async fn send(
response
}
+/// `DescribeAcls` puts a `throttle_time_ms` (`i32`) *before* its error code, unlike the SASL
+/// responses. Reading offset 0 there returns the always-zero high half of the throttle field, so
+/// every assertion comparing it against 0 passes no matter what the gateway answered.
+fn acl_error_code(body: &Bytes) -> i16 {
+ assert!(body.len() >= 6, "DescribeAcls response is too short");
+ i16::from_be_bytes([body[4], body[5]])
+}
+
+/// Decodes a `DescribeAcls` v1 response into
+/// `(resource_type, resource_name, pattern_type, operation, permission_type)` tuples.
+///
+/// Pattern type and permission type are decoded rather than stepped over deliberately. They carry
+/// the security claim of the whole surface: a response that said DENY, or a prefixed pattern,
+/// would otherwise pass every test in this crate while meaning something entirely different.
+fn parse_acl_bindings(body: &Bytes) -> Vec<(i8, String, i8, i8, i8)> {
+ // Every read is bounds-checked by field name. A decoder that walks a response with raw
+ // indexing fails as an index-out-of-bounds naming no field, and the more dangerous case is
+ // quieter still: drift by a few bytes and it returns plausible-looking bindings decoded from
+ // the wrong offsets. That already happened once in this suite, where `error_code` read the
+ // first two bytes of a body whose first field is `throttle_time_ms`, so every assertion
+ // compared 0 against 0 and passed.
+ let need = |at: usize, count: usize, field: &str| {
+ assert!(
+ at + count <= body.len(),
+ "ran past the end of the response reading {field}: wanted {count} byte(s) at {at}, \
+ body holds {}. The decoder is out of step with the response shape.",
+ body.len()
+ );
+ };
+
+ let mut at = 6; // throttle_time_ms + error_code
+ // error_message: nullable string
+ need(at, 2, "error_message length");
+ let len = i16::from_be_bytes([body[at], body[at + 1]]);
+ at += 2;
+ if len >= 0 {
+ at += usize::try_from(len).expect("non-negative string length");
+ }
+ need(at, 4, "resource count");
+ let resource_count = i32::from_be_bytes([body[at], body[at + 1], body[at + 2], body[at + 3]]);
+ at += 4;
+
+ let mut out = Vec::new();
+ for _ in 0..resource_count.max(0) {
+ need(at, 3, "resource type and name length");
+ let resource_type = body[at].cast_signed();
+ at += 1;
+ let name_len = usize::try_from(i16::from_be_bytes([body[at], body[at + 1]]))
+ .expect("non-negative name length");
+ at += 2;
+ need(at, name_len, "resource name");
+ let name = String::from_utf8_lossy(&body[at..at + name_len]).to_string();
+ at += name_len;
+ need(at, 5, "pattern type and acl count");
+ let pattern_type = body[at].cast_signed();
+ at += 1;
+ let acl_count = i32::from_be_bytes([body[at], body[at + 1], body[at + 2], body[at + 3]]);
+ at += 4;
+ for _ in 0..acl_count.max(0) {
+ need(at, 2, "principal length");
+ let principal_len = usize::try_from(i16::from_be_bytes([body[at], body[at + 1]]))
+ .expect("non-negative principal length");
+ at += 2;
+ need(at, principal_len + 2, "principal and host length");
+ at += principal_len;
+ let host_len = usize::try_from(i16::from_be_bytes([body[at], body[at + 1]]))
+ .expect("non-negative host length");
+ at += 2;
+ need(at, host_len + 2, "host, operation and permission type");
+ at += host_len;
+ let operation = body[at].cast_signed();
+ at += 1;
+ let permission_type = body[at].cast_signed();
+ at += 1;
+ out.push((
+ resource_type,
+ name.clone(),
+ pattern_type,
+ operation,
+ permission_type,
+ ));
+ }
+ }
+
+ // Trailing bytes mean the walk is out of step even though every individual read fit, which is
+ // the failure that returns plausible bindings rather than panicking.
+ assert_eq!(
+ at,
+ body.len(),
+ "decoder stopped {} byte(s) short of the end, so the bindings above were read at the \
+ wrong offsets",
+ body.len() - at
+ );
+ out
+}
+
/// First `i16` of a `SaslHandshake` or `SaslAuthenticate` response body is its error code.
fn error_code(body: &Bytes) -> i16 {
assert!(body.len() >= 2, "response body is too short to hold a code");
@@ -621,6 +731,7 @@ async fn given_an_unauthenticated_connection_when_it_goes_quiet_should_be_droppe
let authenticator = Arc::new(FixedCredentialAuthenticator {
username: "alice",
password: "s3cret",
+ permissions: PrincipalPermissions::default(),
});
let (addr, shutdown) = spawn_test_server_with_authenticator(config, authenticator).await;
std::mem::forget(shutdown);
@@ -642,6 +753,7 @@ async fn given_an_authenticated_connection_when_it_goes_quiet_should_keep_the_lo
let authenticator = Arc::new(FixedCredentialAuthenticator {
username: "alice",
password: "s3cret",
+ permissions: PrincipalPermissions::default(),
});
let (addr, shutdown) = spawn_test_server_with_authenticator(config, authenticator).await;
std::mem::forget(shutdown);
@@ -699,7 +811,11 @@ struct UnavailableAuthenticator;
#[async_trait]
impl SaslAuthenticator for UnavailableAuthenticator {
- async fn authenticate(&self, _credentials: &PlainCredentials) -> Result<(), AuthError> {
+ async fn authenticate(
+ &self,
+ _credentials: &PlainCredentials,
+ _slot: SemaphorePermit<'_>,
+ ) -> Result<AuthenticatedPrincipal, AuthError> {
Err(AuthError::Unavailable)
}
}
@@ -882,6 +998,7 @@ async fn given_an_authenticator_without_sasl_enabled_should_refuse_to_start() {
let authenticator = Arc::new(FixedCredentialAuthenticator {
username: "alice",
password: "s3cret",
+ permissions: PrincipalPermissions::default(),
});
let config = GatewayConfig {
sasl_enabled: false,
@@ -905,9 +1022,17 @@ struct StallingAuthenticator {
#[async_trait]
impl SaslAuthenticator for StallingAuthenticator {
- async fn authenticate(&self, _credentials: &PlainCredentials) -> Result<(), AuthError> {
+ async fn authenticate(
+ &self,
+ credentials: &PlainCredentials,
+ _slot: SemaphorePermit<'_>,
+ ) -> Result<AuthenticatedPrincipal, AuthError> {
let _held = self.release.acquire().await;
- Ok(())
+ Ok(AuthenticatedPrincipal {
+ username: credentials.username.clone(),
+ permissions: PrincipalPermissions::default(),
+ permissions_known: true,
+ })
}
}
@@ -949,6 +1074,86 @@ async fn given_all_authentication_slots_are_busy_when_waiting_too_long_should_cl
assert_closed(&mut queued).await;
}
+/// Records the most verifications in flight at once, so a test can pin what bounds them.
+#[derive(Debug, Default)]
+struct ConcurrencyRecordingAuthenticator {
+ in_flight: AtomicUsize,
+ peak: AtomicUsize,
+}
+
+#[async_trait]
+impl SaslAuthenticator for ConcurrencyRecordingAuthenticator {
+ async fn authenticate(
+ &self,
+ credentials: &PlainCredentials,
+ _slot: SemaphorePermit<'_>,
+ ) -> Result<AuthenticatedPrincipal, AuthError> {
+ let in_flight = self.in_flight.fetch_add(1, Ordering::SeqCst) + 1;
+ self.peak.fetch_max(in_flight, Ordering::SeqCst);
+ // Long enough that unbounded verifications would visibly overlap, short enough that a
+ // bounded run still finishes well inside the pre-authentication budget.
+ tokio::time::sleep(Duration::from_millis(50)).await;
+ self.in_flight.fetch_sub(1, Ordering::SeqCst);
+ Ok(AuthenticatedPrincipal {
+ username: credentials.username.clone(),
+ permissions: PrincipalPermissions::default(),
+ permissions_known: true,
+ })
+ }
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn given_a_single_authentication_slot_should_verify_one_credential_at_a_time() {
+ // Iggy hashes with Argon2 inline on its shard threads, which have no blocking pool, so
+ // unbounded concurrent verification is a denial-of-service vector rather than a throughput
+ // question. What pins that here is an authenticator that *completes*: the sibling test above
+ // stalls forever, so its client times out whether or not this semaphore holds, and it passes
+ // just as happily with the permit dropped the moment it is taken.
+ const CLIENTS: i32 = 6;
+ let config = GatewayConfig {
+ max_concurrent_authentications: 1,
+ pre_auth_timeout: Duration::from_secs(10),
+ ..sasl_config()
+ };
+ let authenticator = Arc::new(ConcurrencyRecordingAuthenticator::default());
+ let (addr, shutdown) =
+ spawn_test_server_with_authenticator(config, authenticator.clone()).await;
+ std::mem::forget(shutdown);
+
+ let clients: Vec<_> = (0..CLIENTS)
+ .map(|client| {
+ tokio::spawn(async move {
+ let mut stream = TcpStream::connect(addr).await.expect("connect");
+ handshake_ok(&mut stream).await;
+ let token = plain_token("alice", "s3cret");
+ let body = send(
+ &mut stream,
+ API_KEY_SASL_AUTHENTICATE,
+ AUTHENTICATE_VERSION,
+ client + 2,
+ &authenticate_body(&token),
+ )
+ .await;
+ assert_eq!(
+ error_code(&body),
+ ERROR_NONE,
+ "a queued credential must still be verified, not rejected"
+ );
+ })
+ })
+ .collect();
+ for client in clients {
+ client.await.expect("client task");
+ }
+
+ assert_eq!(
+ authenticator.peak.load(Ordering::SeqCst),
+ 1,
+ "one slot must admit one verification at a time; anything higher means the limit bounds \
+ nothing and a burst of connections reaches Iggy's shard threads unthrottled"
+ );
+}
+
#[tokio::test]
async fn given_a_verification_in_flight_when_shutting_down_should_close_without_waiting_it_out() {
// The connection loop only watches the shutdown token between frames. A verification that
@@ -1072,6 +1277,413 @@ async fn given_a_refused_api_versions_when_retried_lower_should_still_be_answere
);
}
+const API_KEY_DESCRIBE_ACLS: i16 = 29;
+const DESCRIBE_ACLS_VERSION: i16 = 1;
+
+/// `DescribeAcls` v1 filter body: an all-matching filter (`ANY` everywhere, no names).
+fn any_acl_filter_body() -> Bytes {
+ let mut buf = BytesMut::new();
+ buf.put_i8(1); // resource_type_filter = ANY
+ buf.put_i16(-1); // resource_name_filter = null
+ buf.put_i8(1); // pattern_type_filter = ANY
+ buf.put_i16(-1); // principal_filter = null
+ buf.put_i16(-1); // host_filter = null
+ buf.put_i8(1); // operation = ANY
+ buf.put_i8(1); // permission_type = ANY
+ buf.freeze()
+}
+
+/// `DescribeAcls` v3 filter body: flexible framing, so compact nullable strings and a trailing
+/// tagged-fields byte. This is what a real `AdminClient` negotiates against the advertised range,
+/// and it exercises request header v2 and response header v1 as well.
+fn any_acl_filter_body_v3() -> Bytes {
+ let mut buf = BytesMut::new();
+ buf.put_i8(1); // resource_type_filter = ANY
+ buf.put_u8(0); // resource_name_filter = null (compact: varint 0)
+ buf.put_i8(1); // pattern_type_filter = ANY
+ buf.put_u8(0); // principal_filter = null
+ buf.put_u8(0); // host_filter = null
+ buf.put_i8(1); // operation = ANY
+ buf.put_i8(1); // permission_type = ANY
+ buf.put_u8(0); // empty tagged fields
+ buf.freeze()
+}
+
+async fn spawn_gateway_for(permissions: PrincipalPermissions) -> SocketAddr {
+ let authenticator = Arc::new(FixedCredentialAuthenticator {
+ username: "alice",
+ password: "s3cret",
+ permissions,
+ });
+ let (addr, shutdown) = spawn_test_server_with_authenticator(sasl_config(), authenticator).await;
+ std::mem::forget(shutdown);
+ addr
+}
+
+async fn authenticate(stream: &mut TcpStream) {
+ handshake_ok(stream).await;
+ let token = plain_token("alice", "s3cret");
+ let body = send(
+ stream,
+ API_KEY_SASL_AUTHENTICATE,
+ AUTHENTICATE_VERSION,
+ 2,
+ &authenticate_body(&token),
+ )
+ .await;
+ assert_eq!(error_code(&body), ERROR_NONE);
+}
+
+#[tokio::test]
+async fn given_a_principal_with_permissions_when_describing_acls_should_report_them() {
+ let addr = spawn_gateway_for(PrincipalPermissions {
+ poll_messages: true,
+ send_messages: true,
+ read_topics: true,
+ ..PrincipalPermissions::default()
+ })
+ .await;
+ let mut stream = TcpStream::connect(addr).await.expect("connect");
+ authenticate(&mut stream).await;
+
+ let body = send(
+ &mut stream,
+ API_KEY_DESCRIBE_ACLS,
+ DESCRIBE_ACLS_VERSION,
+ 3,
+ &any_acl_filter_body(),
+ )
+ .await;
+ assert_eq!(
+ acl_error_code(&body),
+ ERROR_NONE,
+ "an ACL view is not an error"
+ );
+ // Decoded, not byte-scanned: the resource type, name and operation of every binding are the
+ // thing under test, and a substring scan asserts none of them.
+ let bindings = parse_acl_bindings(&body);
+ assert!(
+ bindings.contains(&(2, "*".to_string(), 3, 3, 3)),
+ "poll_messages must render TOPIC * READ, got {bindings:?}"
+ );
+ // Iggy has no deny rules and nothing here is prefix-scoped, so every binding must say ALLOW on
+ // a LITERAL pattern. Without asserting these two, a response meaning the opposite would pass.
+ assert!(
+ bindings
+ .iter()
+ .all(|(_, _, pattern, _, permission)| *pattern == 3 && *permission == 3),
+ "every binding must be an ALLOW on a LITERAL pattern, got {bindings:?}"
+ );
+ assert!(
+ bindings.contains(&(2, "*".to_string(), 3, 4, 3)),
+ "send_messages must render TOPIC * WRITE, got {bindings:?}"
+ );
+ assert!(
+ bindings.contains(&(3, "*".to_string(), 3, 3, 3)),
+ "a principal that may read topics must get the derived GROUP * READ, got {bindings:?}"
+ );
+ assert!(
+ body.windows(10).any(|w| w == b"User:alice"),
+ "every binding names the authenticated principal"
+ );
+}
+
+#[tokio::test]
+async fn given_only_poll_messages_should_not_render_a_group_binding() {
+ // Regression. Consumer-group operations route through Iggy's topic rule, which never consults
+ // `poll_messages`, so deriving the group binding from polling advertised access Iggy denies.
+ let addr = spawn_gateway_for(PrincipalPermissions {
+ poll_messages: true,
+ ..PrincipalPermissions::default()
+ })
+ .await;
+ let mut stream = TcpStream::connect(addr).await.expect("connect");
+ authenticate(&mut stream).await;
+
+ let body = send(
+ &mut stream,
+ API_KEY_DESCRIBE_ACLS,
+ DESCRIBE_ACLS_VERSION,
+ 3,
+ &any_acl_filter_body(),
+ )
+ .await;
+ let bindings = parse_acl_bindings(&body);
+ assert!(
+ bindings.contains(&(2, "*".to_string(), 3, 3, 3)),
+ "polling still grants TOPIC READ, got {bindings:?}"
+ );
+ assert!(
+ !bindings.iter().any(|(resource, ..)| *resource == 3),
+ "no group binding without a topic read grant, got {bindings:?}"
+ );
+}
+
+#[tokio::test]
+async fn given_manage_servers_should_not_claim_an_unusable_cluster_alter() {
+ // A server flag renders DESCRIBE and nothing else. `manage_servers` cannot be set here on
+ // purpose: Iggy reads it in exactly one rule, as an alias for `read_servers`, so the
+ // projection folds the two together and this layer never sees it apart. Rendering CLUSTER
+ // ALTER from it advertised an ability with nowhere to be used, and the two write ACL APIs it
+ // implies are not even advertised.
+ let addr = spawn_gateway_for(PrincipalPermissions {
+ read_servers: true,
+ ..PrincipalPermissions::default()
+ })
+ .await;
+ let mut stream = TcpStream::connect(addr).await.expect("connect");
+ authenticate(&mut stream).await;
+
+ let body = send(
+ &mut stream,
+ API_KEY_DESCRIBE_ACLS,
+ DESCRIBE_ACLS_VERSION,
+ 3,
+ &any_acl_filter_body(),
+ )
+ .await;
+ let bindings = parse_acl_bindings(&body);
+ assert!(
+ bindings.contains(&(4, "kafka-cluster".to_string(), 3, 8, 3)),
+ "server permissions must still render CLUSTER DESCRIBE, got {bindings:?}"
+ );
+ assert!(
+ !bindings.iter().any(|(.., operation, _)| *operation == 7),
+ "nothing may claim ALTER, got {bindings:?}"
+ );
+}
+
+#[tokio::test]
+async fn given_a_principal_with_no_permissions_should_report_an_empty_view_not_an_error() {
+ // Kafka draws a firm line between "nothing matched" and "the request failed", and an admin
+ // tool prints them very differently.
+ let addr = spawn_gateway_for(PrincipalPermissions::default()).await;
+ let mut stream = TcpStream::connect(addr).await.expect("connect");
+ authenticate(&mut stream).await;
+
+ let body = send(
+ &mut stream,
+ API_KEY_DESCRIBE_ACLS,
+ DESCRIBE_ACLS_VERSION,
+ 3,
+ &any_acl_filter_body(),
+ )
+ .await;
+ assert_eq!(
+ acl_error_code(&body),
+ ERROR_NONE,
+ "an empty ACL set is a successful answer"
+ );
+ assert!(
+ parse_acl_bindings(&body).is_empty(),
+ "no permissions means no bindings at all"
+ );
+}
+
+/// Accepts `alice` but reports that her permissions could not be read, the shape
+/// `IggyAuthenticator` produces when the login succeeds and the follow-up `get_user` does not.
+#[derive(Debug)]
+struct UnreadPermissionsAuthenticator;
+
+#[async_trait]
+impl SaslAuthenticator for UnreadPermissionsAuthenticator {
+ async fn authenticate(
+ &self,
+ credentials: &PlainCredentials,
+ _slot: SemaphorePermit<'_>,
+ ) -> Result<AuthenticatedPrincipal, AuthError> {
+ Ok(AuthenticatedPrincipal {
+ username: credentials.username.clone(),
+ permissions: PrincipalPermissions::default(),
+ permissions_known: false,
+ })
+ }
+}
+
+#[tokio::test]
+async fn given_unread_permissions_when_describing_acls_should_answer_an_error_and_stay_open() {
+ // The fallback permissions are empty, so answering from them would render the same zero
+ // bindings as a principal that genuinely holds nothing. Every other stub reports its
+ // permissions as known, so without this case the two answers could swap unnoticed.
+ let (addr, shutdown) = spawn_test_server_with_authenticator(
+ sasl_config(),
+ Arc::new(UnreadPermissionsAuthenticator),
+ )
+ .await;
+ std::mem::forget(shutdown);
+ let mut stream = TcpStream::connect(addr).await.expect("connect");
+ authenticate(&mut stream).await;
+
+ let body = send(
+ &mut stream,
+ API_KEY_DESCRIBE_ACLS,
+ DESCRIBE_ACLS_VERSION,
+ 3,
+ &any_acl_filter_body(),
+ )
+ .await;
+ assert_eq!(
+ acl_error_code(&body),
+ ERROR_UNKNOWN_SERVER_ERROR,
+ "a view that was never read must not be reported as an empty one"
+ );
+ assert!(
+ parse_acl_bindings(&body).is_empty(),
+ "an error answer carries no bindings"
+ );
+
+ // Kept open: the login was valid, and only the ACL view is missing.
+ let metadata = send(&mut stream, API_KEY_METADATA, 0, 4, &[0, 0, 0, 0]).await;
+ assert!(
+ !metadata.is_empty(),
+ "the connection must keep serving after the ACL error"
+ );
+}
+
+#[tokio::test]
+async fn given_an_unauthenticated_connection_when_describing_acls_should_be_answered_then_closed() {
+ let addr = spawn_gateway_for(PrincipalPermissions::default()).await;
+ let mut stream = TcpStream::connect(addr).await.expect("connect");
+
+ // Answered with a parseable body rather than dropped silently. The key is outside the firewall
+ // table, so the generic error path closes without one, which leaves the client guessing on the
+ // one path an unauthenticated peer can actually reach.
+ let body = send(
+ &mut stream,
+ API_KEY_DESCRIBE_ACLS,
+ DESCRIBE_ACLS_VERSION,
+ 1,
+ &any_acl_filter_body(),
+ )
+ .await;
+ assert_eq!(acl_error_code(&body), ERROR_ILLEGAL_SASL_STATE);
+ assert!(
+ parse_acl_bindings(&body).is_empty(),
+ "a refusal must disclose no bindings"
+ );
+ assert_closed(&mut stream).await;
+}
+
+#[tokio::test]
+async fn given_sasl_enabled_should_advertise_describe_acls() {
+ let addr = spawn_gateway_for(PrincipalPermissions::default()).await;
+ let mut stream = TcpStream::connect(addr).await.expect("connect");
+ let body = send(&mut stream, API_KEY_API_VERSIONS, 1, 1, &[]).await;
+ let rows = parse_api_versions(&body);
+ let acls = rows
+ .iter()
+ .find(|(key, _, _)| *key == API_KEY_DESCRIBE_ACLS)
+ .expect("DescribeAcls must be advertised while SASL is enabled");
+ assert_eq!((acls.1, acls.2), (1, 3));
+}
+
+#[tokio::test]
+async fn given_sasl_disabled_should_not_advertise_describe_acls() {
+ // With no principal there is nothing it could truthfully describe.
+ let (addr, shutdown) = server::spawn_test_server().await;
+ std::mem::forget(shutdown);
+ let mut stream = TcpStream::connect(addr).await.expect("connect");
+ let body = send(&mut stream, API_KEY_API_VERSIONS, 1, 1, &[]).await;
+ assert!(
+ !parse_api_versions(&body)
+ .iter()
+ .any(|(key, _, _)| *key == API_KEY_DESCRIBE_ACLS)
+ );
+}
+
+#[tokio::test]
+async fn given_describe_acls_v3_should_answer_over_the_flexible_framing() {
+ // v3 is the version a real AdminClient negotiates, and the only flexible one: compact strings,
+ // tagged fields, request header v2 and response header v1. v1 exercises none of that.
+ let addr = spawn_gateway_for(PrincipalPermissions {
+ poll_messages: true,
+ read_topics: true,
+ ..PrincipalPermissions::default()
+ })
+ .await;
+ let mut stream = TcpStream::connect(addr).await.expect("connect");
+ authenticate(&mut stream).await;
+
+ let body = send(
+ &mut stream,
+ API_KEY_DESCRIBE_ACLS,
+ 3,
+ 3,
+ &any_acl_filter_body_v3(),
+ )
+ .await;
+ // v3 is flexible, so the decoder above (legacy framing) does not apply; the error code still
+ // sits after the throttle field.
+ assert_eq!(acl_error_code(&body), ERROR_NONE, "v3 must answer cleanly");
+ assert!(
+ body.windows(10).any(|w| w == b"User:alice"),
+ "the v3 response must carry the rendered bindings, not just a header"
+ );
+
+ // Still usable afterwards, which proves the flexible response framing did not desync the
+ // connection's correlation stream.
+ let again = send(&mut stream, API_KEY_METADATA, 0, 4, &[0, 0, 0, 0]).await;
+ assert!(!again.is_empty());
+}
+
+#[tokio::test]
+async fn given_describe_acls_above_the_advertised_range_should_be_refused() {
+ let addr = spawn_gateway_for(PrincipalPermissions::default()).await;
+ let mut stream = TcpStream::connect(addr).await.expect("connect");
+ authenticate(&mut stream).await;
+
+ // v4 has no schema at either end, so there is no parseable body to answer with.
+ let frame = build_request_frame(
+ API_KEY_DESCRIBE_ACLS,
+ 4,
+ 3,
+ Some("sasl-test"),
+ &any_acl_filter_body_v3(),
+ );
+ stream.write_all(&frame).await.expect("write request");
+ assert_closed(&mut stream).await;
+}
+
+#[tokio::test]
+async fn given_a_malformed_acl_filter_should_answer_an_error_and_stay_open() {
+ // The handler's two error branches differ in liveness: a malformed filter is answered and the
+ // connection kept, while an out-of-range version closes. Only the second was covered, so the
+ // reachable one went untested.
+ let addr = spawn_gateway_for(PrincipalPermissions::default()).await;
+ let mut stream = TcpStream::connect(addr).await.expect("connect");
+ authenticate(&mut stream).await;
+
+ // Declares a 64-byte resource name in a body that holds two.
+ let mut truncated = BytesMut::new();
+ truncated.put_i8(1);
+ truncated.put_i16(64);
+ truncated.put_slice(b"xy");
+ let body = send(
+ &mut stream,
+ API_KEY_DESCRIBE_ACLS,
+ DESCRIBE_ACLS_VERSION,
+ 3,
+ &truncated.freeze(),
+ )
+ .await;
+ assert_eq!(
+ acl_error_code(&body),
+ 42,
+ "a malformed filter is INVALID_REQUEST"
+ );
+
+ // Still usable: the client may send a well-formed filter next.
+ let good = send(
+ &mut stream,
+ API_KEY_DESCRIBE_ACLS,
+ DESCRIBE_ACLS_VERSION,
+ 4,
+ &any_acl_filter_body(),
+ )
+ .await;
+ assert_eq!(acl_error_code(&good), ERROR_NONE);
+}
+
#[tokio::test]
async fn given_a_repeated_refused_api_versions_should_run_the_allowance_out() {
// The half a success-only allowance misses: a version this gateway refuses is one the peer