Add SASL/PLAIN password protection to the Kafka container

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;
 
+    /// <summary>
+    /// Initializes a new instance of the <see cref="KafkaServerResource"/> class.
+    /// </summary>
+    /// <param name="name">The name of the resource.</param>
+    /// <param name="userName">A parameter that contains the SASL user name, or <see langword="null"/> to use a default value.</param>
+    /// <param name="password">A parameter that contains the SASL password, or <see langword="null"/> to disable authentication.</param>
+    public KafkaServerResource(string name, ParameterResource? userName, ParameterResource? password) : this(name)
+    {
+        UserNameParameter = userName;
+        PasswordParameter = password;
+    }
+
     /// <summary>
     /// 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 <see cref="InternalEndpoint"/>.
@@ -42,15 +55,66 @@ public class KafkaServerResource(string name) : ContainerResource(name), IResour
     /// </summary>
     public EndpointReference InternalEndpoint => _internalEndpoint ??= new(this, InternalEndpointName, KnownNetworkIdentifiers.DefaultAspireContainerNetwork);
 
+    /// <summary>
+    /// Gets or sets the parameter that contains the SASL user name for the Kafka broker.
+    /// </summary>
+    public ParameterResource? UserNameParameter { get; set; }
+
+    /// <summary>
+    /// Gets a reference to the SASL user name for the Kafka broker.
+    /// </summary>
+    /// <remarks>
+    /// Returns the user name parameter if specified, otherwise returns the default user name "kafka".
+    /// </remarks>
+    public ReferenceExpression UserNameReference =>
+        UserNameParameter is not null ?
+            ReferenceExpression.Create($"{UserNameParameter}") :
+            ReferenceExpression.Create($"{DefaultUserName}");
+
+    /// <summary>
+    /// Gets or sets the parameter that contains the SASL password for the Kafka broker.
+    /// </summary>
+    /// <remarks>
+    /// When <see langword="null"/> the broker listens without authentication.
+    /// </remarks>
+    public ParameterResource? PasswordParameter { get; set; }
+
     /// <summary>
     /// Gets the connection string expression for the Kafka broker.
     /// </summary>
-    public ReferenceExpression ConnectionStringExpression =>
-       ReferenceExpression.Create($"{PrimaryEndpoint.Property(EndpointProperty.HostAndPort)}");
+    /// <remarks>
+    /// When no password is configured the connection string is the bare <c>{host}:{port}</c> of the broker.
+    /// When a password is configured it is a semicolon separated list of Confluent client configuration
+    /// properties: <c>BootstrapServers={host}:{port};SecurityProtocol=SaslPlaintext;SaslMechanism=Plain;SaslUsername={user};SaslPassword="{password}"</c>.
+    /// The password is quoted so that it may contain <c>;</c> and <c>=</c>; it must not contain a double quote.
+    /// </remarks>
+    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<KeyValuePair<string, ReferenceExpression>> 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}"));
+        }
     }
 }

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
 

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;
+
+/// <summary>
+/// Applies an Aspire supplied connection string onto a <see cref="ClientConfig"/>.
+/// </summary>
+/// <remarks>
+/// A connection string is either a bare bootstrap server list, for example <c>localhost:9092</c>, or a semicolon
+/// separated list of Confluent client configuration properties, for example
+/// <c>BootstrapServers=localhost:9092;SecurityProtocol=SaslPlaintext;SaslMechanism=Plain;SaslUsername=kafka;SaslPassword="secret"</c>.
+/// The latter is produced by <c>Aspire.Hosting.Kafka</c> when the broker is password protected.
+/// </remarks>
+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>(securityProtocol, ignoreCase: true, out var parsedSecurityProtocol))
+        {
+            config.SecurityProtocol = parsedSecurityProtocol;
+        }
+
+        if (GetValue(builder, SaslMechanismKey) is string saslMechanism &&
+            Enum.TryParse<SaslMechanism>(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;
+}

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)

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)

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<TKey, TValue>` or `IConsumer<TKey, TValue>` 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:

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";
+
+    /// <summary>
+    /// A connection string in the shape produced by Aspire.Hosting.Kafka when the broker is password protected.
+    /// </summary>
+    public const string TestingSaslConnectionString =
+        $"BootstrapServers={TestingEndpoint};SecurityProtocol=SaslPlaintext;SaslMechanism=Plain;SaslUsername=kafka;SaslPassword=\"{TestingPassword}\"";
 }

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<string, string?>("ConnectionStrings:messaging", CommonHelpers.TestingSaslConnectionString),
+            new KeyValuePair<string, string?>(ConsumerConformanceTests.CreateConfigKey("Aspire:Confluent:Kafka:Consumer", key, "Config:GroupId"), "unused")
+        ]);
+
+        if (useKeyed)
+        {
+            builder.AddKeyedKafkaConsumer<string, string>("messaging");
+        }
+        else
+        {
+            builder.AddKafkaConsumer<string, string>("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;
 }

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<string, string?>("ConnectionStrings:messaging", CommonHelpers.TestingSaslConnectionString)
+        ]);
+
+        if (useKeyed)
+        {
+            builder.AddKeyedKafkaProducer<string, string>("messaging");
+        }
+        else
+        {
+            builder.AddKafkaProducer<string, string>("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<string, string?>("ConnectionStrings:messaging", $"BootstrapServers={CommonHelpers.TestingEndpoint};SaslUsername=kafka;SaslPassword=\"a;b=c\"")
+        ]);
+
+        builder.AddKafkaProducer<string, string>("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;
 }