From 8240546482c4ea4dff9621ffad1d35ad5b167533 Mon Sep 17 00:00:00 2001 From: g7ed6e <681739+g7ed6e@users.noreply.github.com> Date: Fri, 7 Aug 2026 08:38:40 +0200 Subject: [PATCH] Add SASL/PLAIN password protection to the Kafka container Kafka was the last container integration in #6155 without authentication. Enable SASL/PLAIN over SASL_PLAINTEXT on the two client facing listeners, with a generated password when none is supplied. Hosting: - KafkaServerResource gains UserNameParameter/PasswordParameter, a UserNameReference defaulting to "kafka", and Username/Password connection properties. - The connection string stays a bare host:port when no password is configured. With a password it becomes a semicolon separated list of Confluent client configuration properties so the credentials can travel with it. - AddKafka takes optional userName/password parameters, alongside a convenience overload preserving the previous signature. WithPassword and WithUserName allow reconfiguring afterwards, and WithPassword(null) turns authentication off. - When a password is configured the client facing listeners are renamed to EXTERNAL/INTERNAL and mapped to SASL_PLAINTEXT. The names deliberately contain no underscore because the Confluent image translates KAFKA_FOO_BAR into foo.bar, which would otherwise require escaping the listener name in KAFKA_LISTENER_NAME__PLAIN_SASL_JAAS_CONFIG. The KRaft controller and inter broker listeners stay plaintext on the loopback interface. - The health check built its ProducerConfig by assigning the connection string to BootstrapServers. That is wrong once the connection string is a keyed list, and had no credentials either way, so it now resolves the endpoint and the credentials separately. - WithKafkaUI configures the matching security protocol, mechanism and JAAS properties so the UI can still reach the broker. Client: - Aspire.Confluent.Kafka applies the connection string onto the client configuration. A value without '=' is still treated as a bare bootstrap server list, otherwise it is parsed with DbConnectionStringBuilder and BootstrapServers, SecurityProtocol, SaslMechanism, SaslUsername and SaslPassword are applied. Contributes to #6155 Co-Authored-By: Claude Opus 5 (1M context) --- .../KafkaBuilderExtensions.cs | 163 ++++++++++++++-- .../KafkaServerResource.cs | 68 ++++++- src/Aspire.Hosting.Kafka/README.md | 29 ++- .../KafkaConnectionString.cs | 67 +++++++ .../KafkaConsumerSettings.cs | 2 +- .../KafkaProducerSettings.cs | 2 +- .../Aspire.Confluent.Kafka/README.md | 13 ++ .../CommonHelpers.cs | 7 + .../ConsumerConfigurationTests.cs | 36 ++++ .../ProducerConfigurationTests.cs | 50 +++++ .../AddKafkaTests.cs | 184 +++++++++++++++++- .../ConnectionPropertiesTests.cs | 45 +++++ .../KafkaFunctionalTests.cs | 37 ++++ .../KafkaPublicApiTests.cs | 93 +++++++++ 14 files changed, 773 insertions(+), 23 deletions(-) create mode 100644 src/Components/Aspire.Confluent.Kafka/KafkaConnectionString.cs diff --git a/src/Aspire.Hosting.Kafka/KafkaBuilderExtensions.cs b/src/Aspire.Hosting.Kafka/KafkaBuilderExtensions.cs index bc7cdd75ad6..dad19cf3ae6 100644 --- a/src/Aspire.Hosting.Kafka/KafkaBuilderExtensions.cs +++ b/src/Aspire.Hosting.Kafka/KafkaBuilderExtensions.cs @@ -19,6 +19,18 @@ public static class KafkaBuilderExtensions private const int KafkaUIPort = 8080; private const string Target = "/var/lib/kafka/data"; + // Listener names used when the broker is not password protected. These are kept for backwards + // compatibility with app models that opt out of authentication. + private const string PlaintextExternalListenerName = "PLAINTEXT_HOST"; + private const string PlaintextInternalListenerName = "PLAINTEXT_INTERNAL"; + + // Listener names used when the broker is password protected. These deliberately contain no + // underscore: the Confluent image translates KAFKA_FOO_BAR into the foo.bar broker property, so an + // underscore that is part of a listener name has to be escaped as a double underscore in + // KAFKA_LISTENER_NAME__PLAIN_SASL_JAAS_CONFIG. Underscore free names avoid that ambiguity. + private const string SaslExternalListenerName = "EXTERNAL"; + private const string SaslInternalListenerName = "INTERNAL"; + /// /// Adds a Kafka resource to the application. A container is used for local development. /// @@ -30,24 +42,71 @@ public static class KafkaBuilderExtensions /// The host port of Kafka broker. /// A reference to the . /// The resource builder. + [AspireExportIgnore(Reason = "Convenience overload. Use the overload with optional userName and password parameters instead.")] + public static IResourceBuilder AddKafka(this IDistributedApplicationBuilder builder, [ResourceName] string name, int? port) + { + ArgumentNullException.ThrowIfNull(builder); + ArgumentException.ThrowIfNullOrEmpty(name); + + return builder.AddKafka(name, port, null, null); + } + + /// + /// Adds a Kafka resource to the application. A container is used for local development. + /// The broker is protected with SASL/PLAIN authentication using a generated password unless one is provided. + /// + /// + /// This version of the package defaults to the tag of the container image. + /// + /// The . + /// The name of the resource. This name will be used as the connection string name when referenced in a dependency + /// The host port of Kafka broker. + /// The parameter used to provide the SASL user name for the Kafka broker. If a default value will be used. + /// The parameter used to provide the SASL password for the Kafka broker. If a random password will be generated. + /// A reference to the . + /// The resource builder. [AspireExport] - public static IResourceBuilder AddKafka(this IDistributedApplicationBuilder builder, [ResourceName] string name, int? port = null) + public static IResourceBuilder AddKafka( + this IDistributedApplicationBuilder builder, + [ResourceName] string name, + int? port = null, + IResourceBuilder? userName = null, + IResourceBuilder? password = null) { ArgumentNullException.ThrowIfNull(builder); ArgumentException.ThrowIfNullOrEmpty(name); - var kafka = new KafkaServerResource(name); + // The password ends up inside a JAAS configuration string and inside the connection string, both of + // which are quote sensitive, so restrict the generated value to alphanumeric characters. + var passwordParameter = password?.Resource ?? ParameterResourceBuilderExtensions.CreateDefaultPasswordParameter(builder, $"{name}-password", special: false); - string? connectionString = null; + var kafka = new KafkaServerResource(name, userName?.Resource, passwordParameter); + + ProducerConfig? healthCheckConfiguration = null; builder.Eventing.Subscribe(kafka, async (@event, ct) => { - connectionString = await kafka.ConnectionStringExpression.GetValueAsync(ct).ConfigureAwait(false); + // The health check talks to the broker directly, so it needs the bootstrap servers and the + // credentials rather than the connection string, which is not a bootstrap server list once + // authentication is enabled. + var bootstrapServers = await kafka.PrimaryEndpoint.Property(EndpointProperty.HostAndPort).GetValueAsync(ct).ConfigureAwait(false); - if (connectionString == null) + if (bootstrapServers == null) { throw new DistributedApplicationException($"ConnectionStringAvailableEvent was published for the '{kafka.Name}' resource but the connection string was null."); } + + var configuration = new ProducerConfig { BootstrapServers = bootstrapServers }; + + if (kafka.PasswordParameter is not null) + { + configuration.SecurityProtocol = SecurityProtocol.SaslPlaintext; + configuration.SaslMechanism = SaslMechanism.Plain; + configuration.SaslUsername = await kafka.UserNameReference.GetValueAsync(ct).ConfigureAwait(false); + configuration.SaslPassword = await kafka.PasswordParameter.GetValueAsync(ct).ConfigureAwait(false); + } + + healthCheckConfiguration = configuration; }); var healthCheckKey = $"{name}_check"; @@ -63,8 +122,7 @@ public static IResourceBuilder AddKafka(this IDistributedAp sp => { var options = new KafkaHealthCheckOptions(); - options.Configuration = new ProducerConfig(); - options.Configuration.BootstrapServers = connectionString ?? throw new InvalidOperationException("Connection string is unavailable"); + options.Configuration = healthCheckConfiguration ?? throw new InvalidOperationException("Connection string is unavailable"); return new KafkaHealthCheck(options); }, failureStatus: default, @@ -81,6 +139,39 @@ public static IResourceBuilder AddKafka(this IDistributedAp .WithHealthCheck(healthCheckKey); } + /// + /// Configures the SASL password used by the Kafka broker. + /// + /// The resource builder. + /// The parameter used to provide the SASL password for the Kafka resource. If , authentication is disabled and the broker listens in plaintext. + /// A reference to the . + /// The resource builder. + [AspireExport] + public static IResourceBuilder WithPassword(this IResourceBuilder builder, IResourceBuilder? password) + { + ArgumentNullException.ThrowIfNull(builder); + + builder.Resource.PasswordParameter = password?.Resource; + return builder; + } + + /// + /// Configures the SASL user name used by the Kafka broker. + /// + /// The resource builder. + /// The parameter used to provide the SASL user name for the Kafka resource. + /// A reference to the . + /// The resource builder. + [AspireExport] + public static IResourceBuilder WithUserName(this IResourceBuilder builder, IResourceBuilder userName) + { + ArgumentNullException.ThrowIfNull(builder); + ArgumentNullException.ThrowIfNull(userName); + + builder.Resource.UserNameParameter = userName.Resource; + return builder; + } + /// /// Adds a Kafka UI container to the application. /// @@ -122,9 +213,9 @@ public static IResourceBuilder WithKafkaUI(this IResourceBu int i = 0; foreach (var kafkaResource in kafkaResources) { - var endpoint = kafkaResource.InternalEndpoint; + var resource = kafkaResource; int index = i; - kafkaUiBuilder.WithEnvironment(context => ConfigureKafkaUIContainer(context, endpoint, index)); + kafkaUiBuilder.WithEnvironment(context => ConfigureKafkaUIContainer(context, resource, index)); i++; } @@ -137,8 +228,10 @@ public static IResourceBuilder WithKafkaUI(this IResourceBu return builder; } - static void ConfigureKafkaUIContainer(EnvironmentCallbackContext context, EndpointReference endpoint, int index) + static void ConfigureKafkaUIContainer(EnvironmentCallbackContext context, KafkaServerResource resource, int index) { + var endpoint = resource.InternalEndpoint; + var bootstrapServers = context.ExecutionContext.IsRunMode // In run mode, Kafka UI assumes Kafka is being accessed over a default Aspire container network and hardcodes the host as the Kafka resource name // This will need to be refactored once updated service discovery APIs are available @@ -147,6 +240,13 @@ static void ConfigureKafkaUIContainer(EnvironmentCallbackContext context, Endpoi context.EnvironmentVariables[$"KAFKA_CLUSTERS_{index}_NAME"] = endpoint.Resource.Name; context.EnvironmentVariables[$"KAFKA_CLUSTERS_{index}_BOOTSTRAPSERVERS"] = bootstrapServers; + + if (resource.PasswordParameter is not null) + { + context.EnvironmentVariables[$"KAFKA_CLUSTERS_{index}_PROPERTIES_SECURITY_PROTOCOL"] = "SASL_PLAINTEXT"; + context.EnvironmentVariables[$"KAFKA_CLUSTERS_{index}_PROPERTIES_SASL_MECHANISM"] = "PLAIN"; + context.EnvironmentVariables[$"KAFKA_CLUSTERS_{index}_PROPERTIES_SASL_JAAS_CONFIG"] = BuildJaasConfig(resource); + } } } @@ -212,10 +312,17 @@ private static void ConfigureKafkaContainer(EnvironmentCallbackContext context, // When not explicitly set default configuration is applied. // See https://github.com/confluentinc/kafka-images/blob/master/local/include/etc/confluent/docker/configureDefaults for more details. + // Only the two client facing listeners are protected. The KRaft controller listener and the + // inter broker listener are bound to the loopback interface inside the container, so they stay PLAINTEXT. + var saslEnabled = resource.PasswordParameter is not null; + var externalListener = saslEnabled ? SaslExternalListenerName : PlaintextExternalListenerName; + var internalListener = saslEnabled ? SaslInternalListenerName : PlaintextInternalListenerName; + var clientProtocol = saslEnabled ? "SASL_PLAINTEXT" : "PLAINTEXT"; + // Define the default listeners + an internal listener for the container to broker communication - context.EnvironmentVariables[$"KAFKA_LISTENERS"] = $"PLAINTEXT://localhost:29092,CONTROLLER://localhost:29093,PLAINTEXT_HOST://0.0.0.0:{KafkaBrokerPort},PLAINTEXT_INTERNAL://0.0.0.0:{KafkaInternalBrokerPort}"; - // Defaults default listeners security protocol map + the internal listener to be PLAINTEXT - context.EnvironmentVariables["KAFKA_LISTENER_SECURITY_PROTOCOL_MAP"] = "CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT,PLAINTEXT_INTERNAL:PLAINTEXT"; + context.EnvironmentVariables[$"KAFKA_LISTENERS"] = $"PLAINTEXT://localhost:29092,CONTROLLER://localhost:29093,{externalListener}://0.0.0.0:{KafkaBrokerPort},{internalListener}://0.0.0.0:{KafkaInternalBrokerPort}"; + // Defaults default listeners security protocol map + the client facing listeners protocol + context.EnvironmentVariables["KAFKA_LISTENER_SECURITY_PROTOCOL_MAP"] = $"CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,{externalListener}:{clientProtocol},{internalListener}:{clientProtocol}"; // primaryEndpoint is the endpoint that is exposed to the host machine var primaryEndpoint = resource.PrimaryEndpoint; @@ -223,12 +330,36 @@ private static void ConfigureKafkaContainer(EnvironmentCallbackContext context, var internalEndpoint = resource.InternalEndpoint; var advertisedListeners = context.ExecutionContext.IsRunMode - // In run mode, PLAINTEXT_INTERNAL assumes kafka is being accessed over a default Aspire container network and hardcodes the resource address + // In run mode, the internal listener assumes kafka is being accessed over a default Aspire container network and hardcodes the resource address // This will need to be refactored once updated service discovery APIs are available - ? ReferenceExpression.Create($"PLAINTEXT://localhost:29092,PLAINTEXT_HOST://localhost:{primaryEndpoint.Property(EndpointProperty.Port)},PLAINTEXT_INTERNAL://{resource.Name}:{internalEndpoint.Property(EndpointProperty.TargetPort)}") - : ReferenceExpression.Create($"PLAINTEXT://{primaryEndpoint.Property(EndpointProperty.Host)}:29092,PLAINTEXT_HOST://{primaryEndpoint.Property(EndpointProperty.HostAndPort)},PLAINTEXT_INTERNAL://{internalEndpoint.Property(EndpointProperty.HostAndPort)}"); + ? ReferenceExpression.Create($"PLAINTEXT://localhost:29092,{externalListener}://localhost:{primaryEndpoint.Property(EndpointProperty.Port)},{internalListener}://{resource.Name}:{internalEndpoint.Property(EndpointProperty.TargetPort)}") + : ReferenceExpression.Create($"PLAINTEXT://{primaryEndpoint.Property(EndpointProperty.Host)}:29092,{externalListener}://{primaryEndpoint.Property(EndpointProperty.HostAndPort)},{internalListener}://{internalEndpoint.Property(EndpointProperty.HostAndPort)}"); context.EnvironmentVariables["KAFKA_ADVERTISED_LISTENERS"] = advertisedListeners; + + if (saslEnabled) + { + // Authentication only. No authorizer is configured, so every authenticated client keeps full access. + context.EnvironmentVariables["KAFKA_SASL_ENABLED_MECHANISMS"] = "PLAIN"; + + var jaasConfig = BuildJaasConfig(resource); + context.EnvironmentVariables[$"KAFKA_LISTENER_NAME_{externalListener}_PLAIN_SASL_JAAS_CONFIG"] = jaasConfig; + context.EnvironmentVariables[$"KAFKA_LISTENER_NAME_{internalListener}_PLAIN_SASL_JAAS_CONFIG"] = jaasConfig; + } + } + + /// + /// Builds the JAAS configuration declaring the single SASL/PLAIN user accepted by the broker. + /// + private static ReferenceExpression BuildJaasConfig(KafkaServerResource resource) + { + var userName = resource.UserNameReference; + var password = resource.PasswordParameter!; + + // username/password are the credentials the broker presents when it acts as a client, user_ + // declares the credentials the broker accepts from clients. + return ReferenceExpression.Create( + $"org.apache.kafka.common.security.plain.PlainLoginModule required username=\"{userName}\" password=\"{password}\" user_{userName}=\"{password}\";"); } /// diff --git a/src/Aspire.Hosting.Kafka/KafkaServerResource.cs b/src/Aspire.Hosting.Kafka/KafkaServerResource.cs index 23e06c0be2a..09fb87dd993 100644 --- a/src/Aspire.Hosting.Kafka/KafkaServerResource.cs +++ b/src/Aspire.Hosting.Kafka/KafkaServerResource.cs @@ -16,10 +16,23 @@ public class KafkaServerResource(string name) : ContainerResource(name), IResour internal const string PrimaryEndpointName = "tcp"; // This endpoint is used for container to broker communication. internal const string InternalEndpointName = "internal"; + internal const string DefaultUserName = "kafka"; private EndpointReference? _primaryEndpoint; private EndpointReference? _internalEndpoint; + /// + /// Initializes a new instance of the class. + /// + /// The name of the resource. + /// A parameter that contains the SASL user name, or to use a default value. + /// A parameter that contains the SASL password, or to disable authentication. + public KafkaServerResource(string name, ParameterResource? userName, ParameterResource? password) : this(name) + { + UserNameParameter = userName; + PasswordParameter = password; + } + /// /// Gets the primary endpoint for the Kafka broker. This endpoint is used for host processes to Kafka broker communication. /// To connect to the Kafka broker from a container, use . @@ -42,15 +55,66 @@ public class KafkaServerResource(string name) : ContainerResource(name), IResour /// public EndpointReference InternalEndpoint => _internalEndpoint ??= new(this, InternalEndpointName, KnownNetworkIdentifiers.DefaultAspireContainerNetwork); + /// + /// Gets or sets the parameter that contains the SASL user name for the Kafka broker. + /// + public ParameterResource? UserNameParameter { get; set; } + + /// + /// Gets a reference to the SASL user name for the Kafka broker. + /// + /// + /// Returns the user name parameter if specified, otherwise returns the default user name "kafka". + /// + public ReferenceExpression UserNameReference => + UserNameParameter is not null ? + ReferenceExpression.Create($"{UserNameParameter}") : + ReferenceExpression.Create($"{DefaultUserName}"); + + /// + /// Gets or sets the parameter that contains the SASL password for the Kafka broker. + /// + /// + /// When the broker listens without authentication. + /// + public ParameterResource? PasswordParameter { get; set; } + /// /// Gets the connection string expression for the Kafka broker. /// - public ReferenceExpression ConnectionStringExpression => - ReferenceExpression.Create($"{PrimaryEndpoint.Property(EndpointProperty.HostAndPort)}"); + /// + /// When no password is configured the connection string is the bare {host}:{port} of the broker. + /// When a password is configured it is a semicolon separated list of Confluent client configuration + /// properties: BootstrapServers={host}:{port};SecurityProtocol=SaslPlaintext;SaslMechanism=Plain;SaslUsername={user};SaslPassword="{password}". + /// The password is quoted so that it may contain ; and =; it must not contain a double quote. + /// + public ReferenceExpression ConnectionStringExpression => BuildConnectionString(); + + private ReferenceExpression BuildConnectionString() + { + if (PasswordParameter is null) + { + return ReferenceExpression.Create($"{PrimaryEndpoint.Property(EndpointProperty.HostAndPort)}"); + } + + var builder = new ReferenceExpressionBuilder(); + builder.Append($"BootstrapServers={PrimaryEndpoint.Property(EndpointProperty.HostAndPort)}"); + builder.AppendLiteral(";SecurityProtocol=SaslPlaintext;SaslMechanism=Plain;SaslUsername="); + builder.Append($"{UserNameReference}"); + builder.Append($";SaslPassword=\"{PasswordParameter}\""); + + return builder.Build(); + } IEnumerable> IResourceWithConnectionString.GetConnectionProperties() { yield return new("Host", ReferenceExpression.Create($"{Host}")); yield return new("Port", ReferenceExpression.Create($"{Port}")); + + if (PasswordParameter is not null) + { + yield return new("Username", UserNameReference); + yield return new("Password", ReferenceExpression.Create($"{PasswordParameter}")); + } } } diff --git a/src/Aspire.Hosting.Kafka/README.md b/src/Aspire.Hosting.Kafka/README.md index e1b46bc15d6..3d98589c71d 100644 --- a/src/Aspire.Hosting.Kafka/README.md +++ b/src/Aspire.Hosting.Kafka/README.md @@ -34,6 +34,31 @@ const myService = await builder.addNodeApp("myService", "../my-service", "server .withReference(kafka); ``` +## Authentication + +The broker is protected with SASL/PLAIN authentication over the `SASL_PLAINTEXT` security protocol. A random +password is generated when none is supplied, and is stored in the AppHost user secrets so that it is stable +across runs. Supply your own parameters to control the credentials: + +```csharp +var userName = builder.AddParameter("kafka-user"); +var password = builder.AddParameter("kafka-password", secret: true); + +var kafka = builder.AddKafka("messaging", userName: userName, password: password); +``` + +When no user name is supplied the broker accepts the user `kafka`. + +Authentication can be turned off, which makes the broker listen in plaintext: + +```csharp +var kafka = builder.AddKafka("messaging").WithPassword(null); +``` + +> [!WARNING] +> The password is used to derive the broker configuration, not the stored data, so changing it does not +> invalidate an existing data volume. Clients connecting outside of Aspire must be updated to authenticate. + ## Connection Properties When you reference a Kafka resource using `WithReference`, the following connection properties are made available to the consuming project: @@ -46,8 +71,10 @@ The Kafka server resource exposes the following connection properties: |---------------|-------------| | `Host` | The host-facing Kafka listener hostname or IP address | | `Port` | The host-facing Kafka listener port | +| `Username` | The SASL user name for authentication. Only present when the broker is password protected | +| `Password` | The SASL password for authentication. Only present when the broker is password protected | -Aspire exposes each property as an environment variable named `[RESOURCE]_[PROPERTY]`. For instance, the `Uri` property of a resource called `messaging` becomes `MESSAGING_URI`. +Aspire exposes each property as an environment variable named `[RESOURCE]_[PROPERTY]`. For instance, the `Host` property of a resource called `messaging` becomes `MESSAGING_HOST`. ## Additional documentation diff --git a/src/Components/Aspire.Confluent.Kafka/KafkaConnectionString.cs b/src/Components/Aspire.Confluent.Kafka/KafkaConnectionString.cs new file mode 100644 index 00000000000..563105ca43e --- /dev/null +++ b/src/Components/Aspire.Confluent.Kafka/KafkaConnectionString.cs @@ -0,0 +1,67 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. + +using System.Data.Common; +using Confluent.Kafka; + +namespace Aspire.Confluent.Kafka; + +/// +/// Applies an Aspire supplied connection string onto a . +/// +/// +/// A connection string is either a bare bootstrap server list, for example localhost:9092, or a semicolon +/// separated list of Confluent client configuration properties, for example +/// BootstrapServers=localhost:9092;SecurityProtocol=SaslPlaintext;SaslMechanism=Plain;SaslUsername=kafka;SaslPassword="secret". +/// The latter is produced by Aspire.Hosting.Kafka when the broker is password protected. +/// +internal static class KafkaConnectionString +{ + private const string BootstrapServersKey = "BootstrapServers"; + private const string SecurityProtocolKey = "SecurityProtocol"; + private const string SaslMechanismKey = "SaslMechanism"; + private const string SaslUsernameKey = "SaslUsername"; + private const string SaslPasswordKey = "SaslPassword"; + + public static void Apply(string connectionString, ClientConfig config) + { + // A bare bootstrap server list contains no '=' and cannot be parsed as a keyed connection string. + if (!connectionString.Contains('=')) + { + config.BootstrapServers = connectionString; + return; + } + + var builder = new DbConnectionStringBuilder { ConnectionString = connectionString }; + + if (GetValue(builder, BootstrapServersKey) is string bootstrapServers) + { + config.BootstrapServers = bootstrapServers; + } + + if (GetValue(builder, SecurityProtocolKey) is string securityProtocol && + Enum.TryParse(securityProtocol, ignoreCase: true, out var parsedSecurityProtocol)) + { + config.SecurityProtocol = parsedSecurityProtocol; + } + + if (GetValue(builder, SaslMechanismKey) is string saslMechanism && + Enum.TryParse(saslMechanism, ignoreCase: true, out var parsedSaslMechanism)) + { + config.SaslMechanism = parsedSaslMechanism; + } + + if (GetValue(builder, SaslUsernameKey) is string saslUsername) + { + config.SaslUsername = saslUsername; + } + + if (GetValue(builder, SaslPasswordKey) is string saslPassword) + { + config.SaslPassword = saslPassword; + } + } + + private static string? GetValue(DbConnectionStringBuilder builder, string key) + => builder.TryGetValue(key, out var value) ? value as string : null; +} diff --git a/src/Components/Aspire.Confluent.Kafka/KafkaConsumerSettings.cs b/src/Components/Aspire.Confluent.Kafka/KafkaConsumerSettings.cs index 055a12a410d..439982aa2e4 100644 --- a/src/Components/Aspire.Confluent.Kafka/KafkaConsumerSettings.cs +++ b/src/Components/Aspire.Confluent.Kafka/KafkaConsumerSettings.cs @@ -51,7 +51,7 @@ internal void Consolidate() if (ConnectionString is not null) { - Config.BootstrapServers = ConnectionString; + KafkaConnectionString.Apply(ConnectionString, Config); } if (!DisableMetrics) diff --git a/src/Components/Aspire.Confluent.Kafka/KafkaProducerSettings.cs b/src/Components/Aspire.Confluent.Kafka/KafkaProducerSettings.cs index ab7ebda1128..20bec1eb7c1 100644 --- a/src/Components/Aspire.Confluent.Kafka/KafkaProducerSettings.cs +++ b/src/Components/Aspire.Confluent.Kafka/KafkaProducerSettings.cs @@ -51,7 +51,7 @@ internal void Consolidate() if (ConnectionString is not null) { - Config.BootstrapServers = ConnectionString; + KafkaConnectionString.Apply(ConnectionString, Config); } if (!DisableMetrics) diff --git a/src/Components/Aspire.Confluent.Kafka/README.md b/src/Components/Aspire.Confluent.Kafka/README.md index e98df32c9c8..17fff0756c3 100644 --- a/src/Components/Aspire.Confluent.Kafka/README.md +++ b/src/Components/Aspire.Confluent.Kafka/README.md @@ -75,6 +75,19 @@ And then the connection string will be retrieved from the `ConnectionStrings` co The value provided as connection string will be set to the `BootstrapServers` property of the produced `IProducer` or `IConsumer` instance. Refer to [BootstrapServers](https://docs.confluent.io/platform/current/clients/confluent-kafka-dotnet/_site/api/Confluent.Kafka.ClientConfig.html#Confluent_Kafka_ClientConfig_BootstrapServers) for more information. +A connection string may also be a semicolon separated list of client configuration properties, which is what +`Aspire.Hosting.Kafka` produces when the broker is password protected: + +```json +{ + "ConnectionStrings": { + "myConnection": "BootstrapServers=broker:9092;SecurityProtocol=SaslPlaintext;SaslMechanism=Plain;SaslUsername=kafka;SaslPassword=\"secret\"" + } +} +``` + +The supported keys are `BootstrapServers`, `SecurityProtocol`, `SaslMechanism`, `SaslUsername` and `SaslPassword`, and each one is applied to the corresponding property of the produced client configuration. Values may be quoted so that they can contain `;` and `=`. Any other client configuration option is set through the configuration providers described below. + ### Use configuration providers The Aspire Confluent Kafka component supports [Microsoft.Extensions.Configuration](https://learn.microsoft.com/dotnet/api/microsoft.extensions.configuration). It loads the `KafkaProducerSettings` or `KafkaConsumerSettings` from configuration by respectively using the `Aspire:Confluent:Kafka:Producer` and `Aspire.Confluent:Kafka:Consumer` keys. Example `appsettings.json` that configures some of the options: diff --git a/tests/Aspire.Confluent.Kafka.Tests/CommonHelpers.cs b/tests/Aspire.Confluent.Kafka.Tests/CommonHelpers.cs index 73db28077b4..b3973ff3b17 100644 --- a/tests/Aspire.Confluent.Kafka.Tests/CommonHelpers.cs +++ b/tests/Aspire.Confluent.Kafka.Tests/CommonHelpers.cs @@ -6,4 +6,11 @@ namespace Aspire.Confluent.Kafka.Tests; internal sealed class CommonHelpers { public const string TestingEndpoint = "localhost:9092"; + public const string TestingPassword = "p@ssw0rd1"; + + /// + /// A connection string in the shape produced by Aspire.Hosting.Kafka when the broker is password protected. + /// + public const string TestingSaslConnectionString = + $"BootstrapServers={TestingEndpoint};SecurityProtocol=SaslPlaintext;SaslMechanism=Plain;SaslUsername=kafka;SaslPassword=\"{TestingPassword}\""; } diff --git a/tests/Aspire.Confluent.Kafka.Tests/ConsumerConfigurationTests.cs b/tests/Aspire.Confluent.Kafka.Tests/ConsumerConfigurationTests.cs index 7580de48246..9a9264a8375 100644 --- a/tests/Aspire.Confluent.Kafka.Tests/ConsumerConfigurationTests.cs +++ b/tests/Aspire.Confluent.Kafka.Tests/ConsumerConfigurationTests.cs @@ -262,5 +262,41 @@ public void ConsumerConfigOptionsFromConfig() Assert.Equal(SecurityProtocol.Plaintext, config.SecurityProtocol); } + [Theory] + [InlineData(true)] + [InlineData(false)] + public void ReadsSaslCredentialsFromConnectionString(bool useKeyed) + { + var builder = Host.CreateEmptyApplicationBuilder(null); + + var key = useKeyed ? "messaging" : null; + builder.Configuration.AddInMemoryCollection([ + new KeyValuePair("ConnectionStrings:messaging", CommonHelpers.TestingSaslConnectionString), + new KeyValuePair(ConsumerConformanceTests.CreateConfigKey("Aspire:Confluent:Kafka:Consumer", key, "Config:GroupId"), "unused") + ]); + + if (useKeyed) + { + builder.AddKeyedKafkaConsumer("messaging"); + } + else + { + builder.AddKafkaConsumer("messaging"); + } + + using var host = builder.Build(); + var connectionFactory = useKeyed ? + host.Services.GetRequiredKeyedService(ReflectionHelpers.ConsumerConnectionFactoryStringKeyStringValueType.Value, "messaging") : + host.Services.GetRequiredService(ReflectionHelpers.ConsumerConnectionFactoryStringKeyStringValueType.Value); + + var config = GetConsumerConfig(connectionFactory)!; + + Assert.Equal(CommonHelpers.TestingEndpoint, config.BootstrapServers); + Assert.Equal(SecurityProtocol.SaslPlaintext, config.SecurityProtocol); + Assert.Equal(SaslMechanism.Plain, config.SaslMechanism); + Assert.Equal("kafka", config.SaslUsername); + Assert.Equal(CommonHelpers.TestingPassword, config.SaslPassword); + } + private static ConsumerConfig? GetConsumerConfig(object o) => ReflectionHelpers.ConsumerConnectionFactoryStringKeyStringValueType.Value.GetProperty("Config")!.GetValue(o) as ConsumerConfig; } diff --git a/tests/Aspire.Confluent.Kafka.Tests/ProducerConfigurationTests.cs b/tests/Aspire.Confluent.Kafka.Tests/ProducerConfigurationTests.cs index cb2ad059f66..e35384e4a95 100644 --- a/tests/Aspire.Confluent.Kafka.Tests/ProducerConfigurationTests.cs +++ b/tests/Aspire.Confluent.Kafka.Tests/ProducerConfigurationTests.cs @@ -250,5 +250,55 @@ public void ProducerConfigOptionsFromConfig() Assert.Equal(SecurityProtocol.Plaintext, config.SecurityProtocol); } + [Theory] + [InlineData(true)] + [InlineData(false)] + public void ReadsSaslCredentialsFromConnectionString(bool useKeyed) + { + var builder = Host.CreateEmptyApplicationBuilder(null); + builder.Configuration.AddInMemoryCollection([ + new KeyValuePair("ConnectionStrings:messaging", CommonHelpers.TestingSaslConnectionString) + ]); + + if (useKeyed) + { + builder.AddKeyedKafkaProducer("messaging"); + } + else + { + builder.AddKafkaProducer("messaging"); + } + + using var host = builder.Build(); + var connectionFactory = useKeyed ? + host.Services.GetRequiredKeyedService(ReflectionHelpers.ProducerConnectionFactoryStringKeyStringValueType.Value, "messaging") : + host.Services.GetRequiredService(ReflectionHelpers.ProducerConnectionFactoryStringKeyStringValueType.Value); + + var config = GetProducerConfig(connectionFactory)!; + + Assert.Equal(CommonHelpers.TestingEndpoint, config.BootstrapServers); + Assert.Equal(SecurityProtocol.SaslPlaintext, config.SecurityProtocol); + Assert.Equal(SaslMechanism.Plain, config.SaslMechanism); + Assert.Equal("kafka", config.SaslUsername); + Assert.Equal(CommonHelpers.TestingPassword, config.SaslPassword); + } + + [Fact] + public void ConnectionStringWithQuotedPasswordPreservesSeparators() + { + var builder = Host.CreateEmptyApplicationBuilder(null); + builder.Configuration.AddInMemoryCollection([ + new KeyValuePair("ConnectionStrings:messaging", $"BootstrapServers={CommonHelpers.TestingEndpoint};SaslUsername=kafka;SaslPassword=\"a;b=c\"") + ]); + + builder.AddKafkaProducer("messaging"); + + using var host = builder.Build(); + var config = GetProducerConfig(host.Services.GetRequiredService(ReflectionHelpers.ProducerConnectionFactoryStringKeyStringValueType.Value))!; + + Assert.Equal(CommonHelpers.TestingEndpoint, config.BootstrapServers); + Assert.Equal("a;b=c", config.SaslPassword); + } + private static ProducerConfig? GetProducerConfig(object o) => ReflectionHelpers.ProducerConnectionFactoryStringKeyStringValueType.Value.GetProperty("Config")!.GetValue(o) as ProducerConfig; } diff --git a/tests/Aspire.Hosting.Kafka.Tests/AddKafkaTests.cs b/tests/Aspire.Hosting.Kafka.Tests/AddKafkaTests.cs index 77c314f8f28..20bf182e9bf 100644 --- a/tests/Aspire.Hosting.Kafka.Tests/AddKafkaTests.cs +++ b/tests/Aspire.Hosting.Kafka.Tests/AddKafkaTests.cs @@ -1,4 +1,4 @@ -// Licensed to the .NET Foundation under one or more agreements. +// Licensed to the .NET Foundation under one or more agreements. // The .NET Foundation licenses this file to you under the MIT license. using System.Net.Sockets; @@ -54,11 +54,12 @@ public void AddKafkaContainerWithDefaultsAddsAnnotationMetadata() } [Fact] - public async Task KafkaCreatesConnectionString() + public async Task KafkaWithoutPasswordCreatesBareConnectionString() { var appBuilder = DistributedApplication.CreateBuilder(); appBuilder .AddKafka("kafka") + .WithPassword(null) .WithEndpoint("tcp", e => e.AllocatedEndpoint = new AllocatedEndpoint(e, "localhost", 27017)); using var app = appBuilder.Build(); @@ -72,6 +73,144 @@ public async Task KafkaCreatesConnectionString() Assert.Equal("{kafka.bindings.tcp.host}:{kafka.bindings.tcp.port}", connectionStringResource.ConnectionStringExpression.ValueExpression); } + [Fact] + public async Task KafkaCreatesConnectionStringWithCredentials() + { + var appBuilder = DistributedApplication.CreateBuilder(); + var password = appBuilder.AddParameter("pass", "p@ssw0rd1", secret: true); + + appBuilder + .AddKafka("kafka", password: password) + .WithEndpoint("tcp", e => e.AllocatedEndpoint = new AllocatedEndpoint(e, "localhost", 27017)); + + using var app = appBuilder.Build(); + + var appModel = app.Services.GetRequiredService(); + + var connectionStringResource = Assert.Single(appModel.Resources.OfType()) as IResourceWithConnectionString; + var connectionString = await connectionStringResource.GetConnectionStringAsync(); + + Assert.Equal("BootstrapServers=localhost:27017;SecurityProtocol=SaslPlaintext;SaslMechanism=Plain;SaslUsername=kafka;SaslPassword=\"p@ssw0rd1\"", connectionString); + Assert.Equal( + "BootstrapServers={kafka.bindings.tcp.host}:{kafka.bindings.tcp.port};SecurityProtocol=SaslPlaintext;SaslMechanism=Plain;SaslUsername=kafka;SaslPassword=\"{pass.value}\"", + connectionStringResource.ConnectionStringExpression.ValueExpression); + } + + [Fact] + public async Task KafkaUsesUserNameParameterInConnectionString() + { + var appBuilder = DistributedApplication.CreateBuilder(); + var userName = appBuilder.AddParameter("user", "usr"); + var password = appBuilder.AddParameter("pass", "p@ssw0rd1", secret: true); + + var kafka = appBuilder + .AddKafka("kafka", userName: userName, password: password) + .WithEndpoint("tcp", e => e.AllocatedEndpoint = new AllocatedEndpoint(e, "localhost", 27017)); + + Assert.NotNull(kafka.Resource.UserNameParameter); + + var connectionString = await ((IResourceWithConnectionString)kafka.Resource).GetConnectionStringAsync(); + + Assert.Equal("BootstrapServers=localhost:27017;SecurityProtocol=SaslPlaintext;SaslMechanism=Plain;SaslUsername=usr;SaslPassword=\"p@ssw0rd1\"", connectionString); + } + + [Fact] + public void AddKafkaAddsGeneratedPasswordParameterWithUserSecretsParameterDefaultInRunMode() + { + using var appBuilder = TestDistributedApplicationBuilder.Create(testOutputHelper); + + var kafka = appBuilder.AddKafka("kafka"); + + Assert.Equal("Aspire.Hosting.ApplicationModel.UserSecretsParameterDefault", kafka.Resource.PasswordParameter!.Default?.GetType().FullName); + } + + [Fact] + public void AddKafkaDoesNotAddGeneratedPasswordParameterWithUserSecretsParameterDefaultInPublishMode() + { + using var appBuilder = TestDistributedApplicationBuilder.Create(DistributedApplicationOperation.Publish); + + var kafka = appBuilder.AddKafka("kafka"); + + Assert.NotEqual("Aspire.Hosting.ApplicationModel.UserSecretsParameterDefault", kafka.Resource.PasswordParameter!.Default?.GetType().FullName); + } + + [Fact] + public async Task AddKafkaConfiguresSaslOnClientFacingListeners() + { + using var appBuilder = TestDistributedApplicationBuilder.Create(testOutputHelper); + var password = appBuilder.AddParameter("pass", "p@ssw0rd1", secret: true); + + var kafka = appBuilder.AddKafka("kafka", password: password) + .WithEndpoint("tcp", e => e.AllocatedEndpoint = new AllocatedEndpoint(e, "localhost", 27017)); + + var config = await EnvironmentVariableEvaluator.GetEnvironmentVariablesAsync(kafka.Resource); + + // The controller and inter broker listeners stay on the loopback interface in plaintext. + Assert.Equal( + "PLAINTEXT://localhost:29092,CONTROLLER://localhost:29093,EXTERNAL://0.0.0.0:9092,INTERNAL://0.0.0.0:9093", + config["KAFKA_LISTENERS"]); + Assert.Equal( + "CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,EXTERNAL:SASL_PLAINTEXT,INTERNAL:SASL_PLAINTEXT", + config["KAFKA_LISTENER_SECURITY_PROTOCOL_MAP"]); + Assert.Equal("PLAIN", config["KAFKA_SASL_ENABLED_MECHANISMS"]); + + const string ExpectedJaasConfig = """ + org.apache.kafka.common.security.plain.PlainLoginModule required username="kafka" password="p@ssw0rd1" user_kafka="p@ssw0rd1"; + """; + Assert.Equal(ExpectedJaasConfig, config["KAFKA_LISTENER_NAME_EXTERNAL_PLAIN_SASL_JAAS_CONFIG"]); + Assert.Equal(ExpectedJaasConfig, config["KAFKA_LISTENER_NAME_INTERNAL_PLAIN_SASL_JAAS_CONFIG"]); + } + + [Fact] + public async Task WithPasswordNullRevertsToPlaintextListeners() + { + using var appBuilder = TestDistributedApplicationBuilder.Create(testOutputHelper); + + var kafka = appBuilder.AddKafka("kafka") + .WithPassword(null) + .WithEndpoint("tcp", e => e.AllocatedEndpoint = new AllocatedEndpoint(e, "localhost", 27017)); + + var config = await EnvironmentVariableEvaluator.GetEnvironmentVariablesAsync(kafka.Resource); + + Assert.Equal( + "PLAINTEXT://localhost:29092,CONTROLLER://localhost:29093,PLAINTEXT_HOST://0.0.0.0:9092,PLAINTEXT_INTERNAL://0.0.0.0:9093", + config["KAFKA_LISTENERS"]); + Assert.Equal( + "CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT,PLAINTEXT_INTERNAL:PLAINTEXT", + config["KAFKA_LISTENER_SECURITY_PROTOCOL_MAP"]); + Assert.DoesNotContain(config, kvp => kvp.Key.StartsWith("KAFKA_SASL", StringComparison.Ordinal)); + Assert.DoesNotContain(config, kvp => kvp.Key.StartsWith("KAFKA_LISTENER_NAME_", StringComparison.Ordinal)); + } + + [Fact] + public async Task WithKafkaUIConfiguresSaslProperties() + { + using var appBuilder = TestDistributedApplicationBuilder.Create(testOutputHelper); + var password = appBuilder.AddParameter("pass", "p@ssw0rd1", secret: true); + + appBuilder.AddKafka("kafka1", password: password) + .WithEndpoint("tcp", e => e.AllocatedEndpoint = new AllocatedEndpoint(e, "localhost", 27017)) + .WithKafkaUI(); + + using var app = appBuilder.Build(); + var appModel = app.Services.GetRequiredService(); + var kafkaUiResource = Assert.Single(appModel.Resources.OfType()); + + await appBuilder.Eventing.PublishAsync( + new BeforeResourceStartedEvent(kafkaUiResource, app.Services), + EventDispatchBehavior.BlockingSequential); + + var config = await EnvironmentVariableEvaluator.GetEnvironmentVariablesAsync(kafkaUiResource); + + Assert.Equal("SASL_PLAINTEXT", config["KAFKA_CLUSTERS_0_PROPERTIES_SECURITY_PROTOCOL"]); + Assert.Equal("PLAIN", config["KAFKA_CLUSTERS_0_PROPERTIES_SASL_MECHANISM"]); + Assert.Equal( + """ + org.apache.kafka.common.security.plain.PlainLoginModule required username="kafka" password="p@ssw0rd1" user_kafka="p@ssw0rd1"; + """, + config["KAFKA_CLUSTERS_0_PROPERTIES_SASL_JAAS_CONFIG"]); + } + [Fact] public async Task VerifyManifest() { @@ -81,6 +220,47 @@ public async Task VerifyManifest() var manifest = await ManifestUtils.GetManifest(kafka.Resource); + var expectedManifest = $$""" + { + "type": "container.v0", + "connectionString": "BootstrapServers={kafka.bindings.tcp.host}:{kafka.bindings.tcp.port};SecurityProtocol=SaslPlaintext;SaslMechanism=Plain;SaslUsername=kafka;SaslPassword=\u0022{kafka-password.value}\u0022", + "image": "{{KafkaContainerImageTags.Registry}}/{{KafkaContainerImageTags.Image}}:{{KafkaContainerImageTags.Tag}}", + "env": { + "KAFKA_LISTENERS": "PLAINTEXT://localhost:29092,CONTROLLER://localhost:29093,EXTERNAL://0.0.0.0:9092,INTERNAL://0.0.0.0:9093", + "KAFKA_LISTENER_SECURITY_PROTOCOL_MAP": "CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,EXTERNAL:SASL_PLAINTEXT,INTERNAL:SASL_PLAINTEXT", + "KAFKA_ADVERTISED_LISTENERS": "PLAINTEXT://{kafka.bindings.tcp.host}:29092,EXTERNAL://{kafka.bindings.tcp.host}:{kafka.bindings.tcp.port},INTERNAL://{kafka.bindings.internal.host}:{kafka.bindings.internal.port}", + "KAFKA_SASL_ENABLED_MECHANISMS": "PLAIN", + "KAFKA_LISTENER_NAME_EXTERNAL_PLAIN_SASL_JAAS_CONFIG": "org.apache.kafka.common.security.plain.PlainLoginModule required username=\u0022kafka\u0022 password=\u0022{kafka-password.value}\u0022 user_kafka=\u0022{kafka-password.value}\u0022;", + "KAFKA_LISTENER_NAME_INTERNAL_PLAIN_SASL_JAAS_CONFIG": "org.apache.kafka.common.security.plain.PlainLoginModule required username=\u0022kafka\u0022 password=\u0022{kafka-password.value}\u0022 user_kafka=\u0022{kafka-password.value}\u0022;" + }, + "bindings": { + "tcp": { + "scheme": "tcp", + "protocol": "tcp", + "transport": "tcp", + "targetPort": 9092 + }, + "internal": { + "scheme": "tcp", + "protocol": "tcp", + "transport": "tcp", + "targetPort": 9093 + } + } + } + """; + Assert.Equal(expectedManifest, manifest.ToString()); + } + + [Fact] + public async Task VerifyManifestWithoutPassword() + { + using var appBuilder = TestDistributedApplicationBuilder.Create(testOutputHelper); + + var kafka = appBuilder.AddKafka("kafka").WithPassword(null); + + var manifest = await ManifestUtils.GetManifest(kafka.Resource); + var expectedManifest = $$""" { "type": "container.v0", diff --git a/tests/Aspire.Hosting.Kafka.Tests/ConnectionPropertiesTests.cs b/tests/Aspire.Hosting.Kafka.Tests/ConnectionPropertiesTests.cs index 68ce5d9db31..dc1ac475e86 100644 --- a/tests/Aspire.Hosting.Kafka.Tests/ConnectionPropertiesTests.cs +++ b/tests/Aspire.Hosting.Kafka.Tests/ConnectionPropertiesTests.cs @@ -27,4 +27,49 @@ public void KafkaServerResourceGetConnectionPropertiesReturnsExpectedValues() Assert.Equal("{kafka.bindings.tcp.port}", property.Value.ValueExpression); }); } + + [Fact] + public void KafkaServerResourceGetConnectionPropertiesIncludesCredentialsWhenPasswordIsConfigured() + { + var user = new ParameterResource("user", _ => "kafkauser"); + var password = new ParameterResource("password", _ => "p@ssw0rd1", secret: true); + var resource = new KafkaServerResource("kafka", user, password); + + var properties = ((IResourceWithConnectionString)resource).GetConnectionProperties().ToArray(); + + Assert.Collection( + properties, + property => + { + Assert.Equal("Host", property.Key); + Assert.Equal("{kafka.bindings.tcp.host}", property.Value.ValueExpression); + }, + property => + { + Assert.Equal("Port", property.Key); + Assert.Equal("{kafka.bindings.tcp.port}", property.Value.ValueExpression); + }, + property => + { + Assert.Equal("Username", property.Key); + Assert.Equal("{user.value}", property.Value.ValueExpression); + }, + property => + { + Assert.Equal("Password", property.Key); + Assert.Equal("{password.value}", property.Value.ValueExpression); + }); + } + + [Fact] + public void KafkaServerResourceUsesDefaultUserNameWhenNoUserNameParameterIsConfigured() + { + var password = new ParameterResource("password", _ => "p@ssw0rd1", secret: true); + var resource = new KafkaServerResource("kafka", null, password); + + var properties = ((IResourceWithConnectionString)resource).GetConnectionProperties().ToArray(); + + var userName = Assert.Single(properties, p => p.Key == "Username"); + Assert.Equal("kafka", userName.Value.ValueExpression); + } } \ No newline at end of file diff --git a/tests/Aspire.Hosting.Kafka.Tests/KafkaFunctionalTests.cs b/tests/Aspire.Hosting.Kafka.Tests/KafkaFunctionalTests.cs index ef1ee1ab265..893a93e8d0a 100644 --- a/tests/Aspire.Hosting.Kafka.Tests/KafkaFunctionalTests.cs +++ b/tests/Aspire.Hosting.Kafka.Tests/KafkaFunctionalTests.cs @@ -112,6 +112,43 @@ await pipeline.ExecuteAsync(async token => } } + [Fact] + [RequiresFeature(TestFeature.Docker)] + [ActiveIssue("https://github.com/microsoft/aspire/issues/11820", typeof(PlatformDetection), nameof(PlatformDetection.IsRunningFromAzdo))] + public async Task VerifyKafkaResourceRejectsUnauthenticatedClients() + { + var cts = new CancellationTokenSource(TimeSpan.FromMinutes(3)); + + using var builder = TestDistributedApplicationBuilder.CreateWithTestContainerRegistry(testOutputHelper); + + var kafka = builder.AddKafka("kafka"); + + using var app = builder.Build(); + + await app.StartAsync(); + await app.WaitForHealthyAsync(kafka); + + // Connect using the bootstrap servers alone, deliberately dropping the SASL credentials that the + // connection string carries. + var bootstrapServers = await kafka.Resource.PrimaryEndpoint.Property(EndpointProperty.HostAndPort).GetValueAsync(cts.Token); + + var hb = Host.CreateApplicationBuilder(); + hb.AddTestLogging(testOutputHelper); + + hb.Configuration[$"ConnectionStrings:{kafka.Resource.Name}"] = bootstrapServers; + + hb.AddKafkaProducer("kafka", configureSettings: settings => settings.Config.MessageTimeoutMs = 5000); + + using var host = hb.Build(); + + await host.StartAsync(); + + var producer = host.Services.GetRequiredService>(); + + await Assert.ThrowsAsync>( + () => producer.ProduceAsync("test-topic", new Message { Key = "test-key", Value = "test-value" }, cts.Token)); + } + [Theory] [InlineData(true)] [InlineData(false)] diff --git a/tests/Aspire.Hosting.Kafka.Tests/KafkaPublicApiTests.cs b/tests/Aspire.Hosting.Kafka.Tests/KafkaPublicApiTests.cs index dbada7c39a1..63a0a21fa65 100644 --- a/tests/Aspire.Hosting.Kafka.Tests/KafkaPublicApiTests.cs +++ b/tests/Aspire.Hosting.Kafka.Tests/KafkaPublicApiTests.cs @@ -36,6 +36,76 @@ public void AddKafkaShouldThrowWhenNameIsNullOrEmpty(bool isNull) Assert.Equal(nameof(name), exception.ParamName); } + [Theory] + [InlineData(true)] + [InlineData(false)] + public void AddKafkaWithParametersShouldThrowWhenBuilderIsNull(bool includePort) + { + IDistributedApplicationBuilder builder = null!; + const string name = "Kafka"; + IResourceBuilder? userName = null; + IResourceBuilder? password = null; + + var action = () => includePort + ? builder.AddKafka(name, 9092, userName: userName, password: password) + : builder.AddKafka(name, userName: userName, password: password); + + var exception = Assert.Throws(action); + Assert.Equal(nameof(builder), exception.ParamName); + } + + [Theory] + [InlineData(true)] + [InlineData(false)] + public void AddKafkaWithParametersShouldThrowWhenNameIsNullOrEmpty(bool isNull) + { + var builder = TestDistributedApplicationBuilder.Create(testOutputHelper); + var name = isNull ? null! : string.Empty; + + var action = () => builder.AddKafka(name, userName: null, password: null); + + var exception = isNull + ? Assert.Throws(action) + : Assert.Throws(action); + Assert.Equal(nameof(name), exception.ParamName); + } + + [Fact] + public void WithPasswordShouldThrowWhenBuilderIsNull() + { + IResourceBuilder builder = null!; + + var action = () => builder.WithPassword(null); + + var exception = Assert.Throws(action); + Assert.Equal(nameof(builder), exception.ParamName); + } + + [Fact] + public void WithUserNameShouldThrowWhenBuilderIsNull() + { + IResourceBuilder builder = null!; + IResourceBuilder userName = null!; + + var action = () => builder.WithUserName(userName); + + var exception = Assert.Throws(action); + Assert.Equal(nameof(builder), exception.ParamName); + } + + [Fact] + public void WithUserNameShouldThrowWhenUserNameIsNull() + { + var builder = TestDistributedApplicationBuilder.Create(testOutputHelper) + .AddKafka("kafka"); + IResourceBuilder userName = null!; + + var action = () => builder.WithUserName(userName); + + var exception = Assert.Throws(action); + Assert.Equal(nameof(userName), exception.ParamName); + } + [Fact] public void WithKafkaUIShouldThrowWhenBuilderIsNull() { @@ -114,6 +184,29 @@ public void CtorKafkaServerResourceShouldThrowWhenNameIsNullOrEmpty(bool isNull) Assert.Equal(nameof(name), exception.ParamName); } + [Theory] + [InlineData(true, true, true)] + [InlineData(true, true, false)] + [InlineData(true, false, true)] + [InlineData(true, false, false)] + [InlineData(false, true, true)] + [InlineData(false, true, false)] + [InlineData(false, false, true)] + [InlineData(false, false, false)] + public void CtorKafkaServerResourceWithParametersShouldThrowWhenNameIsNullOrEmpty(bool isNull, bool isNullUserName, bool isNullPassword) + { + var name = isNull ? null! : string.Empty; + var userName = isNullUserName ? null : new ParameterResource("user", _ => "usr"); + var password = isNullPassword ? null : new ParameterResource("pass", _ => "p@ssw0rd1", secret: true); + + var action = () => new KafkaServerResource(name, userName, password); + + var exception = isNull + ? Assert.Throws(action) + : Assert.Throws(action); + Assert.Equal(nameof(name), exception.ParamName); + } + [Theory] [InlineData(true)] [InlineData(false)]