Merge remote-tracking branch 'upstream/7.1.x'

ksqldb-rest-app/src/main/java/io/confluent/ksql/api/auth/BasicCallbackHandler.java

@@ -16,42 +16,33 @@
 package io.confluent.ksql.api.auth;
 
 import java.io.IOException;
-import java.util.Objects;
 import javax.security.auth.callback.Callback;
-import javax.security.auth.callback.CallbackHandler;
 import javax.security.auth.callback.NameCallback;
 import javax.security.auth.callback.PasswordCallback;
 import javax.security.auth.callback.TextOutputCallback;
 import javax.security.auth.callback.UnsupportedCallbackException;
+import org.eclipse.jetty.jaas.callback.DefaultCallbackHandler;
 import org.eclipse.jetty.jaas.callback.ObjectCallback;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-class BasicCallbackHandler implements CallbackHandler {
+public class BasicCallbackHandler extends DefaultCallbackHandler {
 
   private static final Logger log = LoggerFactory.getLogger(BasicCallbackHandler.class);
 
-  private final String username;
-  private final String password;
-
-  BasicCallbackHandler(final String username, final String password) {
-    this.username = Objects.requireNonNull(username, "username");
-    this.password = Objects.requireNonNull(password, "password");
-  }
-
   @Override
   public void handle(final Callback[] callbacks)
       throws IOException, UnsupportedCallbackException {
     for (final Callback callback : callbacks) {
       if (callback instanceof NameCallback) {
         final NameCallback nc = (NameCallback) callback;
-        nc.setName(username);
+        nc.setName(getUserName());
       } else if (callback instanceof ObjectCallback) {
         final ObjectCallback oc = (ObjectCallback)callback;
-        oc.setObject(password);
+        oc.setObject(getCredential());
       } else if (callback instanceof PasswordCallback) {
         final PasswordCallback pc = (PasswordCallback) callback;
-        pc.setPassword(password.toCharArray());
+        pc.setPassword(((String) getCredential()).toCharArray());
       } else if (callback instanceof TextOutputCallback) {
         final TextOutputCallback toc = (TextOutputCallback) callback;
         switch (toc.getMessageType()) {
@@ -73,22 +64,4 @@ public void handle(final Callback[] callbacks)
       }
     }
   }
-
-  @Override
-  public boolean equals(final Object o) {
-    if (this == o) {
-      return true;
-    }
-    if (o == null || getClass() != o.getClass()) {
-      return false;
-    }
-    final BasicCallbackHandler that = (BasicCallbackHandler) o;
-    return username.equals(that.username)
-        && password.equals(that.password);
-  }
-
-  @Override
-  public int hashCode() {
-    return Objects.hash(username, password);
-  }
 }

ksqldb-rest-app/src/main/java/io/confluent/ksql/api/auth/JaasAuthProvider.java

@@ -33,10 +33,9 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
-import javax.security.auth.callback.CallbackHandler;
-import javax.security.auth.login.LoginContext;
-import javax.security.auth.login.LoginException;
 import org.apache.commons.collections4.CollectionUtils;
+import org.eclipse.jetty.jaas.JAASLoginService;
+import org.eclipse.jetty.server.UserIdentity;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -54,7 +53,7 @@ public class JaasAuthProvider implements AuthProvider {
   private final String contextName;
 
   public JaasAuthProvider(final Server server, final KsqlRestConfig config) {
-    this(server, config, LoginContext::new);
+    this(server, config, JAASLoginService::new);
   }
 
   @VisibleForTesting
@@ -78,7 +77,7 @@ public JaasAuthProvider(final Server server, final KsqlRestConfig config) {
   @VisibleForTesting
   @FunctionalInterface
   interface LoginContextSupplier {
-    LoginContext get(String name, CallbackHandler callbackHandler) throws LoginException;
+    JAASLoginService get();
   }
 
   @Override
@@ -111,45 +110,47 @@ private void getUser(
       final List<String> allowedRoles,
       final Promise<User> promise
   ) {
-    final LoginContext lc;
+    final JAASLoginService login = loginContextSupplier.get();
+    login.setCallbackHandlerClass(BasicCallbackHandler.class.getName());
+    login.setLoginModuleName(contextName);
+
     try {
-      lc = loginContextSupplier.get(contextName, new BasicCallbackHandler(username, password));
-    } catch (LoginException | SecurityException e) {
-      log.error("Failed to create LoginContext. " + e.getMessage());
-      promise.fail("Failed to create LoginContext.");
-      return;
+      login.start();
+    } catch (final Exception e) {
+      log.error("Could not start login service.", e);
+      promise.fail("Could not start login service.");
     }
 
-    try {
-      lc.login();
-    } catch (LoginException le) {
-      log.error("Failed to log in. " + le.getMessage());
+    final UserIdentity user = login.login(username, password, null);
+
+    if (user == null) {
+      log.error("Failed to log in. ");
       promise.fail("Failed to log in: Invalid username/password.");
       return;
     }
 
     // We do the actual authorization here not in the User class
-    final boolean authorized = validateRoles(lc, allowedRoles);
+    final boolean authorized = validateRoles(user, allowedRoles);
 
     // if the subject from the login context is already a KsqlPrincipal, use the subject
     // (wrapped inside another DefaultKsqlPrincipal) rather than creating a new one
-    final Optional<KsqlPrincipal> ksqlPrincipal = lc.getSubject().getPrincipals().stream()
+    final Optional<KsqlPrincipal> ksqlPrincipal = user.getSubject().getPrincipals().stream()
         .filter(p -> p instanceof KsqlPrincipal)
         .map(p -> (KsqlPrincipal)p)
         .findFirst();
-    final JaasUser user = ksqlPrincipal.isPresent()
+    final JaasUser jaasUser = ksqlPrincipal.isPresent()
         ? new JaasUser(ksqlPrincipal.get(), authorized)
         : new JaasUser(username, password, authorized);
-    promise.complete(user);
+    promise.complete(jaasUser);
   }
 
-  private static boolean validateRoles(final LoginContext lc, final List<String> allowedRoles) {
+  private static boolean validateRoles(final UserIdentity ui, final List<String> allowedRoles) {
     if (allowedRoles.contains("*")) {
       // all users allowed
       return true;
     }
 
-    final Set<String> userRoles = lc.getSubject().getPrincipals().stream()
+    final Set<String> userRoles = ui.getSubject().getPrincipals().stream()
         .map(Principal::getName)
         .collect(Collectors.toSet());
     return !CollectionUtils.intersection(userRoles, allowedRoles).isEmpty();

ksqldb-rest-app/src/test/java/io/confluent/ksql/api/auth/BasicCallbackHandlerTest.java

@@ -46,7 +46,9 @@ public class BasicCallbackHandlerTest {
 
   @Before
   public void setUp() {
-    callbackHandler = new BasicCallbackHandler(USERNAME, PASSWORD);
+    callbackHandler = new BasicCallbackHandler();
+    callbackHandler.setUserName(USERNAME);
+    callbackHandler.setCredential(PASSWORD);
   }
 
   @Test

ksqldb-rest-app/src/test/java/io/confluent/ksql/api/auth/JaasAuthProviderTest.java

@@ -42,7 +42,8 @@
 import java.util.concurrent.CountDownLatch;
 import java.util.stream.Stream;
 import javax.security.auth.Subject;
-import javax.security.auth.login.LoginContext;
+import org.eclipse.jetty.jaas.JAASLoginService;
+import org.eclipse.jetty.server.UserIdentity;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -73,21 +74,27 @@ public class JaasAuthProviderTest {
   @Mock
   private LoginContextSupplier loginContextSupplier;
   @Mock
-  private LoginContext loginContext;
+  private JAASLoginService loginService;
+  @Mock
+  private UserIdentity userIdentity;
   @Mock
   private Subject subject;
 
   private JaasAuthProvider authProvider;
 
   @Before
   public void setUp() throws Exception {
+    final BasicCallbackHandler callbackHandler = new BasicCallbackHandler();
+    callbackHandler.setUserName(USERNAME);
+    callbackHandler.setCredential(PASSWORD);
+
     handleAsyncExecution();
     when(config.getString(KsqlRestConfig.AUTHENTICATION_REALM_CONFIG)).thenReturn(REALM);
     when(authInfo.getString("username")).thenReturn(USERNAME);
     when(authInfo.getString("password")).thenReturn(PASSWORD);
-    when(loginContextSupplier.get(REALM, new BasicCallbackHandler(USERNAME, PASSWORD)))
-        .thenReturn(loginContext);
-    when(loginContext.getSubject()).thenReturn(subject);
+    when(loginContextSupplier.get()).thenReturn(loginService);
+    when(loginService.login(eq(USERNAME), eq(PASSWORD), any())).thenReturn(userIdentity);
+    when(userIdentity.getSubject()).thenReturn(subject);
 
     authProvider = new JaasAuthProvider(server, config, loginContextSupplier);
   }

pom.xml

@@ -139,7 +139,7 @@
         <!-- Temporarily disabling this because it is causing failures in packaging but not CI builds. -->
         <!-- <compile.warnings-flag>-Werror</compile.warnings-flag> -->
         <!-- Only used to provide login module implementation for tests -->
-        <jetty.version>9.4.28.v20200408</jetty.version>
+        <jetty.version>9.4.44.v20210927</jetty.version>
         <git-commit-id-plugin.version>4.9.10</git-commit-id-plugin.version>
         <scala.version>2.13.2</scala.version>
         <apache.io.version>2.7</apache.io.version>