This is an automated email from the ASF dual-hosted git repository. numinnex pushed a commit to branch kafka_proxy_acls in repository https://gitbox.apache.org/repos/asf/iggy.git
commit f8753b2c2de6d86db8171a483414bbd492f2a0f4 Author: Grzegorz Koszyk <[email protected]> AuthorDate: Fri Sep 18 17:18:49 2026 +0200 temp --- .config/nextest.toml | 12 + .github/actions/rust/pre-merge/action.yml | 22 + gateways/kafka/README.md | 22 +- gateways/kafka/docs/ACL_MAPPING.md | 195 ++++++++ gateways/kafka/docs/MANUAL_TESTING.md | 52 ++ gateways/kafka/docs/SCOPE.md | 3 + gateways/kafka/docs/TEST_SUITE.md | 22 + gateways/kafka/src/auth.rs | 207 +++++++- gateways/kafka/src/protocol/acl.rs | 522 ++++++++++++++++++++ gateways/kafka/src/protocol/api.rs | 88 +++- gateways/kafka/src/protocol/bounds_guard.rs | 77 +++ gateways/kafka/src/protocol/mod.rs | 1 + gateways/kafka/src/protocol/responses.rs | 74 ++- gateways/kafka/src/server.rs | 104 +++- gateways/kafka/tests/kafka_client_e2e_tests.rs | 648 +++++++++++++++++++++++++ gateways/kafka/tests/sasl_tests.rs | 558 ++++++++++++++++++++- 16 files changed, 2561 insertions(+), 46 deletions(-) diff --git a/.config/nextest.toml b/.config/nextest.toml index 1ff8ee2c5..46a834b15 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -59,6 +59,18 @@ max-threads = 1 filter = 'binary_id(iggy-gateway-kafka::bridge_iggy_integration_tests)' 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 8b3413b79..1789d9024 100644 --- a/.github/actions/rust/pre-merge/action.yml +++ b/.github/actions/rust/pre-merge/action.yml @@ -393,6 +393,28 @@ 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. Failure stays a warning: + # the test still tries, and the guard above still fails the job if it cannot run. + # 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 retry it inside the test" + 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 a2e8304db..e9af99f9c 100644 --- a/gateways/kafka/README.md +++ b/gateways/kafka/README.md @@ -67,6 +67,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 @@ -120,12 +121,23 @@ Four things to know before switching it on: replicated registration. Verification is deliberately not cached, since caching it per username would let a second connection present any password. Connection churn is therefore server load, bounded by `IGGY_KAFKA_MAX_CONCURRENT_AUTHENTICATIONS`. -- **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. + +A principal sees its own permissions and nobody else's, because the gateway holds no administrative +credentials. 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 3b4d4b962..7edecda54 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 5c28f1d59..becf651ea 100644 --- a/gateways/kafka/docs/SCOPE.md +++ b/gateways/kafka/docs/SCOPE.md @@ -150,6 +150,9 @@ Authentication design ([#3549](https://github.com/apache/iggy/issues/3549)): SASL state machine before dispatch, so a gateway with the feature off refuses them like any other unlisted key and enabling it later 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 e912a421a..206c3b36c 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) | `tests/common/` holds shared helpers (`codec.rs`, `fixtures.rs`, `scope.rs`, `server.rs`, @@ -73,6 +74,27 @@ is test-only primitive encode/decode scaffolding for hand-building legacy/advers --- +## Real-client end-to-end suite + +`kafka_client_e2e_tests.rs` needs two things the rest of the suite does not: Docker, and an +already-built `iggy-server` in the same target directory. Missing either makes it skip with a +printed reason rather than fail, which is what lets `cargo test -p iggy-gateway-kafka` stay usable +without either. + +```bash +cargo build --bin iggy-server +KAFKA_E2E_REQUIRED=1 cargo test -p iggy-gateway-kafka --test kafka_client_e2e_tests +``` + +`KAFKA_E2E_REQUIRED=1` turns a skip into a failure, mirroring `KAFKA_FIXTURES_REQUIRED`, so a CI +job that means to run these cannot report a pass over zero assertions. Set it there. + +The suite shares the `kafka_bridge` nextest group with the bridge tests, so its spawned servers are +serialized against them rather than competing for cores and ports. + +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 37f009efc..7f04d775b 100644 --- a/gateways/kafka/src/auth.rs +++ b/gateways/kafka/src/auth.rs @@ -24,17 +24,20 @@ use std::time::Duration; use async_trait::async_trait; -use iggy::prelude::{AutoLogin, Client, Credentials, IggyClientBuilder, IggyError}; +use iggy::prelude::{ + AutoLogin, Client, Credentials, Identifier, IggyClientBuilder, IggyError, Permissions, +}; use tracing::{debug, warn}; +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 permission read and the teardown have their own, much +/// smaller budgets ([`PERMISSION_READ_TIMEOUT`] and [`TEARDOWN_TIMEOUT`]), so the three together +/// stay inside the caller's pre-authentication budget rather than exceeding it. The caller bounds +/// the whole thing again from outside, because a permit wait is not covered here at all. /// /// A verification that has not answered inside this is indistinguishable, from the Kafka client's /// side, from one that failed, and the client is holding a connection open waiting for it. Shorter @@ -53,12 +56,26 @@ const VERIFY_TIMEOUT: Duration = Duration::from_secs(10); /// hide the failure underneath one the client cannot see. const VERIFY_RECONNECTION_RETRIES: u32 = 1; +/// Budget for the permission read that follows a successful login. +/// +/// Much smaller than [`VERIFY_TIMEOUT`] on purpose, and deliberately small in absolute terms. The +/// caller bounds the whole exchange at its pre-authentication budget, which must also absorb the +/// wait for an authentication slot, so every second spent here is a second that wait does not get. +/// At one second the inner worst case is 12s against a 15s outer budget, leaving the queue three +/// seconds rather than one. +/// +/// It also bounds only *this* future, not the SDK's work. A cancelled call leaves the SDK's own +/// read running on a detached task that holds its connection lock until that task's own deadline, +/// so a shorter budget here does not stop that work, it only stops the caller waiting on it. The +/// login has already succeeded by this point, so abandoning the read costs an ACL view and nothing +/// else. +const PERMISSION_READ_TIMEOUT: Duration = Duration::from_secs(1); + /// Budget for tearing the verification client down again. /// /// Deliberately far shorter than [`VERIFY_TIMEOUT`]. Teardown happens while the caller still holds -/// an authentication permit, so giving it the full verify budget would let one attempt occupy a -/// slot for twice as long as the doc on [`VERIFY_TIMEOUT`] claims the whole operation can take. -/// Nothing is lost by cutting it short: `Drop` aborts the heartbeat task regardless. +/// an authentication permit, so every second here is a second some other connection waits. Nothing +/// is lost by cutting it short: `Drop` aborts the heartbeat task regardless. const TEARDOWN_TIMEOUT: Duration = Duration::from_secs(1); /// Why a SASL exchange did not produce a verified identity. @@ -74,19 +91,45 @@ 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. /// /// # 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, + ) -> Result<AuthenticatedPrincipal, AuthError>; } /// How the verifier reaches Iggy. @@ -219,7 +262,10 @@ 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, + ) -> Result<AuthenticatedPrincipal, AuthError> { let auto_login = AutoLogin::Enabled(Credentials::UsernamePassword( credentials.username.clone(), credentials.password.clone(), @@ -244,7 +290,12 @@ impl SaslAuthenticator for IggyAuthenticator { 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, @@ -265,7 +316,92 @@ impl SaslAuthenticator for IggyAuthenticator { } } - outcome + outcome.map(|(permissions, permissions_known)| AuthenticatedPrincipal { + username: credentials.username.clone(), + permissions, + permissions_known, + }) + } +} + +/// Reads the just-authenticated user's own record and projects its global permissions. +/// +/// A missing record or absent permissions both yield an empty set rather than an error: the +/// credentials were already accepted, so refusing the connection here would reject a valid login +/// over an authorization view it never asked for. +async fn fetch_permissions( + client: &impl Client, + username: &str, +) -> Result<(PrincipalPermissions, bool), AuthError> { + // Unreachable in practice: a username that reached a successful login is already inside + // Identifier's own length bounds. Propagating rather than degrading is still the wrong shape + // for this function, so it degrades like every other failure below. + let Ok(identifier) = Identifier::named(username) else { + warn!("authenticated, but the principal's name is not a valid Iggy identifier"); + return Ok((PrincipalPermissions::default(), false)); + }; + // Deliberately not propagated as a failure. The credentials were already accepted by the login + // above, so turning a stumble on this second round trip into a rejection would answer a correct + // password with `SASL_AUTHENTICATION_FAILED`, which a Kafka client treats as fatal and raises + // to the application. Losing the ACL view is the lesser harm, and it degrades to an empty one. + let fetched = tokio::time::timeout(PERMISSION_READ_TIMEOUT, client.get_user(&identifier)).await; + let mut known = true; + let user = match fetched { + Ok(Ok(None)) => { + known = false; + // 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(Ok(user)) => user, + Ok(Err(error)) => { + known = false; + warn!(%error, "authenticated, but could not read the principal's permissions"); + None + } + Err(_elapsed) => { + known = false; + warn!("authenticated, but timed out reading 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(), + known, + )) +} + +/// 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, + } } } @@ -304,6 +440,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!( diff --git a/gateways/kafka/src/protocol/acl.rs b/gateways/kafka/src/protocol/acl.rs new file mode 100644 index 000000000..fddd4ba0d --- /dev/null +++ b/gateways/kafka/src/protocol/acl.rs @@ -0,0 +1,522 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Rendering an Iggy principal's permissions as Kafka ACL bindings. +//! +//! Pure and synchronous, and deliberately free of any Iggy type: the permissions arrive as +//! [`PrincipalPermissions`], which `auth` fills in. That keeps the protocol layer independent of +//! the SDK and makes the mapping unit-testable without a server. +//! +//! See `docs/ACL_MAPPING.md` for the decisions this implements, including what is not mapped. + +/// Kafka resource types (`org.apache.kafka.common.resource.ResourceType`). `kafka_protocol` carries +/// these as bare `i8` with no enum, so the values live here. +pub mod resource_type { + pub const ANY: i8 = 1; + pub const TOPIC: i8 = 2; + pub const GROUP: i8 = 3; + pub const CLUSTER: i8 = 4; +} + +/// Kafka pattern types (`org.apache.kafka.common.resource.PatternType`). +pub mod pattern_type { + pub const ANY: i8 = 1; + /// Kafka's own "match anything of this type" lookup, which a filter may ask for. + pub const MATCH: i8 = 2; + pub const LITERAL: i8 = 3; +} + +/// Kafka ACL operations (`org.apache.kafka.common.acl.AclOperation`). +pub mod operation { + pub const ANY: i8 = 1; + pub const ALL: i8 = 2; + pub const READ: i8 = 3; + pub const WRITE: i8 = 4; + pub const CREATE: i8 = 5; + pub const DELETE: i8 = 6; + pub const ALTER: i8 = 7; + pub const DESCRIBE: i8 = 8; +} + +/// Kafka ACL permission types (`org.apache.kafka.common.acl.AclPermissionType`). +pub mod permission_type { + pub const ANY: i8 = 1; + pub const ALLOW: i8 = 3; +} + +/// Name Kafka gives the cluster resource. There is exactly one, and it is always called this. +pub const CLUSTER_NAME: &str = "kafka-cluster"; + +/// How Kafka spells "every resource of this type": a literal pattern named `*`. +pub const WILDCARD: &str = "*"; + +/// Host scope on every binding this gateway renders. Iggy has no host-scoped permissions. +pub const ANY_HOST: &str = "*"; + +/// The subset of an Iggy principal's global permissions that has a Kafka meaning. +/// +/// Stream-level flags are folded into the topic ones by the caller: Kafka has no resource above a +/// topic, and every Kafka topic lives inside one Iggy stream, so a stream grant is in practice a +/// grant over the topics a Kafka client can reach. +/// +/// The boolean count mirrors Iggy's own `GlobalPermissions`, which is a flat set of independent +/// grants. Collapsing them into a bitfield would hide which grant is which at every call site for +/// no gain, so the lint is allowed here the way it is elsewhere in this repository. +#[allow(clippy::struct_excessive_bools)] +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct PrincipalPermissions { + /// True when the principal holds either of Iggy's server flags. `manage_servers` is not + /// carried separately: Iggy reads it in exactly one rule, as an alias for this one, so it + /// gates no mutation anywhere and renders nothing of its own. `docs/ACL_MAPPING.md` has the + /// argument, which is the same one that keeps `manage_users` out of the table. + pub read_servers: bool, + pub read_topics: bool, + pub manage_topics: bool, + pub poll_messages: bool, + pub send_messages: bool, +} + +/// One rendered binding. Principal, host and permission type are constant for every binding this +/// gateway produces, so they are not carried here. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct AclBinding { + pub resource_type: i8, + pub resource_name: &'static str, + pub operation: i8, +} + +impl AclBinding { + const fn new(resource_type: i8, resource_name: &'static str, operation: i8) -> Self { + Self { + resource_type, + resource_name, + operation, + } + } +} + +/// Renders a principal's permissions as Kafka ACL bindings. +/// +/// Only global permissions are rendered, as wildcard bindings. Iggy keys its per-stream and +/// per-topic permissions by numeric id, and the topic mapping is one-way, so a named Kafka binding +/// cannot be reconstructed from them without a reverse index that does not exist. The result +/// under-reports rather than over-reports, which is the safe direction for an authorization view. +#[must_use] +pub fn bindings_for(permissions: &PrincipalPermissions) -> Vec<AclBinding> { + let mut bindings = Vec::new(); + + if permissions.read_servers { + bindings.push(AclBinding::new( + resource_type::CLUSTER, + CLUSTER_NAME, + operation::DESCRIBE, + )); + } + if permissions.read_topics { + bindings.push(AclBinding::new( + resource_type::TOPIC, + WILDCARD, + operation::DESCRIBE, + )); + } + if permissions.manage_topics { + for op in [operation::CREATE, operation::DELETE, operation::ALTER] { + bindings.push(AclBinding::new(resource_type::TOPIC, WILDCARD, op)); + } + } + if permissions.poll_messages { + bindings.push(AclBinding::new( + resource_type::TOPIC, + WILDCARD, + operation::READ, + )); + } + // Derived, not stored: Iggy has no group-level permission. Group *membership* operations + // (create, delete, get, join, leave) route through `Permissioner::get_topic` + // (`permissioner_rules/consumer_groups.rs`), which admits on the read and manage flags and + // never consults `poll_messages`. Deriving this from polling instead granted a group to + // principals Iggy denies, which is the over-report the design commits against. + // + // Not every group-shaped operation goes that way: offset commit and fetch route through + // `poll_messages` (`permissioner_rules/consumer_offsets.rs`), while Kafka gates them on this + // same GROUP READ. A principal with polling but no read grant is therefore shown no group + // binding while Iggy would let it commit an offset. That under-reports, which is the safe + // direction, and the alternative over-reports membership. + if permissions.read_topics { + bindings.push(AclBinding::new( + resource_type::GROUP, + WILDCARD, + operation::READ, + )); + } + if permissions.send_messages { + bindings.push(AclBinding::new( + resource_type::TOPIC, + WILDCARD, + operation::WRITE, + )); + } + + bindings +} + +/// The filter carried by a `DescribeAcls` request, already decoded. +#[derive(Debug, Clone)] +pub struct AclFilter { + pub resource_type: i8, + pub resource_name: Option<String>, + pub pattern_type: i8, + pub principal: Option<String>, + pub host: Option<String>, + pub operation: i8, + pub permission_type: i8, +} + +impl AclFilter { + /// Whether this filter selects `binding`, belonging to `principal`. + /// + /// Kafka's `ANY` sentinel matches everything, and an absent string field is the same as `ANY`. + /// A principal filter naming anyone else matches nothing: the gateway holds no administrative + /// credentials and can only ever read the caller's own record, so claiming an empty result for + /// another user is the only honest answer it can give. + #[must_use] + pub fn matches(&self, binding: &AclBinding, principal: &str) -> bool { + if self.resource_type != resource_type::ANY && self.resource_type != binding.resource_type { + return false; + } + // PREFIXED selects nothing, because nothing here is prefix-scoped. + if !matches!( + self.pattern_type, + pattern_type::ANY | pattern_type::MATCH | pattern_type::LITERAL + ) { + return false; + } + if !self.matches_resource_name(binding.resource_name) { + return false; + } + if !matches_name(self.principal.as_deref(), &format!("User:{principal}")) { + 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 + } +} + +#[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 18f63c360..02c4c22b9 100644 --- a/gateways/kafka/src/protocol/api.rs +++ b/gateways/kafka/src/protocol/api.rs @@ -19,20 +19,22 @@ use bytes::{Buf, Bytes}; use kafka_protocol::messages::api_versions_response::ApiVersion; use kafka_protocol::messages::metadata_response::{MetadataResponseBroker, MetadataResponseTopic}; use kafka_protocol::messages::{ - ApiVersionsRequest, ApiVersionsResponse, BrokerId, CreateTopicsRequest, FetchRequest, - ListOffsetsRequest, MetadataRequest, MetadataResponse, ProduceRequest, SaslAuthenticateRequest, - SaslHandshakeRequest, TopicName, + ApiVersionsRequest, ApiVersionsResponse, BrokerId, CreateTopicsRequest, DescribeAclsRequest, + FetchRequest, ListOffsetsRequest, MetadataRequest, MetadataResponse, ProduceRequest, + SaslAuthenticateRequest, SaslHandshakeRequest, TopicName, }; use kafka_protocol::protocol::{Decodable, StrBytes}; use crate::error::{KafkaProtocolError, Result}; +use crate::protocol::acl::{self, AclBinding, AclFilter, PrincipalPermissions}; use crate::protocol::bounds_guard::{ - validate_api_versions_shape, validate_create_topics_shape, validate_fetch_shape, - validate_list_offsets_shape, validate_metadata_shape, validate_produce_shape, - validate_sasl_authenticate_shape, validate_sasl_handshake_shape, + validate_api_versions_shape, validate_create_topics_shape, validate_describe_acls_shape, + validate_fetch_shape, validate_list_offsets_shape, validate_metadata_shape, + validate_produce_shape, validate_sasl_authenticate_shape, validate_sasl_handshake_shape, }; use crate::protocol::responses::{ encode_create_topics_error_response, encode_create_topics_response, + encode_describe_acls_error_response, encode_describe_acls_response, encode_fetch_error_response, encode_fetch_response, encode_list_offsets_error_response, encode_list_offsets_response, encode_message, encode_produce_error_response, encode_produce_response, encode_sasl_authenticate_response, encode_sasl_handshake_response, @@ -46,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; @@ -249,6 +252,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. @@ -259,6 +270,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 aa62613df..444bfeb25 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 header; diff --git a/gateways/kafka/src/protocol/responses.rs b/gateways/kafka/src/protocol/responses.rs index eb2546e19..1d834927c 100644 --- a/gateways/kafka/src/protocol/responses.rs +++ b/gateways/kafka/src/protocol/responses.rs @@ -24,19 +24,21 @@ use bytes::{Bytes, BytesMut}; use kafka_protocol::messages::create_topics_request::CreatableTopic; use kafka_protocol::messages::create_topics_response::CreatableTopicResult; +use kafka_protocol::messages::describe_acls_response::{AclDescription, DescribeAclsResource}; use kafka_protocol::messages::fetch_response::{FetchableTopicResponse, PartitionData}; use kafka_protocol::messages::list_offsets_response::{ ListOffsetsPartitionResponse, ListOffsetsTopicResponse, }; use kafka_protocol::messages::produce_response::{PartitionProduceResponse, TopicProduceResponse}; use kafka_protocol::messages::{ - CreateTopicsRequest, CreateTopicsResponse, FetchRequest, FetchResponse, ListOffsetsRequest, - ListOffsetsResponse, ProduceRequest, ProduceResponse, SaslAuthenticateResponse, - SaslHandshakeResponse, + CreateTopicsRequest, CreateTopicsResponse, DescribeAclsResponse, FetchRequest, FetchResponse, + ListOffsetsRequest, ListOffsetsResponse, ProduceRequest, ProduceResponse, + SaslAuthenticateResponse, SaslHandshakeResponse, }; use kafka_protocol::protocol::{Encodable, StrBytes}; use crate::error::{KafkaProtocolError, Result}; +use crate::protocol::acl::{self, AclBinding}; use crate::protocol::api::{ ERROR_INVALID_PARTITIONS, ERROR_INVALID_REPLICATION_FACTOR, ERROR_NONE, ERROR_NOT_CONTROLLER, ERROR_NOT_LEADER_OR_FOLLOWER, @@ -293,6 +295,72 @@ pub fn encode_sasl_authenticate_response( encode_message(&resp, version, 64) } +// ── DescribeAcls ───────────────────────────────────────────────────────────── + +/// `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(acl::ANY_HOST)) + .with_operation(binding.operation) + .with_permission_type(acl::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(acl::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) +} + // ── CreateTopics ───────────────────────────────────────────────────────────── /// Well-formed `CreateTopics` response with a single placeholder topic. diff --git a/gateways/kafka/src/server.rs b/gateways/kafka/src/server.rs index 545c58c51..edf2698ad 100644 --- a/gateways/kafka/src/server.rs +++ b/gateways/kafka/src/server.rs @@ -31,13 +31,15 @@ use tokio_util::task::TaskTracker; use tracing::{debug, error, info, warn}; use tracing_appender::non_blocking::WorkerGuard; -use crate::auth::{AuthError, SaslAuthenticator}; +use crate::auth::{AuthError, AuthenticatedPrincipal, SaslAuthenticator}; 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, 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, 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}; @@ -431,6 +433,7 @@ struct ConnectionContext<'a> { async fn route_frame( ctx: &ConnectionContext<'_>, sasl_state: &mut SaslState, + principal: &mut Option<AuthenticatedPrincipal>, req: &RequestHeader, body: Bytes, ) -> HandleOutcome { @@ -451,6 +454,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( req.request_api_key, req.request_api_version, @@ -507,7 +519,7 @@ async fn route_frame( HandleOutcome::Close } SaslAction::Authenticate => { - let outcome = authenticate_token( + let (outcome, verified) = authenticate_token( ctx.authenticator, ctx.auth_slots, ctx.config.pre_auth_timeout, @@ -520,6 +532,7 @@ async fn route_frame( // `RespondThenClose`, so the state advances on exactly the accepting branch. if matches!(outcome, HandleOutcome::Respond(_)) { *sasl_state = SaslState::Authenticated; + *principal = verified; } outcome } @@ -556,6 +569,7 @@ async fn handle_connection( } else { SaslState::Authenticated }; + let mut principal: Option<AuthenticatedPrincipal> = None; let ctx = ConnectionContext { config: &config, broker: &broker, @@ -605,7 +619,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, &config, &req, resp_hdr_ver, outcome).await? { return Ok(()); } @@ -624,8 +638,13 @@ async fn authenticate_token( api_version: i16, body: Bytes, peer: &SocketAddr, -) -> HandleOutcome { - let failed = || sasl_authenticate_outcome(api_version, ERROR_SASL_AUTHENTICATION_FAILED, true); +) -> (HandleOutcome, Option<AuthenticatedPrincipal>) { + let failed = || { + ( + sasl_authenticate_outcome(api_version, ERROR_SASL_AUTHENTICATION_FAILED, true), + None, + ) + }; let Some(authenticator) = authenticator else { // Unreachable: `run` refuses to start in this combination. Fail closed anyway, since the @@ -662,13 +681,16 @@ async fn authenticate_token( // code as fatal and surfaces it to the application, and nothing here says the credentials // were wrong. A close reads as a transport failure, which is retriable. warn!(%peer, "authentication did not complete within the pre-authentication budget"); - return HandleOutcome::Close; + return (HandleOutcome::Close, None); }; match result { - Ok(()) => { + Ok(authenticated) => { debug!(%peer, "SASL authentication succeeded"); - 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) => { @@ -682,7 +704,57 @@ async fn authenticate_token( // the account exists. Err(AuthError::Unavailable) => { warn!(%peer, "SASL authentication could not be completed; Iggy is unreachable"); - HandleOutcome::Close + (HandleOutcome::Close, None) + } + } +} + +/// Answers `DescribeAcls` from the permissions captured when this connection authenticated. +/// +/// Only reachable on an authenticated connection: the SASL gate refuses every key but one before a +/// principal exists, and the caller additionally gates this on the feature being on, so a gateway +/// with SASL off never routes here. The `None` arm is a fail-closed guard, not a reachable path. +fn describe_acls( + principal: Option<&AuthenticatedPrincipal>, + api_version: i16, + body: Bytes, + peer: &SocketAddr, +) -> HandleOutcome { + let Some(principal) = principal else { + // `error!` rather than `debug!` on purpose, unlike every other refusal here: reaching this + // means the routing guards above disagree with each other, which is a gateway fault and not + // something a client can provoke. + error!(%peer, "DescribeAcls reached a connection with no authenticated principal"); + // Not `encode_error_for_key`: key 29 is absent from `SUPPORTED_RANGES`, so that helper + // always returns `Close` here and the code would read as if it answers when it cannot. + return respond_describe_acls_error(api_version, ERROR_ILLEGAL_SASL_STATE, true); + }; + // The firewall table cannot cover this key: it is kept out of `SUPPORTED_RANGES` on purpose, + // so the advertised range would otherwise be enforced only by whatever `kafka_protocol`'s + // schema happens to accept. A crate bump adding v4 would start answering v4 while ApiVersions + // still says 3, which is exactly what the sibling SASL keys pin explicitly against. + if !SASL_ADVERTISED_DESCRIBE_ACLS_VERSIONS.contains(&api_version) { + debug!(%peer, api_version, "DescribeAcls version outside the advertised range"); + return HandleOutcome::Close; + } + if !principal.permissions_known { + // The permission read failed after a successful login, so this connection holds no real + // answer. Reporting the empty fallback would tell an operator the principal has no access, + // which is a different statement from "we could not find out". + warn!(%peer, "DescribeAcls asked on a connection whose permissions were never read"); + return respond_describe_acls_error(api_version, ERROR_UNKNOWN_SERVER_ERROR, false); + } + 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) } } } @@ -714,6 +786,12 @@ fn illegal_state_outcome( API_KEY_SASL_AUTHENTICATE => { 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 => { + respond_describe_acls_error(api_version, ERROR_ILLEGAL_SASL_STATE, !keep_open) + } _ => 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..c4b3b4109 --- /dev/null +++ b/gateways/kafka/tests/kafka_client_e2e_tests.rs @@ -0,0 +1,648 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! End-to-end tests driving **real Kafka clients** against the gateway. +//! +//! Every other suite in this crate hand-builds wire frames, which cannot catch a client +//! compatibility problem by construction. One already slipped through that way: advertising +//! `SaslHandshake` from v1 rather than v0 passed every hand-built test and made librdkafka report +//! "SASL Handshake not supported by broker" before it sent anything. These tests exist so that +//! class of bug fails in CI instead of during manual testing. +//! +//! The stack is real on all three sides: a spawned `iggy-server` process, the gateway in-process +//! with a real `IggyAuthenticator`, and a client from a container. They automate categories S and T +//! of `docs/MANUAL_TESTING.md`. +//! +//! Prerequisites are Docker and an already-built `iggy-server`. Missing either skips, the way the +//! wire-fixture suites do, unless `KAFKA_E2E_REQUIRED=1` is set, which turns a skip into a failure +//! so a broken CI step cannot leave these silently green. + +use std::net::SocketAddr; +use std::path::PathBuf; +use std::process::{Child, Command, Stdio}; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use iggy_gateway_kafka::GatewayConfig; +use iggy_gateway_kafka::auth::IggyAuthenticator; + +#[path = "common/server.rs"] +mod server; + +use server::spawn_test_server_with_authenticator; + +const KCAT_IMAGE: &str = "edenhill/kcat:1.7.1"; +const KAFKA_IMAGE: &str = "apache/kafka:3.9.0"; +const ROOT_USER: &str = "iggy"; +const ROOT_PASSWORD: &str = "iggy"; +/// Password for every non-root principal these tests create. +const USER_PASSWORD: &str = "s3cretpass"; + +/// Budget for `iggy-server` to start listening. Generous: it is a cold process start, and a debug +/// build on a loaded machine is not quick. +const SERVER_READY_TIMEOUT: Duration = Duration::from_secs(45); + +/// Reports why the suite cannot run, and whether that is fatal. +/// +/// Returns `true` when the caller should skip. `KAFKA_E2E_REQUIRED=1` makes it panic instead, so a +/// CI job that means to run these fails loudly rather than reporting a pass over zero assertions. +fn skip(reason: &str) -> bool { + assert!( + std::env::var("KAFKA_E2E_REQUIRED").as_deref() != Ok("1"), + "KAFKA_E2E_REQUIRED=1 but the suite cannot run: {reason}" + ); + eprintln!("skipping real-client end-to-end test: {reason}"); + true +} + +fn docker_missing() -> bool { + let available = Command::new("docker") + .arg("info") + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status() + .is_ok_and(|status| status.success()); + if available { + false + } else { + skip("docker is unavailable") + } +} + +/// Locates the already-built `iggy-server` alongside this test binary. Does not build it. +fn iggy_server_binary() -> Option<PathBuf> { + let mut dir = std::env::current_exe().ok()?; + // .../target/<profile>/deps/<test binary> -> .../target/<profile> + dir.pop(); + dir.pop(); + let candidate = dir.join(format!("iggy-server{}", std::env::consts::EXE_SUFFIX)); + candidate.is_file().then_some(candidate) +} + +/// Outcome of waiting for the spawned server to come up. +enum Ready { + Listening, + /// The process exited before it bound, carrying its status. + Exited(String), + TimedOut, +} + +/// A spawned `iggy-server`, killed on drop. +/// +/// The environment recipe mirrors `bridge_iggy_integration_tests.rs`, which explains each setting: +/// the other listeners are off so concurrently spawned servers do not fight over their fixed +/// ports, and the shard pool is capped so a spawned server does not size itself to the whole +/// machine against unrelated packages' tests in the same run. +struct TestServer { + child: Child, + address: String, + http_address: String, + _data_dir: tempfile::TempDir, +} + +impl TestServer { + fn spawn() -> Result<Self, String> { + let Some(binary) = iggy_server_binary() else { + return Err( + "iggy-server is not built; run `cargo build --bin iggy-server` first".to_string(), + ); + }; + let data_dir = tempfile::tempdir().expect("create server data dir"); + let port = free_port(); + let address = format!("127.0.0.1:{port}"); + let http_address = format!("127.0.0.1:{}", free_port()); + + let child = Command::new(binary) + .arg("--fresh") + .env("IGGY_PATH", data_dir.path().display().to_string()) + .env("IGGY_TCP_ADDRESS", &address) + // Left on, unlike the sibling suite: the three-principal procedure creates its + // non-root users over the HTTP API, which is the only administrative surface reachable + // from a test without pulling in the SDK. + .env("IGGY_HTTP_ENABLED", "true") + .env("IGGY_HTTP_ADDRESS", &http_address) + .env("IGGY_QUIC_ENABLED", "false") + .env("IGGY_WEBSOCKET_ENABLED", "false") + .env("IGGY_SHARDING_PIN_CORES", "false") + .env("IGGY_SHARDING_CPU_ALLOCATION", "0..4") + .env("IGGY_ROOT_USERNAME", ROOT_USER) + .env("IGGY_ROOT_PASSWORD", ROOT_PASSWORD) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("spawn iggy-server"); + + let mut server = Self { + child, + address, + http_address, + _data_dir: data_dir, + }; + match server.wait_ready() { + Ready::Listening => Ok(server), + Ready::Exited(status) => Err(format!( + "iggy-server exited during startup with {status}; its output is suppressed, so \ + rerun it by hand with the same environment to see why" + )), + Ready::TimedOut => Err(format!( + "iggy-server never bound {} within {SERVER_READY_TIMEOUT:?}", + server.address + )), + } + } + + fn wait_ready(&mut self) -> Ready { + let deadline = Instant::now() + SERVER_READY_TIMEOUT; + while Instant::now() < deadline { + // Both listeners, not just the data one: the server binds HTTP after TCP, so probing + // TCP alone declares readiness while the provisioning calls that follow would still + // be refused. + if std::net::TcpStream::connect(&self.address).is_ok() + && std::net::TcpStream::connect(&self.http_address).is_ok() + { + return Ready::Listening; + } + // Without this a server that dies at boot burns the whole budget and is then reported + // as a timeout, which sends the reader looking at the wrong thing entirely. + if let Ok(Some(status)) = self.child.try_wait() { + return Ready::Exited(status.to_string()); + } + std::thread::sleep(Duration::from_millis(200)); + } + Ready::TimedOut + } +} + +impl Drop for TestServer { + fn drop(&mut self) { + let _ = self.child.kill(); + let _ = self.child.wait(); + } +} + +/// Binds an ephemeral port and releases it, so the server can take it. +/// +/// A window exists between release and rebind. Acceptable here because this suite is serialized +/// into its own nextest group, so nothing else in it is drawing ports concurrently. +fn free_port() -> u16 { + let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind ephemeral port"); + listener.local_addr().expect("local addr").port() +} + +/// Starts the gateway with SASL on, verifying credentials against `iggy_address`. +async fn spawn_gateway(iggy_address: &str) -> SocketAddr { + let config = GatewayConfig { + sasl_enabled: true, + ..GatewayConfig::default() + }; + let authenticator = Arc::new(IggyAuthenticator::new(iggy_address.to_string())); + let (addr, shutdown) = spawn_test_server_with_authenticator(config, authenticator).await; + // Held for the test's lifetime: dropping the sender shuts the gateway down mid-exchange. + std::mem::forget(shutdown); + addr +} + +/// Runs a container against the host network and returns its combined output. +/// +/// `--network host` is what lets a containerised client reach a gateway bound to the host's +/// loopback. It is Linux-specific, which matches where CI runs. +/// +/// This blocks the calling thread for the life of the container, which is why every test here uses +/// a multi-threaded runtime: the gateway runs as a spawned task, and on the single-threaded runtime +/// `#[tokio::test]` gives by default, this call would starve it and nothing would ever listen. +fn run_client(image: &str, args: &[&str], mounts: &[(&str, &str)]) -> ClientRun { + let mut command = Command::new("docker"); + command.args(["run", "--rm", "--network", "host"]); + for (host, guest) in mounts { + command.args(["-v", &format!("{host}:{guest}:ro")]); + } + command.arg(image).args(args); + let output = command.output().expect("run client container"); + ClientRun { + succeeded: output.status.success(), + text: format!( + "{}{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ), + } +} + +/// What a client container produced, and whether it ran at all. +/// +/// The exit status is carried deliberately. Every negative assertion in this suite is of the form +/// "the output does not contain X", and a container that never started produces output satisfying +/// all of them. Without this a wrong image tag, a malformed config or a Docker failure reads as a +/// passing test. +struct ClientRun { + succeeded: bool, + text: String, +} + +impl ClientRun { + /// Asserts the client itself ran, then hands back its output for content assertions. + fn expect_ran(self, what: &str) -> String { + assert!( + self.succeeded, + "{what}: the client container failed to run, so any assertion about its output would \ + be meaningless. Output: {}", + self.text + ); + self.text + } + + /// For the cases where the client is *expected* to fail, so only the output matters. + fn text(self) -> String { + self.text + } +} + +fn kcat(addr: SocketAddr, username: &str, password: &str, mechanism: &str) -> ClientRun { + run_client( + KCAT_IMAGE, + &[ + "-b", + &addr.to_string(), + "-X", + "security.protocol=SASL_PLAINTEXT", + "-X", + &format!("sasl.mechanisms={mechanism}"), + "-X", + &format!("sasl.username={username}"), + "-X", + &format!("sasl.password={password}"), + "-L", + ], + &[], + ) +} + +/// Creates an Iggy user over the HTTP API with the given global permissions. +/// +/// The spec's three-principal procedure needs non-root accounts, and root holds every flag, so a +/// suite that only ever authenticates as root checks neither the inheritance projection nor the +/// empty view against a real client. +fn create_user(http: &str, token: &str, username: &str, permissions: &str) { + let body = format!( + r#"{{"username":"{username}","password":"{USER_PASSWORD}","status":"active","permissions":{permissions}}}"# + ); + let status = Command::new("curl") + .args([ + "-s", + "-o", + "/dev/null", + "-w", + "%{http_code}", + "-X", + "POST", + &format!("http://{http}/users";), + "-H", + &format!("Authorization: Bearer {token}"), + "-H", + "Content-Type: application/json", + "-d", + &body, + ]) + .output() + .expect("create user"); + let code = String::from_utf8_lossy(&status.stdout).to_string(); + assert!( + code.starts_with('2'), + "creating {username} returned HTTP {code}" + ); +} + +/// Logs in as root over HTTP and returns the bearer token. +fn root_token(http: &str) -> String { + let output = Command::new("curl") + .args([ + "-s", + "-X", + "POST", + &format!("http://{http}/users/login";), + "-H", + "Content-Type: application/json", + "-d", + &format!(r#"{{"username":"{ROOT_USER}","password":"{ROOT_PASSWORD}"}}"#), + ]) + .output() + .expect("root login"); + let body = String::from_utf8_lossy(&output.stdout); + // Avoids a JSON dependency for one field: the token is the value after this key. + let key = "\"token\":\""; + let start = body.find(key).expect("login response carries a token") + key.len(); + let end = start + body[start..].find('"').expect("token is terminated"); + body[start..end].to_string() +} + +/// Writes a JAAS client config for the Java tools and returns its path, kept alive by the handle. +fn java_client_config(username: &str, password: &str) -> (tempfile::TempDir, PathBuf) { + let dir = tempfile::tempdir().expect("create config dir"); + let path = dir.path().join("client.properties"); + std::fs::write( + &path, + format!( + "security.protocol=SASL_PLAINTEXT\n\ + sasl.mechanism=PLAIN\n\ + sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \ + username=\"{username}\" password=\"{password}\";\n" + ), + ) + .expect("write client config"); + (dir, path) +} + +/// Brings up a server and a gateway, or reports why it could not. +async fn stack() -> Option<(TestServer, SocketAddr)> { + if docker_missing() { + return None; + } + let server = match TestServer::spawn() { + Ok(server) => server, + Err(reason) => { + skip(&reason); + return None; + } + }; + let gateway = spawn_gateway(&server.address).await; + Some((server, gateway)) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn given_valid_credentials_when_a_real_client_connects_should_serve_metadata() { + let Some((_server, gateway)) = stack().await else { + return; + }; + let output = kcat(gateway, ROOT_USER, ROOT_PASSWORD, "PLAIN").expect_ran("valid credentials"); + assert!( + output.contains("Metadata for all topics"), + "librdkafka must authenticate and receive metadata, got: {output}" + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn given_a_wrong_password_when_a_real_client_connects_should_report_an_auth_failure() { + let Some((_server, gateway)) = stack().await else { + return; + }; + // kcat exits non-zero here by design, so only its output is asserted. + let output = kcat(gateway, ROOT_USER, "definitely-not-the-password", "PLAIN").text(); + assert!( + output.contains("Authentication failed"), + "a rejected credential must reach the client as an auth failure, got: {output}" + ); + assert!( + !output.contains("Metadata for all topics"), + "nothing may be served to a rejected client" + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn given_an_unsupported_mechanism_should_name_the_supported_one_back() { + let Some((_server, gateway)) = stack().await else { + return; + }; + let output = kcat(gateway, ROOT_USER, ROOT_PASSWORD, "SCRAM-SHA-256").text(); + assert!( + output.contains("PLAIN"), + "the refusal must name what is supported or an operator cannot act on it, got: {output}" + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn given_no_credentials_when_a_real_client_connects_should_be_refused() { + let Some((_server, gateway)) = stack().await else { + return; + }; + // kcat exits non-zero here by design, so only its output is asserted, and the assertion has + // to be a positive one: a container that never started also fails to print the metadata this + // test forbids, which would pass over nothing. librdkafka only reports this particular + // diagnosis after it connected and the broker then dropped it before authenticating, so it + // stands in for the refusal itself. + let output = run_client(KCAT_IMAGE, &["-b", &gateway.to_string(), "-L"], &[]).text(); + assert!( + output.contains("broker might require SASL authentication"), + "an unauthenticated client must be disconnected by the broker, got: {output}" + ); + assert!( + !output.contains("Metadata for all topics"), + "an unauthenticated client must not be served, got: {output}" + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn given_the_java_client_should_authenticate_and_list_the_sasl_apis() { + // The Java client is the one that negotiates `SaslAuthenticate` v2 and sends `ApiVersions` + // both before and after authenticating, none of which kcat exercises identically. + let Some((_server, gateway)) = stack().await else { + return; + }; + let (_dir, config) = java_client_config(ROOT_USER, ROOT_PASSWORD); + let output = run_client( + KAFKA_IMAGE, + &[ + "/opt/kafka/bin/kafka-broker-api-versions.sh", + "--bootstrap-server", + &gateway.to_string(), + "--command-config", + "/tmp/client.properties", + ], + &[( + config.to_str().expect("config path is utf-8"), + "/tmp/client.properties", + )], + ) + .expect_ran("java client api-versions"); + assert!( + output.contains("SaslHandshake(17)"), + "the Java client must authenticate and read the advertisement, got: {output}" + ); + assert!( + output.contains("SaslAuthenticate(36)"), + "both SASL keys must be advertised while the feature is on, got: {output}" + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn given_distinct_principals_when_listing_acls_should_describe_each_differently() { + // The spec's category T procedure. Root alone proves nothing about the mapping, because it + // holds every flag: a fixed set would satisfy it. Each of the others is chosen to fail + // differently if the projection is wrong. The consume-only principal shows the projection + // distinguishes principals and applies Iggy's inheritance; the poll-only one is the single + // case that separates the group binding's real source from `poll_messages`; the produce-only + // one shows a write grant drags in neither a read nor a group; and the ungranted one shows an + // empty view is a successful answer rather than a failure. + let Some((server, gateway)) = stack().await else { + return; + }; + let token = root_token(&server.http_address); + create_user( + &server.http_address, + &token, + "consumer-only", + r#"{"global":{"manage_servers":false,"read_servers":false,"manage_users":false, + "read_users":false,"manage_streams":false,"read_streams":false,"manage_topics":false, + "read_topics":true,"poll_messages":false,"send_messages":false},"streams":null}"#, + ); + create_user(&server.http_address, &token, "no-grants", "null"); + // Discriminates the group-derivation fix. `poll_messages` alone leaves the projected topic + // read false, so Iggy would refuse this principal a consumer group, yet the old derivation + // rendered one from polling. Reverting that fix makes this principal's listing grow a GROUP + // section, which the assertion below catches. + create_user( + &server.http_address, + &token, + "poller-only", + r#"{"global":{"manage_servers":false,"read_servers":false,"manage_users":false, + "read_users":false,"manage_streams":false,"read_streams":false,"manage_topics":false, + "read_topics":false,"poll_messages":true,"send_messages":false},"streams":null}"#, + ); + create_user( + &server.http_address, + &token, + "producer-only", + r#"{"global":{"manage_servers":false,"read_servers":false,"manage_users":false, + "read_users":false,"manage_streams":false,"read_streams":false,"manage_topics":false, + "read_topics":false,"poll_messages":false,"send_messages":true},"streams":null}"#, + ); + + let root = list_acls(gateway, ROOT_USER, ROOT_PASSWORD); + assert!( + root.contains(&format!("principal=User:{ROOT_USER}")), + "every rendered binding names the authenticated principal, got: {root}" + ); + assert!( + root.contains("resourceType=TOPIC") && root.contains("operation=WRITE"), + "root holds every permission, so it must be described as able to write, 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 dbdb227e4..01f6da1a1 100644 --- a/gateways/kafka/tests/sasl_tests.rs +++ b/gateways/kafka/tests/sasl_tests.rs @@ -22,6 +22,7 @@ use std::net::SocketAddr; use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; use std::time::Duration; use async_trait::async_trait; @@ -30,7 +31,8 @@ use tokio::io::AsyncWriteExt; use tokio::net::TcpStream; 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::PlainCredentials; #[path = "common/codec.rs"] @@ -63,16 +65,24 @@ 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, + ) -> 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 +104,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 +170,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"); @@ -552,6 +659,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); @@ -573,6 +681,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); @@ -630,7 +739,10 @@ struct UnavailableAuthenticator; #[async_trait] impl SaslAuthenticator for UnavailableAuthenticator { - async fn authenticate(&self, _credentials: &PlainCredentials) -> Result<(), AuthError> { + async fn authenticate( + &self, + _credentials: &PlainCredentials, + ) -> Result<AuthenticatedPrincipal, AuthError> { Err(AuthError::Unavailable) } } @@ -806,6 +918,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, @@ -829,9 +942,16 @@ struct StallingAuthenticator { #[async_trait] impl SaslAuthenticator for StallingAuthenticator { - async fn authenticate(&self, _credentials: &PlainCredentials) -> Result<(), AuthError> { + async fn authenticate( + &self, + credentials: &PlainCredentials, + ) -> Result<AuthenticatedPrincipal, AuthError> { let _held = self.release.acquire().await; - Ok(()) + Ok(AuthenticatedPrincipal { + username: credentials.username.clone(), + permissions: PrincipalPermissions::default(), + permissions_known: true, + }) } } @@ -873,6 +993,85 @@ 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, + ) -> 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_pre_auth_request_above_the_firewall_should_close_rather_than_answer_it() { // `kafka_protocol`'s encoders reach further than this gateway's firewall, so encoding at the @@ -910,3 +1109,350 @@ async fn given_a_refused_first_api_versions_should_not_spend_the_single_allowanc "the downgrade retry must not be refused for having spent the allowance" ); } + +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" + ); +} + +#[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); +}
