kumaab commented on code in PR #1297: URL: https://github.com/apache/ranger/pull/1297#discussion_r4299151659
########## agents-audit/kafka-client-auth/src/main/java/org/apache/ranger/kafka/auth/RangerKafkaOAuthBearerLoginCallbackHandler.java: ########## @@ -0,0 +1,229 @@ +/* + * 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 + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * 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.ranger.kafka.auth; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.kafka.common.config.ConfigException; +import org.apache.kafka.common.config.SaslConfigs; +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.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.Arrays; +import java.util.Base64; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.stream.Collectors; + +/** + * SASL/OAUTHBEARER login callback handler that presents a JWT read from a local file, such as a + * Kubernetes projected service-account token, to the Kafka broker. + * <p> + * Kafka's bundled {@code OAuthBearerLoginCallbackHandler} also accepts a {@code file://} value for + * {@code sasl.oauthbearer.token.endpoint.url}, but it reads the file once and caches the token, so a + * client fails to re-authenticate after the platform rotates the token in place. This handler never + * caches: every login and every refresh re-reads the file and re-derives {@code sub}, {@code exp} + * and {@code iat}, so Kafka's refresh scheduling always follows the current token. Signature and + * audience validation are the broker's responsibility. + */ +public class RangerKafkaOAuthBearerLoginCallbackHandler implements AuthenticateCallbackHandler, AutoCloseable { + private static final Logger LOG = LoggerFactory.getLogger(RangerKafkaOAuthBearerLoginCallbackHandler.class); + + /* same system property Kafka's bundled handler consults; when set, the token URL must be listed */ + public static final String ALLOWED_URLS_SYSTEM_PROPERTY = "org.apache.kafka.sasl.oauthbearer.allowed.urls"; + public static final int TOKEN_READ_MAX_ATTEMPTS = 5; + public static final long TOKEN_READ_RETRY_SLEEP_MS = 50L; + + private static final String CLAIM_SUBJECT = "sub"; + private static final String CLAIM_EXPIRATION = "exp"; + private static final String CLAIM_ISSUED_AT = "iat"; + private static final ObjectMapper MAPPER = new ObjectMapper(); + + private volatile Path tokenFile; + + @Override + public void configure(Map<String, ?> configs, String saslMechanism, List<AppConfigurationEntry> jaasConfigEntries) { + Object urlValue = configs.get(SaslConfigs.SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL); + String url = urlValue != null ? urlValue.toString().trim() : ""; + + if (url.isEmpty()) { + throw new ConfigException(String.format("%s must be set to a file:// URL for %s", SaslConfigs.SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL, getClass().getSimpleName())); + } + + throwIfUrlNotAllowed(url); + + tokenFile = resolveTokenFile(url); + + LOG.info("{}: reading SASL/OAUTHBEARER token from {}", getClass().getSimpleName(), tokenFile); + } + + @Override + public void handle(Callback[] callbacks) throws IOException, UnsupportedCallbackException { + Path file = tokenFile; + + if (file == null) { + throw new IllegalStateException(String.format("%s is not configured; call configure() first", getClass().getSimpleName())); + } + + for (Callback callback : callbacks) { + if (callback instanceof OAuthBearerTokenCallback) { + RangerOAuthBearerToken token = readToken(file); + + ((OAuthBearerTokenCallback) callback).token(token); + + LOG.info("{}: token loaded from {} for principal={}, expiresAtMs={}", getClass().getSimpleName(), file, token.principalName(), token.lifetimeMs()); Review Comment: done. thanks. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
