Implement review comments

Signed-off-by: JvD_Ericsson <jeff.van.dam@est.tech>

Author: JvD-Ericsson

build.gradle

@@ -303,7 +303,7 @@ project(':cruise-control') {
     implementation 'com.google.code.gson:gson:2.9.0'
     implementation "org.eclipse.jetty:jetty-server:${jettyVersion}"
     implementation 'io.dropwizard.metrics:metrics-jmx:4.2.9'
-    implementation 'com.nimbusds:nimbus-jose-jwt:9.45'
+    implementation 'com.nimbusds:nimbus-jose-jwt:10.0.2'
     implementation 'io.swagger.parser.v3:swagger-parser-v3:2.1.16'
     implementation 'io.github.classgraph:classgraph:4.8.141'
     implementation 'com.google.code.findbugs:jsr305:3.0.2'
@@ -339,6 +339,7 @@ project(':cruise-control') {
     testImplementation 'com.jayway.jsonpath:json-path:2.7.0'
     testImplementation 'org.powermock:powermock-module-junit4:2.0.9'
     testImplementation 'org.powermock:powermock-api-easymock:2.0.9'
+    testImplementation "org.testcontainers:kafka:$testcontainersVersion"
   }
 
   publishing {
@@ -483,7 +484,7 @@ project(':cruise-control-metrics-reporter') {
     testImplementation "org.apache.kafka:kafka-raft:$kafkaVersion"
     testImplementation "org.apache.kafka:kafka-storage:$kafkaVersion"
     testImplementation 'commons-io:commons-io:2.11.0'
-    testImplementation "org.testcontainers:kafka:1.21.3"
+    testImplementation "org.testcontainers:kafka:$testcontainersVersion"
     testOutput sourceSets.test.output
   }
 

cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/KafkaCruiseControlServletApp.java

@@ -22,6 +22,8 @@
 import org.eclipse.jetty.ee10.servlet.ServletHolder;
 import org.eclipse.jetty.util.ssl.SslContextFactory;
 import jakarta.servlet.ServletException;
+import java.nio.file.Files;
+import java.nio.file.Path;
 import java.util.List;
 
 public class KafkaCruiseControlServletApp extends KafkaCruiseControlApp {
@@ -120,12 +122,17 @@ private void maybeConfigureTlsProtocolsAndCiphers(SslContextFactory sslContextFa
     protected void setupWebUi(ServletContextHandler contextHandler) {
         // Placeholder for any static content
         String webuiDir = _config.getString(WebServerConfig.WEBSERVER_UI_DISKPATH_CONFIG);
-        String webuiPathPrefix = _config.getString(WebServerConfig.WEBSERVER_UI_URLPREFIX_CONFIG);
-        DefaultServlet defaultServlet = new DefaultServlet();
-        ServletHolder holderWebapp = new ServletHolder("default", defaultServlet);
-        // holderWebapp.setInitParameter("org.eclipse.jetty.servlet.Default.dirAllowed", "false");
-        holderWebapp.setInitParameter("baseResource", webuiDir);
-        contextHandler.addServlet(holderWebapp, webuiPathPrefix);
+        Path path = Path.of(webuiDir);
+        if (Files.isDirectory(path) && Files.isReadable(path)) {
+            String webUiPathPrefix = _config.getString(WebServerConfig.WEBSERVER_UI_URLPREFIX_CONFIG);
+            DefaultServlet defaultServlet = new DefaultServlet();
+            ServletHolder holderWebapp = new ServletHolder("default", defaultServlet);
+            // holderWebapp.setInitParameter("org.eclipse.jetty.servlet.Default.dirAllowed", "false");
+            contextHandler.setBaseResourceAsString(webuiDir);
+            contextHandler.addServlet(holderWebapp, webUiPathPrefix);
+        } else {
+            LOG.warn("WebUI directory not found or unreadable: {} UI disabled", webuiDir);
+        }
     }
 
     protected ServletContextHandler createContextHandler() {

cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/servlet/UserPermissionsManager.java

@@ -4,23 +4,16 @@
 
 package com.linkedin.kafka.cruisecontrol.servlet;
 
-import java.io.BufferedReader;
-import java.io.IOException;
-import java.io.InputStreamReader;
-import java.io.UncheckedIOException;
-import java.nio.charset.StandardCharsets;
 import java.util.Map;
 import java.util.Set;
 import java.util.HashMap;
-import java.util.HashSet;
 import java.util.stream.Collectors;
 import java.util.Collections;
 import com.linkedin.kafka.cruisecontrol.config.KafkaCruiseControlConfig;
+import org.eclipse.jetty.security.PropertyUserStore;
 import org.eclipse.jetty.security.RolePrincipal;
 import org.eclipse.jetty.security.UserStore;
-import org.eclipse.jetty.security.PropertyUserStore;
 import com.linkedin.kafka.cruisecontrol.config.constants.WebServerConfig;
-import org.eclipse.jetty.server.handler.ResourceHandler;
 import org.eclipse.jetty.util.resource.Resource;
 import org.eclipse.jetty.util.resource.ResourceFactory;
 import org.slf4j.Logger;
@@ -50,16 +43,14 @@ private Map<String, Set<String>> createRolesPerUsersMap() {
         boolean securityEnabled = _config.getBoolean(WebServerConfig.WEBSERVER_SECURITY_ENABLE_CONFIG);
         if (securityEnabled) {
             String privilegesFilePath = _config.getString(WebServerConfig.WEBSERVER_AUTH_CREDENTIALS_FILE_CONFIG);
-            Resource resource = ResourceFactory.of(new ResourceHandler()).newResource(privilegesFilePath);
-            UserStore userStore = createUserStoreFromResource(resource);
+            Resource resource = ResourceFactory.root().newResource(privilegesFilePath);
+            ExposedPropertyUserStore userStore = createUserStoreFromResource(resource);
             startUserStore(userStore);
 
-            Set<String> userNames = parseUsernames(resource);
+            Set<String> userNames = userStore.getUsersNames();
 
             for (String user : userNames) {
-                Set<RolePrincipal> roles = new HashSet<>(userStore.getRolePrincipals(user));
-
-                Set<String> roleNames = roles.stream()
+                Set<String> roleNames = userStore.getRolePrincipals(user).stream()
                         .map(RolePrincipal::getName)
                         .map(String::toUpperCase)
                         .collect(Collectors.toSet());
@@ -99,43 +90,25 @@ public Set<String> getRolesBy(String userName) {
 
     /** Creates UserStore from an external file
      *
-     * @param privilegedResource a filepath containing user privileges information
+     * @param privilegesResource a filepath containing user privileges information
      * @return a UserStore object
      */
-    private UserStore createUserStoreFromResource(Resource privilegedResource) {
-        PropertyUserStore userStore = new PropertyUserStore();
-        userStore.setConfig(privilegedResource);
+    private ExposedPropertyUserStore createUserStoreFromResource(Resource privilegesResource) {
+        ExposedPropertyUserStore userStore = new ExposedPropertyUserStore();
+        userStore.setConfig(privilegesResource);
         return userStore;
     }
 
-    /** Creates a set of usernames from a Resource
-     *
-     * @param resource a Resource containing user privileges information
-     * @return a Set of usernames parsed from the Resource
-     */
-    private static Set<String> parseUsernames(Resource resource) {
-        if (!resource.exists() || !resource.isReadable()) {
-            return Set.of();
-        }
-        Set<String> usernames = new HashSet<>();
-        try (BufferedReader reader = new BufferedReader(new InputStreamReader(resource.newInputStream(), StandardCharsets.UTF_8))) {
-            String line;
-            while ((line = reader.readLine()) != null) {
-                line = line.trim();
-                if (line.isEmpty() || line.startsWith("#")) {
-                    continue;
-                }
-                int colonIndex = line.indexOf(':');
-                if (colonIndex != -1) {
-                    String username = line.substring(0, colonIndex).trim();
-                    if (!username.isEmpty()) {
-                        usernames.add(username);
-                    }
-                }
-            }
-        } catch (IOException e) {
-            throw new UncheckedIOException("Failed to read usernames from " + resource, e);
+    private static class ExposedPropertyUserStore extends PropertyUserStore {
+
+        /**
+         * This method exposes the protected `_users` map inherited from
+         * UserStore, providing direct access to the stored users' names.
+         *
+         * @return the set of users' names
+         */
+        public Set<String> getUsersNames() {
+            return _users.keySet();
         }
-        return usernames;
     }
 }

cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/servlet/security/BasicSecurityProvider.java

@@ -10,7 +10,6 @@
 import org.eclipse.jetty.security.HashLoginService;
 import org.eclipse.jetty.security.LoginService;
 import org.eclipse.jetty.security.authentication.BasicAuthenticator;
-import org.eclipse.jetty.server.handler.ResourceHandler;
 import org.eclipse.jetty.util.resource.Resource;
 import org.eclipse.jetty.util.resource.ResourceFactory;
 
@@ -30,7 +29,7 @@ public void init(KafkaCruiseControlConfig config) {
 
   @Override
   public LoginService loginService() {
-    Resource resource = ResourceFactory.of(new ResourceHandler()).newResource(_userCredentialsFile);
+    Resource resource = ResourceFactory.root().newResource(_userCredentialsFile);
     return new HashLoginService("DefaultLoginService", resource);
   }
 

cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/servlet/security/DefaultRoleSecurityProvider.java

@@ -11,6 +11,7 @@
 import java.util.List;
 import java.util.Set;
 import org.eclipse.jetty.ee10.servlet.security.ConstraintMapping;
+import org.eclipse.jetty.security.Authenticator;
 import org.eclipse.jetty.security.Constraint;
 
 
@@ -65,8 +66,11 @@ public Set<String> roles() {
   }
 
   private ConstraintMapping mapping(CruiseControlEndPoint endpoint, String... roles) {
-    Constraint.Builder builder = new Constraint.Builder();
-    Constraint constraint = builder.roles(roles).name("BASIC").authorization(Constraint.Authorization.SPECIFIC_ROLE).build();
+    Constraint constraint = new Constraint.Builder()
+        .name(Authenticator.BASIC_AUTH)
+        .roles(roles)
+        .authorization(Constraint.Authorization.SPECIFIC_ROLE)
+        .build();
     ConstraintMapping mapping = new ConstraintMapping();
     mapping.setPathSpec(_webServerApiUrlPrefix.replace("*", endpoint.name().toLowerCase()));
     mapping.setConstraint(constraint);

cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/servlet/security/DummyLoginService.java

@@ -45,8 +45,8 @@ public void logout(UserIdentity user) {
         // dummy implementation
     }
 
-    @Override 
-    public UserIdentity login(String u, Object c, Request req, Function<Boolean, Session> getOrCreateSession) { 
+    @Override
+    public UserIdentity login(String u, Object c, Request req, Function<Boolean, Session> getOrCreateSession) {
         return null;
     }
 

cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/servlet/security/RoleProvider.java

@@ -11,13 +11,12 @@
 import org.eclipse.jetty.http.HttpStatus;
 import org.eclipse.jetty.http.HttpURI;
 import org.eclipse.jetty.security.AuthenticationState;
-import org.eclipse.jetty.security.ServerAuthException;
 import org.eclipse.jetty.security.UserIdentity;
 import org.eclipse.jetty.security.authentication.LoginAuthenticator;
+import org.eclipse.jetty.security.internal.DeferredAuthenticationState;
 import org.eclipse.jetty.server.Request;
 import org.eclipse.jetty.server.Response;
 import org.eclipse.jetty.util.Callback;
-import org.eclipse.jetty.util.URIUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import java.text.ParseException;
@@ -74,7 +73,10 @@ public class JwtAuthenticator extends LoginAuthenticator {
   public JwtAuthenticator(String authenticationProviderUrl, String cookieName) {
     _cookieName = cookieName;
     Function<String, Function<Request, String>> urlGen =
-        url -> req -> url.replace(REDIRECT_URL, getRequestURL(req) + getOriginalQueryString(req));
+        url -> req -> {
+          HttpURI httpUri = HttpURI.build(req.getHttpURI()).query(null);
+          return url.replace(REDIRECT_URL, httpUri.asString() + getOriginalQueryString(req));
+        };
     _authenticationProviderUrlGenerator = urlGen.apply(authenticationProviderUrl);
   }
 
@@ -84,13 +86,13 @@ public String getAuthenticationType() {
   }
 
   @Override
-  public AuthenticationState validateRequest(Request request, Response response, Callback callback) throws ServerAuthException {
+  public AuthenticationState validateRequest(Request request, Response response, Callback callback) {
     JWT_LOGGER.trace("Authentication request received for " + request.toString());
 
     String serializedJWT;
     // we'll skip the authentication for CORS preflight requests
     if (HttpMethod.OPTIONS.name().equalsIgnoreCase(request.getMethod())) {
-      return null;
+      return new DeferredAuthenticationState(this);
     }
     serializedJWT = getJwtFromBearerAuthorization(request);
     if (serializedJWT == null) {
@@ -151,21 +153,4 @@ private String getOriginalQueryString(Request req) {
     String originalQueryString = req.getHttpURI().getQuery();
     return (originalQueryString == null) ? "" : "?" + originalQueryString;
   }
-
-  /**
-   * Get the full request URL including scheme, host, port and path but excluding the query string.
-   * 
-   * @param req is the request to process
-   * @return the full request URL
-   */
-  public String getRequestURL(Request req) {
-    final StringBuilder url = new StringBuilder();
-    HttpURI uri = req.getHttpURI();
-    URIUtil.appendSchemeHostPort(url, uri.getScheme(), Request.getServerName(req), Request.getServerPort(req));
-    String path = uri.getPath();
-    if (path != null) {
-      url.append(path);
-    }
-    return url.toString();
-  }
 }