This is an automated email from the ASF dual-hosted git repository. numinnex pushed a commit to branch mtls_kafka_proxy in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 2b987dc65eee028c43e9ba56a8356c207e96e14f Author: Grzegorz Koszyk <[email protected]> AuthorDate: Tue Oct 13 21:04:27 2026 +0200 feat(gateways): add TLS listeners and mTLS client identities to Kafka gateway --- Cargo.lock | 1 + Cargo.toml | 1 + gateways/kafka/Cargo.toml | 1 + gateways/kafka/README.md | 67 +- gateways/kafka/docs/AUTHENTICATION.md | 47 +- gateways/kafka/docs/MANUAL_TESTING.md | 9 +- gateways/kafka/docs/TEST_SUITE.md | 2 +- gateways/kafka/src/auth.rs | 19 +- gateways/kafka/src/bridge/config.rs | 15 + gateways/kafka/src/bridge/iggy_bridge/mod.rs | 6 + gateways/kafka/src/listener.rs | 34 +- gateways/kafka/src/main.rs | 144 ++-- gateways/kafka/src/protocol/api.rs | 5 +- .../kafka/src/protocol/handlers/alter_configs.rs | 13 +- .../kafka/src/protocol/handlers/delete_topics.rs | 11 +- gateways/kafka/src/server.rs | 188 +++-- gateways/kafka/src/tls/identity.rs | 790 ++++++++++++++++++--- gateways/kafka/src/tls/mod.rs | 145 +++- gateways/kafka/tests/common/certs.rs | 8 + .../kafka/tests/delete_topics_real_bridge_tests.rs | 136 +++- gateways/kafka/tests/tls_listener_tests.rs | 316 ++++++++- 21 files changed, 1627 insertions(+), 331 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index cf4b6d4e7..e3d2dfb09 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7123,6 +7123,7 @@ dependencies = [ "lz4", "rcgen", "regex", + "regex-syntax", "rustls", "secrecy", "serde", diff --git a/Cargo.toml b/Cargo.toml index 34b32f0d3..b6efbbf6e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -292,6 +292,7 @@ rand_xoshiro = "0.8.1" rayon = "1.12.0" rcgen = "0.14.10" regex = "1.13.1" +regex-syntax = "0.8.11" reqwest = { version = "0.13.5", default-features = false, features = ["json", "rustls"] } reqwest-middleware = { version = "0.5.2", features = ["json", "query"] } reqwest-retry = "0.9.1" diff --git a/gateways/kafka/Cargo.toml b/gateways/kafka/Cargo.toml index 0451b073b..f1093285d 100644 --- a/gateways/kafka/Cargo.toml +++ b/gateways/kafka/Cargo.toml @@ -48,6 +48,7 @@ kafka-protocol = { workspace = true, features = ["gzip", "lz4", "snappy", "zstd" libc = { workspace = true } lz4 = { workspace = true } regex = { workspace = true } +regex-syntax = { workspace = true } rustls = { workspace = true } secrecy = { workspace = true } serde = { workspace = true } diff --git a/gateways/kafka/README.md b/gateways/kafka/README.md index f7d05e126..6ba7425d6 100644 --- a/gateways/kafka/README.md +++ b/gateways/kafka/README.md @@ -45,18 +45,18 @@ Default bind: a `PLAINTEXT` listener on `127.0.0.1:9093`. Environment variables: | Variable | Default | Description | | --- | --- | --- | -| `IGGY_KAFKA_LISTENERS` | unset | Kafka-style listener list, `PROTOCOL://ip:port` entries separated by commas, one per security protocol: `PLAINTEXT`, `SSL`, `SASL_PLAINTEXT`, `SASL_SSL`. Takes precedence over the three single-listener address variables below (the image bakes `IGGY_KAFKA_BIND_ADDR` in, so they are ignored with a warning) and is refused beside `IGGY_KAFKA_SASL_ENABLED`, since the listener protocols say where SASL runs. See [Listeners and TLS](#listeners-and-tls-4510) | -| `IGGY_KAFKA_ADVERTISED_LISTENERS` | bind address of each listener | `PROTOCOL://host:port` per listener, the address clients are pointed back at. Required for any listener bound to `0.0.0.0`/`::` | +| `IGGY_KAFKA_LISTENERS` | unset | Kafka-style listener list, `PROTOCOL://ip:port` entries separated by commas, one per security protocol: `PLAINTEXT`, `SSL`, `SASL_PLAINTEXT`, `SASL_SSL`. An IPv6 address is bracketed: `SSL://[::1]:9094`. Takes precedence over `IGGY_KAFKA_BIND_ADDR`, which the image bakes in and which is ignored with a warning. Refused beside `IGGY_KAFKA_ADVERTISED_HOST`, `IGGY_KAFKA_ADVERTISED_PORT` (advertise through `IGGY_KAFKA_ADVERTISED_LISTENERS` instead) and `IGGY [...] +| `IGGY_KAFKA_ADVERTISED_LISTENERS` | bind address of each listener | `PROTOCOL://host:port` per listener, the address clients are pointed back at. An IPv6 host is bracketed: `SSL://[2001:db8::1]:9094`. Required for any listener bound to `0.0.0.0`/`::` | | `IGGY_KAFKA_BIND_ADDR` | `127.0.0.1:9093` | Single-listener form: TCP address to listen on | | `IGGY_KAFKA_ADVERTISED_HOST` | bind IP | Single-listener form: hostname/IP clients use to reach this broker (required when binding to `0.0.0.0`/`::`) | | `IGGY_KAFKA_ADVERTISED_PORT` | bind port | Single-listener form: port advertised in Metadata responses | | `IGGY_KAFKA_TLS_CERT_FILE` | unset | PEM certificate chain for every `SSL` and `SASL_SSL` listener, leaf first. Required when one exists, refused when none does | | `IGGY_KAFKA_TLS_KEY_FILE` | unset | PEM private key for the certificate (PKCS#8, PKCS#1 or SEC1) | -| `IGGY_KAFKA_TLS_CLIENT_AUTH` | `none` | Kafka's `ssl.client.auth`: `none`, `requested` or `required`. On an `SSL` listener a verified client certificate names the Iggy user; on `SASL_SSL` it only gates the transport | +| `IGGY_KAFKA_TLS_CLIENT_AUTH` | `none` | Kafka's `ssl.client.auth`: `none`, `requested` or `required`. Applies to `SSL` listeners, where a verified client certificate names the Iggy user. A `SASL_SSL` listener never asks for a client certificate, Kafka's default there. Refused without an `SSL` listener | | `IGGY_KAFKA_TLS_CLIENT_CA_FILE` | unset | PEM roots client certificates must chain to. Required with `requested` or `required`, refused with `none` | | `IGGY_KAFKA_TLS_PRINCIPAL_MAPPING_RULES` | `DEFAULT` | Kafka's `ssl.principal.mapping.rules`, turning a client certificate's subject into an Iggy username. Only with client authentication on an `SSL` listener | | `IGGY_KAFKA_TLS_RELOAD_INTERVAL_SECS` | `60` | How often the certificate, key and client CA files are re-read; a change goes live for new connections without a restart. `0` never re-reads them | -| `IGGY_KAFKA_MAX_CONNECTIONS` | `1024` | Maximum concurrent connections before new ones are rejected | +| `IGGY_KAFKA_MAX_CONNECTIONS` | `1024` | Maximum concurrent connections per listener before new ones are rejected. Each listener has its own allowance, so idle clients on one cannot lock clients out of another. Open connections can therefore reach the number of listeners times this value; size the file-descriptor limit (`ulimit -n`) for that | | `IGGY_KAFKA_MAX_FRAME_SIZE` | `8388608` | Maximum accepted request frame size in bytes | | `IGGY_KAFKA_IDLE_TIMEOUT_SECS` | `600` | Seconds a connection may sit idle before the next frame's length prefix arrives | | `IGGY_KAFKA_READ_TIMEOUT_SECS` | `15` | Seconds allowed to read a frame body once its length prefix arrives | @@ -262,11 +262,21 @@ IGGY_KAFKA_LISTENERS=PLAINTEXT://127.0.0.1:9093,SSL://0.0.0.0:9094,SASL_SSL://0. cargo run -p iggy-gateway-kafka ``` +Each security protocol can be bound once, since the protocol is also the listener's name. + TLS is rustls. One certificate serves every `SSL` and `SASL_SSL` listener, and the files are re-read every `IGGY_KAFKA_TLS_RELOAD_INTERVAL_SECS`: a rotated certificate takes effect for new -connections without a restart, a half-written rotation or a key that does not match leaves the -previous material serving and logs an error. There is no self-signed mode; mint a certificate -(`docs/MANUAL_TESTING.md` has an `openssl` recipe) and give the Kafka client the CA. +connections without a restart, a key that does not match leaves the previous material serving and +logs an error. Rotate by writing the new files elsewhere and renaming them into place: a chain +written in place can be read half written, and if what was read still parses and matches the key, +the truncated chain serves until the next interval. There is no self-signed mode; mint a +certificate (`docs/MANUAL_TESTING.md` has an `openssl` recipe) and give the Kafka client the CA. + +The certificate's subject alternative names must cover every host in +`IGGY_KAFKA_ADVERTISED_LISTENERS`: a client verifies the name of each broker address Metadata +hands it, so a host missing from the certificate fails the client's next connection, not the +first. Client certificates are checked against the CA only; there is no CRL or OCSP revocation +checking, so revoke access by deleting or deactivating the Iggy user a certificate maps to. ```bash kcat -b kafka.example.com:9094 -X security.protocol=SSL -X ssl.ca.location=ca.pem -L @@ -276,23 +286,36 @@ kcat -b kafka.example.com:9095 -X security.protocol=SASL_SSL -X ssl.ca.location= ### Client certificates (mTLS) -`IGGY_KAFKA_TLS_CLIENT_AUTH=required` (or `requested`) makes every TLS listener verify client -certificates against `IGGY_KAFKA_TLS_CLIENT_CA_FILE`. What a verified certificate means depends on -the listener, as it does on a Kafka broker: - -- On an `SSL` listener the certificate names the principal. Its subject, rendered the way Java - does (`CN=alice,OU=eng,O=Example`), goes through `IGGY_KAFKA_TLS_PRINCIPAL_MAPPING_RULES`, and - the result must be the name of an active Iggy user. The gateway looks the user up through the - bridge's own session, so the bridge is required and its Iggy user needs `read_users`. A - certificate naming no such user is closed after the handshake, with the name in the gateway's - log at `warn`. With `requested`, a client that presents no certificate is Kafka's `ANONYMOUS` - principal and is served as the bridge's user, exactly as on a plaintext listener. -- On a `SASL_SSL` listener the certificate only gates the transport. SASL names the principal and - the subject is never mapped. +`IGGY_KAFKA_TLS_CLIENT_AUTH=required` (or `requested`) makes every `SSL` listener verify client +certificates against `IGGY_KAFKA_TLS_CLIENT_CA_FILE`, and the certificate names the principal +there. Its subject, rendered the way Java does (`CN=alice,OU=eng,O=Example`), goes through +`IGGY_KAFKA_TLS_PRINCIPAL_MAPPING_RULES`, and the result must be the name of an active Iggy user. +The gateway looks the user up through the bridge's own session, so the bridge is required and its +Iggy user needs `read_users`. A certificate whose subject no rule maps, or that names no such user, +is closed after the handshake, with the subject or name in the gateway's log at `warn`. + +A `SASL_SSL` listener never asks for a client certificate, which is Kafka's default for that +protocol: SASL names the principal there. + +With `requested`, a client that presents no certificate is Kafka's `ANONYMOUS` principal, which +holds no permissions. Once any listener names principals (a SASL listener, or an `SSL` listener +with client authentication), every connection that proved no identity is `ANONYMOUS`, on a +`PLAINTEXT` listener beside it too. The requests the gateway checks against a principal, +`DeleteTopics` and `AlterConfigs`, need `manage_topics` and are refused for it; everything else is +served as the bridge's user. A gateway whose listeners name nobody serves every request as the +bridge's user. The rules are Kafka's syntax, tried in order: `RULE:pattern/replacement/[LU]` applies when the -pattern matches the whole subject, `DEFAULT` keeps the subject as is. A raw subject is not a valid -Iggy username, so a deployment that maps identities sets at least one rule: +pattern matches the whole subject, `DEFAULT` keeps the subject as is. Iggy accepts any username of +3 to 50 characters, but a raw subject is usually longer than that, and is rarely the name the +operator created, so a deployment that maps identities sets at least one rule. Patterns are a +Java-compatible subset: `\d \s \w` match ASCII only and `.` stops at Java's line terminators, as on +a broker, while `(?i)`, the `u` and `U` flags, `\p{..}`, POSIX classes and `\b` are refused at +startup ([docs/AUTHENTICATION.md](docs/AUTHENTICATION.md) says why). A replacement that +names a group the pattern does not have is refused at startup. That is stricter than Kafka, which +writes such a `$N` literally: `RULE:^CN=(.*)$/$2/` maps to `$2` on a broker. The reverse case, +`/$1\$2/`, makes Kafka throw while the gateway maps `alice` to `alice$2`, which is harmless because +that Iggy user must still exist: ```bash IGGY_KAFKA_TLS_PRINCIPAL_MAPPING_RULES='RULE:^CN=([^,]+),OU=services.*$/svc-$1/L,RULE:^CN=([^,]+).*$/$1/' diff --git a/gateways/kafka/docs/AUTHENTICATION.md b/gateways/kafka/docs/AUTHENTICATION.md index 643642ba0..a217cb83c 100644 --- a/gateways/kafka/docs/AUTHENTICATION.md +++ b/gateways/kafka/docs/AUTHENTICATION.md @@ -227,24 +227,41 @@ Iggy password is capped at 100 bytes with a `u8` length prefix on the wire ## Client certificates -`IGGY_KAFKA_TLS_CLIENT_AUTH` is Kafka's `ssl.client.auth`: `none`, `requested` or `required`. With it on, -every TLS listener verifies the client certificate against `IGGY_KAFKA_TLS_CLIENT_CA_FILE` during the -handshake, and what a verified certificate means follows the listener, as it does on a Kafka broker. +`IGGY_KAFKA_TLS_CLIENT_AUTH` is Kafka's `ssl.client.auth`: `none`, `requested` or `required`. It applies +to `SSL` listeners, which verify the client certificate against `IGGY_KAFKA_TLS_CLIENT_CA_FILE` during the +handshake, and is refused when no `SSL` listener exists. -On a `SASL_SSL` listener the certificate gates the transport only. SASL names the principal, and the -subject is never mapped: Kafka does the same, and a deployment that wants certificate identities uses -`SSL`. +A `SASL_SSL` listener never asks for a client certificate. Kafka applies `ssl.client.auth` to a +`SASL_SSL` listener only when that listener overrides it, and leaves it at `none` otherwise; SASL names +the principal there. A deployment that wants certificate identities uses `SSL`. On an `SSL` listener the certificate names the principal, in three steps the connection takes before its first frame: 1. **Render the subject** the way Java's `X500Principal.getName()` does, RFC 2253 with the most specific RDN first and no spaces: `CN=alice,OU=eng,O=Example`. Operators write mapping rules against Java's - rendering, so the gateway matches it, keyword set and escaping included. + rendering, so the gateway follows JDK 11 to 21's `AVA.toRFC2253String`: its keyword set, its escaping + (NUL as `\00`), Latin-1 for `PrintableString`, `IA5String` and `GeneralString`, and hex for the string + types Java does not decode, `UniversalString` among them. Malformed UTF-8 or BMP values are the one case + not checked against Java. 2. **Map it** through `IGGY_KAFKA_TLS_PRINCIPAL_MAPPING_RULES`, Kafka's `ssl.principal.mapping.rules` syntax: `RULE:pattern/replacement/[LU]` applies when the pattern matches the whole subject, `DEFAULT` - keeps it, the first match wins, and a subject no rule matches closes the connection. The default is - `DEFAULT`, which is not a valid Iggy username, so a deployment sets at least one rule. + keeps it, the first match wins, and a subject no rule matches closes the connection with the subject in + the log at `warn`. The default is `DEFAULT`. Iggy only requires a username of 3 to 50 characters, but a + raw subject is usually longer than that and is rarely the name the operator created, so a deployment + sets at least one rule. A replacement with a bare `$`, or a `${name}` with no such named group, is + refused at startup, where Kafka would fail the first handshake it applies to. One that names a group + the pattern lacks is refused too, which is stricter than Kafka: a broker writes it literally, so + `RULE:^CN=(.*)$/$2/` maps to `$2`. The reverse case, `/$1\$2/`, makes Kafka throw while the gateway + maps `alice` to `alice$2`, which is harmless because that Iggy user must still exist. The replacement is applied as Java's `replaceAll`, to every match in the subject, so + `RULE:(.*)/$1@x/` maps `alice` to `alice@x@x` as it does on a broker. Patterns are the subset of + Java's regex syntax whose meaning the gateway can reproduce: `\d \s \w` and their negations match + ASCII only, and `.` outside `(?s)` stops at `\r`, U+0085, U+2028 and U+2029 as well as `\n`, both as + in Java. Rust would read them over Unicode, so a CA-signed `CN=\u212Aafka` (Kelvin sign) would match + `\w+` and lowercase to `kafka`. Case-insensitive matching (`(?i)`), the `u` and `U` flags, `\p{..}` + and `\P{..}`, POSIX classes such as `[[:alpha:]]`, and word boundaries (`\b`, `\B`) are refused at + startup with the construct named: each has no faithful translation. Non-ASCII names still map, so + `CN=Zoë` under `RULE:^CN=([^,]+).*$/$1/` maps to `Zoë` as on a broker. 3. **Look the name up** through the bridge's own session. The user must exist and be active, or the connection is closed after the handshake with the name in the log at `warn`: a certificate the CA signed for an account that does not exist is a provisioning mistake, not something the client can correct. The @@ -252,9 +269,15 @@ first frame: lookup, including one refused for want of that permission, closes the connection too and never admits the name unchecked. -With `requested`, a client that presents no certificate is Kafka's `ANONYMOUS` principal: served as the -bridge's user, the way a plaintext listener serves everyone, and described by `DescribeAcls` as holding -nothing. With `required`, rustls fails the handshake instead. +With `requested`, a client that presents no certificate is Kafka's `ANONYMOUS` principal, which holds no +permissions: `DescribeAcls` describes it as holding nothing, and `DeleteTopics` and `AlterConfigs`, which +need `manage_topics`, are refused. Everything else is served as the bridge's user. With `required`, rustls +fails the handshake instead. + +The same holds on every listener that names nobody once any listener names principals: a `PLAINTEXT` +listener beside a `SASL_SSL` or mTLS one serves `ANONYMOUS`, so it cannot do what an authenticated user +without `manage_topics` cannot. Only a gateway whose listeners all name nobody leaves those requests +ungated. The principal a certificate yields is the same `AuthenticatedPrincipal` a SASL login yields, with its permissions read at admission, so everything written about the SASL identity above holds: it is a snapshot, diff --git a/gateways/kafka/docs/MANUAL_TESTING.md b/gateways/kafka/docs/MANUAL_TESTING.md index 9354ad493..568c75ace 100644 --- a/gateways/kafka/docs/MANUAL_TESTING.md +++ b/gateways/kafka/docs/MANUAL_TESTING.md @@ -202,7 +202,8 @@ done ``` Then a gateway with every listener, bridged (the bridge looks certificate identities up), with client -certificates requested and the common name as the Iggy username: +certificates requested on the `SSL` listener and the common name as the Iggy username. The `SASL_SSL` +listener asks for no client certificate whatever `IGGY_KAFKA_TLS_CLIENT_AUTH` says: ```bash IGGY_KAFKA_LISTENERS=PLAINTEXT://127.0.0.1:9093,SSL://127.0.0.1:9094,SASL_SSL://127.0.0.1:9095 \ @@ -303,9 +304,9 @@ func main() { } ``` -Run it four ways, with `CA_FILE=/tmp/certs/ca.pem` exported: `go run . 127.0.0.1:19094`, -`SASL_USER=iggy SASL_PASS=iggy go run . 127.0.0.1:19095`, -`go run . 127.0.0.1:19094 /tmp/certs/iggy.pem /tmp/certs/iggy.key` and the same with +Run it four ways, with `CA_FILE=/tmp/certs/ca.pem` exported: `go run . 127.0.0.1:9094`, +`SASL_USER=iggy SASL_PASS=iggy go run . 127.0.0.1:9095`, +`go run . 127.0.0.1:9094 /tmp/certs/iggy.pem /tmp/certs/iggy.key` and the same with `nobody.pem`/`nobody.key`. The first three print the one broker on the port they dialed, the SASL and certificate runs also list `User:iggy`'s bindings, and the last reports the broker closing the connection during `ApiVersions`. diff --git a/gateways/kafka/docs/TEST_SUITE.md b/gateways/kafka/docs/TEST_SUITE.md index ff2508361..5b6178c1a 100644 --- a/gateways/kafka/docs/TEST_SUITE.md +++ b/gateways/kafka/docs/TEST_SUITE.md @@ -78,7 +78,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 | -| [`tls_listener_tests.rs`](../tests/tls_listener_tests.rs) | The `SSL` and `SASL_SSL` listeners over a real socket with certificates minted per test: each listener advertising its own port, a plaintext client refused on the TLS port, SASL inside TLS, client certificates mapped to users through a stub `PrincipalResolver` (`required`, `requested`, unknown user, SASL precedence), certificate reload without a restart, and the startup refusals | No | +| [`tls_listener_tests.rs`](../tests/tls_listener_tests.rs) | The `SSL` and `SASL_SSL` listeners over a real socket with certificates minted per test: each listener advertising its own port, a plaintext client refused on the TLS port, SASL inside TLS, client certificates mapped to users through a stub `PrincipalResolver` (`required`, `requested`, unknown user, a subject no rule maps, a failing or hung lookup, `SASL_SSL` asking for no certificate), a multi-MiB response to a slow reader, t [...] | [`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. With a bridge, kcat writes records, and kcat and the Java consumer read them back through Fetch. Java consumers in one group resume from their commits, across a gateway restart and a member leaving. librdkafka and the Java client connect with `security.pro [...] | [`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, and the bridge dialing a TLS-only server over TLS | No (needs the `iggy-server` binary - see Prerequisites) | | [`list_offsets_real_bridge_tests.rs`](../tests/list_offsets_real_bridge_tests.rs) | ListOffsets (key 2) through the whole handler against a real, spawned `iggy-server`: LATEST, EARLIEST, codes 3 and 43 | No (needs the `iggy-server` binary - see Prerequisites) | diff --git a/gateways/kafka/src/auth.rs b/gateways/kafka/src/auth.rs index b10742091..97eee2e4e 100644 --- a/gateways/kafka/src/auth.rs +++ b/gateways/kafka/src/auth.rs @@ -215,6 +215,23 @@ pub struct AuthenticatedPrincipal { pub permissions_known: bool, } +impl AuthenticatedPrincipal { + /// Kafka's name for a client that proved no identity, as `DescribeAcls` renders it. + pub const ANONYMOUS: &'static str = "ANONYMOUS"; + + /// A connection that proved no identity on a gateway that authenticates somewhere: it holds + /// nothing, and that is a real answer rather than a failed read. It is "authenticated" only in + /// the sense that the connection's identity has been decided. + #[must_use] + pub fn anonymous() -> Self { + Self { + username: Self::ANONYMOUS.to_string(), + permissions: PrincipalPermissions::default(), + permissions_known: true, + } + } +} + /// Verifies Kafka-supplied credentials. /// /// A trait rather than a concrete type so the protocol tests can drive the whole SASL exchange @@ -246,7 +263,7 @@ pub trait SaslAuthenticator: Send + Sync + std::fmt::Debug { /// over its own session; the protocol tests drive the certificate path without an Iggy server /// through a stub. #[async_trait] -pub trait PrincipalResolver: Send + Sync { +pub trait PrincipalResolver: Send + Sync + std::fmt::Debug { /// Returns the principal when `username` names an active Iggy user. /// /// # Errors diff --git a/gateways/kafka/src/bridge/config.rs b/gateways/kafka/src/bridge/config.rs index ed66894a5..7c5e39874 100644 --- a/gateways/kafka/src/bridge/config.rs +++ b/gateways/kafka/src/bridge/config.rs @@ -244,6 +244,21 @@ mod tests { use serial_test::serial; use super::*; + use crate::auth::IggyAuthenticator; + + #[test] + fn given_the_iggy_tls_variables_should_be_listed_by_every_reader() { + for var in IggyTls::KNOWN_ENV_VARS { + assert!( + IggyBridgeConfig::KNOWN_ENV_VARS.contains(var), + "{var} is read by the bridge and must be in its list" + ); + assert!( + IggyAuthenticator::KNOWN_ENV_VARS.contains(var), + "{var} is read by the SASL credential check and must be in its list" + ); + } + } fn test_config() -> IggyBridgeConfig { IggyBridgeConfig { diff --git a/gateways/kafka/src/bridge/iggy_bridge/mod.rs b/gateways/kafka/src/bridge/iggy_bridge/mod.rs index 2fa920d87..b77f9e370 100644 --- a/gateways/kafka/src/bridge/iggy_bridge/mod.rs +++ b/gateways/kafka/src/bridge/iggy_bridge/mod.rs @@ -264,6 +264,12 @@ async fn connect_client(config: &IggyBridgeConfig) -> Result<IggyClient, BridgeE Ok(client) } +impl std::fmt::Debug for IggyBridge { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("IggyBridge").finish_non_exhaustive() + } +} + /// The bridge's own session answers for the users client certificates name. It runs as the /// bridge's Iggy user, so that account needs `read_users` for any certificate to be admitted. #[async_trait] diff --git a/gateways/kafka/src/listener.rs b/gateways/kafka/src/listener.rs index 1fe2c580c..221b66c8b 100644 --- a/gateways/kafka/src/listener.rs +++ b/gateways/kafka/src/listener.rs @@ -288,12 +288,18 @@ fn split_entry<'a>(var: &str, entry: &'a str) -> Result<(SecurityProtocol, &'a s } /// Splits `host:port`, with `[v6]:port` for an IPv6 literal, into a bracket-free host and the -/// port text. +/// port text. An unbracketed host containing `:` is refused, since its port cannot be told apart. fn split_host_port(address: &str) -> Option<(&str, &str)> { - let (host, port) = address.rsplit_once(':')?; - let host = host - .strip_prefix('[') - .map_or(host, |inner| inner.strip_suffix(']').unwrap_or(host)); + let (host, port) = if let Some(bracketed) = address.strip_prefix('[') { + let (host, rest) = bracketed.split_once(']')?; + (host, rest.strip_prefix(':')?) + } else { + let (host, port) = address.rsplit_once(':')?; + if host.contains(':') { + return None; + } + (host, port) + }; if host.is_empty() { return None; } @@ -389,6 +395,24 @@ mod tests { } } + #[test] + fn given_an_unbracketed_ipv6_advertised_host_should_refuse() { + let mut listeners = parse_listeners("SSL://0.0.0.0:9094").expect("valid"); + for raw in ["SSL://fe80::1", "SSL://fe80::1:9094", "SSL://[fe80::1]9094"] { + let error = apply_advertised_listeners(&mut listeners, raw) + .err() + .unwrap_or_else(|| panic!("'{raw}' must be rejected")) + .to_string(); + assert!( + error.contains("IGGY_KAFKA_ADVERTISED_LISTENERS"), + "'{raw}' rejection must name the variable, got: {error}" + ); + } + apply_advertised_listeners(&mut listeners, "SSL://[fe80::1]:9094") + .expect("a bracketed IPv6 literal is valid"); + assert_eq!(listeners[0].advertised_host.as_deref(), Some("fe80::1")); + } + #[test] fn given_a_wildcard_bind_without_an_advertised_host_should_refuse_and_name_the_listener() { let listener = ListenerConfig::new(SecurityProtocol::SaslSsl, "0.0.0.0:9095"); diff --git a/gateways/kafka/src/main.rs b/gateways/kafka/src/main.rs index 9a138da72..dfb24d6f5 100644 --- a/gateways/kafka/src/main.rs +++ b/gateways/kafka/src/main.rs @@ -291,39 +291,42 @@ fn load_config() -> Result<GatewayConfig, String> { Iggy cluster needs its own value, or they hand out colliding producer ids" ); } - reject_unread_iggy_tls(config.sasl_enabled(), bridge_enabled()?)?; + reject_unread_iggy_tls( + config.sasl_enabled(), + bridge_enabled()?, + env_var("IGGY_KAFKA_LISTENERS").is_some(), + )?; Ok(config) } /// The listeners, from `IGGY_KAFKA_LISTENERS` or from the single-listener variables. /// -/// With a listener list, the single-listener address variables are ignored with a warning rather -/// than refused: the gateway's own image bakes `IGGY_KAFKA_BIND_ADDR=0.0.0.0:9093` in, and -/// refusing it would make `IGGY_KAFKA_LISTENERS` unusable there. `IGGY_KAFKA_SASL_ENABLED` is -/// refused beside a list all the same. It is a security switch, and with a list the protocol of -/// each listener says whether it runs SASL, so a `true` left over from the single-listener form -/// would otherwise read as SASL on while nothing ran it. +/// With a listener list, `IGGY_KAFKA_BIND_ADDR` is ignored with a warning rather than refused: +/// the gateway's own image bakes `IGGY_KAFKA_BIND_ADDR=0.0.0.0:9093` in, and refusing it would +/// make `IGGY_KAFKA_LISTENERS` unusable there. Every other single-listener variable is refused +/// beside a list, since ignoring it would advertise or authenticate differently from what its +/// operator set. fn load_listeners() -> Result<Vec<ListenerConfig>, String> { - let addressing: Vec<&str> = SINGLE_LISTENER_ENV_VARS - .iter() - .copied() - .filter(|var| *var != "IGGY_KAFKA_SASL_ENABLED" && env_var(var).is_some()) - .collect(); let advertised = env_var("IGGY_KAFKA_ADVERTISED_LISTENERS"); if let Some(raw) = env_var("IGGY_KAFKA_LISTENERS") { - if env_var("IGGY_KAFKA_SASL_ENABLED").is_some() { - return Err( - "IGGY_KAFKA_LISTENERS and IGGY_KAFKA_SASL_ENABLED are both set; with a listener \ - list, a SASL_PLAINTEXT or SASL_SSL listener is what turns SASL on" - .to_string(), - ); + let refused: Vec<&str> = SINGLE_LISTENER_ENV_VARS + .iter() + .copied() + .filter(|var| *var != "IGGY_KAFKA_BIND_ADDR" && env_var(var).is_some()) + .collect(); + if !refused.is_empty() { + return Err(format!( + "IGGY_KAFKA_LISTENERS is set together with {}, which belong to the \ + single-listener form; advertise through IGGY_KAFKA_ADVERTISED_LISTENERS, and \ + turn SASL on with a SASL_PLAINTEXT or SASL_SSL listener", + refused.join(", ") + )); } - if !addressing.is_empty() { + if env_var("IGGY_KAFKA_BIND_ADDR").is_some() { warn!( - "{} set but IGGY_KAFKA_LISTENERS takes precedence; they are ignored (bind and \ - advertise through IGGY_KAFKA_LISTENERS and IGGY_KAFKA_ADVERTISED_LISTENERS)", - addressing.join(", ") + "IGGY_KAFKA_BIND_ADDR set but IGGY_KAFKA_LISTENERS takes precedence; it is \ + ignored (bind through IGGY_KAFKA_LISTENERS)" ); } let mut listeners = parse_listeners(&raw).map_err(|e| e.to_string())?; @@ -398,6 +401,16 @@ fn load_tls(listeners: &[ListenerConfig]) -> Result<Option<TlsConfig>, String> { .map_err(|e| format!("invalid IGGY_KAFKA_TLS_CLIENT_AUTH `{raw}`: {e}"))?, None => ClientAuth::None, }; + let has_ssl_listener = listeners + .iter() + .any(|listener| listener.protocol.authenticates_by_certificate()); + if client_auth.is_enabled() && !has_ssl_listener { + return Err(format!( + "IGGY_KAFKA_TLS_CLIENT_AUTH is {client_auth} but no SSL listener is configured; it \ + applies to SSL listeners only, and a SASL_SSL listener asks for no client \ + certificate" + )); + } match ( client_auth.is_enabled(), env_var("IGGY_KAFKA_TLS_CLIENT_CA_FILE"), @@ -419,11 +432,7 @@ fn load_tls(listeners: &[ListenerConfig]) -> Result<Option<TlsConfig>, String> { (false, None) => {} } if let Some(raw) = env_var("IGGY_KAFKA_TLS_PRINCIPAL_MAPPING_RULES") { - let maps_identities = client_auth.is_enabled() - && listeners - .iter() - .any(|listener| listener.protocol.authenticates_by_certificate()); - if !maps_identities { + if !client_auth.is_enabled() { return Err( "IGGY_KAFKA_TLS_PRINCIPAL_MAPPING_RULES is set but no SSL listener verifies \ client certificates; the rules apply to an SSL listener with \ @@ -452,7 +461,11 @@ fn load_tls(listeners: &[ListenerConfig]) -> Result<Option<TlsConfig>, String> { /// The bridge and the SASL credential check are the two things that dial Iggy. With both off, /// accepting the switches lets an operator believe the link to Iggy is encrypted when there is no /// link, the same mistake `IggyTls::from_env` refuses to start for. -fn reject_unread_iggy_tls(sasl_enabled: bool, bridge_enabled: bool) -> Result<(), String> { +fn reject_unread_iggy_tls( + sasl_enabled: bool, + bridge_enabled: bool, + listener_list: bool, +) -> Result<(), String> { if sasl_enabled || bridge_enabled { return Ok(()); } @@ -464,9 +477,14 @@ fn reject_unread_iggy_tls(sasl_enabled: bool, bridge_enabled: bool) -> Result<() if set.is_empty() { return Ok(()); } + let sasl_switch = if listener_list { + "no listener in IGGY_KAFKA_LISTENERS runs SASL" + } else { + "IGGY_KAFKA_SASL_ENABLED is not true" + }; Err(format!( - "{} set but neither IGGY_KAFKA_SASL_ENABLED nor IGGY_KAFKA_BRIDGE_ENABLED is true; only \ - the SASL credential check and the bridge dial Iggy, so nothing would be encrypted", + "{} set but {sasl_switch} and IGGY_KAFKA_BRIDGE_ENABLED is not true; only the SASL \ + credential check and the bridge dial Iggy, so nothing would be encrypted", set.join(", ") )) } @@ -662,15 +680,25 @@ mod tests { unsafe { std::env::set_var("IGGY_KAFKA_IGGY_TLS_ENABLED", "true"); } - let nothing_reads_it = reject_unread_iggy_tls(false, false); - let sasl_reads_it = reject_unread_iggy_tls(true, false); - let bridge_reads_it = reject_unread_iggy_tls(false, true); + let nothing_reads_it = reject_unread_iggy_tls(false, false, false); + let nothing_reads_it_in_a_list = reject_unread_iggy_tls(false, false, true); + let sasl_reads_it = reject_unread_iggy_tls(true, false, false); + let bridge_reads_it = reject_unread_iggy_tls(false, true, false); unsafe { std::env::remove_var("IGGY_KAFKA_IGGY_TLS_ENABLED"); } assert!( - nothing_reads_it.is_err(), - "nothing reads the TLS switch with SASL and the bridge off, so accepting it hides that" + nothing_reads_it + .expect_err( + "nothing reads the TLS switch with SASL and the bridge off, so accepting it \ + hides that" + ) + .contains("IGGY_KAFKA_SASL_ENABLED") + ); + let in_a_list = nothing_reads_it_in_a_list.expect_err("refused with a listener list too"); + assert!( + !in_a_list.contains("IGGY_KAFKA_SASL_ENABLED"), + "a listener list refuses that switch, so naming it misleads, got: {in_a_list}" ); assert!(sasl_reads_it.is_ok()); assert!(bridge_reads_it.is_ok()); @@ -706,14 +734,31 @@ mod tests { #[test] #[serial] - fn given_both_listener_forms_should_refuse_the_sasl_switch_and_ignore_the_addresses() { - let error = load_with(&[ - ("IGGY_KAFKA_LISTENERS", "PLAINTEXT://127.0.0.1:9093"), - ("IGGY_KAFKA_SASL_ENABLED", "true"), - ]) - .unwrap_err(); - assert!(error.contains("IGGY_KAFKA_LISTENERS"), "got: {error}"); - assert!(error.contains("IGGY_KAFKA_SASL_ENABLED"), "got: {error}"); + fn given_both_listener_forms_should_refuse_all_but_the_bind_address() { + for var in [ + "IGGY_KAFKA_SASL_ENABLED", + "IGGY_KAFKA_ADVERTISED_HOST", + "IGGY_KAFKA_ADVERTISED_PORT", + ] { + let value = if var == "IGGY_KAFKA_SASL_ENABLED" { + "true" + } else { + "9093" + }; + let error = load_with(&[ + ("IGGY_KAFKA_LISTENERS", "PLAINTEXT://127.0.0.1:9093"), + (var, value), + ]) + .unwrap_err(); + assert!(error.contains("IGGY_KAFKA_LISTENERS"), "got: {error}"); + assert!(error.contains(var), "got: {error}"); + if var != "IGGY_KAFKA_SASL_ENABLED" { + assert!( + error.contains("IGGY_KAFKA_ADVERTISED_LISTENERS"), + "got: {error}" + ); + } + } // The image bakes a bind address in, so a listener list has to win over it. let config = load_with(&[ @@ -813,6 +858,19 @@ mod tests { "got: {no_auth}" ); + let auth_without_ssl = load_with(&[ + ("IGGY_KAFKA_LISTENERS", "SASL_SSL://0.0.0.0:9095"), + ("IGGY_KAFKA_TLS_CERT_FILE", "/etc/iggy/kafka.pem"), + ("IGGY_KAFKA_TLS_KEY_FILE", "/etc/iggy/kafka.key"), + ("IGGY_KAFKA_TLS_CLIENT_AUTH", "required"), + ("IGGY_KAFKA_TLS_CLIENT_CA_FILE", "/etc/iggy/ca.pem"), + ]) + .unwrap_err(); + assert!( + auth_without_ssl.contains("no SSL listener"), + "client authentication applies to SSL listeners only, got: {auth_without_ssl}" + ); + let rules_without_auth = with(&[("IGGY_KAFKA_TLS_PRINCIPAL_MAPPING_RULES", "DEFAULT")]).unwrap_err(); assert!( diff --git a/gateways/kafka/src/protocol/api.rs b/gateways/kafka/src/protocol/api.rs index ef4f169cd..65c64fdec 100644 --- a/gateways/kafka/src/protocol/api.rs +++ b/gateways/kafka/src/protocol/api.rs @@ -483,8 +483,9 @@ pub async fn handle_request_bounded( /// [`handle_request_bounded`] for one request of `connection`. /// -/// `principal` is `None` when SASL is off, or for a key the connection reached before -/// authenticating - only `crate::server::route_frame` ever has a principal to pass. +/// `principal` is `None` when no listener of the gateway names principals, or for a key the +/// connection reached before authenticating. Only `crate::server::route_frame` ever has a +/// principal to pass. pub async fn handle_connection_request( state: &GatewayState, connection: &ConnectionState, diff --git a/gateways/kafka/src/protocol/handlers/alter_configs.rs b/gateways/kafka/src/protocol/handlers/alter_configs.rs index 8386db6f9..a7e52ceb6 100644 --- a/gateways/kafka/src/protocol/handlers/alter_configs.rs +++ b/gateways/kafka/src/protocol/handlers/alter_configs.rs @@ -123,10 +123,10 @@ pub async fn handle( ) } -/// Denies every resource when `principal` is known but is not allowed to alter topic -/// configuration, or when its permissions could not be read. `None` means SASL is off, which -/// this gateway treats as no principal to enforce against - the same choice the rest of this -/// gateway makes when authentication itself is not configured. +/// Denies every resource when `principal` is not allowed to alter topic configuration, or when +/// its permissions could not be read. `None` means no listener of this gateway names principals, +/// so there is nothing to enforce against; once one does, a connection that proved no identity +/// carries [`AuthenticatedPrincipal::anonymous`] instead and is denied here. fn authorize( principal: Option<&AuthenticatedPrincipal>, resources: &[AlterConfigsResource], @@ -144,9 +144,8 @@ fn authorize( "AlterConfigs asked on a connection whose permissions were never read; denying" ); } - let message = StrBytes::from_static_str( - "the authenticated principal is not authorized to alter topic configuration", - ); + let message = + StrBytes::from_static_str("the principal is not authorized to alter topic configuration"); Some( resources .iter() diff --git a/gateways/kafka/src/protocol/handlers/delete_topics.rs b/gateways/kafka/src/protocol/handlers/delete_topics.rs index bd41fd5ae..7d5318a0e 100644 --- a/gateways/kafka/src/protocol/handlers/delete_topics.rs +++ b/gateways/kafka/src/protocol/handlers/delete_topics.rs @@ -152,10 +152,10 @@ pub async fn handle( respond_or_close(encode_message(&resp, api_version, 256), "DeleteTopics") } -/// Denies every requested topic when `principal` is known but is not allowed to delete topic -/// data, or when its permissions could not be read. `None` means SASL is off, which this -/// gateway treats as no principal to enforce against - the same choice the rest of this -/// gateway makes when authentication itself is not configured. +/// Denies every requested topic when `principal` is not allowed to delete topic data, or when +/// its permissions could not be read. `None` means no listener of this gateway names principals, +/// so there is nothing to enforce against; once one does, a connection that proved no identity +/// carries [`AuthenticatedPrincipal::anonymous`] instead and is denied here. fn authorize( principal: Option<&AuthenticatedPrincipal>, topic_names: &[TopicName], @@ -173,8 +173,7 @@ fn authorize( "DeleteTopics asked on a connection whose permissions were never read; denying" ); } - let message = - StrBytes::from_static_str("the authenticated principal is not authorized to delete topics"); + let message = StrBytes::from_static_str("the principal is not authorized to delete topics"); Some( topic_names .iter() diff --git a/gateways/kafka/src/server.rs b/gateways/kafka/src/server.rs index 6000a8d54..3610a2690 100644 --- a/gateways/kafka/src/server.rs +++ b/gateways/kafka/src/server.rs @@ -25,7 +25,7 @@ use kafka_protocol::messages::RequestHeader; use kafka_protocol::protocol::Decodable; use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; use tokio::net::{TcpListener, TcpStream}; -use tokio::sync::{Semaphore, broadcast}; +use tokio::sync::{Semaphore, SemaphorePermit, broadcast}; use tokio::task::JoinSet; use tokio::time::{timeout, timeout_at}; use tokio_util::sync::CancellationToken; @@ -42,7 +42,6 @@ use crate::bridge::IggyBridge; use crate::error::{KafkaProtocolError, Result}; use crate::group::{GroupCoordinator, GroupCoordinatorConfig}; use crate::listener::{ListenerConfig, SecurityProtocol}; -use crate::protocol::acl::PrincipalPermissions; use crate::protocol::api::{ API_KEY_DESCRIBE_ACLS, API_KEY_SASL_AUTHENTICATE, API_KEY_SASL_HANDSHAKE, BrokerAdvertise, ConnectionState, DEFAULT_KAFKA_PORT, ERROR_ILLEGAL_SASL_STATE, ERROR_INVALID_REQUEST, @@ -55,17 +54,13 @@ use crate::protocol::api::{ use crate::protocol::handlers::api_versions::Advertisement; use crate::protocol::header::{request_header_version, response_header_version}; use crate::protocol::sasl::{ - PlainCredentials, SASL_AUTHENTICATE_MAX_VERSION, SASL_HANDSHAKE_VERSION, SaslAction, SaslState, - parse_plain, + SASL_AUTHENTICATE_MAX_VERSION, SASL_HANDSHAKE_VERSION, SaslAction, SaslState, parse_plain, }; -use crate::tls::{TlsConfig, TlsState}; +use crate::tls::{IdentityError, TlsConfig, TlsState}; const READ_CHUNK: usize = 65536; const GATEWAY_LOG_TARGET: &str = env!("CARGO_CRATE_NAME"); -/// Kafka's name for a client that proved no identity, as `DescribeAcls` renders it. -const ANONYMOUS_PRINCIPAL: &str = "ANONYMOUS"; - /// Builds the log filter, forcing the Iggy SDK quiet unless the operator asked otherwise. /// /// This is a credential-disclosure control, not noise reduction. The SDK logs the username it @@ -117,7 +112,8 @@ pub struct GatewayConfig { /// Required exactly when one of those exists. pub tls: Option<TlsConfig>, pub max_frame_size: usize, - /// Maximum concurrent connections accepted before new ones are rejected. + /// Maximum concurrent connections each listener accepts before rejecting new ones. Every + /// listener has its own allowance, so clients of one cannot use up another's. pub max_connections: usize, /// Bound on how long an accepted connection may sit idle before sending the next /// frame's length prefix. Kafka brokers default `connections.max.idle.ms` to 10 minutes; @@ -344,6 +340,10 @@ struct Listener { /// Whether this listener names a principal, by SASL or by a verified client certificate. /// Decides whether `DescribeAcls` is advertised and answered. acls_enabled: bool, + /// Whether a connection here that names no principal is served as + /// [`AuthenticatedPrincipal::anonymous`]. True on every listener without SASL once any + /// listener of the gateway names principals, so the permission gates hold there too. + anonymous_without_principal: bool, } impl Listener { @@ -421,7 +421,7 @@ impl KafkaGateway { listeners: Vec<BoundListener>, mut shutdown: broadcast::Receiver<()>, ) -> Result<()> { - self.check_sasl_wiring()?; + self.check_sasl_wiring(&listeners)?; let tls = self.check_tls_wiring(&listeners)?; if listeners.is_empty() { return Err(KafkaProtocolError::InvalidConfig( @@ -436,6 +436,12 @@ impl KafkaGateway { let acls_by_certificate = tls .as_ref() .is_some_and(|tls| tls.config().client_auth.is_enabled()); + let names_principals = |protocol: SecurityProtocol| { + protocol.uses_sasl() || (protocol.authenticates_by_certificate() && acls_by_certificate) + }; + let any_names_principals = listeners + .iter() + .any(|listener| names_principals(listener.config.protocol)); let mut bound = Vec::with_capacity(listeners.len()); for BoundListener { config, socket } in listeners { @@ -454,8 +460,8 @@ impl KafkaGateway { Arc::new(Listener { protocol, broker, - acls_enabled: protocol.uses_sasl() - || (protocol.authenticates_by_certificate() && acls_by_certificate), + acls_enabled: names_principals(protocol), + anonymous_without_principal: any_names_principals && !protocol.uses_sasl(), }), socket, )); @@ -478,7 +484,6 @@ impl KafkaGateway { self.config.max_concurrent_authentications, )); let tracker = TaskTracker::new(); - let conn_limiter = Arc::new(Semaphore::new(self.config.max_connections)); let drain_timeout = self.config.shutdown_drain_timeout; if let Some(tls) = &tls { @@ -495,7 +500,7 @@ impl KafkaGateway { state: Arc::clone(&state), auth: Arc::clone(&shared_auth), tracker: tracker.clone(), - conn_limiter: Arc::clone(&conn_limiter), + conn_limiter: Arc::new(Semaphore::new(self.config.max_connections)), cancel: cancel.clone(), tls: tls.clone(), })); @@ -521,14 +526,18 @@ impl KafkaGateway { Err(join_error) => Err(io::Error::other(join_error).into()), }, }; - drain(&tracker, &cancel, drain_timeout).await; + // The accept loops stop first, so no connection is spawned after the drain starts waiting. + cancel.cancel(); accept_loops.shutdown().await; + drain(&tracker, &cancel, drain_timeout).await; outcome } /// Refuses to start when the SASL listeners and the configured authenticator disagree. - fn check_sasl_wiring(&self) -> Result<()> { - let sasl_enabled = self.config.sasl_enabled(); + fn check_sasl_wiring(&self, listeners: &[BoundListener]) -> Result<()> { + let sasl_enabled = listeners + .iter() + .any(|listener| listener.config.protocol.uses_sasl()); if !sasl_enabled && self.authenticator.is_some() { // The mirror of the guard below, and the quieter mistake: a verifier attached while no // listener runs SASL means every connection is served unauthenticated, with nothing @@ -576,11 +585,20 @@ impl KafkaGateway { .into(), )); } - let maps_identities = tls_config.client_auth.is_enabled() - && listeners - .iter() - .any(|listener| listener.config.protocol.authenticates_by_certificate()); - if maps_identities && self.principal_resolver.is_none() && self.bridge.is_none() { + let has_ssl_listener = listeners + .iter() + .any(|listener| listener.config.protocol.authenticates_by_certificate()); + if tls_config.client_auth.is_enabled() && !has_ssl_listener { + return Err(KafkaProtocolError::InvalidConfig(format!( + "client authentication is '{}' but no SSL listener is configured; it applies to \ + SSL listeners only, and a SASL_SSL listener asks for no client certificate", + tls_config.client_auth + ))); + } + if tls_config.client_auth.is_enabled() + && self.principal_resolver.is_none() + && self.bridge.is_none() + { // A certificate names an Iggy user, and only an Iggy session can say whether that // user exists. Admitting the name unchecked would let any certificate from the CA // claim any account. @@ -615,10 +633,13 @@ async fn accept_loop(this: AcceptLoop) -> Result<()> { () = this.cancel.cancelled() => return Ok(()), accepted = this.socket.accept() => accepted, }; + if this.cancel.is_cancelled() { + return Ok(()); + } match accepted { Ok((stream, peer)) => { let Ok(permit) = Arc::clone(&this.conn_limiter).try_acquire_owned() else { - warn!(%peer, max_connections = this.config.max_connections, "connection limit reached, rejecting"); + warn!(%peer, protocol = %this.listener.protocol, max_connections = this.config.max_connections, "listener connection limit reached, rejecting"); continue; }; if let Err(e) = stream.set_nodelay(true) { @@ -677,9 +698,16 @@ async fn serve_connection(stream: TcpStream, connection: Connection) { // unfinished handshake holds a `max_connections` permit. let handshake = timeout( connection.config.pre_auth_timeout, - tls.acceptor().accept(stream), + tls.acceptor(connection.listener.protocol).accept(stream), ); - let tls_stream = match handshake.await { + let handshake = tokio::select! { + () = connection.cancel.cancelled() => { + debug!(%peer, "TLS handshake abandoned by shutdown"); + return; + } + handshake = handshake => handshake, + }; + let tls_stream = match handshake { Ok(Ok(tls_stream)) => tls_stream, Ok(Err(error)) => { // debug!, not warn!: a plaintext client on the TLS port, an untrusted client @@ -695,6 +723,12 @@ async fn serve_connection(stream: TcpStream, connection: Connection) { let certificate_name = if connection.listener.protocol.authenticates_by_certificate() { match tls.principal_name(tls_stream.get_ref().1) { Ok(name) => name, + // warn!: the CA signed this certificate, so a subject no rule maps is a mapping + // rule the operator has to fix. + Err(IdentityError::NoMatchingRule { subject }) => { + warn!(%peer, subject, "client certificate subject matches no principal mapping rule; closing"); + return; + } Err(error) => { debug!(%peer, %error, "client certificate names no principal; closing"); return; @@ -937,6 +971,8 @@ where return Ok(()); }; Some(resolved) + } else if listener.anonymous_without_principal { + Some(AuthenticatedPrincipal::anonymous()) } else { None }; @@ -1020,32 +1056,33 @@ async fn resolve_certificate_principal( error!(%peer, "a client certificate named a principal but nothing can look it up"); return None; }; + // The lookup is an Iggy call per handshake, so it shares the authentication slots with SASL. + let lookup = within_auth_budget(ctx, "certificate identity lookup", |slot| async move { + let _slot = slot; + resolver.resolve(username).await + }); let outcome = tokio::select! { () = ctx.cancel.cancelled() => { debug!(%peer, "certificate identity lookup abandoned by shutdown"); return None; } - outcome = timeout(ctx.config.pre_auth_timeout, resolver.resolve(username)) => outcome, + outcome = lookup => outcome?, }; match outcome { - Ok(Ok(principal)) => { + Ok(principal) => { debug!(%peer, username, "client certificate authenticated"); Some(principal) } - Ok(Err(AuthError::Rejected)) => { + Err(AuthError::Rejected) => { // warn!, not debug!: the certificate chained to the trusted CA, so this is an account // the operator meant to exist, or a mapping rule that produced the wrong name. warn!(%peer, username, "client certificate names no active Iggy user; closing"); None } - Ok(Err(AuthError::Unavailable)) => { + Err(AuthError::Unavailable) => { warn!(%peer, username, "could not look up the certificate's Iggy user; closing"); None } - Err(_elapsed) => { - warn!(%peer, username, "certificate identity lookup did not finish within the pre-authentication budget"); - None - } } } @@ -1097,7 +1134,9 @@ async fn authenticate_token( debug!(%peer, "SASL authentication abandoned by shutdown"); return (HandleOutcome::Close, None); } - verified = verify_within_budget(ctx, authenticator, &credentials) => verified, + verified = within_auth_budget(ctx, "SASL authentication", |slot| { + authenticator.authenticate(&credentials, slot) + }) => verified, }; let Some(result) = verified else { @@ -1138,9 +1177,9 @@ async fn authenticate_token( /// /// Only reachable on a listener that names principals: elsewhere the key is not advertised and /// falls through to ordinary dispatch, which refuses it. On a SASL listener the SASL gate refuses -/// every key but one before a principal exists, so a missing principal there is a gateway fault. -/// On an `SSL` listener that only requests a certificate, a client that presented none is -/// Kafka's anonymous principal, and is described as holding nothing. +/// every key but one before a principal exists, and every other such listener starts each +/// connection with a principal, anonymous when no certificate named one. A missing principal +/// is therefore a gateway fault. fn describe_acls( ctx: &ConnectionContext<'_>, principal: Option<&AuthenticatedPrincipal>, @@ -1148,26 +1187,14 @@ fn describe_acls( body: Bytes, ) -> HandleOutcome { let peer = ctx.peer; - let anonymous; - let principal = match principal { - Some(principal) => principal, - None if ctx.listener.protocol.uses_sasl() => { - // `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); - } - None => { - anonymous = AuthenticatedPrincipal { - username: ANONYMOUS_PRINCIPAL.to_string(), - permissions: PrincipalPermissions::default(), - permissions_known: true, - }; - &anonymous - } + 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 @@ -1199,32 +1226,37 @@ fn describe_acls( } } -/// Waits for an authentication slot, then verifies, each within the pre-authentication budget. +/// Waits for an authentication slot, then runs `work` holding it, each within the +/// pre-authentication budget. `None` means a budget ran out. /// /// Two budgets rather than one shared deadline. The wait needs its own bound so a queued /// connection cannot hold a `max_connections` permit for as long as the backlog takes to drain, -/// which is the invariant `pre_auth_timeout` is documented to enforce. The verification's budget -/// starts once the slot is held, because cutting a verification short does not stop the login it -/// started inside Iggy: a verification that inherited whatever the queue left of a shared deadline -/// would be abandoned mid-hash and hand its slot to the next one while Iggy is still working. -async fn verify_within_budget( - ctx: &ConnectionContext<'_>, - authenticator: &dyn SaslAuthenticator, - credentials: &PlainCredentials, -) -> Option<std::result::Result<AuthenticatedPrincipal, AuthError>> { +/// which is the invariant `pre_auth_timeout` is documented to enforce. The work's budget starts +/// once the slot is held, because cutting it short does not stop the call it started inside Iggy: +/// work that inherited whatever the queue left of a shared deadline would be abandoned mid-hash +/// and hand its slot to the next one while Iggy is still working. +async fn within_auth_budget<'a, T, Work, Fut>( + ctx: &ConnectionContext<'a>, + operation: &'static str, + work: Work, +) -> Option<T> +where + Work: FnOnce(SemaphorePermit<'a>) -> Fut, + Fut: Future<Output = T>, +{ let peer = ctx.peer; let budget = ctx.config.pre_auth_timeout; let Ok(acquired) = timeout(budget, ctx.auth_slots.acquire()).await else { - warn!(%peer, "no authentication slot came free within the pre-authentication budget"); + warn!(%peer, operation, "no authentication slot came free within the pre-authentication budget"); return None; }; // Acquire fails only once the semaphore is closed, which this gateway never does. let Ok(slot) = acquired else { - error!(%peer, "authentication slots unavailable"); + error!(%peer, operation, "authentication slots unavailable"); return None; }; - let Ok(result) = timeout(budget, authenticator.authenticate(credentials, slot)).await else { - warn!(%peer, "authentication did not complete within the pre-authentication budget"); + let Ok(result) = timeout(budget, work(slot)).await else { + warn!(%peer, operation, "did not complete within the pre-authentication budget"); return None; }; Some(result) @@ -1379,11 +1411,11 @@ async fn close_gracefully<S: AsyncRead + AsyncWrite + Unpin>( budget: Duration, cancel: &CancellationToken, ) { - if stream.shutdown().await.is_err() { - return; - } let mut discard = [0u8; 4096]; let drain_input = async { + if stream.shutdown().await.is_err() { + return; + } while let Ok(read) = stream.read(&mut discard).await { if read == 0 { break; @@ -1427,7 +1459,13 @@ async fn send_response<S: AsyncWrite + Unpin>( prefix.put_u8(0); // empty tagged fields } let mut frame = prefix.freeze().chain(body); - timeout(write_timeout, stream.write_all_buf(&mut frame)) + // The flush matters on TLS: the stream reports a write done once rustls holds the record, + // and the tail of a large response would otherwise sit unsent until the next write. + let write = async { + stream.write_all_buf(&mut frame).await?; + stream.flush().await + }; + timeout(write_timeout, write) .await .map_err(|_| io::Error::new(io::ErrorKind::TimedOut, "write timeout"))??; Ok(()) diff --git a/gateways/kafka/src/tls/identity.rs b/gateways/kafka/src/tls/identity.rs index e7917fe90..877e5061a 100644 --- a/gateways/kafka/src/tls/identity.rs +++ b/gateways/kafka/src/tls/identity.rs @@ -19,15 +19,23 @@ //! //! Kafka hands `ssl.principal.mapping.rules` the subject as Java's `X500Principal.getName()` //! renders it: RFC 2253, most specific RDN first, no spaces after the commas. Operators write -//! their rules against that rendering, so this one matches it rather than x509-parser's own -//! `Display`, which keeps the certificate's order and pads with spaces. +//! their rules against that rendering, so this one follows `sun.security.x509.AVA`'s +//! `toRFC2253String` (JDK 11 to 21) rather than x509-parser's own `Display`, which keeps the +//! certificate's order and pads with spaces. Malformed UTF-8 and BMP values are the one case not +//! checked against Java: both sides substitute U+FFFD, possibly not one for one. use std::fmt::Write as _; use std::str::FromStr; use regex::{Captures, Regex}; +use regex_syntax::ast::{ + self, Assertion, AssertionKind, Ast, ClassPerl, ClassPerlKind, ClassSetItem, Flag, Flags, Span, + Visitor, +}; use thiserror::Error; use x509_parser::asn1_rs::{Any, Tag, ToDer}; +use x509_parser::error::X509Error; +use x509_parser::nom; use x509_parser::parse_x509_certificate; use x509_parser::x509::{AttributeTypeAndValue, RelativeDistinguishedName, X509Name}; @@ -35,19 +43,86 @@ use x509_parser::x509::{AttributeTypeAndValue, RelativeDistinguishedName, X509Na #[derive(Debug, Error)] pub enum IdentityError { #[error("the client certificate could not be parsed: {0}")] - Certificate(String), + Certificate(#[from] nom::Err<X509Error>), #[error("no principal mapping rule matches the certificate subject '{subject}'")] NoMatchingRule { subject: String }, } +/// Why `ssl.principal.mapping.rules` text did not parse. +#[derive(Debug, Error)] +pub enum MappingRuleError { + #[error("the rule list is empty; the default is DEFAULT")] + Empty, + #[error("the rule list ends with a comma")] + TrailingComma, + #[error("'{0}' is neither DEFAULT nor RULE:pattern/replacement/[LU]")] + UnknownRule(String), + #[error("expected ',' or the end of the list before '{0}'")] + MissingSeparator(String), + #[error("rule '{0}' has no '/' after its pattern")] + UnterminatedPattern(String), + #[error("rule '{0}' has no '/' after its replacement")] + UnterminatedReplacement(String), + #[error("rule pattern '{pattern}' is not a valid regex: {source}")] + InvalidPattern { + pattern: String, + #[source] + source: regex::Error, + }, + #[error("rule pattern '{pattern}' uses '{construct}': {reason}")] + JavaIncompatible { + pattern: String, + construct: String, + reason: JavaIncompatibility, + }, + #[error("replacement '{replacement}' of rule pattern '{pattern}' is invalid: {source}")] + InvalidReplacement { + pattern: String, + replacement: String, + #[source] + source: ReplacementError, + }, +} + +/// A pattern construct whose Rust `regex` meaning admits subjects Java's `Pattern` rejects. +#[derive(Debug, Error, PartialEq, Eq, Clone, Copy)] +pub enum JavaIncompatibility { + #[error( + "the case-insensitive flag folds Unicode in Rust (the Kelvin sign matches 'k'), ASCII in Java" + )] + CaseInsensitive, + #[error("the 'u' and 'U' flags mean different things in Rust and Java")] + UnicodeFlag, + #[error("Unicode property classes are not supported")] + UnicodeClass, + #[error("Java reads '[:name:]' as a set of literal characters, not a POSIX class")] + PosixClass, + #[error("word boundaries are not supported")] + WordBoundary, +} + +/// A replacement that `java.util.regex.Matcher` would throw on when the rule is applied. +#[derive(Debug, Error, PartialEq, Eq)] +pub enum ReplacementError { + #[error("a trailing '\\' escapes nothing")] + DanglingEscape, + #[error("'$' is followed by neither a group number nor '{{name}}'")] + IllegalGroupReference, + #[error("the pattern has no group {0}")] + NoGroup(usize), + #[error("'${{' has no closing '}}'")] + UnterminatedGroupName, + #[error("the pattern has no group named '{0}'")] + NoNamedGroup(String), +} + /// The subject of a DER certificate, as Kafka's principal builder sees it. /// /// # Errors /// /// Returns [`IdentityError::Certificate`] when `certificate` is not a DER X.509 certificate. pub fn subject_distinguished_name(certificate: &[u8]) -> Result<String, IdentityError> { - let (_, parsed) = parse_x509_certificate(certificate) - .map_err(|error| IdentityError::Certificate(error.to_string()))?; + let (_, parsed) = parse_x509_certificate(certificate)?; Ok(render_rfc2253(parsed.subject())) } @@ -76,17 +151,16 @@ fn render_rdn(rdn: &RelativeDistinguishedName<'_>) -> String { out } -/// `KEYWORD=value` for the attribute types RFC 2253 and Java give a keyword, `OID=#hex` for the -/// rest, and `#hex` of the DER for any value that is not a printable string. Java hex-encodes -/// in exactly those cases, and a rule written against its output has to keep matching here. +/// `KEYWORD=value` for the attribute types Java gives a keyword, `OID=#hex` for the rest, and +/// `#hex` of the DER for a value whose string type Java does not decode. fn render_attribute(out: &mut String, attribute: &AttributeTypeAndValue<'_>) { let value = attribute.attr_value(); if let Some(keyword) = rfc2253_keyword(&attribute.attr_type().to_string()) { out.push_str(keyword); out.push('='); match string_value(value) { - Some(text) if !text.chars().any(char::is_control) => escape_into(out, &text), - _ => push_hex(out, value), + Some(text) => escape_into(out, &text), + None => push_hex(out, value), } } else { let _ = write!(out, "{}=", attribute.attr_type()); @@ -112,22 +186,27 @@ fn rfc2253_keyword(oid: &str) -> Option<&'static str> { }) } -/// The text of a directory string, for the string types Java decodes rather than hex-encodes. +/// The text of a directory string, decoded with the charset `AVA.getCharset` picks for its tag, +/// or `None` for the tags `AVA.isDerString` rejects, which Java renders as hex +/// (`NumericString` and `UniversalString` among them). fn string_value(value: &Any<'_>) -> Option<String> { + let data = value.data; match value.tag() { - Tag::Utf8String | Tag::PrintableString | Tag::Ia5String | Tag::NumericString => { - std::str::from_utf8(value.data).ok().map(str::to_owned) + // Latin-1, not ASCII: Java keeps a non-ASCII byte in these types as its own code point. + Tag::PrintableString | Tag::Ia5String | Tag::GeneralString | Tag::T61String => { + Some(data.iter().map(|&byte| char::from(byte)).collect()) } + Tag::Utf8String => Some(String::from_utf8_lossy(data).into_owned()), Tag::BmpString => { - let units: Option<Vec<u16>> = value - .data + let units: Vec<u16> = data .chunks(2) - .map(|pair| <[u8; 2]>::try_from(pair).ok().map(u16::from_be_bytes)) + .map(|pair| match pair { + [high, low] => u16::from_be_bytes([*high, *low]), + _ => 0xFFFD, + }) .collect(); - String::from_utf16(&units?).ok() + Some(String::from_utf16_lossy(&units)) } - // T.61 is close enough to Latin-1 for the subjects that still carry it. - Tag::T61String => Some(value.data.iter().map(|&byte| char::from(byte)).collect()), _ => None, } } @@ -142,18 +221,35 @@ fn push_hex(out: &mut String, value: &Any<'_>) { } } -/// Backslash-escapes what Java's RFC 2253 rendering escapes: `,=+<>#;"\` anywhere, and a space -/// at either end. +/// Backslash-escapes what Java's RFC 2253 rendering escapes: `,=+<>#;"\` anywhere, NUL as `\00`, +/// and every space or carriage return in the leading and trailing runs. Everything else, other +/// control characters included, is written as is. fn escape_into(out: &mut String, text: &str) { - let last = text.chars().count().saturating_sub(1); - for (index, character) in text.chars().enumerate() { - let edge_space = character == ' ' && (index == 0 || index == last); - if edge_space - || matches!( - character, - ',' | '=' | '+' | '<' | '>' | '#' | ';' | '"' | '\\' - ) - { + let mut escaped: Vec<char> = Vec::with_capacity(text.len()); + for character in text.chars() { + if character == '\0' { + escaped.extend(['\\', '0', '0']); + continue; + } + if matches!( + character, + ',' | '=' | '+' | '<' | '>' | '#' | ';' | '"' | '\\' + ) { + escaped.push('\\'); + } + escaped.push(character); + } + let is_edge_space = |character: &char| matches!(character, ' ' | '\r'); + let lead = escaped + .iter() + .position(|character| !is_edge_space(character)) + .unwrap_or(escaped.len()); + let trail = escaped + .iter() + .rposition(|character| !is_edge_space(character)); + for (index, character) in escaped.into_iter().enumerate() { + let in_trailing_run = trail.is_none_or(|trail| index > trail); + if index < lead || in_trailing_run { out.push('\\'); } out.push(character); @@ -161,7 +257,16 @@ fn escape_into(out: &mut String, text: &str) { } /// Kafka's `ssl.principal.mapping.rules`: `RULE:pattern/replacement/[LU]` and `DEFAULT`, comma -/// separated, tried in order. The first rule whose pattern matches the whole subject decides. +/// separated, tried in order. +/// +/// A rule applies when its pattern matches the whole subject (Java's `matches()`). Its result is +/// then Java's `subject.replaceAll(pattern, replacement)`: every match of the unanchored pattern +/// in the subject is replaced, an empty match at the end included, so `RULE:(.*)/$1@x/` maps +/// `alice` to `alice@x@x` exactly as Kafka does. +/// +/// Patterns are a Java-compatible subset: `\d \s \w` are rewritten to ASCII classes and `.` to +/// exclude Java's line terminators, and constructs whose Rust meaning is wider than Java's are +/// refused as [`MappingRuleError::JavaIncompatible`]. #[derive(Debug, Clone)] pub struct PrincipalMappingRules { rules: Vec<Rule>, @@ -172,17 +277,21 @@ enum Rule { /// The subject string itself. Default, Rewrite { - /// The operator's pattern anchored at both ends, since Java's `matches()` demands the - /// whole subject. whole: Regex, - /// The pattern as written, for the replacement pass. Java runs `replaceAll` over the - /// subject with it unanchored, and `a|ab` on `ab` replaces `a` alone that way. finder: Regex, - replacement: String, + replacement: Vec<Piece>, case: Case, }, } +/// One part of a parsed replacement. +#[derive(Debug, Clone, PartialEq, Eq)] +enum Piece { + Literal(String), + Group(usize), + Named(String), +} + #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum Case { Keep, @@ -213,11 +322,9 @@ impl PrincipalMappingRules { if !whole.is_match(subject) { return None; } - let rewritten = finder.replace_all(subject, |captures: &Captures<'_>| { - expand_replacement(replacement, captures) - }); + let rewritten = replace_all(finder, subject, replacement); Some(match case { - Case::Keep => rewritten.into_owned(), + Case::Keep => rewritten, Case::Lower => rewritten.to_lowercase(), Case::Upper => rewritten.to_uppercase(), }) @@ -227,66 +334,220 @@ impl PrincipalMappingRules { } impl FromStr for PrincipalMappingRules { - type Err = String; + type Err = MappingRuleError; fn from_str(raw: &str) -> Result<Self, Self::Err> { let mut rules = Vec::new(); let mut rest = raw.trim_start(); if rest.is_empty() { - return Err("the rule list is empty; the default is DEFAULT".to_string()); + return Err(MappingRuleError::Empty); } while !rest.is_empty() { if let Some(after) = rest.strip_prefix("DEFAULT") { rules.push(Rule::Default); rest = after; } else if let Some(after) = rest.strip_prefix("RULE:") { - let (pattern, after) = take_until_slash(after) - .ok_or_else(|| format!("rule '{rest}' has no '/' after its pattern"))?; - let (replacement, after) = take_until_slash(after) - .ok_or_else(|| format!("rule '{rest}' has no '/' after its replacement"))?; - let (case, after) = match after.chars().next() { - Some('L') => (Case::Lower, &after[1..]), - Some('U') => (Case::Upper, &after[1..]), - _ => (Case::Keep, after), - }; - let finder = Regex::new(&pattern).map_err(|error| { - format!("rule pattern '{pattern}' is not a valid regex: {error}") - })?; - let whole = Regex::new(&format!("^(?:{pattern})$")).map_err(|error| { - format!("rule pattern '{pattern}' is not a valid regex: {error}") - })?; - rules.push(Rule::Rewrite { - whole, - finder, - replacement, - case, - }); + let (rule, after) = parse_rewrite(rest, after)?; + rules.push(rule); rest = after; } else { - return Err(format!( - "'{rest}' is neither DEFAULT nor RULE:pattern/replacement/[LU]" - )); + return Err(MappingRuleError::UnknownRule(rest.to_string())); } rest = rest.trim_start(); match rest.strip_prefix(',') { Some(after) => { rest = after.trim_start(); if rest.is_empty() { - return Err("the rule list ends with a comma".to_string()); + return Err(MappingRuleError::TrailingComma); } } None if rest.is_empty() => {} - None => { - return Err(format!( - "expected ',' or the end of the list before '{rest}'" - )); - } + None => return Err(MappingRuleError::MissingSeparator(rest.to_string())), } } Ok(Self { rules }) } } +/// Parses the `pattern/replacement/[LU]` after `RULE:`, returning the rule and what follows it. +fn parse_rewrite<'a>(rule: &str, body: &'a str) -> Result<(Rule, &'a str), MappingRuleError> { + let (pattern, after) = take_until_slash(body) + .ok_or_else(|| MappingRuleError::UnterminatedPattern(rule.to_string()))?; + let (replacement, after) = take_until_slash(after) + .ok_or_else(|| MappingRuleError::UnterminatedReplacement(rule.to_string()))?; + let (case, after) = match after.chars().next() { + Some('L') => (Case::Lower, &after[1..]), + Some('U') => (Case::Upper, &after[1..]), + _ => (Case::Keep, after), + }; + let invalid_pattern = |source| MappingRuleError::InvalidPattern { + pattern: pattern.clone(), + source, + }; + let java = java_semantics(&pattern)?; + let finder = Regex::new(&java).map_err(invalid_pattern)?; + let whole = Regex::new(&format!("^(?:{java})$")).map_err(invalid_pattern)?; + let pieces = parse_replacement(&replacement, &finder).map_err(|source| { + MappingRuleError::InvalidReplacement { + pattern: pattern.clone(), + replacement: replacement.clone(), + source, + } + })?; + Ok(( + Rule::Rewrite { + whole, + finder, + replacement: pieces, + case, + }, + after, + )) +} + +/// Rewrites `pattern` so Rust matches what `java.util.regex.Pattern` matches without flags: +/// `\d \s \w` and their negations become ASCII classes, and `.` outside `(?s)` also stops at +/// `\r`, U+0085, U+2028 and U+2029. Constructs with no faithful rewrite are refused with their +/// span. +fn java_semantics(pattern: &str) -> Result<String, MappingRuleError> { + let parsed = ast::parse::Parser::new().parse(pattern).map_err(|error| { + MappingRuleError::InvalidPattern { + pattern: pattern.to_string(), + source: regex::Error::Syntax(error.to_string()), + } + })?; + let rewriter = JavaRewriter { + dot_all: vec![false], + edits: Vec::new(), + }; + let edits = ast::visit(&parsed, rewriter).map_err(|(span, reason)| { + MappingRuleError::JavaIncompatible { + pattern: pattern.to_string(), + construct: pattern[span.start.offset..span.end.offset].to_string(), + reason, + } + })?; + let mut rewritten = pattern.to_string(); + for (span, replacement) in edits.iter().rev() { + rewritten.replace_range(span.start.offset..span.end.offset, replacement); + } + Ok(rewritten) +} + +const JAVA_DOT: &str = r"[^\n\r\x{85}\x{2028}\x{2029}]"; + +/// Collects `(span, replacement)` edits in pattern order. +struct JavaRewriter { + /// Whether `.` matches line terminators, one entry per open group. + dot_all: Vec<bool>, + edits: Vec<(Span, String)>, +} + +impl JavaRewriter { + fn apply_flags(&mut self, flags: &Flags) -> Result<(), (Span, JavaIncompatibility)> { + for item in &flags.items { + match item.kind { + ast::FlagsItemKind::Flag(Flag::CaseInsensitive) + if flags.flag_state(Flag::CaseInsensitive) == Some(true) => + { + return Err((item.span, JavaIncompatibility::CaseInsensitive)); + } + ast::FlagsItemKind::Flag(Flag::Unicode | Flag::SwapGreed) => { + return Err((item.span, JavaIncompatibility::UnicodeFlag)); + } + _ => {} + } + } + if let Some(dot_all) = flags.flag_state(Flag::DotMatchesNewLine) + && let Some(current) = self.dot_all.last_mut() + { + *current = dot_all; + } + Ok(()) + } + + fn perl(&mut self, class: &ClassPerl, bracketed: bool) { + let name = match class.kind { + ClassPerlKind::Digit => "digit", + ClassPerlKind::Space => "space", + ClassPerlKind::Word => "word", + }; + let negation = if class.negated { "^" } else { "" }; + let ascii = format!("[:{negation}{name}:]"); + let replacement = if bracketed { + ascii + } else { + format!("[{ascii}]") + }; + self.edits.push((class.span, replacement)); + } +} + +impl Visitor for JavaRewriter { + type Output = Vec<(Span, String)>; + type Err = (Span, JavaIncompatibility); + + fn finish(self) -> Result<Self::Output, Self::Err> { + Ok(self.edits) + } + + fn visit_pre(&mut self, node: &Ast) -> Result<(), Self::Err> { + match node { + Ast::Group(group) => { + let inherited = self.dot_all.last().copied().unwrap_or(false); + self.dot_all.push(inherited); + if let Some(flags) = group.flags() { + self.apply_flags(flags)?; + } + } + Ast::Flags(set) => self.apply_flags(&set.flags)?, + Ast::Dot(span) => { + if !self.dot_all.last().copied().unwrap_or(false) { + self.edits.push((**span, JAVA_DOT.to_string())); + } + } + Ast::ClassPerl(class) => self.perl(class, false), + Ast::ClassUnicode(class) => { + return Err((class.span, JavaIncompatibility::UnicodeClass)); + } + Ast::Assertion(assertion) => refuse_word_boundary(assertion)?, + _ => {} + } + Ok(()) + } + + fn visit_post(&mut self, node: &Ast) -> Result<(), Self::Err> { + if let Ast::Group(_) = node { + self.dot_all.pop(); + } + Ok(()) + } + + fn visit_class_set_item_pre(&mut self, item: &ClassSetItem) -> Result<(), Self::Err> { + match item { + ClassSetItem::Perl(class) => self.perl(class, true), + ClassSetItem::Unicode(class) => { + return Err((class.span, JavaIncompatibility::UnicodeClass)); + } + ClassSetItem::Ascii(class) => { + return Err((class.span, JavaIncompatibility::PosixClass)); + } + _ => {} + } + Ok(()) + } +} + +const fn refuse_word_boundary(assertion: &Assertion) -> Result<(), (Span, JavaIncompatibility)> { + match assertion.kind { + AssertionKind::StartLine + | AssertionKind::EndLine + | AssertionKind::StartText + | AssertionKind::EndText => Ok(()), + _ => Err((assertion.span, JavaIncompatibility::WordBoundary)), + } +} + /// Reads up to the first unescaped `/`, unescaping `\/` on the way and keeping every other /// escape for the regex engine. Returns the text and what follows the slash. fn take_until_slash(input: &str) -> Option<(String, &str)> { @@ -309,57 +570,122 @@ fn take_until_slash(input: &str) -> Option<(String, &str)> { None } -/// Expands a replacement the way `java.util.regex.Matcher` does: `$N` is a group, taking as -/// many digits as still name an existing group, `${name}` a named group, and `\x` a literal -/// `x`. A group that did not participate expands to nothing. -fn expand_replacement(replacement: &str, captures: &Captures<'_>) -> String { - let mut out = String::new(); +/// Parses a replacement the way `java.util.regex.Matcher.appendReplacement` reads it: `$N` is a +/// group, taking as many digits as still name an existing group, `${name}` a named group, and +/// `\x` a literal `x`. +/// +/// Stricter than Kafka, which writes `$N` literally when N exceeds a nonzero group count: that is +/// refused here. The reverse, `$1\$2`, makes Kafka throw while this maps `alice` to `alice$2`, +/// which is harmless because that Iggy user must still exist. +fn parse_replacement(replacement: &str, finder: &Regex) -> Result<Vec<Piece>, ReplacementError> { + let group_count = finder.captures_len() - 1; + let mut pieces = Vec::new(); + let mut literal = String::new(); let mut chars = replacement.chars().peekable(); while let Some(character) = chars.next() { match character { - '\\' => { - if let Some(escaped) = chars.next() { - out.push(escaped); - } - } - '$' => match chars.peek() { - Some('{') => { - chars.next(); - let name: String = chars.by_ref().take_while(|&c| c != '}').collect(); - if let Some(found) = captures.name(&name) { - out.push_str(found.as_str()); - } - } - Some(digit) if digit.is_ascii_digit() => { - let mut group = 0usize; - let mut taken_any = false; - while let Some(&next) = chars.peek() { - let Some(value) = next.to_digit(10) else { - break; - }; - let candidate = group * 10 + value as usize; - if taken_any && candidate >= captures.len() { - break; + '\\' => literal.push(chars.next().ok_or(ReplacementError::DanglingEscape)?), + '$' => { + let piece = match chars.next() { + Some('{') => { + let mut name = String::new(); + loop { + match chars.next() { + Some('}') => break, + Some(next) => name.push(next), + None => return Err(ReplacementError::UnterminatedGroupName), + } + } + if !finder.capture_names().flatten().any(|known| known == name) { + return Err(ReplacementError::NoNamedGroup(name)); } - group = candidate; - taken_any = true; - chars.next(); + Piece::Named(name) } - if let Some(found) = captures.get(group) { - out.push_str(found.as_str()); + Some(digit) if digit.is_ascii_digit() => { + let mut group = digit.to_digit(10).map_or(0, |value| value as usize); + if group > group_count { + return Err(ReplacementError::NoGroup(group)); + } + while let Some(value) = chars.peek().and_then(|next| next.to_digit(10)) { + let candidate = group * 10 + value as usize; + if candidate > group_count { + break; + } + group = candidate; + chars.next(); + } + Piece::Group(group) } + _ => return Err(ReplacementError::IllegalGroupReference), + }; + if !literal.is_empty() { + pieces.push(Piece::Literal(std::mem::take(&mut literal))); } - _ => out.push('$'), - }, - other => out.push(other), + pieces.push(piece); + } + other => literal.push(other), } } + if !literal.is_empty() { + pieces.push(Piece::Literal(literal)); + } + Ok(pieces) +} + +/// Java's `Matcher.replaceAll`: after an empty match the search resumes one character later, and +/// an empty match right after a non-empty one still counts. Java steps one UTF-16 unit rather than +/// one character, so the two differ only when an empty match lands before a supplementary-plane +/// character. +fn replace_all(finder: &Regex, subject: &str, replacement: &[Piece]) -> String { + let mut out = String::with_capacity(subject.len()); + let mut copied_up_to = 0; + let mut search_from = 0; + while search_from <= subject.len() { + let Some(captures) = finder.captures_at(subject, search_from) else { + break; + }; + let found = captures + .get(0) + .map_or(search_from..search_from, |m| m.range()); + out.push_str(&subject[copied_up_to..found.start]); + expand_into(&mut out, replacement, &captures); + copied_up_to = found.end; + search_from = if found.is_empty() { + match subject[found.end..].chars().next() { + Some(next) => found.end + next.len_utf8(), + None => break, + } + } else { + found.end + }; + } + out.push_str(&subject[copied_up_to..]); out } +/// A group that did not participate in the match expands to nothing. +fn expand_into(out: &mut String, replacement: &[Piece], captures: &Captures<'_>) { + for piece in replacement { + match piece { + Piece::Literal(text) => out.push_str(text), + Piece::Group(group) => { + if let Some(found) = captures.get(*group) { + out.push_str(found.as_str()); + } + } + Piece::Named(name) => { + if let Some(found) = captures.name(name) { + out.push_str(found.as_str()); + } + } + } + } +} + #[cfg(test)] mod tests { use rcgen::{CertificateParams, DistinguishedName, DnType, DnValue, KeyPair}; + use x509_parser::prelude::FromDer; use super::*; @@ -409,6 +735,75 @@ mod tests { assert_eq!(subject, r"CN=\ a b\ "); } + #[test] + fn given_runs_of_edge_spaces_should_escape_every_one_of_them() { + let subject = subject_of(|params| { + params + .distinguished_name + .push(DnType::CommonName, DnValue::Utf8String(" a ".to_string())); + }); + assert_eq!(subject, r"CN=\ \ a\ \ "); + } + + #[test] + fn given_a_control_character_should_write_it_raw() { + let subject = subject_of(|params| { + params + .distinguished_name + .push(DnType::CommonName, DnValue::Utf8String("a\tb".to_string())); + }); + assert_eq!(subject, "CN=a\tb"); + } + + #[test] + fn given_a_numeric_string_should_render_it_as_hex() { + // Name { RDN { CN, NumericString "123" } }; rcgen cannot emit a NumericString. + let der = [ + 0x30, 0x0e, 0x31, 0x0c, 0x30, 0x0a, 0x06, 0x03, 0x55, 0x04, 0x03, 0x12, 0x03, 0x31, + 0x32, 0x33, + ]; + let (_, name) = X509Name::from_der(&der).expect("a DER name"); + assert_eq!(render_rfc2253(&name), "CN=#1203313233"); + } + + /// `Name { RDN { CN, value } }` around one DER-encoded value, for string types rcgen cannot + /// emit or contents it refuses. + fn common_name_with(value: &[u8]) -> String { + let attribute_len = u8::try_from(5 + value.len()).expect("a short value"); + let mut der = vec![0x30, attribute_len + 4, 0x31, attribute_len + 2]; + der.extend([0x30, attribute_len, 0x06, 0x03, 0x55, 0x04, 0x03]); + der.extend(value); + let (_, name) = X509Name::from_der(&der).expect("a DER name"); + render_rfc2253(&name) + } + + #[test] + fn given_non_ascii_bytes_in_an_ascii_string_type_should_decode_them_as_latin_1() { + for tag in [0x13, 0x16, 0x1b] { + assert_eq!( + common_name_with(&[tag, 0x03, b'a', 0xe9, b'b']), + "CN=a\u{e9}b", + "tag {tag:#04x}" + ); + } + } + + #[test] + fn given_a_universal_string_should_render_it_as_hex() { + assert_eq!( + common_name_with(&[0x1c, 0x04, 0x00, 0x00, 0x00, b'a']), + "CN=#1c0400000061" + ); + } + + #[test] + fn given_a_nul_character_should_escape_it_as_two_hex_digits() { + assert_eq!( + common_name_with(&[0x0c, 0x03, b'a', 0x00, b'b']), + "CN=a\\00b" + ); + } + #[test] fn given_an_attribute_without_a_keyword_should_render_its_oid_and_hex_value() { let subject = subject_of(|params| { @@ -506,6 +901,63 @@ mod tests { assert_eq!(named.map("CN=bob,O=x").as_deref(), Some("bob")); } + #[test] + fn given_a_pattern_that_also_matches_empty_should_replace_like_java_replace_all() { + let rules: PrincipalMappingRules = "RULE:(.*)/$1@x/".parse().expect("valid"); + assert_eq!(rules.map("alice").as_deref(), Some("alice@x@x")); + } + + #[test] + fn given_more_digits_than_groups_should_take_the_longest_existing_group() { + let rules: PrincipalMappingRules = "RULE:^CN=(\\w+)$/$12/".parse().expect("valid"); + assert_eq!(rules.map("CN=alice").as_deref(), Some("alice2")); + } + + #[test] + fn given_a_group_beyond_the_patterns_count_should_refuse_what_kafka_writes_literally() { + match "RULE:^CN=(.*)$/$2/".parse::<PrincipalMappingRules>() { + Err(MappingRuleError::InvalidReplacement { source, .. }) => { + assert_eq!(source, ReplacementError::NoGroup(2)); + } + other => panic!("Kafka maps this to '$2'; the gateway refuses it, got {other:?}"), + } + } + + #[test] + fn given_an_escaped_dollar_after_a_group_should_write_it_literally() { + let rules: PrincipalMappingRules = "RULE:^CN=(.*)$/$1\\$2/".parse().expect("valid"); + assert_eq!(rules.map("CN=alice").as_deref(), Some("alice$2")); + } + + #[test] + fn given_a_replacement_java_would_throw_on_should_refuse_the_rule() { + for (raw, expected) in [ + ( + "RULE:^CN=(.*)$/$x/", + ReplacementError::IllegalGroupReference, + ), + ( + "RULE:^CN=(.*)$/a$/", + ReplacementError::IllegalGroupReference, + ), + ( + "RULE:^CN=(?P<user>.*)$/${nope}/", + ReplacementError::NoNamedGroup("nope".to_string()), + ), + ( + "RULE:^CN=(?P<user>.*)$/${user/", + ReplacementError::UnterminatedGroupName, + ), + ] { + match raw.parse::<PrincipalMappingRules>() { + Err(MappingRuleError::InvalidReplacement { source, .. }) => { + assert_eq!(source, expected, "for '{raw}'"); + } + other => panic!("'{raw}' must be refused for its replacement, got {other:?}"), + } + } + } + #[test] fn given_a_malformed_rule_list_should_refuse() { for raw in [ @@ -525,4 +977,126 @@ mod tests { ); } } + + #[test] + fn given_a_kelvin_sign_should_not_match_an_ascii_word_class() { + let rules: PrincipalMappingRules = r"RULE:^CN=(\w+)$/$1/L".parse().expect("valid"); + assert_eq!(rules.map("CN=\u{212A}afka"), None); + assert_eq!(rules.map("CN=Kafka").as_deref(), Some("kafka")); + let bracketed: PrincipalMappingRules = r"RULE:^CN=([\w.]+)$/$1/".parse().expect("valid"); + assert_eq!(bracketed.map("CN=\u{212A}afka"), None); + assert_eq!(bracketed.map("CN=k.afka").as_deref(), Some("k.afka")); + } + + #[test] + fn given_non_ascii_digits_or_spaces_should_not_match_ascii_classes() { + let digits: PrincipalMappingRules = r"RULE:^CN=u(\d+)$/$1/".parse().expect("valid"); + assert_eq!(digits.map("CN=u\u{0661}\u{0662}"), None); + assert_eq!(digits.map("CN=u12").as_deref(), Some("12")); + let spaces: PrincipalMappingRules = r"RULE:^CN=a\sb$/ab/".parse().expect("valid"); + assert_eq!(spaces.map("CN=a\u{00A0}b"), None); + assert_eq!(spaces.map("CN=a b").as_deref(), Some("ab")); + let negated: PrincipalMappingRules = r"RULE:^CN=(\D+)$/$1/".parse().expect("valid"); + assert_eq!( + negated.map("CN=\u{0661}").as_deref(), + Some("\u{0661}"), + "Java's \\D matches every non-ASCII digit" + ); + } + + #[test] + fn given_a_java_line_terminator_should_not_match_dot_outside_dotall() { + let rules: PrincipalMappingRules = "RULE:^CN=(.*)$/$1/".parse().expect("valid"); + for terminator in ['\r', '\u{85}', '\u{2028}', '\u{2029}'] { + assert_eq!(rules.map(&format!("CN=a{terminator}b")), None); + } + let dot_all: PrincipalMappingRules = "RULE:(?s)^CN=(.*)$/$1/".parse().expect("valid"); + assert_eq!(dot_all.map("CN=a\u{2028}b").as_deref(), Some("a\u{2028}b")); + let scoped: PrincipalMappingRules = "RULE:^CN=(?s:(.))(.)$/$1$2/".parse().expect("valid"); + assert_eq!(scoped.map("CN=\u{85}x").as_deref(), Some("\u{85}x")); + assert_eq!(scoped.map("CN=x\u{85}"), None, "(?s:) ends with its group"); + } + + #[test] + fn given_a_non_ascii_name_should_still_map_like_kafka() { + for raw in ["RULE:^CN=([^,]+).*$/$1/", "RULE:^CN=(.*)$/$1/"] { + let rules: PrincipalMappingRules = raw.parse().expect("valid"); + assert_eq!( + rules.map("CN=Zo\u{eb}").as_deref(), + Some("Zo\u{eb}"), + "{raw}" + ); + assert_eq!( + rules.map("CN=Jos\u{e9}").as_deref(), + Some("Jos\u{e9}"), + "{raw}" + ); + } + } + + #[test] + fn given_a_construct_without_java_semantics_should_refuse_the_rule() { + for (raw, construct, expected) in [ + ( + "RULE:(?i)^CN=(\u{17F}vc)$/$1/", + "i", + JavaIncompatibility::CaseInsensitive, + ), + ( + "RULE:^CN=(?i:kafka)$/$1/", + "i", + JavaIncompatibility::CaseInsensitive, + ), + ( + "RULE:(?u)^CN=(.*)$/$1/", + "u", + JavaIncompatibility::UnicodeFlag, + ), + ( + "RULE:(?U)^CN=(.*)$/$1/", + "U", + JavaIncompatibility::UnicodeFlag, + ), + ( + r"RULE:^CN=(\pL+)$/$1/", + r"\pL", + JavaIncompatibility::UnicodeClass, + ), + ( + r"RULE:^CN=([\p{Greek}a]+)$/$1/", + r"\p{Greek}", + JavaIncompatibility::UnicodeClass, + ), + ( + "RULE:^CN=([[:alpha:]]+)$/$1/", + "[:alpha:]", + JavaIncompatibility::PosixClass, + ), + ( + r"RULE:^CN=\b(.*)$/$1/", + r"\b", + JavaIncompatibility::WordBoundary, + ), + ( + r"RULE:^CN=(.*)\B$/$1/", + r"\B", + JavaIncompatibility::WordBoundary, + ), + ] { + match raw.parse::<PrincipalMappingRules>() { + Err(MappingRuleError::JavaIncompatible { + construct: found, + reason, + .. + }) => { + assert_eq!((found.as_str(), reason), (construct, expected), "{raw}"); + } + other => panic!("'{raw}' must be refused, got {other:?}"), + } + } + let case_sensitive: PrincipalMappingRules = "RULE:(?-i)^CN=(.*)$/$1/" + .parse() + .expect("(?-i) is Java's default"); + assert_eq!(case_sensitive.map("CN=a").as_deref(), Some("a")); + } } diff --git a/gateways/kafka/src/tls/mod.rs b/gateways/kafka/src/tls/mod.rs index 85c3ef4f0..873ecef62 100644 --- a/gateways/kafka/src/tls/mod.rs +++ b/gateways/kafka/src/tls/mod.rs @@ -35,10 +35,14 @@ use tokio_util::sync::CancellationToken; use tracing::{error, info}; use crate::error::{KafkaProtocolError, Result}; +use crate::listener::SecurityProtocol; pub mod identity; -pub use identity::{IdentityError, PrincipalMappingRules, subject_distinguished_name}; +pub use identity::{ + IdentityError, MappingRuleError, PrincipalMappingRules, ReplacementError, + subject_distinguished_name, +}; /// How often the certificate files are re-read when nothing else is configured. /// @@ -97,6 +101,10 @@ impl fmt::Display for ClientAuth { } /// TLS material and policy shared by every `SSL` and `SASL_SSL` listener. +/// +/// `client_auth` applies to `SSL` listeners only. A `SASL_SSL` listener never asks for a client +/// certificate, as Kafka leaves `ssl.client.auth` at `none` there unless that listener +/// overrides it. #[derive(Debug, Clone)] pub struct TlsConfig { /// PEM certificate chain, leaf first. @@ -146,7 +154,7 @@ impl TlsConfig { } /// The bytes last read from disk, kept so a reload can tell whether anything changed. -#[derive(PartialEq, Eq)] +#[derive(Clone, PartialEq, Eq)] struct Material { certificate: Vec<u8>, key: Vec<u8>, @@ -165,12 +173,51 @@ impl Material { .transpose()?, }) } + + async fn read_async(config: &TlsConfig) -> Result<Self> { + let client_ca = match config.client_ca_file.as_deref() { + Some(path) => Some(read_file_async(path).await?), + None => None, + }; + Ok(Self { + certificate: read_file_async(&config.cert_file).await?, + key: read_file_async(&config.key_file).await?, + client_ca, + }) + } } fn read_file(path: &Path) -> Result<Vec<u8>> { - fs::read(path).map_err(|error| { - KafkaProtocolError::InvalidConfig(format!("cannot read {}: {error}", path.display())) - }) + fs::read(path).map_err(|error| unreadable(path, &error)) +} + +async fn read_file_async(path: &Path) -> Result<Vec<u8>> { + tokio::fs::read(path) + .await + .map_err(|error| unreadable(path, &error)) +} + +fn unreadable(path: &Path, error: &std::io::Error) -> KafkaProtocolError { + KafkaProtocolError::InvalidConfig(format!("cannot read {}: {error}", path.display())) +} + +/// The two server configurations one set of material builds: `SSL` listeners verify client +/// certificates as `client_auth` says, `SASL_SSL` listeners never ask for one. +struct ServerConfigs { + ssl: Arc<ServerConfig>, + sasl_ssl: Arc<ServerConfig>, +} + +impl ServerConfigs { + fn build(config: &TlsConfig, material: &Material) -> Result<Self> { + let ssl = Arc::new(build_server_config(config, config.client_auth, material)?); + let sasl_ssl = if config.client_auth.is_enabled() { + Arc::new(build_server_config(config, ClientAuth::None, material)?) + } else { + Arc::clone(&ssl) + }; + Ok(Self { ssl, sasl_ssl }) + } } /// The live TLS configuration, swapped under running listeners when the files change. @@ -180,7 +227,7 @@ fn read_file(path: &Path) -> Result<Vec<u8>> { /// lock in the accept path: the swap is an atomic pointer exchange. pub struct TlsState { config: TlsConfig, - current: ArcSwap<ServerConfig>, + current: ArcSwap<ServerConfigs>, loaded: Mutex<Material>, } @@ -210,10 +257,10 @@ impl TlsState { ))); } let material = Material::read(&config)?; - let server_config = build_server_config(&config, &material)?; + let server_configs = ServerConfigs::build(&config, &material)?; Ok(Self { config, - current: ArcSwap::from_pointee(server_config), + current: ArcSwap::from_pointee(server_configs), loaded: Mutex::new(material), }) } @@ -223,30 +270,38 @@ impl TlsState { &self.config } - /// An acceptor over the configuration current right now. + /// An acceptor over the configuration current right now, for a listener speaking + /// `protocol`. Only an `SSL` listener asks for a client certificate. #[must_use] - pub fn acceptor(&self) -> TlsAcceptor { - TlsAcceptor::from(self.current.load_full()) + pub fn acceptor(&self, protocol: SecurityProtocol) -> TlsAcceptor { + let current = self.current.load(); + let config = if protocol.authenticates_by_certificate() { + ¤t.ssl + } else { + ¤t.sasl_ssl + }; + TlsAcceptor::from(Arc::clone(config)) } /// Re-reads the files and swaps the configuration if they changed. /// /// Returns `Ok(true)` when new material went live. A read or build failure leaves the current /// material serving: a rotation that is half written, or a certificate that does not match - /// its key, must not take the listener down. + /// its key, must not take the listener down. Material that failed to build is retried on every + /// call, so each tick reports the failure until the files are fixed. /// /// # Errors /// /// The same conditions as [`Self::load`]. - pub fn reload(&self) -> Result<bool> { - let material = Material::read(&self.config)?; - let unchanged = *self.loaded.lock().unwrap_or_else(PoisonError::into_inner) == material; - if unchanged { + pub async fn reload(&self) -> Result<bool> { + let material = Material::read_async(&self.config).await?; + if *self.loaded.lock().unwrap_or_else(PoisonError::into_inner) == material { return Ok(false); } - let server_config = build_server_config(&self.config, &material)?; - self.current.store(Arc::new(server_config)); + let server_configs = ServerConfigs::build(&self.config, &material)?; + // Recorded only after a successful build, so a failed build is retried every tick. *self.loaded.lock().unwrap_or_else(PoisonError::into_inner) = material; + self.current.store(Arc::new(server_configs)); Ok(true) } @@ -261,7 +316,7 @@ impl TlsState { () = cancel.cancelled() => return, () = tokio::time::sleep(interval) => {} } - match self.reload() { + match self.reload().await { Ok(true) => info!( cert_file = %self.config.cert_file.display(), "TLS certificate reloaded; new connections use the new material" @@ -306,7 +361,11 @@ impl TlsState { } } -fn build_server_config(config: &TlsConfig, material: &Material) -> Result<ServerConfig> { +fn build_server_config( + config: &TlsConfig, + client_auth: ClientAuth, + material: &Material, +) -> Result<ServerConfig> { // Named explicitly rather than taken from the process default: the Iggy SDK installs its own // provider lazily on its first TLS dial, and a listener that came up first would otherwise // decide for it, or panic because neither had chosen yet. @@ -327,11 +386,7 @@ fn build_server_config(config: &TlsConfig, material: &Material) -> Result<Server let builder = ServerConfig::builder_with_provider(Arc::clone(&provider)) .with_safe_default_protocol_versions() .map_err(|error| KafkaProtocolError::InvalidConfig(error.to_string()))?; - let builder = match ( - config.client_auth, - &material.client_ca, - &config.client_ca_file, - ) { + let builder = match (client_auth, &material.client_ca, &config.client_ca_file) { (ClientAuth::None, _, _) => builder.with_no_client_auth(), (mode, Some(ca_pem), Some(ca_file)) => { let mut roots = RootCertStore::empty(); @@ -430,18 +485,18 @@ mod tests { } } - #[test] - fn given_a_certificate_and_key_should_load_and_report_no_change_on_reload() { + #[tokio::test] + async fn given_a_certificate_and_key_should_load_and_report_no_change_on_reload() { let dir = tempfile::tempdir().expect("tempdir"); let (cert, key) = self_signed("first"); let cert_file = write(dir.path(), "cert.pem", &cert); let key_file = write(dir.path(), "key.pem", &key); let state = TlsState::load(TlsConfig::new(cert_file, key_file)).expect("loads"); - assert!(!state.reload().expect("reload reads the same files")); + assert!(!state.reload().await.expect("reload reads the same files")); } - #[test] - fn given_rotated_files_should_swap_and_keep_the_old_material_on_a_broken_rotation() { + #[tokio::test] + async fn given_rotated_files_should_swap_and_keep_the_old_material_on_a_broken_rotation() { let dir = tempfile::tempdir().expect("tempdir"); let (cert, key) = self_signed("first"); let cert_file = write(dir.path(), "cert.pem", &cert); @@ -454,6 +509,7 @@ mod tests { fs::write(&cert_file, &next_cert).expect("rotate certificate"); let error = state .reload() + .await .expect_err("a certificate that does not match the key must not go live"); assert!(error.to_string().contains("pair"), "got: {error}"); assert!( @@ -462,9 +518,34 @@ mod tests { ); fs::write(&key_file, &next_key).expect("rotate key"); - assert!(state.reload().expect("the completed rotation loads")); + assert!(state.reload().await.expect("the completed rotation loads")); assert!(!Arc::ptr_eq(&before, &state.current.load_full())); - assert!(!state.reload().expect("nothing changed since")); + assert!(!state.reload().await.expect("nothing changed since")); + } + + #[tokio::test] + async fn given_a_broken_rotation_should_report_it_on_every_reload_until_fixed() { + let dir = tempfile::tempdir().expect("tempdir"); + let (cert, key) = self_signed("first"); + let cert_file = write(dir.path(), "cert.pem", &cert); + let key_file = write(dir.path(), "key.pem", &key); + let state = TlsState::load(TlsConfig::new(&cert_file, &key_file)).expect("loads"); + + let (next_cert, next_key) = self_signed("second"); + fs::write(&cert_file, &next_cert).expect("rotate certificate"); + for attempt in 1..=2 { + let error = state + .reload() + .await + .expect_err("a mismatched pair must fail on every reload, not only the first"); + assert!( + error.to_string().contains("pair"), + "attempt {attempt}: {error}" + ); + } + + fs::write(&key_file, &next_key).expect("rotate key"); + assert!(state.reload().await.expect("the fixed rotation loads")); } #[test] diff --git a/gateways/kafka/tests/common/certs.rs b/gateways/kafka/tests/common/certs.rs index c7297fde9..7fa0dc601 100644 --- a/gateways/kafka/tests/common/certs.rs +++ b/gateways/kafka/tests/common/certs.rs @@ -173,6 +173,14 @@ pub async fn connect_tls( config: Arc<ClientConfig>, ) -> Result<TlsStream<TcpStream>, std::io::Error> { let tcp = TcpStream::connect(addr).await?; + connect_tls_over(tcp, config).await +} + +/// Runs the client handshake over a socket the caller already connected. +pub async fn connect_tls_over( + tcp: TcpStream, + config: Arc<ClientConfig>, +) -> Result<TlsStream<TcpStream>, std::io::Error> { let server_name = ServerName::try_from("localhost").expect("valid server name"); TlsConnector::from(config).connect(server_name, tcp).await } diff --git a/gateways/kafka/tests/delete_topics_real_bridge_tests.rs b/gateways/kafka/tests/delete_topics_real_bridge_tests.rs index 81adc88db..800880bbd 100644 --- a/gateways/kafka/tests/delete_topics_real_bridge_tests.rs +++ b/gateways/kafka/tests/delete_topics_real_bridge_tests.rs @@ -28,26 +28,44 @@ use std::sync::Arc; +use async_trait::async_trait; use bytes::Bytes; use iggy::prelude::{Identifier, StreamClient, TopicClient}; use serial_test::serial; +use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt}; +use tokio::net::TcpStream; +use tokio::sync::SemaphorePermit; use tokio_util::sync::CancellationToken; +use iggy_gateway_kafka::auth::{AuthError, AuthenticatedPrincipal, SaslAuthenticator}; use iggy_gateway_kafka::bridge::IggyBridge; use iggy_gateway_kafka::group::{GroupCoordinator, GroupCoordinatorConfig}; +use iggy_gateway_kafka::listener::SecurityProtocol; use iggy_gateway_kafka::protocol::api::{ - ERROR_INVALID_REQUEST, ERROR_NONE, ERROR_POLICY_VIOLATION, ERROR_UNKNOWN_TOPIC_OR_PARTITION, - GatewayState, + API_KEY_DELETE_TOPICS, ERROR_INVALID_REQUEST, ERROR_NONE, ERROR_POLICY_VIOLATION, + ERROR_TOPIC_AUTHORIZATION_FAILED, ERROR_UNKNOWN_TOPIC_OR_PARTITION, GatewayState, }; use iggy_gateway_kafka::protocol::handlers::delete_topics; +use iggy_gateway_kafka::protocol::sasl::PlainCredentials; +use iggy_gateway_kafka::tls::{ClientAuth, TlsConfig}; +use iggy_gateway_kafka::{GatewayConfig, KafkaGateway}; +#[path = "common/certs.rs"] +mod certs; #[path = "common/codec.rs"] mod codec; #[path = "common/iggy_server.rs"] mod iggy_server; +#[path = "common/server.rs"] +mod server; +#[path = "common/tcp.rs"] +mod tcp; +use certs::{TestPki, connect_tls}; use codec::{Decoder, Encoder}; use iggy_server::{TestServer, raw_client}; +use server::{listener, spawn_gateway}; +use tcp::{build_request_frame, parse_response_payload, read_response_frame}; const REQUEST_VERSION: i16 = 5; const TEST_MAX_FRAME_SIZE: usize = 8 * 1024 * 1024; @@ -332,3 +350,117 @@ async fn delete_topics_over_cap_error_message_names_the_limit() { "error_message must name the actual cap" ); } + +/// Refuses every login: these tests only need a SASL listener to exist beside the one they use. +#[derive(Debug)] +struct NoLogins; + +#[async_trait] +impl SaslAuthenticator for NoLogins { + async fn authenticate( + &self, + _credentials: &PlainCredentials, + _slot: SemaphorePermit<'_>, + ) -> Result<AuthenticatedPrincipal, AuthError> { + Err(AuthError::Rejected) + } +} + +/// Sends a v5 `DeleteTopics` for `names` over a gateway connection. +async fn delete_over<S: AsyncRead + AsyncWrite + Unpin>( + stream: &mut S, + names: &[&str], +) -> Vec<(Option<String>, i16)> { + let frame = build_request_frame( + API_KEY_DELETE_TOPICS, + REQUEST_VERSION, + 1, + Some("anonymous"), + &build_request(names), + ); + stream.write_all(&frame).await.expect("write DeleteTopics"); + let payload = read_response_frame(stream, TEST_MAX_FRAME_SIZE).await; + let (_, body) = parse_response_payload(API_KEY_DELETE_TOPICS, REQUEST_VERSION, payload); + decode_all_results(body) +} + +/// A bridge to `server` with one topic, `orders`, to delete. +async fn bridge_with_a_topic(server: &TestServer) -> Arc<IggyBridge> { + let bridge = IggyBridge::connect(server.test_config()) + .await + .expect("bridge should connect to a ready server"); + bridge + .create_kafka_topic("orders", 1) + .await + .expect("create the topic to delete"); + Arc::new(bridge) +} + +fn tls_material(pki: &TestPki, dir: &std::path::Path) -> TlsConfig { + let (cert_file, key_file) = pki.server_cert("gateway").write(dir, "server"); + TlsConfig::new(cert_file, key_file) +} + +#[tokio::test] +#[serial] +async fn given_requested_client_auth_when_no_certificate_is_presented_should_deny_delete_topics() { + let data_dir = tempfile::tempdir().expect("tempdir"); + let server = TestServer::spawn(data_dir.path()).await; + let bridge = bridge_with_a_topic(&server).await; + let certs_dir = tempfile::tempdir().expect("certs dir"); + let pki = TestPki::new(); + let ca_file = pki.write_ca(certs_dir.path()); + let config = GatewayConfig { + listeners: vec![listener(SecurityProtocol::Ssl)], + tls: Some( + tls_material(&pki, certs_dir.path()).with_client_auth(ClientAuth::Requested, ca_file), + ), + ..GatewayConfig::default() + }; + let (addrs, _shutdown) = spawn_gateway( + move |config| KafkaGateway::new(config).with_bridge(Some(bridge)), + config, + ); + + let mut stream = connect_tls(addrs[0], pki.client_config(None)) + .await + .expect("a requested certificate may be withheld"); + assert_eq!( + delete_over(&mut stream, &["orders"]).await, + vec![(Some("orders".to_string()), ERROR_TOPIC_AUTHORIZATION_FAILED)], + "a client that proved nothing holds less than any user, not more" + ); +} + +#[tokio::test] +#[serial] +async fn given_a_plaintext_listener_beside_sasl_ssl_should_deny_delete_topics_there() { + let data_dir = tempfile::tempdir().expect("tempdir"); + let server = TestServer::spawn(data_dir.path()).await; + let bridge = bridge_with_a_topic(&server).await; + let certs_dir = tempfile::tempdir().expect("certs dir"); + let pki = TestPki::new(); + let config = GatewayConfig { + listeners: vec![ + listener(SecurityProtocol::Plaintext), + listener(SecurityProtocol::SaslSsl), + ], + tls: Some(tls_material(&pki, certs_dir.path())), + ..GatewayConfig::default() + }; + let (addrs, _shutdown) = spawn_gateway( + move |config| { + KafkaGateway::new(config) + .with_bridge(Some(bridge)) + .with_authenticator(Arc::new(NoLogins)) + }, + config, + ); + + let mut stream = TcpStream::connect(addrs[0]).await.expect("connect"); + assert_eq!( + delete_over(&mut stream, &["orders"]).await, + vec![(Some("orders".to_string()), ERROR_TOPIC_AUTHORIZATION_FAILED)], + "the plaintext listener must not bypass what the SASL listener enforces" + ); +} diff --git a/gateways/kafka/tests/tls_listener_tests.rs b/gateways/kafka/tests/tls_listener_tests.rs index a922dd534..6530edaa6 100644 --- a/gateways/kafka/tests/tls_listener_tests.rs +++ b/gateways/kafka/tests/tls_listener_tests.rs @@ -23,13 +23,14 @@ //! certificate path against a real one is in `kafka_client_e2e_tests.rs`. use std::collections::HashMap; +use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Arc, Mutex}; use std::time::Duration; use async_trait::async_trait; use bytes::Bytes; -use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt}; -use tokio::net::TcpStream; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; +use tokio::net::{TcpSocket, TcpStream}; use tokio::sync::SemaphorePermit; use iggy_gateway_kafka::auth::{ @@ -51,7 +52,7 @@ mod server; #[path = "common/tcp.rs"] mod tcp; -use certs::{Issued, TestPki, connect_tls}; +use certs::{Issued, TestPki, connect_tls, connect_tls_over}; use codec::{Decoder, Encoder}; use server::{listener, spawn_gateway}; use tcp::{ @@ -144,6 +145,43 @@ impl PrincipalResolver for KnownUsers { } } +/// Answers every lookup the same way, or never answers. +#[derive(Debug)] +enum BrokenResolver { + Unavailable, + Hangs, +} + +#[async_trait] +impl PrincipalResolver for BrokenResolver { + async fn resolve(&self, _username: &str) -> Result<AuthenticatedPrincipal, AuthError> { + match self { + Self::Unavailable => Err(AuthError::Unavailable), + Self::Hangs => std::future::pending().await, + } + } +} + +/// Knows every user, after a fixed delay, and counts the lookups it started. +#[derive(Debug)] +struct SlowUsers { + delay: Duration, + started: AtomicUsize, +} + +#[async_trait] +impl PrincipalResolver for SlowUsers { + async fn resolve(&self, username: &str) -> Result<AuthenticatedPrincipal, AuthError> { + self.started.fetch_add(1, Ordering::SeqCst); + tokio::time::sleep(self.delay).await; + Ok(AuthenticatedPrincipal { + username: username.to_string(), + permissions: topic_reader(), + permissions_known: true, + }) + } +} + /// Accepts `USERNAME`/`PASSWORD` and nothing else. #[derive(Debug)] struct FixedCredentials; @@ -539,7 +577,8 @@ async fn given_requested_client_auth_when_no_certificate_is_presented_should_ser } #[tokio::test] -async fn given_a_sasl_ssl_listener_with_client_auth_should_take_the_principal_from_sasl() { +async fn given_client_auth_when_a_client_without_a_certificate_uses_sasl_ssl_should_authenticate_by_sasl() + { let pki = Pki::new(); let users = KnownUsers::with("certificate-user", topic_reader()); let resolver = Arc::clone(&users); @@ -550,18 +589,20 @@ async fn given_a_sasl_ssl_listener_with_client_auth_should_take_the_principal_fr .with_principal_resolver(resolver) }, quick_config( - vec![SecurityProtocol::SaslSsl], + vec![SecurityProtocol::Ssl, SecurityProtocol::SaslSsl], Some( pki.tls_with_client_auth(ClientAuth::Required) .with_principal_mapping("RULE:^CN=([^,]+).*$/$1/".parse().expect("valid")), ), ), ); + let [ssl, sasl_ssl] = addrs[..] else { + panic!("two listeners") + }; - let certificate = pki.authority.client_cert("certificate-user", "Example"); - let mut stream = connect_tls(addrs[0], pki.authority.client_config(Some(&certificate))) + let mut stream = connect_tls(sasl_ssl, pki.authority.client_config(None)) .await - .expect("mTLS handshake"); + .expect("SASL_SSL asks for no client certificate, as on a Kafka broker"); assert_eq!( sasl_plain(&mut stream, USERNAME, PASSWORD).await, ERROR_NONE @@ -569,13 +610,166 @@ async fn given_a_sasl_ssl_listener_with_client_auth_should_take_the_principal_fr let (code, body) = describe_acls(&mut stream).await; assert_eq!(code, ERROR_NONE); assert!( - mentions(&body, &format!("User:{USERNAME}")) && !mentions(&body, "certificate-user"), - "with SASL the certificate gates the transport only, as on a Kafka SASL_SSL listener" + mentions(&body, &format!("User:{USERNAME}")), + "the SASL principal is the one described" ); assert!( users.asked().is_empty(), - "the certificate's subject is never looked up on a SASL listener" + "nothing is looked up by certificate on a SASL listener" + ); + + let Ok(mut refused) = connect_tls(ssl, pki.authority.client_config(None)).await else { + return; + }; + let frame = build_request_frame(API_KEY_API_VERSIONS, 1, 1, Some("anonymous"), &[]); + let _ = refused.write_all(&frame).await; + assert_eq!( + read_byte_with_timeout(&mut refused, Duration::from_secs(5)).await, + ByteRead::Closed, + "the SSL listener beside it still requires a certificate" + ); +} + +#[tokio::test] +async fn given_a_large_response_when_the_tls_client_reads_slowly_should_deliver_all_of_it() { + let pki = Pki::new(); + // Small fixed buffers on both ends turn off autotuning, so the gateway's socket is full + // whenever rustls hands it the end of the response. + let socket = bind_listener("127.0.0.1:0").expect("bind"); + socket2::SockRef::from(&socket) + .set_send_buffer_size(16 * 1024) + .expect("send buffer"); + let addr = socket.local_addr().expect("local addr"); + let (_shutdown, rx) = tokio::sync::broadcast::channel(1); + let config = GatewayConfig { + write_timeout: Duration::from_secs(30), + ..quick_config(vec![SecurityProtocol::Ssl], Some(pki.tls())) + }; + tokio::spawn(KafkaGateway::new(config).run( + vec![BoundListener::new(listener(SecurityProtocol::Ssl), socket)], + rx, + )); + let client = TcpSocket::new_v4().expect("socket"); + client + .set_recv_buffer_size(16 * 1024) + .expect("receive buffer"); + let tcp = client.connect(addr).await.expect("connect"); + let mut stream = connect_tls_over(tcp, pki.authority.client_config(None)) + .await + .expect("TLS handshake"); + + // The stub echoes every requested topic name back, so the answer is about as large as the + // request. + let name = "t".repeat(16 * 1024); + let names: Vec<&str> = std::iter::repeat_n(name.as_str(), 128).collect(); + let request = build_metadata_legacy_request(&names); + let frame = build_request_frame(API_KEY_METADATA, 0, 7, Some("tls-test"), &request); + stream.write_all(&frame).await.expect("write request"); + + let read = tokio::time::timeout(Duration::from_secs(20), async { + let mut prefix = [0u8; 4]; + stream.read_exact(&mut prefix).await.expect("length prefix"); + let frame_len = usize::try_from(i32::from_be_bytes(prefix)).expect("positive length"); + let mut chunk = vec![0u8; 16 * 1024]; + let mut received = 0; + while received < frame_len { + let read = stream.read(&mut chunk).await.expect("read"); + assert_ne!(read, 0, "closed after {received} of {frame_len} bytes"); + received += read; + tokio::time::sleep(Duration::from_millis(1)).await; + } + frame_len + }) + .await; + let frame_len = read.expect("the whole response arrives; its tail must not wait for a flush"); + assert!(frame_len > 2 * 1024 * 1024, "got only {frame_len} bytes"); +} + +#[tokio::test] +async fn given_required_client_auth_when_no_rule_maps_the_subject_should_close() { + let pki = Pki::new(); + let users = KnownUsers::with(USERNAME, topic_reader()); + let resolver = Arc::clone(&users); + let (addrs, _shutdown) = spawn_gateway( + move |config| KafkaGateway::new(config).with_principal_resolver(resolver), + quick_config( + vec![SecurityProtocol::Ssl], + Some( + pki.tls_with_client_auth(ClientAuth::Required) + .with_principal_mapping( + "RULE:^CN=([^,]+),O=Allowed$/$1/".parse().expect("valid"), + ), + ), + ), + ); + + let unmapped = pki.authority.client_cert(USERNAME, "Example"); + let mut stream = connect_tls(addrs[0], pki.authority.client_config(Some(&unmapped))) + .await + .expect("the CA signed the certificate, so the handshake succeeds"); + let frame = build_request_frame(API_KEY_API_VERSIONS, 1, 1, Some("unmapped"), &[]); + let _ = stream.write_all(&frame).await; + assert_eq!( + read_byte_with_timeout(&mut stream, Duration::from_secs(5)).await, + ByteRead::Closed, + "a subject no rule maps names nobody, and is not served as anonymous" ); + assert!(users.asked().is_empty(), "nothing was looked up"); +} + +#[tokio::test] +async fn given_a_client_that_never_starts_the_handshake_should_close_it_within_the_budget() { + let pki = Pki::new(); + let (addrs, _shutdown) = spawn_gateway( + KafkaGateway::new, + GatewayConfig { + pre_auth_timeout: Duration::from_millis(300), + ..quick_config(vec![SecurityProtocol::Ssl], Some(pki.tls())) + }, + ); + + let mut stream = TcpStream::connect(addrs[0]).await.expect("connect"); + assert_eq!( + read_byte_with_timeout(&mut stream, Duration::from_secs(3)).await, + ByteRead::Closed, + "a silent client must not hold a connection slot past the pre-authentication budget" + ); +} + +#[tokio::test] +async fn given_a_lookup_that_fails_or_hangs_should_close_the_certificate_connection() { + for resolver in [BrokenResolver::Unavailable, BrokenResolver::Hangs] { + let pki = Pki::new(); + let label = format!("{resolver:?}"); + let resolver = Arc::new(resolver); + let (addrs, _shutdown) = spawn_gateway( + move |config| KafkaGateway::new(config).with_principal_resolver(resolver), + GatewayConfig { + pre_auth_timeout: Duration::from_millis(300), + ..quick_config( + vec![SecurityProtocol::Ssl], + Some( + pki.tls_with_client_auth(ClientAuth::Required) + .with_principal_mapping( + "RULE:^CN=([^,]+).*$/$1/".parse().expect("valid"), + ), + ), + ) + }, + ); + + let certificate = pki.authority.client_cert(USERNAME, "Example"); + let mut stream = connect_tls(addrs[0], pki.authority.client_config(Some(&certificate))) + .await + .expect("mTLS handshake"); + let frame = build_request_frame(API_KEY_API_VERSIONS, 1, 1, Some("lookup"), &[]); + let _ = stream.write_all(&frame).await; + assert_eq!( + read_byte_with_timeout(&mut stream, Duration::from_secs(3)).await, + ByteRead::Closed, + "{label}: an identity that cannot be confirmed admits nothing" + ); + } } #[tokio::test] @@ -661,6 +855,106 @@ async fn given_certificate_identities_without_a_lookup_should_refuse_to_start() ); } +#[tokio::test] +async fn given_a_lookup_queued_behind_another_should_get_its_own_budget_once_it_holds_the_slot() { + // Each lookup fits the budget alone, but the queued one's wait plus its own lookup does not: + // it is served only if the budget restarts when the slot is acquired. + let budget = Duration::from_millis(1000); + let pki = Pki::new(); + let users = Arc::new(SlowUsers { + delay: Duration::from_millis(800), + started: AtomicUsize::new(0), + }); + let resolver = Arc::clone(&users); + let (addrs, _shutdown) = spawn_gateway( + move |config| KafkaGateway::new(config).with_principal_resolver(resolver), + GatewayConfig { + pre_auth_timeout: budget, + max_concurrent_authentications: 1, + ..quick_config( + vec![SecurityProtocol::Ssl], + Some( + pki.tls_with_client_auth(ClientAuth::Required) + .with_principal_mapping("RULE:^CN=([^,]+).*$/$1/".parse().expect("valid")), + ), + ) + }, + ); + let certificate = pki.authority.client_cert(USERNAME, "Example"); + let client_config = pki.authority.client_config(Some(&certificate)); + + let mut first = connect_tls(addrs[0], Arc::clone(&client_config)) + .await + .expect("first mTLS handshake"); + while users.started.load(Ordering::SeqCst) == 0 { + tokio::time::sleep(Duration::from_millis(5)).await; + } + let mut queued = connect_tls(addrs[0], client_config) + .await + .expect("queued mTLS handshake"); + + let (first_rows, queued_rows) = + tokio::join!(api_versions(&mut first), api_versions(&mut queued)); + assert!(advertises(&first_rows, API_KEY_DESCRIBE_ACLS)); + assert!( + advertises(&queued_rows, API_KEY_DESCRIBE_ACLS), + "the queued lookup must not inherit what the wait left of a shared deadline" + ); +} + +#[tokio::test] +async fn given_one_listener_at_its_connection_limit_should_still_serve_another_listener() { + let (addrs, _shutdown) = spawn_gateway( + KafkaGateway::new, + GatewayConfig { + max_connections: 1, + ..quick_config( + vec![SecurityProtocol::Plaintext, SecurityProtocol::Plaintext], + None, + ) + }, + ); + + let mut held = TcpStream::connect(addrs[0]).await.expect("connect"); + assert!(advertises(&api_versions(&mut held).await, API_KEY_METADATA)); + let mut refused = TcpStream::connect(addrs[0]).await.expect("connect"); + assert_eq!( + read_byte_with_timeout(&mut refused, Duration::from_secs(3)).await, + ByteRead::Closed, + "the first listener is full" + ); + + let mut other = TcpStream::connect(addrs[1]).await.expect("connect"); + assert!( + advertises(&api_versions(&mut other).await, API_KEY_METADATA), + "each listener has its own connection pool" + ); +} + +#[tokio::test] +async fn given_client_auth_without_an_ssl_listener_should_refuse_to_start() { + let pki = Pki::new(); + let socket = bind_listener("127.0.0.1:0").expect("bind"); + let (_tx, rx) = tokio::sync::broadcast::channel(1); + let config = quick_config( + vec![SecurityProtocol::SaslSsl], + Some(pki.tls_with_client_auth(ClientAuth::Required)), + ); + let error = KafkaGateway::new(config) + .with_authenticator(Arc::new(FixedCredentials)) + .run( + vec![BoundListener::new( + listener(SecurityProtocol::SaslSsl), + socket, + )], + rx, + ) + .await + .unwrap_err() + .to_string(); + assert!(error.contains("no SSL listener"), "got: {error}"); +} + /// Kept for the suites that pin the single-listener shape: a `SASL_PLAINTEXT` listener on an /// ephemeral port is what `IGGY_KAFKA_SASL_ENABLED=true` used to mean. #[tokio::test]
