Merge master

confluentinc · Dec 19, 2024 · 49afa24 · 49afa24
2 parents d63ca5f + aef7faa
commit 49afa24
Show file tree

Hide file tree

Showing 26 changed files with 432 additions and 152 deletions.
diff --git a/.github/pull_request_template.md b/.github/pull_request_template.md
@@ -0,0 +1,43 @@
+<!--
+Suggested PR template: Fill/delete/add sections as needed. Optionally delete any commented block.
+-->
+What
+----
+<!--
+Briefly describe **what** you have changed and **why**.
+Optionally include implementation strategy.
+-->
+
+Checklist
+------------------
+- [ ] Contains customer facing changes? Including API/behavior changes <!-- This can help identify if it has introduced any breaking changes -->
+- [ ] Did you add sufficient unit test and/or integration test coverage for this PR?
+  - If not, please explain why it is not required
+
+References
+----------
+JIRA: 
+<!--
+Copy&paste links: to Jira ticket, other PRs, issues, Slack conversations...
+For code bumps: link to PR, tag or GitHub `/compare/master...master`
+-->
+
+Test & Review
+------------
+<!--
+Has it been tested? how?
+Copy&paste any handy instructions, steps or requirements that can save time to the reviewer or any reader.
+-->
+
+Open questions / Follow-ups
+--------------------------
+<!--
+Optional: anything open to discussion for the reviewer, out of scope, or follow-ups.
+-->
+
+<!--
+Review stakeholders
+------------------
+<!--
+Optional: mention stakeholders or if special context that is required to review.
+-->
diff --git a/3RD_PARTY.md b/3RD_PARTY.md
@@ -10,3 +10,4 @@ To add your project, open a pull request!
 - [Multi Schema Avro Deserializer](https://github.com/ycherkes/multi-schema-avro-desrializer) - Avro deserializer for reading messages serialized with multiple schemas.
 - [OpenSleigh.Transport.Kafka](https://github.com/mizrael/OpenSleigh/tree/develop/src/OpenSleigh.Transport.Kafka) - A Kafka Transport for OpenSleigh, a distributed saga management library.
 - [SlimMessageBus.Host.Kafka](https://github.com/zarusz/SlimMessageBus) - Apache Kafka transport for SlimMessageBus (lightweight message bus for .NET)
+- [Kafka Core](https://github.com/ffernandolima/confluent-kafka-core-dotnet) - Kafka Core empowers developers to build robust .NET applications on top of Confluent Kafka, focusing on simplicity, maintainability, and extensibility with intuitive abstractions and builders.
diff --git a/src/Confluent.SchemaRegistry.Encryption.Aws/AwsKmsClient.cs b/src/Confluent.SchemaRegistry.Encryption.Aws/AwsKmsClient.cs
@@ -12,7 +12,7 @@ public class AwsKmsClient : IKmsClient
     {
         private AmazonKeyManagementServiceClient kmsClient;
         private string keyId;
-        
+
         public string KekId { get; }
 
         public AwsKmsClient(string kekId, AWSCredentials credentials)
@@ -36,7 +36,7 @@ public AwsKmsClient(string kekId, AWSCredentials credentials)
 
         public bool DoesSupport(string uri)
         {
-            return uri.StartsWith(AwsKmsDriver.Prefix); 
+            return KekId == uri;
         }
 
         public async Task<byte[]> Encrypt(byte[] plaintext)

diff --git a/src/Confluent.SchemaRegistry.Encryption.Aws/AwsKmsDriver.cs b/src/Confluent.SchemaRegistry.Encryption.Aws/AwsKmsDriver.cs
@@ -1,5 +1,7 @@
-using System.Collections.Generic;
+using System;
+using System.Collections.Generic;
 using Amazon.Runtime;
+using Amazon.Runtime.CredentialManagement;
 
 namespace Confluent.SchemaRegistry.Encryption.Aws
 {
@@ -13,20 +15,73 @@ public static void Register()
         public static readonly string Prefix = "aws-kms://";
         public static readonly string AccessKeyId = "access.key.id";
         public static readonly string SecretAccessKey = "secret.access.key";
-
+        public static readonly string Profile = "profile";
+        public static readonly string RoleArn = "role.arn";
+        public static readonly string RoleSessionName = "role.session.name";
+        public static readonly string RoleExternalId = "role.external.id";
+
         public string GetKeyUrlPrefix()
         {
             return Prefix;
         }
 
         public IKmsClient NewKmsClient(IDictionary<string, string> config, string keyUrl)
         {
+            config.TryGetValue(RoleArn, out string roleArn);
+            if (roleArn == null)
+            {
+                roleArn = Environment.GetEnvironmentVariable("AWS_ROLE_ARN");
+            }
+            config.TryGetValue(RoleSessionName, out string roleSessionName);
+            if (roleSessionName == null)
+            {
+                roleSessionName = Environment.GetEnvironmentVariable("AWS_ROLE_SESSION_NAME");
+            }
+            config.TryGetValue(RoleExternalId, out string roleExternalId);
+            if (roleExternalId == null)
+            {
+                roleExternalId = Environment.GetEnvironmentVariable("AWS_ROLE_EXTERNAL_ID");
+            }
             AWSCredentials credentials = null;
             if (config.TryGetValue(AccessKeyId, out string accessKeyId) 
                 && config.TryGetValue(SecretAccessKey, out string secretAccessKey))
             {
                 credentials = new BasicAWSCredentials(accessKeyId, secretAccessKey);
             }
+            else if (config.TryGetValue(Profile, out string profile))
+            {
+                var credentialProfileStoreChain = new CredentialProfileStoreChain();
+                if (credentialProfileStoreChain.TryGetAWSCredentials(
+                        profile, out AWSCredentials creds))
+                    credentials = creds;
+            }
+            if (credentials == null)
+            {
+                credentials = FallbackCredentialsFactory.GetCredentials();
+            }
+            if (roleArn != null)
+            {
+                if (string.IsNullOrEmpty(roleExternalId))
+                {
+                    credentials = new AssumeRoleAWSCredentials(
+                        credentials,
+                        roleArn,
+                        roleSessionName ?? "confluent-encrypt");
+                }
+                else
+                {
+                    var options = new AssumeRoleAWSCredentialsOptions
+                    {
+                        ExternalId = roleExternalId
+                    };
+
+                    credentials = new AssumeRoleAWSCredentials(
+                        credentials,
+                        roleArn,
+                        roleSessionName ?? "confluent-encrypt",
+                        options);
+                }
+            }
             return new AwsKmsClient(keyUrl, credentials);
         }
     }

diff --git a/src/Confluent.SchemaRegistry.Encryption.Aws/Confluent.SchemaRegistry.Encryption.Aws.csproj b/src/Confluent.SchemaRegistry.Encryption.Aws/Confluent.SchemaRegistry.Encryption.Aws.csproj
@@ -29,7 +29,8 @@
     </ItemGroup>
 
     <ItemGroup>
-        <PackageReference Include="AWSSDK.KeyManagementService" Version="3.7.302.19" />
+        <PackageReference Include="AWSSDK.KeyManagementService" Version="3.7.400.61" />
+        <PackageReference Include="AWSSDK.SecurityToken" Version="3.7.401.10" />
     </ItemGroup>
 
     <ItemGroup>

diff --git a/src/Confluent.SchemaRegistry.Encryption.Azure/AzureKmsClient.cs b/src/Confluent.SchemaRegistry.Encryption.Azure/AzureKmsClient.cs
@@ -25,7 +25,7 @@ public AzureKmsClient(string kekId, TokenCredential tokenCredential)
 
         public bool DoesSupport(string uri)
         {
-            return uri.StartsWith(AzureKmsDriver.Prefix); 
+            return KekId == uri;
         }
 
         public async Task<byte[]> Encrypt(byte[] plaintext)

diff --git a/src/Confluent.SchemaRegistry.Encryption.Gcp/GcpKmsClient.cs b/src/Confluent.SchemaRegistry.Encryption.Gcp/GcpKmsClient.cs
@@ -36,7 +36,7 @@ public GcpKmsClient(string kekId, GoogleCredential credential)
 
         public bool DoesSupport(string uri)
         {
-            return uri.StartsWith(GcpKmsDriver.Prefix);
+            return KekId == uri;
         }
 
         public async Task<byte[]> Encrypt(byte[] plaintext)

diff --git a/src/Confluent.SchemaRegistry.Encryption.HcVault/HcVaultKmsClient.cs b/src/Confluent.SchemaRegistry.Encryption.HcVault/HcVaultKmsClient.cs
@@ -21,11 +21,6 @@ public class HcVaultKmsClient : IKmsClient
 
         public HcVaultKmsClient(string kekId, string ns, string tokenId)
         {
-            if (tokenId == null)
-            {
-                tokenId = Environment.GetEnvironmentVariable("VAULT_TOKEN");
-                ns = Environment.GetEnvironmentVariable("VAULT_NAMESPACE");
-            }
             KekId = kekId;
             Namespace = ns;
             TokenId = tokenId;
@@ -53,7 +48,7 @@ public HcVaultKmsClient(string kekId, string ns, string tokenId)
 
         public bool DoesSupport(string uri)
         {
-            return uri.StartsWith(HcVaultKmsDriver.Prefix);
+            return KekId == uri;
         }
 
         public async Task<byte[]> Encrypt(byte[] plaintext)

diff --git a/src/Confluent.SchemaRegistry.Encryption.HcVault/HcVaultKmsDriver.cs b/src/Confluent.SchemaRegistry.Encryption.HcVault/HcVaultKmsDriver.cs
@@ -23,6 +23,11 @@ public IKmsClient NewKmsClient(IDictionary<string, string> config, string keyUrl
         {
             config.TryGetValue(TokenId, out string tokenId);
             config.TryGetValue(Namespace, out string ns);
+            if (tokenId == null)
+            {
+                tokenId = Environment.GetEnvironmentVariable("VAULT_TOKEN");
+                ns = Environment.GetEnvironmentVariable("VAULT_NAMESPACE");
+            }
             return new HcVaultKmsClient(keyUrl, ns, tokenId);
         }
     }

diff --git a/src/Confluent.SchemaRegistry.Encryption/LocalKmsClient.cs b/src/Confluent.SchemaRegistry.Encryption/LocalKmsClient.cs
@@ -16,14 +16,6 @@ public class LocalKmsClient : IKmsClient
 
         public LocalKmsClient(string secret)
         {
-            if (secret == null)
-            {
-                secret = Environment.GetEnvironmentVariable("LOCAL_SECRET");
-            }
-            if (secret == null)
-            {
-                throw new ArgumentNullException("Cannot load secret");
-            }
             Secret = secret;
             cryptor = new Cryptor(DekFormat.AES128_GCM);
             byte[] rawKey = Hkdf.DeriveKey(

diff --git a/src/Confluent.SchemaRegistry.Encryption/LocalKmsDriver.cs b/src/Confluent.SchemaRegistry.Encryption/LocalKmsDriver.cs
@@ -22,6 +22,14 @@ public string GetKeyUrlPrefix()
         public IKmsClient NewKmsClient(IDictionary<string, string> config, string keyUrl)
         {
             config.TryGetValue(Secret, out string secret);
+            if (secret == null)
+            {
+                secret = Environment.GetEnvironmentVariable("LOCAL_SECRET");
+            }
+            if (secret == null)
+            {
+                throw new ArgumentNullException("Cannot load secret");
+            }
             return new LocalKmsClient(secret);
         }
     }

diff --git a/src/Confluent.SchemaRegistry.Serdes.Avro/GenericSerializerImpl.cs b/src/Confluent.SchemaRegistry.Serdes.Avro/GenericSerializerImpl.cs
@@ -31,9 +31,10 @@ namespace Confluent.SchemaRegistry.Serdes
 {
     internal class GenericSerializerImpl : AsyncSerializer<GenericRecord, Avro.Schema>
     {
-        private Dictionary<Avro.Schema, string> knownSchemas = new Dictionary<global::Avro.Schema, string>();
-        private HashSet<KeyValuePair<string, string>> registeredSchemas = new HashSet<KeyValuePair<string, string>>();
-        private Dictionary<string, int> schemaIds = new Dictionary<string, int>();
+        private Dictionary<Avro.Schema, string> knownSchemas =
+            new Dictionary<global::Avro.Schema, string>();
+        private Dictionary<KeyValuePair<string, string>, int> registeredSchemas =
+            new Dictionary<KeyValuePair<string, string>, int>();
 
         public GenericSerializerImpl(
             ISchemaRegistryClient schemaRegistryClient,
@@ -99,12 +100,10 @@ public async Task<byte[]> Serialize(string topic, Headers headers, GenericRecord
                     // something more sophisticated than the below + not allow 
                     // the misuse to keep happening without warning.
                     if (knownSchemas.Count > schemaRegistryClient.MaxCachedSchemas ||
-                        registeredSchemas.Count > schemaRegistryClient.MaxCachedSchemas ||
-                        schemaIds.Count > schemaRegistryClient.MaxCachedSchemas)
+                        registeredSchemas.Count > schemaRegistryClient.MaxCachedSchemas)
                     {
                         knownSchemas.Clear();
                         registeredSchemas.Clear();
-                        schemaIds.Clear();
                     }
 
                     // Determine a schema string corresponding to the schema object.
@@ -139,41 +138,18 @@ public async Task<byte[]> Serialize(string topic, Headers headers, GenericRecord
                     {
                         schemaId = latestSchema.Id;
                     }
-                    else if (!registeredSchemas.Contains(subjectSchemaPair))
+                    else if (!registeredSchemas.TryGetValue(subjectSchemaPair, out schemaId))
                     {
-                        int newSchemaId;
-
                         // first usage: register/get schema to check compatibility
-                        if (autoRegisterSchema)
-                        {
-                            newSchemaId = await schemaRegistryClient
+                        schemaId = autoRegisterSchema
+                            ? await schemaRegistryClient
                                 .RegisterSchemaAsync(subject, writerSchemaString, normalizeSchemas)
+                                .ConfigureAwait(continueOnCapturedContext: false)
+                            : await schemaRegistryClient
+                                .GetSchemaIdAsync(subject, writerSchemaString, normalizeSchemas)
                                 .ConfigureAwait(continueOnCapturedContext: false);
-                        }
-                        else
-                        {
-                            newSchemaId = await schemaRegistryClient.GetSchemaIdAsync(subject, writerSchemaString, normalizeSchemas)
-                                .ConfigureAwait(continueOnCapturedContext: false);
-                        }
-
-                        if (!schemaIds.ContainsKey(writerSchemaString))
-                        {
-                            schemaIds.Add(writerSchemaString, newSchemaId);
-                        }
-                        else if (schemaIds[writerSchemaString] != newSchemaId)
-                        {
-                            schemaIds.Clear();
-                            registeredSchemas.Clear();
-                            throw new KafkaException(new Error(isKey ? ErrorCode.Local_KeySerialization : ErrorCode.Local_ValueSerialization, $"Duplicate schema registration encountered: Schema ids {schemaIds[writerSchemaString]} and {newSchemaId} are associated with the same schema."));
-                        }
-
-                        registeredSchemas.Add(subjectSchemaPair);
 
-                        schemaId = schemaIds[writerSchemaString];
-                    }
-                    else
-                    {
-                        schemaId = schemaIds[writerSchemaString];
+                        registeredSchemas.Add(subjectSchemaPair, schemaId);
                     }
                 }
                 finally