2 * ============LICENSE_START=======================================================
3 * Copyright (C) 2023 Nordix Foundation.
4 * ================================================================================
5 * Licensed under the Apache License, Version 2.0 (the "License");
6 * you may not use this file except in compliance with the License.
7 * You may obtain a copy of the License at
9 * http://www.apache.org/licenses/LICENSE-2.0
11 * Unless required by applicable law or agreed to in writing, software
12 * distributed under the License is distributed on an "AS IS" BASIS,
13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 * See the License for the specific language governing permissions and
15 * limitations under the License.
17 * ============LICENSE_END=========================================================
20 package org.oran.datafile.oauth2;
22 import java.io.IOException;
25 import javax.security.auth.callback.Callback;
26 import javax.security.auth.callback.UnsupportedCallbackException;
27 import javax.security.auth.login.AppConfigurationEntry;
29 import org.apache.kafka.common.security.auth.AuthenticateCallbackHandler;
30 import org.apache.kafka.common.security.auth.SaslExtensions;
31 import org.apache.kafka.common.security.auth.SaslExtensionsCallback;
32 import org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule;
33 import org.apache.kafka.common.security.oauthbearer.OAuthBearerTokenCallback;
34 import org.oran.datafile.exceptions.DatafileTaskException;
35 import org.slf4j.Logger;
36 import org.slf4j.LoggerFactory;
38 public class OAuthKafkaAuthenticateLoginCallbackHandler implements AuthenticateCallbackHandler {
39 private final Logger logger = LoggerFactory.getLogger(OAuthKafkaAuthenticateLoginCallbackHandler.class);
41 private boolean isConfigured = false;
44 public void configure(Map<String, ?> map, String saslMechanism, List<AppConfigurationEntry> jaasConfigEntries) {
46 if (!OAuthBearerLoginModule.OAUTHBEARER_MECHANISM.equals(saslMechanism))
47 throw new IllegalArgumentException(String.format("Unexpected SASL mechanism: %s", saslMechanism));
48 if (Objects.requireNonNull(jaasConfigEntries).size() != 1 || jaasConfigEntries.get(0) == null)
49 throw new IllegalArgumentException(String.format(
50 "Must supply exactly 1 non-null JAAS mechanism configuration (size was %d)", jaasConfigEntries.size()));
56 /*This method intentionally left empty.
57 Close functionality will be implemented later.*/
61 public void handle(Callback[] callbacks) throws IOException, UnsupportedCallbackException {
63 if (!this.isConfigured)
64 throw new IllegalStateException("Callback handler not configured");
65 for (Callback callback : callbacks) {
66 logger.debug("callback {}", callback);
67 if (callback instanceof OAuthBearerTokenCallback oauthBearerTokenCallback) {
68 handleCallback(oauthBearerTokenCallback);
69 } else if (callback instanceof SaslExtensionsCallback saslExtensionsCallback) {
70 handleCallback(saslExtensionsCallback);
72 logger.error("Unsupported callback: {}", callback);
73 throw new UnsupportedCallbackException(callback);
78 private void handleCallback(SaslExtensionsCallback callback) {
79 callback.extensions(SaslExtensions.empty());
82 private void handleCallback(OAuthBearerTokenCallback callback) {
84 if (callback.token() != null) {
85 throw new DatafileTaskException("Callback had a token already");
88 String accessToken = SecurityContext.getInstance().getBearerAuthToken();
89 OAuthBearerTokenJwt token = OAuthBearerTokenJwt.create(accessToken);
91 callback.token(token);
92 } catch (Exception e) {
93 logger.error("Could not handle login callback: {}", e.getMessage());
97 public boolean isConfigured() {