Enhance Bearer Authentication Support with Azure IMDS

CHANGELOG.md

@@ -1,3 +1,12 @@
+# 2.12.0
+
+## Enhancements
+
+* References librdkafka.redist 2.12.0. Refer to the [librdkafka v2.12.0 release notes](https://github.com/confluentinc/librdkafka/releases/tag/v2.12.0) for more information.
+* OAuth OIDC method for Schema Registry metadata based authentication with
+  an Azure IMDS endpoint using an attached managed identity as principal (#2523).
+
+
 # 2.11.1
 
 ## Enhancements

examples/OAuthOIDCAzureIMDS/OAuthOIDCAzureIMDS.csproj

@@ -0,0 +1,16 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <ProjectTypeGuids>{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}</ProjectTypeGuids>
+    <AssemblyName>OAuthOIDC</AssemblyName>
+    <TargetFramework>net8.0</TargetFramework>
+    <OutputType>Exe</OutputType>
+    <LangVersion>7.1</LangVersion>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <!-- nuget package reference: <PackageReference Include="Confluent.SchemaRegistry.Serdes.Json" Version="2.11.1" /> -->
+    <ProjectReference Include="../../src/Confluent.SchemaRegistry.Serdes.Json/Confluent.SchemaRegistry.Serdes.Json.csproj" />
+  </ItemGroup>
+
+</Project>

examples/OAuthOIDCAzureIMDS/Program.cs

@@ -0,0 +1,190 @@
+// Copyright 2022 Confluent Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+// Refer to LICENSE for more information.
+
+using System;
+using System.Collections.Generic;
+using System.Text;
+using System.Text.RegularExpressions;
+using System.Threading;
+using System.Threading.Tasks;
+using Confluent.Kafka;
+using Confluent.Kafka.Admin;
+using Confluent.Kafka.SyncOverAsync;
+using Newtonsoft.Json;
+using Confluent.SchemaRegistry;
+using Confluent.SchemaRegistry.Serdes;
+
+/// <summary>
+///     An example demonstrating how to produce a message to 
+///     a topic, and then reading it back again using a consumer.
+///     The authentication uses the OpenID Connect method of the OAUTHBEARER SASL mechanism.
+///     The token is acquired from the Azure Instance Metadata Service (IMDS)
+///     using metadata based secret-less authentication.
+/// </summary>
+namespace Confluent.Kafka.Examples.OAuthOIDCAzureIMDS
+{
+
+    class User
+    {
+        [JsonRequired] // use Newtonsoft.Json annotations
+        [JsonProperty("name")]
+        public string Name { get; set; }
+
+        [JsonRequired]
+        [JsonProperty("favorite_color")]
+        public string FavoriteColor { get; set; }
+
+        [JsonProperty("favorite_number")]
+        public long FavoriteNumber { get; set; }
+    }
+
+    public class Program
+    {
+        private const string azureIMDSQueryParams = "api-version=&resource=&client_id=";
+        private const string kafkaLogicalCluster = "your-logical-cluster";
+        private const string identityPoolId = "your-identity-pool-id";
+
+        public static async Task Main(string[] args)
+        {
+            if (args.Length != 3)
+            {
+                Console.WriteLine("Usage: .. brokerList schemaRegistryUrl");
+                return;
+            }
+            var bootstrapServers = args[1];
+            var schemaRegistryUrl = args[2];
+            var topicName = Guid.NewGuid().ToString();
+            var groupId = Guid.NewGuid().ToString();
+
+            var commonConfig = new ClientConfig
+            {
+                BootstrapServers = bootstrapServers,
+                SecurityProtocol = SecurityProtocol.SaslPlaintext,
+                SaslMechanism = SaslMechanism.OAuthBearer,
+                SaslOauthbearerMethod = SaslOauthbearerMethod.Oidc,
+                SaslOauthbearerMetadataAuthenticationType = SaslOauthbearerMetadataAuthenticationType.AzureIMDS,
+                SaslOauthbearerConfig = $"query={azureIMDSQueryParams}",
+                SaslOauthbearerExtensions = $"logicalCluster={kafkaLogicalCluster},identityPoolId={identityPoolId}"
+            };
+
+            var consumerConfig = new ConsumerConfig
+            {
+                BootstrapServers = bootstrapServers,
+                SecurityProtocol = SecurityProtocol.SaslPlaintext,
+                SaslMechanism = SaslMechanism.OAuthBearer,
+                SaslOauthbearerMethod = SaslOauthbearerMethod.Oidc,
+                GroupId = groupId,
+                AutoOffsetReset = AutoOffsetReset.Earliest,
+                EnableAutoOffsetStore = false
+            };
+
+            var schemaRegistryConfig = new SchemaRegistryConfig
+            {
+                Url = schemaRegistryUrl,
+                BearerAuthCredentialsSource = BearerAuthCredentialsSource.OAuthBearerAzureIMDS,
+                BearerAuthTokenEndpointQuery = azureIMDSQueryParams,
+                BearerAuthLogicalCluster = kafkaLogicalCluster,
+                BearerAuthIdentityPoolId = identityPoolId
+            };
+
+            try
+            {
+                createTopic(commonConfig, topicName);
+            }
+            catch (CreateTopicsException e)
+            {
+                Console.WriteLine($"An error occurred creating topic {e.Results[0].Topic}: {e.Results[0].Error.Reason}");
+                Environment.Exit(1);
+            }
+
+            using (var schemaRegistry = new CachedSchemaRegistryClient(schemaRegistryConfig))
+            using (var producer = new ProducerBuilder<Null, User>(commonConfig)
+                .SetValueSerializer(new JsonSerializer<User>(schemaRegistry))
+                .Build())
+            using (var consumer = new ConsumerBuilder<Ignore, User>(consumerConfig)
+                .SetValueDeserializer(new JsonDeserializer<User>(schemaRegistry).AsSyncOverAsync()).Build())
+            {
+                consumer.Subscribe(topicName);
+
+                var cancelled = false;
+                CancellationTokenSource cts = new CancellationTokenSource();
+
+                Console.CancelKeyPress += (_, e) =>
+                {
+                    e.Cancel = true; // prevent the process from terminating.
+                    cancelled = true;
+                    cts.Cancel();
+                };
+
+                try
+                {
+                    while (!cancelled)
+                    {
+                        var msg = new User
+                        {
+                            Name = "user-" + Guid.NewGuid().ToString(),
+                            FavoriteColor = "blue",
+                            FavoriteNumber = 7
+                        };
+
+                        try
+                        {
+                            var deliveryReport = await producer.ProduceAsync(topicName, new Message<Null, User> { Value = msg });
+                            Console.WriteLine($"Produced message to {deliveryReport.TopicPartitionOffset}, {msg}");
+                        }
+                        catch (ProduceException<Null, User> e)
+                        {
+                            Console.WriteLine($"failed to deliver message: {e.Message} [{e.Error.Code}]");
+                        }
+
+                        try
+                        {
+                            var consumeResult = consumer.Consume(cts.Token);
+                            Console.WriteLine($"Received message at {consumeResult.TopicPartitionOffset}: {consumeResult.Message.Value}");
+                            try
+                            {
+                                consumer.StoreOffset(consumeResult);
+                            }
+                            catch (KafkaException e)
+                            {
+                                Console.WriteLine($"Store Offset error: {e.Error.Reason}");
+                            }
+                        }
+                        catch (ConsumeException e)
+                        {
+                            Console.WriteLine($"Consume error: {e.Error.Reason}");
+                        }
+                    }
+                }
+                catch (OperationCanceledException)
+                {
+                    Console.WriteLine("Closing consumer.");
+                    consumer.Close();
+                }
+            }
+        }
+
+        private static void createTopic(ClientConfig config, String topicName)
+        {
+            using (var adminClient = new AdminClientBuilder(config).Build())
+            {
+                adminClient.CreateTopicsAsync(new TopicSpecification[] {
+                            new TopicSpecification { Name = topicName, ReplicationFactor = 3, NumPartitions = 1 } }).Wait(); ;
+            }
+        }
+    }
+
+}

examples/SchemaRegistryOAuth/Program.cs

@@ -50,9 +50,9 @@ public ExampleBearerAuthProvider(string token, string logicalCluster, string ide
 
         public static async Task Main(string[] args)
         {
-            if (args.Length != 9)
+            if (args.Length < 9)
             {
-                Console.WriteLine("Usage: .. schemaRegistryUrl clientId clientSecret scope tokenEndpoint logicalCluster identityPool token");
+                Console.WriteLine("Usage: .. schemaRegistryUrl clientId clientSecret scope tokenEndpoint logicalCluster identityPool token [azureIMDSQueryParams]");
                 return;
             }
             string schemaRegistryUrl = args[1];
@@ -63,10 +63,15 @@ public static async Task Main(string[] args)
             string logicalCluster = args[6];
             string identityPool = args[7];
             string token = args[8];
+            string? azureIMDSQueryParams = null;
+            if (args.Length >= 10)
+            { 
+                azureIMDSQueryParams = args[9];
+            }
 
             //using BearerAuthCredentialsSource.OAuthBearer
             var clientCredentialsSchemaRegistryConfig = new SchemaRegistryConfig
-            {   
+            {
                 Url = schemaRegistryUrl,
                 BearerAuthCredentialsSource = BearerAuthCredentialsSource.OAuthBearer,
                 BearerAuthClientId = clientId,
@@ -85,7 +90,7 @@ public static async Task Main(string[] args)
 
             //using BearerAuthCredentialsSource.StaticToken
             var staticSchemaRegistryConfig = new SchemaRegistryConfig
-            {   
+            {
                 Url = schemaRegistryUrl,
                 BearerAuthCredentialsSource = BearerAuthCredentialsSource.StaticToken,
                 BearerAuthToken = token,
@@ -101,7 +106,7 @@ public static async Task Main(string[] args)
 
             //Using BearerAuthCredentialsSource.Custom
             var customSchemaRegistryConfig = new SchemaRegistryConfig
-            {   
+            {
                 Url = schemaRegistryUrl,
                 BearerAuthCredentialsSource = BearerAuthCredentialsSource.Custom
             };
@@ -112,6 +117,27 @@ public static async Task Main(string[] args)
                 var subjects = await schemaRegistry.GetAllSubjectsAsync();
                 Console.WriteLine(string.Join(", ", subjects));
             }
+
+            if (azureIMDSQueryParams == null)
+            {
+                return;
+            }
+
+            //Using BearerAuthCredentialsSource.OAuthOIDCAzureIMDS
+            var azureIMDSSchemaRegistryConfig = new SchemaRegistryConfig
+            {
+                Url = schemaRegistryUrl,
+                BearerAuthCredentialsSource = BearerAuthCredentialsSource.OAuthBearerAzureIMDS,
+                BearerAuthTokenEndpointQuery = azureIMDSQueryParams,
+                BearerAuthLogicalCluster = logicalCluster,
+                BearerAuthIdentityPoolId = identityPool,
+            };
+
+            using (var schemaRegistry = new CachedSchemaRegistryClient(azureIMDSSchemaRegistryConfig))
+            {
+                var subjects = await schemaRegistry.GetAllSubjectsAsync();
+                Console.WriteLine(string.Join(", ", subjects));
+            }
         }
     }
 }

src/ConfigGen/Program.cs

@@ -320,6 +320,7 @@ static string ConfigNameToDotnetName(string configName)
             { "resolve_canonical_bootstrap_servers_only", "ResolveCanonicalBootstrapServersOnly"},
             { "client_credentials", "ClientCredentials"},
             { "urn:ietf:params:oauth:grant-type:jwt-bearer", "JwtBearer"},
+            { "azure_imds", "AzureIMDS"},
         };
 
         static string EnumNameToDotnetName(string enumName)

src/Confluent.Kafka/Config_gen.cs

@@ -1,4 +1,4 @@
-// *** Auto-generated from librdkafka v2.11.1 *** - do not modify manually.
+// *** Auto-generated from librdkafka dev_oauthbearer_metadata_based *** - do not modify manually.
 //
 // Copyright 2018-2022 Confluent Inc.
 //
@@ -203,6 +203,22 @@ public enum SaslOauthbearerAssertionAlgorithm
         ES256
     }
 
+    /// <summary>
+    ///     SaslOauthbearerMetadataAuthenticationType enum values
+    /// </summary>
+    public enum SaslOauthbearerMetadataAuthenticationType
+    {
+        /// <summary>
+        ///     None
+        /// </summary>
+        None,
+
+        /// <summary>
+        ///     AzureIMDS
+        /// </summary>
+        AzureIMDS
+    }
+
     /// <summary>
     ///     PartitionAssignmentStrategy enum values
     /// </summary>
@@ -1318,6 +1334,16 @@ public Acks? Acks
         /// </summary>
         public string SaslOauthbearerAssertionJwtTemplateFile { get { return Get("sasl.oauthbearer.assertion.jwt.template.file"); } set { this.SetObject("sasl.oauthbearer.assertion.jwt.template.file", value); } }
 
+        /// <summary>
+        /// <![CDATA[
+        ///     Type of metadata-based authentication to use for OAUTHBEARER/OIDC `azure_imds` authenticates using the Azure IMDS endpoint. Sets a default value for `sasl.oauthbearer.token.endpoint.url` if missing. Configuration values specific of chosen authentication type can be passed through `sasl.oauthbearer.config`.
+        ///
+        ///     default: none
+        ///     importance: low
+        /// ]]>
+        /// </summary>
+        public SaslOauthbearerMetadataAuthenticationType? SaslOauthbearerMetadataAuthenticationType { get { return (SaslOauthbearerMetadataAuthenticationType?)GetEnum(typeof(SaslOauthbearerMetadataAuthenticationType), "sasl.oauthbearer.metadata.authentication.type"); } set { this.SetObject("sasl.oauthbearer.metadata.authentication.type", value); } }
+
         /// <summary>
         /// <![CDATA[
         ///     List of plugin libraries to load (; separated). The library search path is platform dependent (see dlopen(3) for Unix and LoadLibrary() for Windows). If no filename extension is specified the platform-specific extension (such as .dll or .so) will be appended automatically.