From 4d2a0e5e71decc0639064907bef055796a68e98f Mon Sep 17 00:00:00 2001
From: Diogo Costa <diogoabcosta89@gmail.com>
Date: Wed, 3 Apr 2024 11:43:25 +0100
Subject: [PATCH] feat: add oauth

feat: add oauth
---
 .../Configuration/SaslMechanism.cs            |  3 ++
 .../Configuration/SaslOauthbearerMethod.cs    | 12 ++++++
 .../Configuration/SecurityInformation.cs      | 42 ++++++++++++++++++-
 .../Extensions/ConfigurationExtensions.cs     | 16 ++++++-
 4 files changed, 71 insertions(+), 2 deletions(-)
 create mode 100644 src/KafkaFlow.Abstractions/Configuration/SaslOauthbearerMethod.cs

diff --git a/src/KafkaFlow.Abstractions/Configuration/SaslMechanism.cs b/src/KafkaFlow.Abstractions/Configuration/SaslMechanism.cs
index fdd22fd3f..08a599af9 100644
--- a/src/KafkaFlow.Abstractions/Configuration/SaslMechanism.cs
+++ b/src/KafkaFlow.Abstractions/Configuration/SaslMechanism.cs
@@ -14,4 +14,7 @@ public enum SaslMechanism
 
     /// <summary>SCRAM-SHA-512</summary>
     ScramSha512,
+
+    /// <summary>OAUTH-BEARER</summary>
+    OAuthBearer,
 }
diff --git a/src/KafkaFlow.Abstractions/Configuration/SaslOauthbearerMethod.cs b/src/KafkaFlow.Abstractions/Configuration/SaslOauthbearerMethod.cs
new file mode 100644
index 000000000..26d64c25e
--- /dev/null
+++ b/src/KafkaFlow.Abstractions/Configuration/SaslOauthbearerMethod.cs
@@ -0,0 +1,12 @@
+namespace KafkaFlow.Configuration
+{
+    /// <summary>SaslOauthbearerMethod enum values</summary>
+    public enum SaslOauthbearerMethod
+    {
+        /// <summary>Default</summary>
+        Default,
+
+        /// <summary>Oidc</summary>
+        Oidc,
+    }
+}
diff --git a/src/KafkaFlow.Abstractions/Configuration/SecurityInformation.cs b/src/KafkaFlow.Abstractions/Configuration/SecurityInformation.cs
index e18ef0898..f6b614289 100644
--- a/src/KafkaFlow.Abstractions/Configuration/SecurityInformation.cs
+++ b/src/KafkaFlow.Abstractions/Configuration/SecurityInformation.cs
@@ -209,4 +209,44 @@ public class SecurityInformation
     ///     importance: low
     /// </summary>
     public bool? EnableSaslOauthbearerUnsecureJwt { get; set; }
-}
+
+    /// <summary>
+    ///     Gets or sets the SaslOauthbearerMethod
+    ///
+    ///     default: false
+    ///     importance: low
+    /// </summary>
+    public SaslOauthbearerMethod? SaslOauthbearerMethod { get; set; }
+
+    /// <summary>
+    ///     Gets or sets the SaslOauthbearerClientId
+    ///
+    ///     default: false
+    ///     importance: low
+    /// </summary>
+    public string SaslOauthbearerClientId { get; set; }
+
+    /// <summary>
+    ///     Gets or sets the SaslOauthbearerClientSecret
+    ///
+    ///     default: false
+    ///     importance: low
+    /// </summary>
+    public string SaslOauthbearerClientSecret { get; set; }
+
+    /// <summary>
+    ///     Gets or sets the SaslOauthbearerTokenEndpointUrl
+    ///
+    ///     default: false
+    ///     importance: low
+    /// </summary>
+    public string SaslOauthbearerTokenEndpointUrl { get; set; }
+
+    /// <summary>
+    ///     Gets or sets the SaslOauthbearerScope
+    ///
+    ///     default: false
+    ///     importance: low
+    /// </summary>
+    public string SaslOauthbearerScope { get; set; }
+}
\ No newline at end of file
diff --git a/src/KafkaFlow/Extensions/ConfigurationExtensions.cs b/src/KafkaFlow/Extensions/ConfigurationExtensions.cs
index 205f2cbc5..3b90cb3f9 100644
--- a/src/KafkaFlow/Extensions/ConfigurationExtensions.cs
+++ b/src/KafkaFlow/Extensions/ConfigurationExtensions.cs
@@ -2,6 +2,7 @@
 using Confluent.Kafka;
 using KafkaFlow.Configuration;
 using SaslMechanism = KafkaFlow.Configuration.SaslMechanism;
+using SaslOauthbearerMethod = KafkaFlow.Configuration.SaslOauthbearerMethod;
 using SecurityProtocol = KafkaFlow.Configuration.SecurityProtocol;
 using SslEndpointIdentificationAlgorithm = KafkaFlow.Configuration.SslEndpointIdentificationAlgorithm;
 
@@ -15,6 +16,14 @@ internal static class ConfigurationExtensions
         SaslMechanism.Plain => Confluent.Kafka.SaslMechanism.Plain,
         SaslMechanism.ScramSha256 => Confluent.Kafka.SaslMechanism.ScramSha256,
         SaslMechanism.ScramSha512 => Confluent.Kafka.SaslMechanism.ScramSha512,
+        SaslMechanism.OAuthBearer => Confluent.Kafka.SaslMechanism.OAuthBearer,
+        _ => throw new ArgumentOutOfRangeException()
+    };
+
+    public static Confluent.Kafka.SaslOauthbearerMethod ToConfluent(this SaslOauthbearerMethod method) => method switch
+    {
+        SaslOauthbearerMethod.Default => Confluent.Kafka.SaslOauthbearerMethod.Default,
+        SaslOauthbearerMethod.Oidc => Confluent.Kafka.SaslOauthbearerMethod.Oidc,
         _ => throw new ArgumentOutOfRangeException()
     };
 
@@ -70,5 +79,10 @@ public static void ReadSecurityInformationFrom(this ClientConfig config, Cluster
         config.SaslMechanism = securityInformation.SaslMechanism?.ToConfluent();
         config.SaslUsername = securityInformation.SaslUsername;
         config.SaslPassword = securityInformation.SaslPassword;
+        config.SaslOauthbearerMethod = securityInformation.SaslOauthbearerMethod?.ToConfluent();
+        config.SaslOauthbearerClientId = securityInformation.SaslOauthbearerClientId;
+        config.SaslOauthbearerClientSecret = securityInformation.SaslOauthbearerClientSecret;
+        config.SaslOauthbearerTokenEndpointUrl = securityInformation.SaslOauthbearerTokenEndpointUrl;
+        config.SaslOauthbearerScope = securityInformation.SaslOauthbearerScope;
     }
-}
+}
\ No newline at end of file