This is an automated email from the ASF dual-hosted git repository.
nixonrodrigues pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/atlas.git
The following commit(s) were added to refs/heads/master by this push:
new 18fbc57ee ATLAS-5409: Atlas Kafka clients authenticate with
OAUTHBEARER using K… (#762)
18fbc57ee is described below
commit 18fbc57ee2f4fc45ab791a60c55f0ea5e10accc2
Author: Nixon Rodrigues <[email protected]>
AuthorDate: Wed Sep 23 18:00:07 2026 +0530
ATLAS-5409: Atlas Kafka clients authenticate with OAUTHBEARER using K…
(#762)
* ATLAS-5409: Atlas Kafka clients authenticate with OAUTHBEARER using
Kubernetes projected SA tokens
* ATLAS-5409 Rename of Handler Class to AtlasKafkaOAuthHandler
---
distro/src/conf/atlas-application.properties | 7 +
.../org/apache/atlas/kafka/KafkaNotification.java | 8 +-
.../security/kafka/AtlasKafkaOAuthHandler.java | 264 +++++++++++++++++++++
.../security/kafka/AtlasKafkaOAuthHandlerTest.java | 168 +++++++++++++
.../rest/web/rest/NotificationREST.java | 2 +
5 files changed, 448 insertions(+), 1 deletion(-)
diff --git a/distro/src/conf/atlas-application.properties
b/distro/src/conf/atlas-application.properties
index b9f3a6d0a..f4363702f 100755
--- a/distro/src/conf/atlas-application.properties
+++ b/distro/src/conf/atlas-application.properties
@@ -111,6 +111,13 @@ atlas.notification.hook.retry.interval=1000
#atlas.notification.kafka.service.principal=kafka/[email protected]
#atlas.notification.kafka.keytab.location=/etc/security/keytabs/kafka.service.keytab
+# Enable for Kafka clusters using Kubernetes projected service-account tokens
(OAUTHBEARER)
+#atlas.kafka.security.protocol=SASL_SSL
+#atlas.kafka.sasl.mechanism=OAUTHBEARER
+#atlas.kafka.sasl.oauthbearer.token.endpoint.url=file:///var/run/secrets/kafka/kafka.<audience-id>
+#atlas.kafka.sasl.login.callback.handler.class=org.apache.atlas.security.kafka.AtlasKafkaOAuthHandler
+# Ensure the Atlas process user is in the group that owns the projected token
file (e.g. kafka-sa-token).
+
## Server port configuration
#atlas.server.http.port=21000
#atlas.server.https.port=21443
diff --git
a/notification/src/main/java/org/apache/atlas/kafka/KafkaNotification.java
b/notification/src/main/java/org/apache/atlas/kafka/KafkaNotification.java
index e75ecfafd..8e6e5826d 100644
--- a/notification/src/main/java/org/apache/atlas/kafka/KafkaNotification.java
+++ b/notification/src/main/java/org/apache/atlas/kafka/KafkaNotification.java
@@ -449,7 +449,13 @@ public class KafkaNotification extends
AbstractNotification implements Service {
ret = (KafkaProducer)
producersByCriteria.get(producerCriteria);
if (ret == null) {
- ret = new KafkaProducer(properties);
+ try {
+ ret = new KafkaProducer(properties);
+ } catch (Exception e) {
+ LOG.error("Failed to create KafkaProducer for
criteria={}", producerCriteria, e);
+
+ throw e;
+ }
producersByCriteria.put(producerCriteria, ret);
}
diff --git
a/notification/src/main/java/org/apache/atlas/security/kafka/AtlasKafkaOAuthHandler.java
b/notification/src/main/java/org/apache/atlas/security/kafka/AtlasKafkaOAuthHandler.java
new file mode 100644
index 000000000..6809e97b5
--- /dev/null
+++
b/notification/src/main/java/org/apache/atlas/security/kafka/AtlasKafkaOAuthHandler.java
@@ -0,0 +1,264 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.
+ */
+package org.apache.atlas.security.kafka;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.kafka.common.security.auth.AuthenticateCallbackHandler;
+import org.apache.kafka.common.security.auth.SaslExtensions;
+import org.apache.kafka.common.security.auth.SaslExtensionsCallback;
+import org.apache.kafka.common.security.oauthbearer.OAuthBearerToken;
+import org.apache.kafka.common.security.oauthbearer.OAuthBearerTokenCallback;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.security.auth.callback.Callback;
+import javax.security.auth.callback.UnsupportedCallbackException;
+import javax.security.auth.login.AppConfigurationEntry;
+
+import java.io.IOException;
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.Base64;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Kafka SASL/OAUTHBEARER callback handler for Kubernetes projected
service-account
+ * tokens. Reads a JWT from a {@code file://} URL configured via
+ * {@link #SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL} and supplies it to the
+ * Kafka client without OAuth 2.0 scope validation. The broker validates the
JWT.
+ * <p>
+ * Kubernetes rotates projected SA tokens in place (same path, new contents).
Kafka's
+ * {@code ExpiringCredentialRefreshingLogin} re-invokes this handler on each
login /
+ * refresh; {@link #handle(Callback[])} always re-reads the token file and
builds a
+ * new {@link OAuthBearerToken} from the current JWT (including updated {@code
exp}).
+ * Token bytes are never cached between callbacks.
+ */
+public class AtlasKafkaOAuthHandler implements AuthenticateCallbackHandler {
+ private static final Logger LOG =
LoggerFactory.getLogger(AtlasKafkaOAuthHandler.class);
+
+ /** Kafka client config key for the OAuth bearer token endpoint / file
URL. */
+ static final String SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL =
"sasl.oauthbearer.token.endpoint.url";
+
+ /** Retries while kubelet atomically replaces the projected token file. */
+ private static final int TOKEN_READ_MAX_ATTEMPTS = 5;
+ private static final long TOKEN_READ_RETRY_SLEEP_MS = 50L;
+
+ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+ private Path tokenFilePath;
+
+ @Override
+ public void configure(Map<String, ?> configs, String saslMechanism,
List<AppConfigurationEntry> jaasConfigEntries) {
+ Object tokenEndpointUrl =
configs.get(SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL);
+
+ if (tokenEndpointUrl == null) {
+ throw new IllegalArgumentException("Missing required Kafka config:
" + SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL);
+ }
+
+ tokenFilePath = resolveTokenFilePath(tokenEndpointUrl.toString());
+ }
+
+ @Override
+ public void handle(Callback[] callbacks) throws IOException,
UnsupportedCallbackException {
+ if (tokenFilePath == null) {
+ throw new IllegalStateException("AtlasKafkaOAuthHandler is not
configured");
+ }
+
+ for (Callback callback : callbacks) {
+ if (callback instanceof OAuthBearerTokenCallback) {
+ handleTokenCallback((OAuthBearerTokenCallback) callback);
+ } else if (callback instanceof SaslExtensionsCallback) {
+ ((SaslExtensionsCallback)
callback).extensions(SaslExtensions.empty());
+ } else {
+ throw new UnsupportedCallbackException(callback);
+ }
+ }
+ }
+
+ @Override
+ public void close() {
+ tokenFilePath = null;
+ }
+
+ private void handleTokenCallback(OAuthBearerTokenCallback callback) throws
IOException {
+ String rawToken = readTokenFromProjectedVolume();
+
+ JwtClaims claims = parseJwtClaims(rawToken);
+ callback.token(new AtlasKafkaOAuthBearerToken(rawToken, claims));
+
+ LOG.info("AtlasKafkaOAuthHandler: token set OK for principal={},
lifetimeMs={}",
+ claims.principalName, claims.lifetimeMs);
+ }
+
+ /**
+ * Reads the current JWT from the projected volume. Called on every Kafka
OAuth login / refresh.
+ */
+ private String readTokenFromProjectedVolume() throws IOException {
+ IOException lastFailure = null;
+
+ for (int attempt = 1; attempt <= TOKEN_READ_MAX_ATTEMPTS; attempt++) {
+ try {
+ String rawToken = readTokenOnce();
+
+ parseJwtClaims(rawToken);
+
+ return rawToken;
+ } catch (IOException e) {
+ lastFailure = e;
+
+ if (attempt < TOKEN_READ_MAX_ATTEMPTS) {
+ LOG.debug("K8sOAuth: token read/parse failed (attempt
{}/{}), retrying: {}",
+ attempt, TOKEN_READ_MAX_ATTEMPTS, e.getMessage());
+
+ sleepBeforeTokenReadRetry();
+ }
+ }
+ }
+
+ throw lastFailure != null ? lastFailure : new IOException("Failed to
read projected SA token from " + tokenFilePath);
+ }
+
+ private String readTokenOnce() throws IOException {
+ byte[] tokenBytes = Files.readAllBytes(tokenFilePath);
+ String rawToken = new String(tokenBytes,
StandardCharsets.UTF_8).trim();
+
+ if (rawToken.isEmpty()) {
+ throw new IOException("Projected SA token file is empty: " +
tokenFilePath);
+ }
+
+ return rawToken;
+ }
+
+ private static void sleepBeforeTokenReadRetry() throws IOException {
+ try {
+ Thread.sleep(TOKEN_READ_RETRY_SLEEP_MS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IOException("Interrupted while waiting to re-read
projected SA token", e);
+ }
+ }
+
+ static Path resolveTokenFilePath(String tokenEndpointUrl) {
+ URI uri = URI.create(tokenEndpointUrl);
+
+ if (!"file".equalsIgnoreCase(uri.getScheme())) {
+ throw new IllegalArgumentException("AtlasKafkaOAuthHandler
supports only file:// token URLs, got: " + tokenEndpointUrl);
+ }
+
+ Path path = Paths.get(uri);
+
+ if (!Files.isRegularFile(path)) {
+ throw new IllegalArgumentException("Token file does not exist: " +
path);
+ }
+
+ return path;
+ }
+
+ static JwtClaims parseJwtClaims(String jwt) throws IOException {
+ String[] parts = jwt.split("\\.");
+
+ if (parts.length < 2) {
+ throw new IOException("Malformed JWT: expected at least header and
payload sections");
+ }
+
+ byte[] payloadBytes = Base64.getUrlDecoder().decode(parts[1]);
+ JsonNode claimsNode = OBJECT_MAPPER.readTree(payloadBytes);
+
+ JsonNode subNode = claimsNode.get("sub");
+ JsonNode expNode = claimsNode.get("exp");
+
+ if (subNode == null || subNode.asText().trim().isEmpty()) {
+ throw new IOException("JWT is missing required 'sub' claim");
+ }
+
+ if (expNode == null || !expNode.isNumber()) {
+ throw new IOException("JWT is missing required numeric 'exp'
claim");
+ }
+
+ Long startTimeMs = null;
+ JsonNode iatNode = claimsNode.get("iat");
+
+ if (iatNode != null && iatNode.isNumber()) {
+ startTimeMs = toEpochMillis(iatNode);
+ }
+
+ return new JwtClaims(subNode.asText(), toEpochMillis(expNode),
startTimeMs);
+ }
+
+ private static long toEpochMillis(JsonNode epochSecondsNode) {
+ return Math.round(epochSecondsNode.asDouble() * 1000);
+ }
+
+ static final class JwtClaims {
+ private final String principalName;
+ private final long lifetimeMs;
+ private final Long startTimeMs;
+
+ JwtClaims(String principalName, long lifetimeMs, Long startTimeMs) {
+ this.principalName = principalName;
+ this.lifetimeMs = lifetimeMs;
+ this.startTimeMs = startTimeMs;
+ }
+ }
+
+ private static final class AtlasKafkaOAuthBearerToken implements
OAuthBearerToken {
+ private final String value;
+ private final String principalName;
+ private final long lifetimeMs;
+ private final Long startTimeMs;
+
+ AtlasKafkaOAuthBearerToken(String rawJwt, JwtClaims claims) {
+ this.value = rawJwt;
+ this.principalName = claims.principalName;
+ this.lifetimeMs = claims.lifetimeMs;
+ this.startTimeMs = claims.startTimeMs;
+ }
+
+ @Override
+ public String value() {
+ return value;
+ }
+
+ @Override
+ public Set<String> scope() {
+ return Collections.emptySet();
+ }
+
+ @Override
+ public long lifetimeMs() {
+ return lifetimeMs;
+ }
+
+ @Override
+ public String principalName() {
+ return principalName;
+ }
+
+ @Override
+ public Long startTimeMs() {
+ return startTimeMs;
+ }
+ }
+}
diff --git
a/notification/src/test/java/org/apache/atlas/security/kafka/AtlasKafkaOAuthHandlerTest.java
b/notification/src/test/java/org/apache/atlas/security/kafka/AtlasKafkaOAuthHandlerTest.java
new file mode 100644
index 000000000..b9eb43b3f
--- /dev/null
+++
b/notification/src/test/java/org/apache/atlas/security/kafka/AtlasKafkaOAuthHandlerTest.java
@@ -0,0 +1,168 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.
+ */
+package org.apache.atlas.security.kafka;
+
+import org.apache.kafka.common.security.oauthbearer.OAuthBearerToken;
+import org.apache.kafka.common.security.oauthbearer.OAuthBearerTokenCallback;
+import org.testng.annotations.Test;
+
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Base64;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotEquals;
+import static org.testng.Assert.assertTrue;
+
+public class AtlasKafkaOAuthHandlerTest {
+ @Test
+ public void testHandleTokenFromFileWithoutScopeClaim() throws Exception {
+ Path tokenFile = Files.createTempFile("k8s-sa-token", ".jwt");
+ String jwt = buildJwt("system:serviceaccount:ns:atlas",
1_700_000_000L, 1_700_003_600L);
+
+ Files.write(tokenFile, jwt.getBytes(StandardCharsets.UTF_8));
+
+ try {
+ AtlasKafkaOAuthHandler handler = new AtlasKafkaOAuthHandler();
+ Map<String, Object> configs = new HashMap<>();
+
+
configs.put(AtlasKafkaOAuthHandler.SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL,
"file://" + tokenFile);
+
+ handler.configure(configs, "OAUTHBEARER", Collections.emptyList());
+
+ OAuthBearerTokenCallback callback = new OAuthBearerTokenCallback();
+
+ handler.handle(new javax.security.auth.callback.Callback[]
{callback});
+
+ OAuthBearerToken token = callback.token();
+
+ assertEquals(token.value(), jwt);
+ assertEquals(token.principalName(),
"system:serviceaccount:ns:atlas");
+ assertEquals(token.lifetimeMs(), 1_700_003_600_000L);
+ assertEquals(token.startTimeMs(),
Long.valueOf(1_700_000_000_000L));
+ assertTrue(token.scope().isEmpty());
+ } finally {
+ Files.deleteIfExists(tokenFile);
+ }
+ }
+
+ @Test(expectedExceptions = IllegalArgumentException.class)
+ public void testConfigureRejectsNonFileUrl() {
+ AtlasKafkaOAuthHandler handler = new AtlasKafkaOAuthHandler();
+ Map<String, Object> configs = new HashMap<>();
+
+
configs.put(AtlasKafkaOAuthHandler.SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL,
"https://example.com/token";);
+
+ handler.configure(configs, "OAUTHBEARER", Collections.emptyList());
+ }
+
+ @Test
+ public void testHandleReloadsTokenAfterKubernetesRotation() throws
Exception {
+ Path tokenFile = Files.createTempFile("k8s-sa-token-rotate", ".jwt");
+ String jwtV1 = buildJwt("system:serviceaccount:ns:atlas",
1_700_000_000L, 1_700_003_600L);
+ String jwtV2 = buildJwt("system:serviceaccount:ns:atlas",
1_700_003_601L, 1_700_007_201L);
+
+ Files.write(tokenFile, jwtV1.getBytes(StandardCharsets.UTF_8));
+
+ try {
+ AtlasKafkaOAuthHandler handler = new AtlasKafkaOAuthHandler();
+ Map<String, Object> configs = new HashMap<>();
+
+
configs.put(AtlasKafkaOAuthHandler.SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL,
"file://" + tokenFile);
+
+ handler.configure(configs, "OAUTHBEARER", Collections.emptyList());
+
+ OAuthBearerTokenCallback firstCallback = new
OAuthBearerTokenCallback();
+ handler.handle(new javax.security.auth.callback.Callback[]
{firstCallback});
+ OAuthBearerToken firstToken = firstCallback.token();
+
+ assertEquals(firstToken.value(), jwtV1);
+ assertEquals(firstToken.lifetimeMs(), 1_700_003_600_000L);
+
+ Files.write(tokenFile, jwtV2.getBytes(StandardCharsets.UTF_8));
+
+ OAuthBearerTokenCallback refreshCallback = new
OAuthBearerTokenCallback();
+ handler.handle(new javax.security.auth.callback.Callback[]
{refreshCallback});
+ OAuthBearerToken refreshedToken = refreshCallback.token();
+
+ assertNotEquals(refreshedToken.value(), jwtV1);
+ assertEquals(refreshedToken.value(), jwtV2);
+ assertEquals(refreshedToken.lifetimeMs(), 1_700_007_201_000L);
+ assertEquals(refreshedToken.principalName(),
firstToken.principalName());
+ } finally {
+ Files.deleteIfExists(tokenFile);
+ }
+ }
+
+ @Test
+ public void testHandleTokenFromFileWithoutIatClaim() throws Exception {
+ Path tokenFile = Files.createTempFile("k8s-sa-token-no-iat", ".jwt");
+ String jwt = buildJwt("system:serviceaccount:ns:atlas", null,
1_700_003_600L);
+
+ Files.write(tokenFile, jwt.getBytes(StandardCharsets.UTF_8));
+
+ try {
+ AtlasKafkaOAuthHandler handler = new AtlasKafkaOAuthHandler();
+ Map<String, Object> configs = new HashMap<>();
+
+
configs.put(AtlasKafkaOAuthHandler.SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL,
"file://" + tokenFile);
+
+ handler.configure(configs, "OAUTHBEARER", Collections.emptyList());
+
+ OAuthBearerTokenCallback callback = new OAuthBearerTokenCallback();
+
+ handler.handle(new javax.security.auth.callback.Callback[]
{callback});
+
+ OAuthBearerToken token = callback.token();
+
+ assertEquals(token.principalName(),
"system:serviceaccount:ns:atlas");
+ assertEquals(token.lifetimeMs(), 1_700_003_600_000L);
+ assertEquals(token.startTimeMs(), null);
+ } finally {
+ Files.deleteIfExists(tokenFile);
+ }
+ }
+
+ private static String buildJwt(String subject, Long issuedAtSeconds, long
expirationSeconds) {
+ String header = base64Url("{\"alg\":\"RS256\",\"typ\":\"JWT\"}");
+ String payload = base64Url(buildPayload(subject, issuedAtSeconds,
expirationSeconds));
+
+ return header + "." + payload + ".signature";
+ }
+
+ private static String buildPayload(String subject, Long issuedAtSeconds,
long expirationSeconds) {
+ StringBuilder payload = new StringBuilder();
+
payload.append("{\"sub\":\"").append(subject).append("\",\"exp\":").append(expirationSeconds);
+
+ if (issuedAtSeconds != null) {
+ payload.append(",\"iat\":").append(issuedAtSeconds);
+ }
+
+
payload.append(",\"iss\":\"https://kubernetes.default.svc.cluster.local\"}";);
+
+ return payload.toString();
+ }
+
+ private static String base64Url(String value) {
+ return
Base64.getUrlEncoder().withoutPadding().encodeToString(value.getBytes(StandardCharsets.UTF_8));
+ }
+}
diff --git
a/rest-notification-webapp/src/main/java/org/apache/atlas/notification/rest/web/rest/NotificationREST.java
b/rest-notification-webapp/src/main/java/org/apache/atlas/notification/rest/web/rest/NotificationREST.java
index 14b9062ca..bbecba3a6 100644
---
a/rest-notification-webapp/src/main/java/org/apache/atlas/notification/rest/web/rest/NotificationREST.java
+++
b/rest-notification-webapp/src/main/java/org/apache/atlas/notification/rest/web/rest/NotificationREST.java
@@ -108,6 +108,8 @@ public class NotificationREST {
String concatenatedMessage =
StringUtils.join(failedMessages, "\n");
throw new
AtlasBaseException(AtlasErrorCode.NOTIFICATION_EXCEPTION, exception,
concatenatedMessage);
+ } catch (Exception exception) {
+ throw new
AtlasBaseException(AtlasErrorCode.NOTIFICATION_EXCEPTION, exception,
exception.getMessage());
}
}
