Kerberos auth to local rules support

cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/config/constants/WebServerConfig.java

@@ -314,6 +314,18 @@ public final class WebServerConfig {
       + "to authenticate clients. This principal is stored in spnego.keytab.file. This must be a fully qualified principal "
       + "in the service/host@REALM format (service is usually HTTP).";
 
+  /**
+   * <code>spnego.principal.to.local.rules</code>
+   */
+  public static final String SPNEGO_PRINCIPAL_TO_LOCAL_RULES_CONFIG =
+          "spnego.principal.to.local.rules";
+  public static final String DEFAULT_SPNEGO_PRINCIPAL_TO_LOCAL_RULES = null;
+  public static final String SPNEGO_PRINCIPAL_TO_LOCAL_RULES_DOC = "A list of rules for mapping from principal "
+          + "names to short names (typically operating system usernames). The rules are evaluated in order and the "
+          + "first rule that matches a principal name is used to map it to a short name. Any later rules in the list are "
+          + "ignored. By default, principal names of the form <code>{username}/{hostname}@{REALM}</code> are mapped "
+          + "to <code>{username}</code>. When not specified, the short name will be used.";
+
   /**
    * <code>trusted.proxy.services</code>
    */
@@ -560,6 +572,11 @@ public static ConfigDef define(ConfigDef configDef) {
                             DEFAULT_SPNEGO_PRINCIPAL,
                             ConfigDef.Importance.MEDIUM,
                             SPNEGO_PRINCIPAL_DOC)
+                    .define(SPNEGO_PRINCIPAL_TO_LOCAL_RULES_CONFIG,
+                            ConfigDef.Type.LIST,
+                            DEFAULT_SPNEGO_PRINCIPAL_TO_LOCAL_RULES,
+                            ConfigDef.Importance.MEDIUM,
+                            SPNEGO_PRINCIPAL_TO_LOCAL_RULES_DOC)
                     .define(TRUSTED_PROXY_SERVICES_CONFIG,
                             ConfigDef.Type.LIST,
                             DEFAULT_TRUSTED_PROXY_SERVICES,

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

@@ -0,0 +1,40 @@
+/*
+ * Copyright 2023 LinkedIn Corp. Licensed under the BSD 2-Clause License (the "License"). See License in the project root for license information.
+ */
+
+package com.linkedin.kafka.cruisecontrol.servlet.security;
+
+import org.eclipse.jetty.security.SpnegoUserIdentity;
+import org.eclipse.jetty.security.SpnegoUserPrincipal;
+import org.eclipse.jetty.security.authentication.AuthorizationService;
+import org.eclipse.jetty.server.UserIdentity;
+import org.eclipse.jetty.util.security.Credential;
+import javax.security.auth.Subject;
+import javax.servlet.http.HttpServletRequest;
+import java.security.Principal;
+
+public class DummyAuthorizationService implements AuthorizationService {
+
+    private static final Credential NO_CREDENTIAL = new Credential() {
+        @Override
+        public boolean check(Object credentials) {
+            return false;
+        }
+    };
+
+    @Override
+    public UserIdentity getUserIdentity(HttpServletRequest request, String name) {
+        return createUserIdentity(name);
+    }
+
+    private UserIdentity createUserIdentity(String username) {
+        Principal userPrincipal = new SpnegoUserPrincipal(username, "");
+        Subject subject = new Subject();
+        subject.getPrincipals().add(userPrincipal);
+        subject.getPrivateCredentials().add(NO_CREDENTIAL);
+        subject.setReadOnly();
+
+        return new SpnegoUserIdentity(subject, userPrincipal, null);
+    }
+
+}

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

@@ -0,0 +1,64 @@
+/*
+ * Copyright 2023 LinkedIn Corp. Licensed under the BSD 2-Clause License (the "License"). See License in the project root for license information.
+ */
+
+package com.linkedin.kafka.cruisecontrol.servlet.security.spnego;
+
+import java.util.Objects;
+
+public class PrincipalName {
+    private final String _primary;
+    private final String _instance;
+    private final String _realm;
+
+    public PrincipalName(String primary, String instance, String realm) {
+        _primary = Objects.requireNonNull(primary, "primary must not be null");
+        _instance = instance;
+        _realm = realm;
+    }
+
+    public PrincipalName(String primary) {
+        _primary = Objects.requireNonNull(primary, "primary must not be null");
+        _instance = null;
+        _realm = null;
+    }
+
+    public String getPrimary() {
+        return _primary;
+    }
+
+    public String getInstance() {
+        return _instance;
+    }
+
+    public String getRealm() {
+        return _realm;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || !Objects.equals(getClass(), o.getClass())) {
+            return false;
+        }
+        PrincipalName principalName = (PrincipalName) o;
+        return _primary.equals(principalName._primary) && Objects.equals(_instance, principalName._instance)
+                && Objects.equals(_realm, principalName._realm);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(_primary, _instance, _realm);
+    }
+
+    @Override
+    public String toString() {
+        return "PrincipalName{"
+                + "primary='" + _primary + '\''
+                + ", instance='" + _instance + '\''
+                + ", realm='" + _realm + '\''
+                + '}';
+    }
+}

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

@@ -0,0 +1,60 @@
+/*
+ * Copyright 2023 LinkedIn Corp. Licensed under the BSD 2-Clause License (the "License"). See License in the project root for license information.
+ */
+
+package com.linkedin.kafka.cruisecontrol.servlet.security.spnego;
+
+import org.apache.kafka.common.config.ConfigDef.Validator;
+import org.apache.kafka.common.config.ConfigException;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+public class PrincipalValidator implements Validator {
+    private static final Pattern PRINCIPAL_REGEX =
+            Pattern.compile("(?<primary>[^/\\s@]+)(/(?<instance>[\\w.-]+))?(@(?<realm>(\\S+)))?");
+
+    private final boolean _instanceRequired;
+    private final boolean _realmRequired;
+
+    public PrincipalValidator(boolean instanceRequired, boolean realmRequired) {
+        _instanceRequired = instanceRequired;
+        _realmRequired = realmRequired;
+    }
+
+    /**
+     * Creates a PrincipalName object.
+     * @param configName The name of the configuration
+     * @param principal The principal which will be the base of the PrincipalName object
+     * @return PrincipalName object
+     */
+    public static PrincipalName parsePrincipal(String configName, String principal) {
+        Matcher matcher = PRINCIPAL_REGEX.matcher(principal);
+        if (!matcher.matches()) {
+            throw new ConfigException(configName, principal, "Invalid principal");
+        }
+        String primary = matcher.group("primary");
+        String instance = matcher.group("instance");
+        String realm = matcher.group("realm");
+        return new PrincipalName(primary, instance, realm);
+    }
+
+    @Override
+    public void ensureValid(String name, Object value) {
+        if (value == null) {
+            return;
+        }
+
+        if (!(value instanceof String)) {
+            throw new ConfigException(name, value, "Value must be string");
+        }
+
+        String strVal = (String) value;
+        PrincipalName principalName = parsePrincipal(name, strVal);
+        if (_instanceRequired && principalName.getInstance() == null) {
+            throw new ConfigException(name, strVal, "Principal must contain the instance section");
+        }
+        if (_realmRequired && principalName.getRealm() == null) {
+            throw new ConfigException(name, strVal, "Principal must contain the realm section");
+        }
+    }
+}