Merge pull request #8 from ssijbabu/workloadidentity

Added azure workload identity

Author: qboileau

README.md

@@ -75,6 +75,24 @@ The rest of the parameters are read from the environment variables.
 - `AZURE_CLIENT_ID` / `AZURE_USERNAME` / `AZURE_PASSWORD` / `AZURE_TENANT_ID` : for username password authentication
 
 
+```properties
+sasl.login.callback.handler.class=io.conduktor.kafka.security.oauthbearer.azure.AzureManagedIdentityCallbackHandler
+sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required scope="https://<resource>/.default";
+```
+
+### Workload Identity Authentication
+
+Use Azure workload identity environment variables to configure token auth bearer retriever.
+More details on Azure identity [WorkloadIdentityCredential documentation](https://learn.microsoft.com/en-us/java/api/com.azure.identity.workloadidentitycredential?view=azure-java-stable)
+
+Use `io.conduktor.kafka.security.oauthbearer.azure.AzureManagedIdentityCallbackHandler` as the callback handler class and provide
+the following required parameters in the `sasl.jaas.config` property :
+- `scope` : The [scope](https://learn.microsoft.com/en-us/entra/identity-platform/scopes-oidc#the-default-scope) of the token
+
+The rest of the parameters are read from the environment variables.
+- `AZURE_CLIENT_ID` / `AZURE_TENANT_ID` / `AZURE_FEDERATED_TOKEN_FILE` : for workload identity authentication
+
+
 ```properties
 sasl.login.callback.handler.class=io.conduktor.kafka.security.oauthbearer.azure.AzureManagedIdentityCallbackHandler
 sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required scope="https://<resource>/.default";

src/main/java/io/conduktor/kafka/security/oauthbearer/azure/AzureIdentityAccessTokenRetriever.java

@@ -5,6 +5,7 @@
 import com.azure.identity.ClientCertificateCredential;
 import com.azure.identity.ClientCertificateCredentialBuilder;
 import com.azure.identity.EnvironmentCredentialBuilder;
+import com.azure.identity.WorkloadIdentityCredentialBuilder;
 import org.apache.kafka.common.config.ConfigException;
 import org.apache.kafka.common.errors.AuthenticationException;
 import org.apache.kafka.common.security.oauthbearer.internals.secured.AccessTokenRetriever;
@@ -57,9 +58,10 @@ public String retrieve() {
         try {
             // See https://learn.microsoft.com/en-us/entra/identity-platform/v2-oauth2-client-creds-grant-flow#second-case-access-token-request-with-a-certificate
             // See https://learn.microsoft.com/en-us/java/api/overview/azure/identity-readme?view=azure-java-stable#credential-classes
-            var chainedTokenCredentialBuilder = new ChainedTokenCredentialBuilder();
-            clientCertificateCredentials.ifPresent(chainedTokenCredentialBuilder::addFirst);
+            var chainedTokenCredentialBuilder = new ChainedTokenCredentialBuilder();            
             chainedTokenCredentialBuilder.addLast(new EnvironmentCredentialBuilder().build());
+            chainedTokenCredentialBuilder.addLast(new WorkloadIdentityCredentialBuilder().build());            
+            clientCertificateCredentials.ifPresent(chainedTokenCredentialBuilder::addLast);
 
             var clientCredentials = chainedTokenCredentialBuilder.build();
             return clientCredentials.getTokenSync(new TokenRequestContext().setScopes(scopes)).getToken();