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 a5e4b6bb77392ecf57ab6ff99a596c3c79b69263 Merge: b277988ac c70c286cd Author: Grzegorz Koszyk <[email protected]> AuthorDate: Thu Oct 15 10:13:33 2026 +0200 merge master .../actions/python-maturin/pre-merge/action.yml | 2 +- .github/workflows/_common.yml | 4 +- .github/workflows/coverage-baseline.yml | 4 +- Cargo.lock | 2482 ++-- Cargo.toml | 98 +- bdd/go/tests/tcp_test/messages_feature_send.go | 4 +- bdd/go/tests/tcp_test/offset_feature_delete.go | 2 +- bdd/rust/tests/steps/leader_redirection.rs | 3 +- core/ai/mcp/src/service/mod.rs | 6 +- core/bench/dashboard/server/src/github/client.rs | 2 +- core/bench/src/analytics/report_builder.rs | 13 +- core/binary_protocol/src/codes.rs | 6 +- core/binary_protocol/src/consensus/command.rs | 11 +- core/binary_protocol/src/consensus/header.rs | 426 +- core/binary_protocol/src/consensus/mod.rs | 7 +- core/binary_protocol/src/consensus/request.rs | 19 +- core/binary_protocol/src/dispatch.rs | 4 +- core/binary_protocol/src/lib.rs | 20 +- core/binary_protocol/src/primitives/identifier.rs | 3 + .../src/primitives/partition_history.rs | 7 +- ...{get_send_context.rs => get_partition_route.rs} | 22 +- core/binary_protocol/src/requests/messages/mod.rs | 4 +- .../partitions/retire_consumer_group_owners.rs | 8 +- core/binary_protocol/src/responses/messages/mod.rs | 2 + .../src/responses/messages/partition_route.rs | 157 + .../src/responses/messages/poll_routing.rs | 1 + .../src/responses/system/get_cluster_metadata.rs | 45 +- .../binary_cluster/get_cluster_metadata.rs | 7 +- core/common/src/error/iggy_error.rs | 81 +- core/common/src/lib.rs | 4 +- core/common/src/traits/binary_impls/messages.rs | 19 +- core/common/src/traits/binary_impls/mod.rs | 58 +- .../traits/binary_impls/personal_access_tokens.rs | 39 +- core/common/src/traits/binary_impls/users.rs | 39 +- core/common/src/traits/binary_transport.rs | 71 +- core/common/src/traits/client.rs | 28 + core/common/src/traits/mod.rs | 2 +- core/common/src/types/cluster/node.rs | 8 +- core/common/src/wire_conversions.rs | 20 + core/configs/src/server_config/cluster.rs | 11 +- core/connectors/sinks/rabbitmq_sink/src/lib.rs | 42 +- core/consensus/src/client_table.rs | 15 +- core/consensus/src/impls.rs | 232 +- core/consensus/src/lib.rs | 7 + core/consensus/src/plane_helpers.rs | 285 +- core/integration/src/harness/handle/mcp.rs | 4 +- .../cli/system/test_cluster_metadata_command.rs | 4 +- .../tests/cluster/client_table_adversarial.rs | 1 + .../tests/cluster/client_table_restart.rs | 1 + core/integration/tests/cluster/partition_dedup.rs | 1 + .../tests/cluster/partition_primary_routing.rs | 570 +- .../tests/cluster/partition_state_transfer.rs | 100 +- .../connectors/fixtures/rabbitmq/container.rs | 14 +- .../fixtures/redshift/redshift_mock/handler.rs | 4 +- .../fixtures/redshift/redshift_mock/parser.rs | 8 +- core/integration/tests/connectors/s3/s3_sink.rs | 1 + .../verify_consumer_group_partition_assignment.rs | 89 +- .../tests/sdk/consumer_group_membership.rs | 15 +- core/integration/tests/server/http_client.rs | 17 + .../tests/server/http_read_your_writes.rs | 19 +- core/integration/tests/server/http_vsr.rs | 82 +- core/integration/tests/server/mod.rs | 3 + .../tests/server/pipelined_frames_vsr.rs | 303 + core/integration/tests/server/raw_tcp.rs | 7 + .../server/scenarios/authentication_scenario.rs | 6 +- core/journal/src/lib.rs | 13 +- core/journal/src/prepare_journal.rs | 75 +- core/message_bus/src/config.rs | 21 +- core/message_bus/src/lib.rs | 27 + core/message_bus/src/transports/mod.rs | 8 +- core/message_bus/src/transports/quic.rs | 860 +- core/metadata/src/impls/metadata.rs | 293 +- core/metadata/src/stm/consumer_group.rs | 346 +- core/metadata/src/stm/lifecycle.rs | 389 +- core/metadata/src/stm/mod.rs | 80 +- core/metadata/src/stm/result.rs | 153 +- core/metadata/src/stm/snapshot.rs | 28 +- core/metadata/src/stm/stream.rs | 141 +- core/partitions/Cargo.toml | 2 +- core/partitions/src/consumer_offset_capacity.rs | 14 + core/partitions/src/iggy_index_writer.rs | 13 +- core/partitions/src/iggy_partition.rs | 11751 ++++++++++++------- core/partitions/src/install_backup.rs | 84 +- core/partitions/src/io.rs | 144 +- core/partitions/src/journal.rs | 63 +- core/partitions/src/lib.rs | 11 +- core/partitions/src/log.rs | 7 +- core/partitions/src/messages_writer.rs | 13 +- core/partitions/src/offset_recovery.rs | 62 +- core/partitions/src/offset_storage.rs | 31 + core/partitions/src/partition_storage.rs | 104 +- core/partitions/src/persistence.rs | 67 +- core/partitions/src/segment_recovery.rs | 3 +- core/partitions/src/state_transfer.rs | 187 +- core/partitions/src/types.rs | 10 +- core/sdk/src/client_wrappers/binary_client.rs | 21 + core/sdk/src/clients/client.rs | 110 +- core/sdk/src/clients/consumer.rs | 1 + core/sdk/src/clients/producer.rs | 10 +- core/sdk/src/driver.rs | 1325 +++ .../src/{poll_routing.rs => driver/poll_router.rs} | 185 +- core/sdk/src/driver/route_cache.rs | 381 + core/sdk/src/leader_aware.rs | 611 +- core/sdk/src/lib.rs | 4 +- core/sdk/src/{leader_aware.rs => links/connect.rs} | 524 +- core/sdk/src/links/dial.rs | 324 + core/sdk/src/links/fake_cluster.rs | 442 + core/sdk/src/links/mod.rs | 448 + core/sdk/src/links/mux.rs | 815 ++ core/sdk/src/links/quic.rs | 335 + core/sdk/src/links/replicas.rs | 1571 +++ core/sdk/src/quic/quic_client.rs | 791 +- core/sdk/src/session.rs | 16 +- core/sdk/src/session_owner.rs | 601 + core/sdk/src/session_owner/lanes.rs | 606 + core/sdk/src/tcp/mod.rs | 4 - core/sdk/src/tcp/tcp_client.rs | 1527 +-- core/sdk/src/tcp/tcp_connection_stream.rs | 108 - core/sdk/src/tcp/tcp_connection_stream_kind.rs | 66 - core/sdk/src/tcp/tcp_stream.rs | 27 - core/sdk/src/tcp/tcp_tls_connection_stream.rs | 105 - core/sdk/src/vsr.rs | 396 +- core/sdk/src/websocket/mod.rs | 4 - core/sdk/src/websocket/websocket_client.rs | 912 +- .../src/websocket/websocket_connection_stream.rs | 212 - core/sdk/src/websocket/websocket_stream.rs | 36 - core/sdk/src/websocket/websocket_stream_kind.rs | 58 - .../websocket/websocket_tls_connection_stream.rs | 217 - core/server/README.md | 22 +- core/server/config.toml | 33 +- core/server/src/boot/mod.rs | 61 +- core/server/src/boot/topology.rs | 12 +- core/server/src/cluster_meta.rs | 39 + core/server/src/consumer_group.rs | 9 +- core/server/src/consumer_group/liveness.rs | 10 +- core/server/src/dispatch/authz.rs | 10 +- core/server/src/dispatch/failure.rs | 73 +- core/server/src/dispatch/host.rs | 40 +- core/server/src/dispatch/mod.rs | 1482 ++- core/server/src/dispatch/partition.rs | 866 +- core/server/src/dispatch/reads.rs | 93 +- core/server/src/dispatch/submit.rs | 1386 ++- core/server/src/dispatch/test_support.rs | 91 +- core/server/src/http.rs | 38 +- core/server/src/http/error.rs | 17 +- core/server/src/http/forward.rs | 965 +- core/server/src/http/handlers.rs | 268 +- core/server/src/http/reply.rs | 8 + core/server/src/http/state.rs | 7 + core/server/src/http/submit.rs | 98 +- core/server/src/lib.rs | 8 +- core/server/src/namespace.rs | 34 +- core/server/src/partition_helpers.rs | 5 +- core/server/src/partition_reconciler.rs | 702 +- core/server/src/reply_frame.rs | 12 +- core/server/src/rewrite.rs | 42 +- core/server_common/src/consensus_message.rs | 16 +- core/shard/src/coordinator.rs | 5 + core/shard/src/lib.rs | 1224 +- core/shard/src/partition_io.rs | 2694 ++++- core/shard/src/poll.rs | 87 +- core/shard/src/poll/completion_tests.rs | 229 +- core/simulator/src/bin/workload-fuzz.rs | 64 +- core/simulator/src/client.rs | 27 +- core/simulator/src/deps.rs | 21 +- core/simulator/src/lib.rs | 2468 +++- core/simulator/src/packet.rs | 3 +- core/simulator/src/replica.rs | 6 +- core/simulator/src/storage/tests.rs | 209 +- core/simulator/src/workload/actions.rs | 4 + core/simulator/src/workload/auditor.rs | 18 +- core/simulator/src/workload/invariants.rs | 30 +- core/simulator/src/workload/mod.rs | 2481 +++- .../src/workload/ops/join_fixture_group.rs | 96 + .../src/workload/ops/leave_fixture_group.rs | 84 + core/simulator/src/workload/ops/mod.rs | 12 +- .../src/workload/ops/replace_fixture_topic.rs | 112 + .../src/workload/ops/store_stale_owner_offset.rs | 84 + core/simulator/src/workload/options.rs | 53 +- core/simulator/src/workload/oracle.rs | 384 +- core/simulator/src/workload/shadow.rs | 550 +- core/simulator/src/workload/state_checker.rs | 568 +- foreign/cpp/Cargo.toml | 6 +- foreign/cpp/src/lib.rs | 1 + foreign/cpp/src/type_conversion.rs | 1 + foreign/csharp/Directory.Packages.props | 8 +- .../Fixtures/VsrCluster.cs | 9 + .../Helpers/ConsumerGroupAssignment.cs | 4 +- .../Vsr/VsrMetadataTests.cs | 3 +- foreign/csharp/Iggy_SDK/Contracts/ClusterNode.cs | 3 + .../csharp/Iggy_SDK/Contracts/PartitionRoute.cs | 14 +- .../Exceptions/IggyInvalidStatusCodeException.cs | 7 + .../TcpMessageStream.PollRouting.cs | 449 +- .../Implementations/TcpMessageStream.Vsr.cs | 73 +- .../IggyClient/Implementations/TcpMessageStream.cs | 101 +- foreign/csharp/Iggy_SDK/Mappers/BinaryMapper.cs | 21 +- foreign/csharp/Iggy_SDK/Utils/CommandCodes.cs | 2 +- foreign/csharp/Iggy_SDK/Vsr/VsrConnection.cs | 61 +- foreign/csharp/Iggy_SDK/Vsr/VsrError.cs | 13 +- foreign/csharp/Iggy_SDK/Vsr/VsrHeader.cs | 4 +- foreign/csharp/Iggy_SDK/Vsr/VsrOperation.cs | 2 +- foreign/csharp/Iggy_SDK/Vsr/VsrReplyDecoder.cs | 3 +- .../ConsumerTests/OffsetContinuationTests.cs | 9 +- .../Iggy_SDK_Tests/MapperTests/BinaryMapper.cs | 59 +- .../VsrTests/EndpointFailoverTests.cs | 249 +- .../Iggy_SDK_Tests/VsrTests/GroupPollingTests.cs | 1 + foreign/csharp/Iggy_SDK_Tests/VsrTests/MockNode.cs | 41 +- .../VsrTests/PartitionContextTests.cs | 54 +- .../Iggy_SDK_Tests/VsrTests/PrimaryPollingTests.cs | 487 +- .../VsrTests/VsrReplyDecoderTests.cs | 10 +- .../Iggy_SDK_Tests/VsrTests/VsrTestPayloads.cs | 5 + foreign/go/README.md | 2 +- foreign/go/client/tcp/cluster.go | 12 + foreign/go/client/tcp/tcp_core.go | 73 +- foreign/go/client/tcp/tcp_core_review_test.go | 2 +- foreign/go/client/tcp/tcp_core_test.go | 35 +- foreign/go/client/tcp/tcp_failover_test.go | 4 +- foreign/go/client/tcp/tcp_messaging.go | 174 +- foreign/go/client/tcp/tcp_offset_management.go | 27 +- foreign/go/client/tcp/tcp_partition_management.go | 3 + foreign/go/client/tcp/tcp_poll_routing.go | 248 +- foreign/go/client/tcp/tcp_poll_routing_test.go | 498 +- foreign/go/client/tcp/tcp_stream_management.go | 1 + foreign/go/client/tcp/tcp_testing_test.go | 19 +- foreign/go/client/tcp/tcp_topic_cache.go | 44 +- foreign/go/client/tcp/tcp_topic_cache_test.go | 71 +- foreign/go/contracts/cluster_metadata_test.go | 56 +- foreign/go/contracts/cluster_node.go | 10 +- foreign/go/contracts/partitions.go | 26 + foreign/go/errors/errors.yaml | 4 + foreign/go/errors/errors_gen.go | 15 + foreign/go/internal/command/code.go | 2 +- foreign/go/internal/vsr/envelope.go | 5 + foreign/go/internal/vsr/envelope_test.go | 4 +- foreign/go/internal/vsr/session.go | 46 +- foreign/go/internal/vsr/session_test.go | 40 +- foreign/go/tests/e2e_test.go | 29 +- .../iggy/client/async/tcp/AsyncIggyTcpClient.java | 69 +- .../iggy/client/async/tcp/AsyncTcpConnection.java | 137 +- .../iggy/client/async/tcp/ClientRoutingState.java | 36 +- .../client/async/tcp/ConsumerOffsetsTcpClient.java | 49 +- .../iggy/client/async/tcp/MessagesTcpClient.java | 146 +- .../iggy/client/async/tcp/PartitionContexts.java | 191 - .../apache/iggy/client/async/tcp/PollRouter.java | 429 +- .../client/async/tcp/RawPartitionRequests.java | 12 +- .../client/async/tcp/vsr/ConsensusSession.java | 21 +- .../iggy/client/async/tcp/vsr/VsrHeaders.java | 2 +- .../client/async/tcp/vsr/VsrRequestEncoder.java | 13 +- .../java/org/apache/iggy/cluster/ClusterNode.java | 14 +- .../org/apache/iggy/exception/IggyErrorCode.java | 1 + .../org/apache/iggy/partition/PartitionRoute.java | 47 + .../org/apache/iggy/serde/BytesDeserializer.java | 5 +- .../java/org/apache/iggy/serde/CommandCode.java | 2 +- .../AsyncIggyTcpClientEndpointFailoverTest.java | 1 + .../AsyncIggyTcpClientPartitionContextTest.java | 141 +- .../tcp/AsyncIggyTcpClientRawRequestTest.java | 192 +- .../AsyncIggyTcpClientSendContextCacheTest.java | 99 +- .../AsyncIggyTcpClientTransientFailoverTest.java | 128 +- .../tcp/AsyncTcpConnectionLifecycleBusyTest.java | 52 +- .../client/async/tcp/ClientRoutingStateTest.java | 40 - .../iggy/client/async/tcp/LeaderAwarenessTest.java | 2 +- .../iggy/client/async/tcp/PollRouterTest.java | 102 +- .../async/tcp/vsr/VsrRequestEncoderTest.java | 35 +- .../client/blocking/http/ObjectMapperTest.java | 19 + .../apache/iggy/partition/PartitionRouteTest.java | 65 + .../apache/iggy/serde/BytesDeserializerTest.java | 3 + foreign/node/src/client/client.socket.test.ts | 521 +- foreign/node/src/client/client.socket.ts | 461 +- foreign/node/src/e2e/tcp.cluster.e2e.ts | 1 + foreign/node/src/wire/cluster/cluster.utils.ts | 6 +- foreign/node/src/wire/command.code.ts | 2 +- foreign/node/src/wire/error.code.ts | 1 + .../src/wire/message/send-messages.command.test.ts | 52 +- .../node/src/wire/message/send-messages.command.ts | 56 +- foreign/node/src/wire/offset/get-offset.command.ts | 2 +- foreign/node/src/wire/offset/offset.utils.ts | 12 +- .../src/wire/partition/partition-route.test.ts | 46 + foreign/node/src/wire/partition/partition-route.ts | 51 + foreign/node/src/wire/vsr/index.ts | 2 +- foreign/node/src/wire/vsr/session.ts | 5 +- foreign/node/src/wire/vsr/vsr.test.ts | 4 +- foreign/python/Cargo.toml | 6 +- foreign/swift/Package.resolved | 60 + foreign/swift/Package.swift | 24 +- foreign/swift/README.md | 74 +- .../Sources/Iggy/Client/ClientConfiguration.swift | 142 + foreign/swift/Sources/Iggy/Client/ClientCore.swift | 1082 ++ .../Sources/Iggy/Client/IggyClient+Resources.swift | 195 + foreign/swift/Sources/Iggy/Client/IggyClient.swift | 239 + foreign/swift/Sources/Iggy/Errors/IggyError.swift | 28 + .../swift/Sources/Iggy/Errors/IggyErrorCode.swift | 8 + foreign/swift/Sources/Iggy/Models/Resources.swift | 4 +- .../swift/Sources/Iggy/Models/TopicOptions.swift | 3 + .../swift/Sources/Iggy/Protocol/CommandCode.swift | 16 +- foreign/swift/Sources/Iggy/Protocol/Requests.swift | 6 +- .../swift/Sources/Iggy/Protocol/Responses.swift | 3 +- .../swift/Sources/Iggy/Transport/Endpoint.swift | 155 + .../Sources/Iggy/Transport/TCPConnection.swift | 295 + .../swift/Sources/Iggy/Transport/TLSOptions.swift | 67 + .../swift/Sources/Iggy/Utilities/AsyncMutex.swift | 201 + .../Tests/IggyTests/ClientTransportTests.swift | 1168 ++ .../swift/Tests/IggyTests/ConcurrencyTests.swift | 209 + .../swift/Tests/IggyTests/ConfigurationTests.swift | 120 + foreign/swift/Tests/IggyTests/Fixtures/golden.json | 8 +- foreign/swift/Tests/IggyTests/ProtocolTests.swift | 6 +- .../Tests/IggyTests/ResponseVectorTests.swift | 2 + .../Tests/IggyTests/Support/FakeVSRServer.swift | 331 + .../Tests/IggyTests/Support/TestHelpers.swift | 64 + foreign/swift/Tools/golden-vectors/src/main.rs | 2 + gateways/kafka/README.md | 65 +- gateways/kafka/docs/CONFIGS.md | 31 +- gateways/kafka/docs/IDEMPOTENCE.md | 37 +- gateways/kafka/docs/MANUAL_TESTING.md | 4 +- gateways/kafka/docs/SCOPE.md | 11 +- gateways/kafka/src/bridge/config.rs | 87 +- gateways/kafka/src/bridge/iggy_bridge/fetch.rs | 34 +- .../kafka/src/bridge/iggy_bridge/group_offsets.rs | 20 +- gateways/kafka/src/bridge/iggy_bridge/mod.rs | 114 +- gateways/kafka/src/bridge/iggy_bridge/produce.rs | 819 +- gateways/kafka/src/bridge/mod.rs | 6 +- gateways/kafka/src/main.rs | 43 +- gateways/kafka/src/protocol/api.rs | 52 +- .../kafka/src/protocol/handlers/create_topics.rs | 125 +- .../src/protocol/handlers/describe_configs.rs | 42 +- .../src/protocol/handlers/init_producer_id.rs | 46 +- gateways/kafka/src/protocol/handlers/mod.rs | 2 +- gateways/kafka/src/protocol/handlers/produce.rs | 479 +- .../kafka/src/protocol/handlers/topic_config.rs | 55 +- gateways/kafka/src/records.rs | 149 +- gateways/kafka/src/server.rs | 27 +- .../kafka/tests/bridge_iggy_integration_tests.rs | 6 +- gateways/kafka/tests/common/iggy_server.rs | 19 +- gateways/kafka/tests/configs_java_client_tests.rs | 5 +- gateways/kafka/tests/configs_real_bridge_tests.rs | 18 +- .../kafka/tests/create_topics_real_bridge_tests.rs | 45 + gateways/kafka/tests/java/ConfigsProbe.java | 17 +- gateways/kafka/tests/kafka_client_e2e_tests.rs | 5 +- gateways/kafka/tests/produce_real_bridge_tests.rs | 189 +- gateways/kafka/tests/response_negative_tests.rs | 11 +- 339 files changed, 49991 insertions(+), 14847 deletions(-) diff --cc Cargo.lock index e3d2dfb09,aa6908221..a23619e46 --- a/Cargo.lock +++ b/Cargo.lock @@@ -7130,19 -7258,17 +7263,19 @@@ dependencies = "serde_json", "serial_test", "snap", - "socket2 0.6.5", + "socket2", "tempfile", - "thiserror 2.0.20", + "thiserror 2.0.21", "tokio", + "tokio-rustls", "tokio-util", - "toml 1.1.6+spec-1.1.0", + "toml 1.1.8+spec-1.1.0", "tracing", "tracing-appender", "tracing-subscriber", "uuid", - "x509-parser 0.18.1", - "zstd 0.13.3", ++ "x509-parser", + "zstd 0.14.2", ] [[package]] diff --cc gateways/kafka/README.md index bfcac9934,3c83aa194..bd5e985ba --- a/gateways/kafka/README.md +++ b/gateways/kafka/README.md @@@ -61,10 -53,10 +61,10 @@@ Default bind: a `PLAINTEXT` listener o | `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 | | `IGGY_KAFKA_WRITE_TIMEOUT_SECS` | `10` | Seconds allowed to write a response frame | - | `IGGY_KAFKA_SHUTDOWN_DRAIN_TIMEOUT_SECS` | `25` | Seconds graceful shutdown waits for in-flight connections before abandoning them | + | `IGGY_KAFKA_SHUTDOWN_DRAIN_TIMEOUT_SECS` | `25` | Seconds graceful shutdown waits for in-flight connections before abandoning them. The bridge then signs its Iggy clients out, 5 s at most | | `IGGY_KAFKA_INSTANCE_ID` | `0` | This gateway's number among the gateways fronting one Iggy cluster. It is the high half of every producer id `InitProducerId` hands out, and Kafka requires those to be cluster-unique, so give every gateway its own value. A single gateway can leave it at `0`. | | `IGGY_KAFKA_BRIDGE_ENABLED` | `false` | Connect the Iggy bridge at startup. While false every API answers with its stub, and the `IGGY_KAFKA_IGGY_*` variables below are read by nothing. A failed connection is fatal, not a downgrade to stubs. | -| `IGGY_KAFKA_SASL_ENABLED` | `false` | Require SASL/PLAIN authentication before serving any other API (`true` or `false`, nothing else) | +| `IGGY_KAFKA_SASL_ENABLED` | `false` | Single-listener form: require SASL/PLAIN before serving any other API (`true` or `false`, nothing else). With `IGGY_KAFKA_LISTENERS`, a `SASL_PLAINTEXT` or `SASL_SSL` listener says the same | | `IGGY_KAFKA_PRE_AUTH_TIMEOUT_SECS` | `15` | Seconds an unauthenticated connection may sit between frames. Waiting for an authentication slot and the verification itself each get this budget, the verification's starting once it holds a slot. Separate from the 10-minute idle timeout that applies once authenticated | | `IGGY_KAFKA_MAX_CONCURRENT_AUTHENTICATIONS` | `4` | Credential verifications the gateway runs at once, across all connections. Each costs a password hash on an Iggy shard thread. This bounds the gateway's side only: a check that times out frees its slot while its hash keeps running inside Iggy. Size it below the Iggy node's shard count | diff --cc gateways/kafka/src/bridge/config.rs index 7c5e39874,4f77661b1..7fc284277 --- a/gateways/kafka/src/bridge/config.rs +++ b/gateways/kafka/src/bridge/config.rs @@@ -38,64 -38,25 +39,83 @@@ const DEFAULT_IGGY_USERNAME: &str = "ig /// Iggy's default `message_bus.max_message_size`. pub const DEFAULT_MAX_MESSAGE_SIZE: u64 = 64 * 1024 * 1024; +/// How the gateway reaches Iggy: the bridge's own clients and the SASL credential check alike. +/// +/// Separate from the gateway's own listener security. A deployment can terminate TLS on the Kafka +/// side and still speak plain TCP to a co-located Iggy, or the reverse, and the two are configured +/// independently because they protect different hops. +#[derive(Debug, Clone, Default)] +pub struct IggyTls { + pub enabled: bool, + /// Name checked against the server certificate. Empty means derive it from the address. + pub domain: String, + /// PEM roots to trust. Unset uses the SDK's bundled roots, not the system trust store. + pub ca_file: Option<String>, +} + +impl IggyTls { + /// The `IGGY_KAFKA_*` vars [`Self::from_env`] reads. + pub const KNOWN_ENV_VARS: &'static [&'static str] = &[ + "IGGY_KAFKA_IGGY_TLS_ENABLED", + "IGGY_KAFKA_IGGY_TLS_DOMAIN", + "IGGY_KAFKA_IGGY_TLS_CA_FILE", + ]; + + /// Reads the transport security of the Iggy hop from the environment. + /// + /// # Errors + /// + /// Returns a message naming the offending variable when `IGGY_KAFKA_IGGY_TLS_ENABLED` is set + /// to anything but `true` or `false`, or when material is configured with the switch off. + /// Defaulting a mistyped security switch to off is how a deployment ends up sending + /// passwords in the clear while believing it does not. + pub fn from_env() -> Result<Self, String> { + // Unset reads as off; only a value that is set and unrecognised is an error, which is + // what `parse_bool` decides for every switch in this gateway. + let enabled = match std::env::var("IGGY_KAFKA_IGGY_TLS_ENABLED") { + Ok(raw) => parse_bool("IGGY_KAFKA_IGGY_TLS_ENABLED", &raw)?, + Err(_) => false, + }; + let domain = std::env::var("IGGY_KAFKA_IGGY_TLS_DOMAIN").unwrap_or_default(); + let ca_file = std::env::var("IGGY_KAFKA_IGGY_TLS_CA_FILE").ok(); + // Configuring TLS material and leaving the switch off is the same "believes TLS is on" + // mistake the value check above exists to catch, reached through the unset path instead of + // a typo. Silently ignoring it would send credentials in the clear. + if !enabled && (!domain.is_empty() || ca_file.is_some()) { + return Err( + "IGGY_KAFKA_IGGY_TLS_DOMAIN or IGGY_KAFKA_IGGY_TLS_CA_FILE is set but \ + IGGY_KAFKA_IGGY_TLS_ENABLED is not true; the connection to Iggy would be in the \ + clear despite the TLS configuration" + .to_string(), + ); + } + Ok(Self { + enabled, + domain, + ca_file, + }) + } +} + + /// How many producers get an Iggy session of their own, and how long one stays open unused. + #[derive(Debug, Clone, Copy, PartialEq, Eq)] + pub struct ProduceSessionLimits { + /// Sessions open at once, each an Iggy connection. Producers past it share one more. With 0 + /// every producer shares that one, and sends run one at a time. + pub max: usize, + /// A session unused this long closes. + pub idle: Duration, + } + + impl Default for ProduceSessionLimits { + fn default() -> Self { + Self { + max: 256, + idle: Duration::from_mins(10), + } + } + } + /// Connection + topic-mapping config for [`IggyBridge`](crate::bridge::iggy_bridge::IggyBridge). /// /// `Debug` is safe to derive: `password` is `SecretString`, which redacts on `Debug` by design @@@ -109,8 -70,7 +129,9 @@@ pub struct IggyBridgeConfig pub topic_mapping: TopicMapping, /// Iggy's `message_bus.max_message_size`, in bytes. Larger partitions answer 10. pub max_message_size: u64, + /// Transport security of every client the bridge opens. + pub tls: IggyTls, + pub produce_sessions: ProduceSessionLimits, } impl IggyBridgeConfig { @@@ -129,9 -89,8 +150,11 @@@ "IGGY_KAFKA_IGGY_STREAM", "IGGY_KAFKA_TOPIC_MAP_PATH", "IGGY_KAFKA_IGGY_MAX_MESSAGE_SIZE", + "IGGY_KAFKA_IGGY_TLS_ENABLED", + "IGGY_KAFKA_IGGY_TLS_DOMAIN", + "IGGY_KAFKA_IGGY_TLS_CA_FILE", + "IGGY_KAFKA_PRODUCE_SESSIONS", + "IGGY_KAFKA_PRODUCE_SESSION_IDLE_SECS", ]; /// Builds config from `IGGY_KAFKA_*` env vars, defaulting to the Iggy server's own @@@ -226,7 -186,24 +250,25 @@@ })?, }; + let tls = IggyTls::from_env().map_err(BridgeError::InvalidConfig)?; + let mut produce_sessions = ProduceSessionLimits::default(); + if let Ok(value) = std::env::var("IGGY_KAFKA_PRODUCE_SESSIONS") { + produce_sessions.max = value.parse().map_err(|_| { + BridgeError::InvalidConfig(format!( + "IGGY_KAFKA_PRODUCE_SESSIONS '{value}' is not a count, for example 256" + )) + })?; + } + if let Ok(value) = std::env::var("IGGY_KAFKA_PRODUCE_SESSION_IDLE_SECS") { + // 0 would close a session between two requests of a busy producer. + let seconds = value.parse().ok().filter(|seconds| *seconds > 0); + produce_sessions.idle = seconds.map(Duration::from_secs).ok_or_else(|| { + BridgeError::InvalidConfig(format!( + "IGGY_KAFKA_PRODUCE_SESSION_IDLE_SECS '{value}' is not a number of seconds \ + above 0, for example 600" + )) + })?; + } Ok(Self { address, @@@ -234,7 -211,7 +276,8 @@@ password: SecretString::from(password), topic_mapping, max_message_size, + tls, + produce_sessions, }) } } @@@ -268,7 -230,7 +311,8 @@@ mod tests topic_mapping: TopicMapping::new("kafka".to_string(), std::collections::HashMap::new()) .expect("valid mapping for this test's fixture data"), max_message_size: DEFAULT_MAX_MESSAGE_SIZE, + tls: IggyTls::default(), + produce_sessions: ProduceSessionLimits::default(), } } @@@ -303,9 -265,8 +347,11 @@@ std::env::remove_var("IGGY_KAFKA_IGGY_STREAM"); std::env::remove_var("IGGY_KAFKA_TOPIC_MAP_PATH"); std::env::remove_var("IGGY_KAFKA_IGGY_MAX_MESSAGE_SIZE"); + std::env::remove_var("IGGY_KAFKA_IGGY_TLS_ENABLED"); + std::env::remove_var("IGGY_KAFKA_IGGY_TLS_DOMAIN"); + std::env::remove_var("IGGY_KAFKA_IGGY_TLS_CA_FILE"); + std::env::remove_var("IGGY_KAFKA_PRODUCE_SESSIONS"); + std::env::remove_var("IGGY_KAFKA_PRODUCE_SESSION_IDLE_SECS"); } let result = IggyBridgeConfig::from_env(); unsafe { @@@ -314,10 -275,7 +360,11 @@@ let config = result.expect("valid config from documented defaults alone"); assert_eq!(config.max_message_size, DEFAULT_MAX_MESSAGE_SIZE); + assert!( + !config.tls.enabled, + "the Iggy hop is in the clear unless switched on" + ); + assert_eq!(config.produce_sessions, ProduceSessionLimits::default()); assert_eq!(config.address, "127.0.0.1:8090"); assert_eq!(config.username, "iggy"); assert_eq!(config.topic_mapping.default_stream(), "kafka"); diff --cc gateways/kafka/src/bridge/iggy_bridge/fetch.rs index e404d8fc4,fd42f5030..924b07ef2 --- a/gateways/kafka/src/bridge/iggy_bridge/fetch.rs +++ b/gateways/kafka/src/bridge/iggy_bridge/fetch.rs @@@ -129,24 -153,7 +153,24 @@@ impl LazyClient result } + /// [`Self::until`] on a task of its own, so the deadline still drops a stuck client when the + /// caller stops waiting first. + pub(super) async fn until_detached<T, Call>( + &self, + deadline: Instant, + call: Call, + ) -> Result<T, BridgeError> + where + T: Send + 'static, + Call: Future<Output = Result<T, BridgeError>> + Send + 'static, + { + let client = self.clone(); + tokio::spawn(async move { client.until(deadline, call).await }) + .await + .unwrap_or(Err(BridgeError::Timeout)) + } + - /// Shuts the client down, if it connected. + /// Signs the client out and shuts it down, if it connected. pub(super) async fn close(&self) -> Result<(), BridgeError> { let Some(client) = self.0.lock().await.take() else { return Ok(()); diff --cc gateways/kafka/src/bridge/iggy_bridge/mod.rs index 18ee64c97,fc068e8ec..d945a6f0f --- a/gateways/kafka/src/bridge/iggy_bridge/mod.rs +++ b/gateways/kafka/src/bridge/iggy_bridge/mod.rs @@@ -19,15 -19,14 +19,16 @@@ use std::future::Future use std::sync::Arc; use std::time::Duration; +use async_trait::async_trait; use iggy::prelude::{ AutoLogin, Client, Credentials, Identifier, IggyClient, IggyClientBuilder, IggyError, + UserClient, }; use tokio::sync::{OwnedSemaphorePermit, Semaphore, oneshot}; use tokio::time::{Instant, timeout_at}; - use tracing::{info, warn}; -use tracing::{debug, info}; ++use tracing::{debug, info, warn}; +use crate::auth::{AuthError, AuthenticatedPrincipal, PrincipalResolver, resolve_principal}; use crate::bridge::config::IggyBridgeConfig; use crate::bridge::error::BridgeError; use crate::bridge::topic_map::validate_kafka_topic_name; @@@ -116,9 -130,10 +132,10 @@@ async fn permit_by(turn: &Arc<Semaphore /// Owns the connected `IggyClient`s and resolves Kafka topics against them. /// - /// One lockstep client serves Produce, Metadata and `CreateTopics` for every Kafka connection, so - /// those Iggy calls run one at a time. Topic probes and certificate user lookups each use a client - /// of their own, Fetch polls one per read slot, and offset calls two per offset slot. A Produce pool is a TODO in `docs/SCOPE.md`. + /// Each producer sends on a client of its own, one request at a time. Producers past the session + /// limit share one more. Metadata, `CreateTopics`, `DeleteTopics`, `DescribeConfigs` and -/// `AlterConfigs` share another. Topic probes use a client of their own, Fetch polls one per read -/// slot, and offset calls two per offset slot. ++/// `AlterConfigs` share another. Topic probes and certificate user lookups each use a client of ++/// their own, Fetch polls one per read slot, and offset calls two per offset slot. /// /// Each client signs in at the metadata leader, and a view change or a refused call can move it to /// another node. So in a cluster, two clients can read replicas that are not at the same offset. @@@ -169,10 -186,9 +190,10 @@@ impl IggyBridge Ok(Self { client: Arc::new(client), config, - send_slot: Arc::new(Semaphore::new(1)), + produce_pool, fetch_pool: Arc::new(FetchPool::new()), probe_client: LazyClient::default(), + lookup_client: LazyClient::default(), offset_pool: OffsetPool::new(), }) } @@@ -226,20 -243,56 +248,67 @@@ /// [`BridgeError::Iggy`] if the underlying client reports a shutdown failure (e.g. the socket /// was already in a state that rejects a clean shutdown). Every client is shut down even when /// one fails, and the error is the first one. - pub async fn close(self) -> Result<(), BridgeError> { - let fetch_clients = self.fetch_pool.close().await; - let probe_client = self.probe_client.close().await; - let lookup_client = self.lookup_client.close().await; - let offset_clients = self.offset_pool.close().await; - let shared_client = with_request_timeout(self.client.shutdown()).await; - fetch_clients + pub async fn close(&self) -> Result<(), BridgeError> { + // At once, so a client that is slow to close does not use up the time of the others. - let (produce_clients, fetch_clients, probe_client, offset_clients, shared_client) = tokio::join!( ++ let ( ++ produce_clients, ++ fetch_clients, ++ probe_client, ++ lookup_client, ++ offset_clients, ++ shared_client, ++ ) = tokio::join!( + self.produce_pool.close(), + self.fetch_pool.close(), + self.probe_client.close(), ++ self.lookup_client.close(), + self.offset_pool.close(), + sign_out_and_shut_down(&self.client), + ); + produce_clients + .and(fetch_clients) .and(probe_client) + .and(lookup_client) .and(offset_clients) .and(shared_client) } } + #[cfg(test)] + impl IggyBridge { + /// A bridge that never dialed, so a test needs no server. Each Iggy call on it fails at once. + pub(crate) fn offline() -> Self { + let address = "127.0.0.1:1".to_string(); + let client = IggyClientBuilder::new() + .with_tcp() + .with_server_address(address.clone()) + .build() + .expect("building a client dials nothing"); + let config = IggyBridgeConfig { + address, + username: "iggy".to_string(), + password: secrecy::SecretString::from("iggy"), + topic_mapping: crate::bridge::TopicMapping::new( + "kafka".to_string(), + std::collections::HashMap::new(), + ) + .expect("a valid mapping"), + max_message_size: crate::bridge::DEFAULT_MAX_MESSAGE_SIZE, + produce_sessions: crate::bridge::ProduceSessionLimits::default(), ++ tls: crate::bridge::IggyTls::default(), + }; + Self { + client: Arc::new(client), + produce_pool: Arc::new(ProducePool::new(config.produce_sessions)), + config, + fetch_pool: Arc::new(FetchPool::new()), + probe_client: LazyClient::default(), ++ lookup_client: LazyClient::default(), + offset_pool: OffsetPool::new(), + } + } + } + /// A client for `config`, connected and signed in. See [`IggyBridge::connect`]. async fn connect_client(config: &IggyBridgeConfig) -> Result<IggyClient, BridgeError> { if config.address.trim().is_empty() { @@@ -250,21 -303,16 +319,24 @@@ let credentials = Credentials::UsernamePassword(config.username.clone(), config.password.clone()); + // A send over the SDK's 8 KiB write buffer leaves as two writes, header then payload. With + // Nagle on, the payload waits for the server's delayed ACK of the header: about 40 ms a send. - let client = IggyClientBuilder::new() + let mut builder = IggyClientBuilder::new() .with_tcp() .with_server_address(config.address.clone()) + .with_no_delay() .with_auto_sign_in(AutoLogin::Enabled(credentials)) - .with_reconnection_max_retries(Some(RECONNECTION_RETRIES)) - .build() - .map_err(BridgeError::Iggy)?; + .with_reconnection_max_retries(Some(RECONNECTION_RETRIES)); + if config.tls.enabled { + builder = builder.with_tls_enabled(true); + if !config.tls.domain.is_empty() { + builder = builder.with_tls_domain(config.tls.domain.clone()); + } + if let Some(ref ca_file) = config.tls.ca_file { + builder = builder.with_tls_ca_file(ca_file.clone()); + } + } + let client = builder.build().map_err(BridgeError::Iggy)?; with_request_timeout(client.connect()).await?; Ok(client) } diff --cc gateways/kafka/src/bridge/mod.rs index ac7796759,8b25a8ae7..6f9e76ef6 --- a/gateways/kafka/src/bridge/mod.rs +++ b/gateways/kafka/src/bridge/mod.rs @@@ -26,10 -26,10 +26,10 @@@ pub mod error pub mod iggy_bridge; pub mod topic_map; - pub use config::{DEFAULT_MAX_MESSAGE_SIZE, IggyBridgeConfig, IggyTls}; -pub use config::{DEFAULT_MAX_MESSAGE_SIZE, IggyBridgeConfig, ProduceSessionLimits}; ++pub use config::{DEFAULT_MAX_MESSAGE_SIZE, IggyBridgeConfig, IggyTls, ProduceSessionLimits}; pub use error::BridgeError; pub use iggy_bridge::{ - IggyBridge, KafkaTopicMetadata, OFFSET_GROUP_PREFIX, OffsetCalls, StreamTopicCache, - TopicCreationOutcome, TopicLoad, TopicTarget, + IggyBridge, KafkaTopicMetadata, OFFSET_GROUP_PREFIX, OffsetCalls, ProduceTurn, Producer, + StreamTopicCache, TopicCreationOutcome, TopicLoad, TopicTarget, }; pub use topic_map::{TopicMapping, TopicOverride}; diff --cc gateways/kafka/src/main.rs index dfb24d6f5,5eaf8fcf5..0d755d3c8 --- a/gateways/kafka/src/main.rs +++ b/gateways/kafka/src/main.rs @@@ -42,30 -38,26 +42,30 @@@ async fn main() -> Result<(), Box<dyn s let _tracing_guard = init_tracing(); let config = load_config()?; - let sasl_enabled = config.sasl_enabled; + let sasl_enabled = config.sasl_enabled(); + let plaintext_sasl = config + .listeners + .iter() + .any(|listener| listener.protocol == SecurityProtocol::SaslPlaintext); let bridge = connect_bridge(sasl_enabled).await?; - let listener = bind_listener(&config.bind_addr) - .map_err(|e| format!("failed to bind {}: {e}", config.bind_addr))?; + let listeners = bind_listeners(&config).map_err(|e| e.to_string())?; - let mut server = KafkaGateway::new(config).with_bridge(bridge); + let mut server = KafkaGateway::new(config).with_bridge(bridge.clone()); if sasl_enabled { let authenticator = IggyAuthenticator::from_env()?; - // Unconditional, because the Kafka listener has no TLS at any setting. This is the hop the - // Kafka client's password actually crosses, and gating the warning on the Iggy hop meant - // encrypting that one silenced the only notice about this one. - warn!( - "SASL/PLAIN sends the password in the clear and the Kafka listener has no TLS; only \ - run this on a trusted network until listener TLS lands" - ); + if plaintext_sasl { + // This is the hop the Kafka client's password actually crosses. The warning is tied + // to the listener that exposes it, not to the Iggy hop, so encrypting the Iggy side + // does not silence the only notice about this one. + warn!( + "SASL/PLAIN on the SASL_PLAINTEXT listener sends the password in the clear; use \ + a SASL_SSL listener outside a trusted network" + ); + } if !authenticator.is_tls_enabled() { warn!( - "the credential check's link to Iggy is also unencrypted; set \ - IGGY_KAFKA_IGGY_TLS_ENABLED=true to encrypt it. The bridge's own connection has \ - no TLS at any setting" + "the credential check's link to Iggy is unencrypted; set \ + IGGY_KAFKA_IGGY_TLS_ENABLED=true to encrypt it" ); } info!("SASL/PLAIN enabled; credentials verified against {authenticator}"); @@@ -73,19 -65,40 +73,40 @@@ } let (tx, rx) = broadcast::channel(1); - let mut server_task = tokio::spawn(async move { server.run(listener, rx).await }); + let mut server_task = tokio::spawn(async move { server.run(listeners, rx).await }); - tokio::select! { - result = &mut server_task => { - return Ok(result??); - } - () = shutdown_signal() => { - let _ = tx.send(()); - } - } + let stopped_alone = tokio::select! { + result = &mut server_task => Some(result), + () = shutdown_signal() => None, + }; + let outcome = if let Some(outcome) = stopped_alone { + outcome + } else { + let _ = tx.send(()); + server_task.await + }; + close_bridge(bridge.as_deref()).await; + Ok(outcome??) + } + + /// How long shutdown waits for the Iggy clients to sign out. The connection drain before it can + /// take 25 s of a 30 s grace period. + const BRIDGE_CLOSE_TIMEOUT: Duration = Duration::from_secs(5); - server_task.await??; - Ok(()) + /// Signs the bridge's Iggy clients out and shuts them down. Without it Iggy keeps each client's + /// session until its lease runs out. + async fn close_bridge(bridge: Option<&IggyBridge>) { + let Some(bridge) = bridge else { + return; + }; + match tokio::time::timeout(BRIDGE_CLOSE_TIMEOUT, bridge.close()).await { + Ok(Ok(())) => info!("Iggy bridge closed"), + Ok(Err(error)) => warn!("the Iggy bridge did not close cleanly: {error}"), + Err(_elapsed) => warn!( + "the Iggy bridge did not close within {BRIDGE_CLOSE_TIMEOUT:?}: Iggy ends the \ + remaining sessions when their leases run out" + ), + } } /// The `IGGY_KAFKA_*` vars `load_config` itself reads. `IGGY_KAFKA_` is a diff --cc gateways/kafka/src/protocol/api.rs index 65c64fdec,e6038a06d..1ff796965 --- a/gateways/kafka/src/protocol/api.rs +++ b/gateways/kafka/src/protocol/api.rs @@@ -319,12 -319,16 +320,12 @@@ pub fn supported_api_ranges() -> &'stat /// `bridge` is `None` until `IGGY_KAFKA_BRIDGE_ENABLED` turns it on. A handler that finds `None` /// answers with its stub, so APIs can be wired one at a time. /// - /// Every Iggy call but a Fetch poll, a topic probe and an offset call goes through one lockstep - /// `IggyClient`, so Kafka connections serialize behind whichever of those calls is in flight. The - /// `Arc` does not change that. See the README's "Concurrency ceiling". + /// Produce sends on one Iggy client per producer. Metadata, `CreateTopics`, `DeleteTopics`, + /// `DescribeConfigs` and `AlterConfigs` share one lockstep `IggyClient`, so those calls run one + /// at a time across connections. See the README's "Concurrency ceiling". pub struct GatewayState { - pub broker: BrokerAdvertise, pub bridge: Option<Arc<IggyBridge>>, pub max_frame_size: usize, - /// Whether `SaslHandshake` and `SaslAuthenticate` are advertised and routed. Kept on the - /// shared state so `ApiVersions` can answer without a widened handler signature. - pub sasl_enabled: bool, /// Shared across every connection this gateway serves: a producer id has to be unique for /// the process, not for the connection that asked for it. pub producer_ids: ProducerIdAllocator, @@@ -353,10 -369,17 +364,15 @@@ impl GatewayState groups: GroupCoordinator, ) -> Self { Self { - broker, bridge, max_frame_size, - sasl_enabled, producer_ids: ProducerIdAllocator::new(instance_id), - produce_slots: Semaphore::const_new(PRODUCE_SLOTS), + produce_memory: Arc::new(Semaphore::new( + PRODUCE_CAPS + .saturating_mul(max_frame_size) + .min(Semaphore::MAX_PERMITS), + )), + produce_large: Arc::new(Semaphore::new(PRODUCE_LARGE)), groups, loading: Mutex::default(), probe_board: ProbeBoard::default(), @@@ -406,28 -433,10 +436,32 @@@ impl Default for ConnectionId } impl ConnectionState { + /// State for a connection accepted on the listener that advertises `broker`. + /// + /// `acls_enabled` is at least `sasl_enabled`: a SASL listener always names a principal. + #[must_use] + pub fn for_listener(broker: BrokerAdvertise, sasl_enabled: bool, acls_enabled: bool) -> Self { + Self { + broker, + sasl_enabled, + acls_enabled: acls_enabled || sasl_enabled, + ..Self::default() + } + } + + /// What `ApiVersions` tells this connection about the listener-dependent keys. + #[must_use] + pub const fn advertisement(&self) -> Advertisement { + Advertisement { + sasl: self.sasl_enabled, + acls: self.acls_enabled, + } + } + + pub(crate) const fn id(&self) -> u64 { + self.id.0 + } + pub(crate) fn note_heartbeat(&self, member: GroupMember) { *self .heartbeat_member diff --cc gateways/kafka/src/protocol/handlers/mod.rs index 3d73d7a56,941fe8e74..3748b3a1a --- a/gateways/kafka/src/protocol/handlers/mod.rs +++ b/gateways/kafka/src/protocol/handlers/mod.rs @@@ -111,13 -111,13 +111,13 @@@ pub async fn dispatch body: Bytes, ) -> HandleOutcome { match api_key { - API_KEY_PRODUCE => produce::handle(state, api_version, body).await, + API_KEY_PRODUCE => produce::handle(state, connection, principal, api_version, body).await, API_KEY_FETCH => fetch::handle(state, connection, api_version, body).await, API_KEY_LIST_OFFSETS => list_offsets::handle(state, api_version, body).await, - API_KEY_METADATA => metadata::handle(state, api_version, body).await, + API_KEY_METADATA => metadata::handle(state, connection, api_version, body).await, API_KEY_OFFSET_COMMIT => offset_commit::handle(state, connection, api_version, body).await, API_KEY_OFFSET_FETCH => offset_fetch::handle(state, connection, api_version, body).await, - API_KEY_API_VERSIONS => api_versions::handle(state, api_version, body).await, + API_KEY_API_VERSIONS => api_versions::handle(connection, api_version, body).await, API_KEY_CREATE_TOPICS => create_topics::handle(state, api_version, body).await, API_KEY_DELETE_TOPICS => delete_topics::handle(state, principal, api_version, body).await, API_KEY_DESCRIBE_CONFIGS => describe_configs::handle(state, api_version, body).await, diff --cc gateways/kafka/src/protocol/handlers/produce.rs index a18b9c388,bb2e24c5d..d006c9b4a --- a/gateways/kafka/src/protocol/handlers/produce.rs +++ b/gateways/kafka/src/protocol/handlers/produce.rs @@@ -575,7 -690,11 +690,10 @@@ mod tests Compression, NO_PARTITION_LEADER_EPOCH, NO_PRODUCER_EPOCH, NO_PRODUCER_ID, NO_SEQUENCE, Record, RecordBatchEncoder, RecordEncodeOptions, TimestampType, }; + use tokio_util::sync::CancellationToken; + use crate::group::{GroupCoordinator, GroupCoordinatorConfig}; + use crate::protocol::acl::PrincipalPermissions; - use crate::protocol::api::BrokerAdvertise; use crate::records::encode_batch; const CREATE_TIME: i64 = 1_700_000_000_123; @@@ -694,6 -813,289 +812,287 @@@ ); } + fn request_of(partitions: Vec<PartitionProduceData>) -> ProduceRequest { + ProduceRequest::default().with_topic_data(vec![ + TopicProduceData::default() + .with_name(TopicName(StrBytes::from_static_str(TOPIC))) + .with_partition_data(partitions), + ]) + } + + /// The default frame. + const FRAME: usize = 8 * 1024 * 1024; + /// Long enough for an admission that is free to complete. Time is paused in these tests. + const WAIT: Duration = Duration::from_millis(50); + + /// A one-record batch that carries `producer_id`. + fn batch_of(producer_id: i64) -> Bytes { + let mut records = [record(0, b"v")]; + records[0].producer_id = producer_id; + records[0].producer_epoch = 0; + records[0].sequence = 0; + compressed_batch(&records, Compression::None) + } + + fn principal(username: &str) -> AuthenticatedPrincipal { + AuthenticatedPrincipal { + username: username.to_string(), + permissions: PrincipalPermissions::default(), + permissions_known: true, + } + } + + #[test] + fn given_a_producer_id_this_gateway_gave_out_when_keyed_should_take_it_with_its_user() { - let state = GatewayState::stub(BrokerAdvertise::default(), FRAME); ++ let state = GatewayState::stub(FRAME); + let id = state.producer_ids.allocate().expect("an id"); + let request = request_of(vec![entry(0, Some(batch_of(id)))]); + let connection = ConnectionState::default(); + + assert_eq!( + producer_of(&state, &connection, None, &request), + Producer::Id { id, user: None } + ); + assert_eq!( + producer_of(&state, &connection, Some(&principal("ann")), &request), + Producer::Id { + id, + user: Some("ann".to_string()) + }, + "another user's request with this id gets another session" + ); + } + + #[test] + fn given_a_producer_id_a_client_made_up_when_keyed_should_take_the_connection() { - let state = GatewayState::stub(BrokerAdvertise::default(), FRAME); ++ let state = GatewayState::stub(FRAME); + let connection = ConnectionState::default(); + let issued = state.producer_ids.allocate().expect("an id"); + + for made_up in [7, issued + 1] { + let request = request_of(vec![entry(0, Some(batch_of(made_up)))]); + assert_eq!( + producer_of(&state, &connection, None, &request), + Producer::Connection(connection.id()), + "{made_up} picks no session, and opens none of its own" + ); + } + } + + #[test] + fn given_a_batch_without_a_producer_id_when_keyed_should_take_the_connection() { - let state = GatewayState::stub(BrokerAdvertise::default(), FRAME); ++ let state = GatewayState::stub(FRAME); + let batch = compressed_batch(&[record(0, b"v")], Compression::None); + let request = request_of(vec![entry(0, Some(batch))]); + let connection = ConnectionState::default(); + + assert_eq!( + producer_of(&state, &connection, None, &request), + Producer::Connection(connection.id()) + ); + assert_ne!( + ConnectionState::default().id(), + connection.id(), + "each connection keys a session of its own" + ); + } + + fn cost_of(partitions: Vec<PartitionProduceData>, max_frame_size: usize) -> usize { + decode_cost(&request_of(partitions), max_frame_size) + } + + #[test] + fn given_an_uncompressed_batch_when_costed_should_take_its_bytes() { + let batch = encode_batch(&mut [record(0, &[7; 1000]), record(1, &[7; 1000])]).unwrap(); + assert_eq!( + cost_of(vec![entry(0, Some(batch.clone()))], FRAME), + batch.len() + ); + } + + #[test] + fn given_small_records_when_costed_should_take_a_record_slot_each() { + let mut records: Vec<Record> = (0..100).map(|offset| record(offset, b"v")).collect(); + let batch = encode_batch(&mut records).unwrap(); + assert!(batch.len() < 100 * FRAME_BYTES_PER_RECORD); + assert_eq!( + cost_of(vec![entry(0, Some(batch))], FRAME), + 100 * FRAME_BYTES_PER_RECORD + ); + } + + #[test] + fn given_records_with_many_headers_when_costed_should_take_the_slots_the_decode_charges() { + let mut heavy = record(0, b"v"); + heavy.headers = (0..300) + .map(|index| (StrBytes::from_string(format!("h{index}")), None)) + .collect(); + let batch = encode_batch(&mut [heavy]).unwrap(); + let cost = cost_of(vec![entry(0, Some(batch.clone()))], FRAME); + + // One slot for the record and one for each three headers. + assert_eq!(cost, 101 * FRAME_BYTES_PER_RECORD); + assert!( + cost > batch.len(), + "its bytes alone would admit it for less" + ); + let slots = cost / FRAME_BYTES_PER_RECORD; + let entry = entry(0, Some(batch)); + assert!( + plan_with(&partition_budget(FRAME, slots), Zstd::Allowed, &entry).is_ok(), + "the decode takes no more than the request was admitted for" + ); + assert_eq!( + plan_with(&partition_budget(FRAME, slots - 1), Zstd::Allowed, &entry).unwrap_err(), + ERROR_MESSAGE_TOO_LARGE + ); + } + + #[test] + fn given_a_compressed_batch_when_costed_should_take_a_whole_cap() { + let batch = gzip_batch(&[record(0, b"v")]); + assert_eq!( + cost_of(vec![entry(0, Some(batch))], FRAME), + FRAME, + "it may inflate to the cap" + ); + } + + #[test] + fn given_several_partitions_when_costed_should_take_the_largest() { + let small = compressed_batch(&[record(0, &[7; 100])], Compression::None); + let large = compressed_batch(&[record(0, &[7; 5000])], Compression::None); + let partitions = vec![ + entry(0, Some(small)), + entry(1, Some(large.clone())), + entry(2, None), + ]; + assert_eq!( + cost_of(partitions, FRAME), + large.len(), + "partitions decode one at a time" + ); + } + + #[test] + fn given_a_partition_past_the_cap_when_costed_should_take_one_cap() { + let batch = compressed_batch(&[record(0, &[7; 5000])], Compression::None); + assert_eq!(cost_of(vec![entry(0, Some(batch))], 4096), 4096); + } + + /// Admits four requests of `cost`, which takes every large slot. + async fn large_slots_taken(state: &GatewayState, cost: usize) -> Vec<Admission> { + let mut admitted = Vec::new(); + for _ in 0..4 { + admitted.push(admit(state, cost).await.unwrap()); + } + admitted + } + + #[tokio::test(start_paused = true)] + async fn given_the_large_slots_taken_when_a_small_request_comes_should_admit_it() { - let state = GatewayState::stub(BrokerAdvertise::default(), FRAME); ++ let state = GatewayState::stub(FRAME); + let _large = large_slots_taken(&state, FRAME).await; + + let fifth = admit(&state, CODEC_OFF_WORKER_BYTES); + assert!( + tokio::time::timeout(WAIT, fifth).await.is_err(), + "a fifth large request waits its turn" + ); + let small = admit(&state, CODEC_OFF_WORKER_BYTES - 1); + assert!( + tokio::time::timeout(WAIT, small).await.is_ok(), + "four requests of a whole cap each leave a small one its memory" + ); + } + + #[tokio::test(start_paused = true)] + async fn given_the_large_slots_taken_when_a_large_request_waits_should_hold_no_memory() { - let state = GatewayState::stub(BrokerAdvertise::default(), FRAME); ++ let state = GatewayState::stub(FRAME); + let _large = large_slots_taken(&state, CODEC_OFF_WORKER_BYTES).await; + let free = state.produce_memory.available_permits(); + + let waiting = admit(&state, FRAME); + tokio::pin!(waiting); + assert!(tokio::time::timeout(WAIT, &mut waiting).await.is_err()); + assert_eq!( + state.produce_memory.available_permits(), + free, + "small requests can still take that memory" + ); + } + + /// A gateway whose bridge never dialed. A request that stops before its first send needs no + /// server. + fn offline_state() -> GatewayState { + GatewayState::new( - BrokerAdvertise::default(), + Some(Arc::new(IggyBridge::offline())), + FRAME, - false, + 0, + GroupCoordinator::new(GroupCoordinatorConfig::default(), CancellationToken::new()), + ) + } + + /// One large Produce v9 request of `connection` for partition 0, through the whole handler. + async fn handled( + state: &GatewayState, + connection: &ConnectionState, + acks: i16, + ) -> HandleOutcome { + let value = vec![7; CODEC_OFF_WORKER_BYTES]; + let batch = compressed_batch(&[record(0, &value)], Compression::None); + let request = request_of(vec![entry(0, Some(batch))]) + .with_acks(acks) + .with_timeout_ms(50); + let mut body = BytesMut::new(); + request.encode(&mut body, 9).expect("encode the request"); + handle(state, connection, None, 9, body.freeze()).await + } + + #[tokio::test(start_paused = true)] + async fn given_the_producers_earlier_send_still_running_when_handled_should_answer_timed_out() { + let state = offline_state(); + let bridge = state.bridge.as_deref().expect("a bridge"); + let connection = ConnectionState::default(); + let far = tokio::time::Instant::now() + Duration::from_secs(60); + let _earlier_send = bridge + .produce_turn(Producer::Connection(connection.id()), far) + .await + .expect("a free turn"); + let free = |state: &GatewayState| { + ( + state.produce_large.available_permits(), + state.produce_memory.available_permits(), + ) + }; + let before = free(&state); + + let waiting = handled(&state, &connection, ACKS_ALL); + tokio::pin!(waiting); + assert!( + tokio::time::timeout(Duration::from_millis(25), &mut waiting) + .await + .is_err(), + "it waits for the turn until its own deadline" + ); + assert_eq!( + free(&state), + before, + "and holds no large slot and no memory while it waits" + ); + let mut answer = waiting.await.expect_response("acks=-1 is answered"); + let response = ProduceResponse::decode(&mut answer, 9).expect("decode the response"); + let code = response.responses[0].partition_responses[0].error_code; + assert_eq!( + code, ERROR_REQUEST_TIMED_OUT, + "it must not overtake that send" + ); + assert!( + handled(&state, &connection, ACKS_NONE).await.is_close(), + "acks=0 has no response to carry the 7" + ); + } + #[test] fn given_an_idempotent_batch_when_planned_should_convert_it() { let mut records = [record(0, b"v")]; diff --cc gateways/kafka/src/server.rs index 683feb4cf,65a82570e..ab3e71d52 --- a/gateways/kafka/src/server.rs +++ b/gateways/kafka/src/server.rs @@@ -928,48 -629,58 +928,67 @@@ async fn route_frame } } - /// The frame loop of one accepted connection, plain or TLS. -async fn handle_connection( - stream: TcpStream, - config: Arc<GatewayConfig>, - peer: SocketAddr, - state: Arc<GatewayState>, - shared_auth: Arc<SharedAuth>, - cancel: CancellationToken, -) -> Result<()> { - debug!(%peer, "connection accepted"); - let connection = ConnectionState::default(); - let bridge = state.bridge.clone(); - let served = serve_connection( - stream, - config, - peer, - state, - shared_auth, - cancel, - &connection, - ) - .await; ++/// Serves one accepted connection, plain or TLS, then ends its producer session. ++async fn handle_connection<S>( ++ stream: S, ++ connection: Connection, ++ certificate_name: Option<String>, ++) -> Result<()> ++where ++ S: AsyncRead + AsyncWrite + Unpin + Send, ++{ ++ let connection_state = connection.listener.connection_state(); ++ let bridge = connection.state.bridge.clone(); ++ let served = serve_frames(stream, connection, certificate_name, &connection_state).await; + // A producer without a producer id has its Produce session by connection, so it ends here. + if let Some(bridge) = bridge { - bridge.end_connection(connection.id()); ++ bridge.end_connection(connection_state.id()); + } + served + } + -/// Reads and answers the frames of `connection` until it closes. -async fn serve_connection( - mut stream: TcpStream, - config: Arc<GatewayConfig>, - peer: SocketAddr, - state: Arc<GatewayState>, - shared_auth: Arc<SharedAuth>, - cancel: CancellationToken, - connection: &ConnectionState, -) -> Result<()> { - // A gateway with SASL off starts every connection already authenticated, so the dispatch path - // below has one question to ask rather than two. - let mut sasl_state = if config.sasl_enabled { ++/// The frame loop of one accepted connection. +/// +/// `certificate_name` is the Iggy username the client certificate mapped to, on an `SSL` +/// listener that verifies certificates. It is resolved against Iggy before the first frame, and +/// a name that resolves to no user closes the connection, so a certificate from the trusted CA +/// never admits a client as an account that does not exist. - async fn handle_connection<S>( ++async fn serve_frames<S>( + mut stream: S, + connection: Connection, + certificate_name: Option<String>, ++ connection_state: &ConnectionState, +) -> Result<()> +where + S: AsyncRead + AsyncWrite + Unpin + Send, +{ + let Connection { + peer, + listener, + config, + state, + auth: shared_auth, + cancel, + tls: _, + } = connection; + debug!(%peer, protocol = %listener.protocol, "connection accepted"); + + // A listener without SASL starts every connection already authenticated, so the dispatch + // path below has one question to ask rather than two. + let mut sasl_state = if listener.protocol.uses_sasl() { SaslState::new() } else { SaslState::Authenticated }; - let connection_state = listener.connection_state(); - let mut principal: Option<AuthenticatedPrincipal> = None; let ctx = ConnectionContext { config: &config, + listener: &listener, state: &state, - connection: &connection_state, - connection, ++ connection: connection_state, authenticator: shared_auth.authenticator.as_deref(), + resolver: shared_auth.resolver.as_deref(), auth_slots: &shared_auth.slots, + lookup_slots: &shared_auth.lookup_slots, failed_logins: &shared_auth.failed_logins, peer: &peer, cancel: &cancel, diff --cc gateways/kafka/tests/bridge_iggy_integration_tests.rs index 371d4cbac,a61aacdf5..1456627ae --- a/gateways/kafka/tests/bridge_iggy_integration_tests.rs +++ b/gateways/kafka/tests/bridge_iggy_integration_tests.rs @@@ -30,12 -30,10 +30,12 @@@ use secrecy::SecretString use serial_test::serial; use iggy_gateway_kafka::bridge::{ - BridgeError, DEFAULT_MAX_MESSAGE_SIZE, IggyBridge, IggyBridgeConfig, IggyTls, TopicMapping, - TopicOverride, - BridgeError, DEFAULT_MAX_MESSAGE_SIZE, IggyBridge, IggyBridgeConfig, ProduceSessionLimits, - TopicMapping, TopicOverride, ++ BridgeError, DEFAULT_MAX_MESSAGE_SIZE, IggyBridge, IggyBridgeConfig, IggyTls, ++ ProduceSessionLimits, TopicMapping, TopicOverride, }; +#[path = "common/certs.rs"] +mod certs; #[path = "common/iggy_server.rs"] mod iggy_server; @@@ -426,7 -394,7 +426,8 @@@ async fn connect_to_unreachable_iggy_re topic_mapping: TopicMapping::new("kafka".to_string(), HashMap::new()) .expect("valid mapping for this test's fixture data"), max_message_size: DEFAULT_MAX_MESSAGE_SIZE, + tls: IggyTls::default(), + produce_sessions: ProduceSessionLimits::default(), }; let result = IggyBridge::connect(config).await; @@@ -451,7 -419,7 +452,8 @@@ async fn connect_to_a_black_hole_addres topic_mapping: TopicMapping::new("kafka".to_string(), HashMap::new()) .expect("valid mapping for this test's fixture data"), max_message_size: DEFAULT_MAX_MESSAGE_SIZE, + tls: IggyTls::default(), + produce_sessions: ProduceSessionLimits::default(), }; let start = tokio::time::Instant::now(); diff --cc gateways/kafka/tests/common/iggy_server.rs index 6ed6433c3,414519ffb..5ee5ae129 --- a/gateways/kafka/tests/common/iggy_server.rs +++ b/gateways/kafka/tests/common/iggy_server.rs @@@ -34,7 -34,7 +34,7 @@@ use iggy::prelude::{AutoLogin, Client, use secrecy::SecretString; use iggy_gateway_kafka::bridge::{ - DEFAULT_MAX_MESSAGE_SIZE, IggyBridgeConfig, IggyTls, TopicMapping, - DEFAULT_MAX_MESSAGE_SIZE, IggyBridgeConfig, ProduceSessionLimits, TopicMapping, ++ DEFAULT_MAX_MESSAGE_SIZE, IggyBridgeConfig, IggyTls, ProduceSessionLimits, TopicMapping, }; /// Desired slot count - the actual count `port_band()` computes may be smaller if the machine's @@@ -363,7 -346,7 +375,8 @@@ impl TestServer topic_mapping: TopicMapping::new("kafka".to_string(), HashMap::new()) .expect("valid mapping for this test's fixture data"), max_message_size: DEFAULT_MAX_MESSAGE_SIZE, + tls: IggyTls::default(), + produce_sessions: ProduceSessionLimits::default(), } } } diff --cc gateways/kafka/tests/kafka_client_e2e_tests.rs index ae1688e12,1ed5d3054..bb1dcd174 --- a/gateways/kafka/tests/kafka_client_e2e_tests.rs +++ b/gateways/kafka/tests/kafka_client_e2e_tests.rs @@@ -51,14 -51,11 +51,15 @@@ use tokio::task::JoinHandle use iggy_gateway_kafka::auth::IggyAuthenticator; use iggy_gateway_kafka::bridge::{ - DEFAULT_MAX_MESSAGE_SIZE, IggyBridge, IggyBridgeConfig, IggyTls, TopicMapping, - DEFAULT_MAX_MESSAGE_SIZE, IggyBridge, IggyBridgeConfig, ProduceSessionLimits, TopicMapping, ++ DEFAULT_MAX_MESSAGE_SIZE, IggyBridge, IggyBridgeConfig, IggyTls, ProduceSessionLimits, ++ TopicMapping, }; -use iggy_gateway_kafka::server::bind_listener; +use iggy_gateway_kafka::listener::{ListenerConfig, SecurityProtocol}; +use iggy_gateway_kafka::server::{BoundListener, bind_listener}; use iggy_gateway_kafka::{GatewayConfig, KafkaGateway}; +#[path = "common/certs.rs"] +mod certs; #[path = "common/server.rs"] mod server; @@@ -342,7 -334,7 +343,8 @@@ async fn start_bridged_gateway(server: topic_mapping: TopicMapping::new("kafka".to_string(), HashMap::new()) .expect("the default mapping is valid"), max_message_size: DEFAULT_MAX_MESSAGE_SIZE, + tls: IggyTls::default(), + produce_sessions: ProduceSessionLimits::default(), }; let bridge = IggyBridge::connect(config) .await @@@ -1408,341 -1400,3 +1410,342 @@@ fn records_by_partition(output: &[Strin } read } + +/// The TLS-side stack: a bridged gateway with an `SSL` and a `SASL_SSL` listener, client +/// certificates requested and mapped to Iggy users by common name, and the material every +/// client needs on disk. +struct TlsStack { + _server: TestServer, + ssl: SocketAddr, + sasl_ssl: SocketAddr, + _certs_dir: tempfile::TempDir, + ca_file: PathBuf, + /// A certificate whose common name is `TLS_USER`, an Iggy user the stack creates. + user_cert: Issued, + /// A certificate the CA signed for a name no Iggy user has. + stranger_cert: Issued, +} + +/// The Iggy user a client certificate maps to. Created with a topic read so `DescribeAcls` +/// renders a binding that names it. +const TLS_USER: &str = "cert-alice"; + +/// Brings up the TLS stack, or reports why it could not. +async fn tls_stack() -> Option<TlsStack> { + let server = ready_server()?; + let token = root_token(&server.http_address); + create_user( + &server.http_address, + &token, + TLS_USER, + r#"{"global":{"manage_servers":false,"read_servers":false,"manage_users":false, + "read_users":false,"manage_streams":false,"read_streams":false,"manage_topics":false, + "read_topics":true,"poll_messages":false,"send_messages":false},"streams":null}"#, + ); + + let certs_dir = tempfile::tempdir().expect("certs dir"); + let pki = TestPki::new(); + let (cert_file, key_file) = pki.server_cert("gateway").write(certs_dir.path(), "server"); + let ca_file = pki.write_ca(certs_dir.path()); + let user_cert = pki.client_cert(TLS_USER, "Iggy"); + let stranger_cert = pki.client_cert("nobody", "Iggy"); + + let bridge = IggyBridge::connect(IggyBridgeConfig { + address: server.address.clone(), + username: ROOT_USER.to_string(), + password: SecretString::from(ROOT_PASSWORD.to_string()), + topic_mapping: TopicMapping::new("kafka".to_string(), HashMap::new()) + .expect("the default mapping is valid"), + max_message_size: DEFAULT_MAX_MESSAGE_SIZE, ++ produce_sessions: ProduceSessionLimits::default(), + tls: IggyTls::default(), + }) + .await + .expect("the bridge connects to a ready server"); + bridge + .ensure_stream_and_topic(TOPIC, 1) + .await + .expect("create the topic"); + let authenticator = Arc::new(IggyAuthenticator::new(server.address.clone())); + let config = GatewayConfig { + listeners: vec![ + ListenerConfig::new(SecurityProtocol::Ssl, "127.0.0.1:0"), + ListenerConfig::new(SecurityProtocol::SaslSsl, "127.0.0.1:0"), + ], + tls: Some( + TlsConfig::new(cert_file, key_file) + .with_client_auth(ClientAuth::Requested, &ca_file) + .with_principal_mapping("RULE:^CN=([^,]+).*$/$1/".parse().expect("valid rule")), + ), + ..GatewayConfig::default() + }; + let (addrs, shutdown) = spawn_listeners( + move |config| { + KafkaGateway::new(config) + .with_bridge(Some(Arc::new(bridge))) + .with_authenticator(authenticator) + }, + config, + ); + // Held for the test's lifetime: dropping the sender shuts the gateway down mid-exchange. + std::mem::forget(shutdown); + Some(TlsStack { + _server: server, + ssl: addrs[0], + sasl_ssl: addrs[1], + _certs_dir: certs_dir, + ca_file, + user_cert, + stranger_cert, + }) +} + +impl TlsStack { + /// kcat against `addr` with `extra` librdkafka properties, trusting the stack's CA. + fn kcat_tls(&self, addr: SocketAddr, extra: &[&str], client: Option<&Issued>) -> ClientRun { + let dir = tempfile::tempdir().expect("client certs dir"); + let mut args = vec![ + "-b".to_string(), + addr.to_string(), + "-X".to_string(), + "ssl.ca.location=/tmp/certs/ca.pem".to_string(), + ]; + let mut mounts = vec![(self.ca_file.display().to_string(), "/tmp/certs/ca.pem")]; + let (cert_file, key_file) = client.map_or_else( + || (PathBuf::new(), PathBuf::new()), + |issued| issued.write(dir.path(), "client"), + ); + if client.is_some() { + args.extend([ + "-X".to_string(), + "ssl.certificate.location=/tmp/certs/client.pem".to_string(), + "-X".to_string(), + "ssl.key.location=/tmp/certs/client.key".to_string(), + ]); + mounts.push((cert_file.display().to_string(), "/tmp/certs/client.pem")); + mounts.push((key_file.display().to_string(), "/tmp/certs/client.key")); + } + for property in extra { + args.extend(["-X".to_string(), (*property).to_string()]); + } + args.push("-L".to_string()); + let args: Vec<&str> = args.iter().map(String::as_str).collect(); + let mounts: Vec<(&str, &str)> = mounts + .iter() + .map(|(host, guest)| (host.as_str(), *guest)) + .collect(); + run_client(KCAT_IMAGE, &args, &mounts) + } + + /// Writes a Java client config for TLS with `extra` properties, plus the CA and, when + /// given, a PEM keystore. Returns the mounts the container needs. + fn java_tls_config( + &self, + extra: &str, + client: Option<&Issued>, + ) -> (tempfile::TempDir, Vec<(String, &'static str)>) { + let dir = tempfile::tempdir().expect("create config dir"); + let mut mounts = vec![(self.ca_file.display().to_string(), "/tmp/certs/ca.pem")]; + let mut properties = format!( + "ssl.truststore.type=PEM\n\ + ssl.truststore.location=/tmp/certs/ca.pem\n\ + request.timeout.ms={JAVA_CLIENT_TIMEOUT_MS}\n\ + default.api.timeout.ms={JAVA_CLIENT_TIMEOUT_MS}\n\ + {extra}\n" + ); + if let Some(issued) = client { + let bundle = issued.write_bundle(dir.path(), "client"); + mounts.push((bundle.display().to_string(), "/tmp/certs/client.pem")); + properties.push_str( + "ssl.keystore.type=PEM\n\ + ssl.keystore.location=/tmp/certs/client.pem\n", + ); + } + let path = dir.path().join("client.properties"); + std::fs::write(&path, properties).expect("write client config"); + mounts.push((path.display().to_string(), "/tmp/client.properties")); + (dir, mounts) + } + + fn java( + &self, + tool: &str, + addr: SocketAddr, + extra: &str, + client: Option<&Issued>, + ) -> ClientRun { + let (_dir, mounts) = self.java_tls_config(extra, client); + let mounts: Vec<(&str, &str)> = mounts + .iter() + .map(|(host, guest)| (host.as_str(), *guest)) + .collect(); + let bootstrap = addr.to_string(); + let mut args = vec![ + tool, + "--bootstrap-server", + &bootstrap, + "--command-config", + "/tmp/client.properties", + ]; + if tool.ends_with("kafka-acls.sh") { + args.push("--list"); + } + run_client(KAFKA_IMAGE, &args, &mounts) + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn given_an_ssl_listener_when_librdkafka_connects_with_ssl_should_serve_metadata() { + let Some(stack) = tls_stack().await else { + return; + }; + let output = stack + .kcat_tls(stack.ssl, &["security.protocol=SSL"], None) + .expect_ran("kcat over SSL"); + assert!( + output.contains("Metadata for all topics"), + "librdkafka must handshake against the gateway certificate and be served, got: {output}" + ); + assert!( + output.contains(&format!("127.0.0.1:{}", stack.ssl.port())), + "the broker advertised must be the SSL listener the client used, got: {output}" + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn given_a_sasl_ssl_listener_when_librdkafka_authenticates_should_serve_metadata() { + let Some(stack) = tls_stack().await else { + return; + }; + let sasl = |password: &str| { + stack.kcat_tls( + stack.sasl_ssl, + &[ + "security.protocol=SASL_SSL", + "sasl.mechanisms=PLAIN", + &format!("sasl.username={ROOT_USER}"), + &format!("sasl.password={password}"), + ], + None, + ) + }; + let output = sasl(ROOT_PASSWORD).expect_ran("kcat over SASL_SSL"); + assert!( + output.contains("Metadata for all topics"), + "librdkafka must authenticate inside TLS and be served, got: {output}" + ); + assert!( + output.contains(&format!("127.0.0.1:{}", stack.sasl_ssl.port())), + "the broker advertised must be the SASL_SSL listener the client used, got: {output}" + ); + + // kcat exits non-zero here by design, so only its output is asserted. + let rejected = sasl("definitely-not-the-password").text(); + assert!( + rejected.contains("Authentication failed"), + "a rejected credential must reach the client as an auth failure, got: {rejected}" + ); + assert!( + !rejected.contains("Metadata for all topics"), + "nothing may be served to a rejected client" + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn given_a_client_certificate_when_librdkafka_connects_should_be_admitted_as_its_user() { + let Some(stack) = tls_stack().await else { + return; + }; + let output = stack + .kcat_tls( + stack.ssl, + &["security.protocol=SSL"], + Some(&stack.user_cert), + ) + .expect_ran("kcat with a client certificate"); + assert!( + output.contains("Metadata for all topics"), + "a certificate naming an Iggy user must be admitted, got: {output}" + ); + + let stranger = stack.kcat_tls( + stack.ssl, + &["security.protocol=SSL"], + Some(&stack.stranger_cert), + ); + assert!( + !stranger.text.contains("Metadata for all topics"), + "a certificate naming no Iggy user must not be served, got: {}", + stranger.text + ); + assert!( + !stranger.succeeded, + "kcat must fail when the gateway closes on it, got: {}", + stranger.text + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn given_the_java_client_with_ssl_and_sasl_ssl_should_connect_through_both_listeners() { + let Some(stack) = tls_stack().await else { + return; + }; + let ssl = stack + .java( + "/opt/kafka/bin/kafka-broker-api-versions.sh", + stack.ssl, + "security.protocol=SSL", + None, + ) + .expect_ran("java client over SSL"); + assert!( + ssl.contains(&format!("127.0.0.1:{}", stack.ssl.port())), + "the Java client reads Metadata and then dials the advertised SSL listener, got: {ssl}" + ); + assert!( + ssl.contains("SaslHandshake(17): UNSUPPORTED"), + "an SSL listener advertises no SASL keys, got: {ssl}" + ); + + let sasl_ssl = stack + .java( + "/opt/kafka/bin/kafka-broker-api-versions.sh", + stack.sasl_ssl, + &format!( + "security.protocol=SASL_SSL\n\ + sasl.mechanism=PLAIN\n\ + sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule \ + required username=\"{ROOT_USER}\" password=\"{ROOT_PASSWORD}\";" + ), + None, + ) + .expect_ran("java client over SASL_SSL"); + assert!( + sasl_ssl.contains("SaslHandshake(17): 0 to 1") + && sasl_ssl.contains("SaslAuthenticate(36): 0 to 2"), + "the Java client must authenticate inside TLS and read the advertisement, got: {sasl_ssl}" + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn given_the_java_client_with_a_client_certificate_should_be_described_as_its_user() { + let Some(stack) = tls_stack().await else { + return; + }; + let listing = stack + .java( + "/opt/kafka/bin/kafka-acls.sh", + stack.ssl, + "security.protocol=SSL", + Some(&stack.user_cert), + ) + .expect_ran("java kafka-acls over mTLS"); + assert!( + listing.contains(&format!("principal=User:{TLS_USER}")), + "the certificate's common name is the principal described, got: {listing}" + ); + assert!( + grants(&listing, "TOPIC", "READ"), + "the mapped user's read_topics grant renders, got: {listing}" + ); +} diff --cc gateways/kafka/tests/produce_real_bridge_tests.rs index fb3e3071a,6420aa155..3a70c1825 --- a/gateways/kafka/tests/produce_real_bridge_tests.rs +++ b/gateways/kafka/tests/produce_real_bridge_tests.rs @@@ -35,11 -37,12 +37,11 @@@ use kafka_protocol::records:: use serial_test::serial; use tokio_util::sync::CancellationToken; - use iggy_gateway_kafka::bridge::IggyBridge; + use iggy_gateway_kafka::bridge::{IggyBridge, Producer}; use iggy_gateway_kafka::group::{GroupCoordinator, GroupCoordinatorConfig}; use iggy_gateway_kafka::protocol::api::{ - API_KEY_PRODUCE, ERROR_NONE, ERROR_NOT_LEADER_OR_FOLLOWER, ERROR_UNKNOWN_TOPIC_OR_PARTITION, - GatewayState, handle_request_bounded, - API_KEY_INIT_PRODUCER_ID, API_KEY_PRODUCE, BrokerAdvertise, ERROR_NONE, - ERROR_NOT_LEADER_OR_FOLLOWER, ERROR_UNKNOWN_TOPIC_OR_PARTITION, GatewayState, - handle_request_bounded, ++ API_KEY_INIT_PRODUCER_ID, API_KEY_PRODUCE, ERROR_NONE, ERROR_NOT_LEADER_OR_FOLLOWER, ++ ERROR_UNKNOWN_TOPIC_OR_PARTITION, GatewayState, handle_request_bounded, }; use iggy_gateway_kafka::records::{KEY_HEADER, MAPPING_VERSION, VERSION_HEADER, from_iggy};
