This is an automated email from the ASF dual-hosted git repository. asf-gitbox-commits pushed a commit to branch atlas-2.6 in repository https://gitbox.apache.org/repos/asf/atlas.git
commit e50d3a1c7cf8a5cfa5e6ec018885467940376f98 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 +++++++++++++ .../apache/atlas/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 d02f9e3898..77c0797aa3 100755 --- a/distro/src/conf/atlas-application.properties +++ b/distro/src/conf/atlas-application.properties @@ -114,6 +114,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 3c5ded1453..9ecde2e41c 100644 --- a/notification/src/main/java/org/apache/atlas/kafka/KafkaNotification.java +++ b/notification/src/main/java/org/apache/atlas/kafka/KafkaNotification.java @@ -438,7 +438,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 0000000000..68cfea65f8 --- /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.NO_SASL_EXTENSIONS); + } 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 0000000000..b9eb43b3f1 --- /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/webapp/src/main/java/org/apache/atlas/web/rest/NotificationREST.java b/webapp/src/main/java/org/apache/atlas/web/rest/NotificationREST.java index d0c8a736fa..62150c21a7 100644 --- a/webapp/src/main/java/org/apache/atlas/web/rest/NotificationREST.java +++ b/webapp/src/main/java/org/apache/atlas/web/rest/NotificationREST.java @@ -106,6 +106,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()); } }
