Introduce LoginType and refactor KerberosName in order to make authToLocal rules configurable

clients/src/main/java/org/apache/kafka/clients/ClientUtils.java

@@ -20,6 +20,7 @@
 import java.util.concurrent.atomic.AtomicReference;
 
 import org.apache.kafka.common.network.ChannelBuilders;
+import org.apache.kafka.common.network.LoginType;
 import org.apache.kafka.common.network.Mode;
 import org.apache.kafka.common.protocol.SecurityProtocol;
 import org.apache.kafka.common.network.ChannelBuilder;
@@ -76,7 +77,7 @@ public static ChannelBuilder createChannelBuilder(Map<String, ?> configs) {
         SecurityProtocol securityProtocol = SecurityProtocol.valueOf((String) configs.get(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG));
         if (securityProtocol != SecurityProtocol.SSL && securityProtocol != SecurityProtocol.PLAINTEXT && securityProtocol != SecurityProtocol.SASL_PLAINTEXT)
             throw new ConfigException("Invalid SecurityProtocol " + CommonClientConfigs.SECURITY_PROTOCOL_CONFIG);
-        return ChannelBuilders.create(securityProtocol, Mode.CLIENT, configs);
+        return ChannelBuilders.create(securityProtocol, Mode.CLIENT, LoginType.CLIENT, configs);
     }
 
 }

clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java

@@ -308,6 +308,7 @@ public class ConsumerConfig extends AbstractConfig {
                                 .define(SaslConfigs.SASL_KERBEROS_TICKET_RENEW_WINDOW_FACTOR, Type.DOUBLE, SaslConfigs.DEFAULT_KERBEROS_TICKET_RENEW_WINDOW_FACTOR, Importance.LOW, SaslConfigs.SASL_KERBEROS_TICKET_RENEW_WINDOW_FACTOR_DOC)
                                 .define(SaslConfigs.SASL_KERBEROS_TICKET_RENEW_JITTER, Type.DOUBLE, SaslConfigs.DEFAULT_KERBEROS_TICKET_RENEW_JITTER, Importance.LOW, SaslConfigs.SASL_KERBEROS_TICKET_RENEW_JITTER_DOC)
                                 .define(SaslConfigs.SASL_KERBEROS_MIN_TIME_BEFORE_RELOGIN, Type.LONG, SaslConfigs.DEFAULT_KERBEROS_MIN_TIME_BEFORE_RELOGIN, Importance.LOW, SaslConfigs.SASL_KERBEROS_MIN_TIME_BEFORE_RELOGIN_DOC)
+                                .define(SaslConfigs.AUTH_TO_LOCAL, Type.LIST, SaslConfigs.DEFAULT_AUTH_TO_LOCAL, Importance.MEDIUM, SaslConfigs.AUTH_TO_LOCAL_DOC)
                                 .define(REQUEST_TIMEOUT_MS_CONFIG,
                                         Type.INT,
                                         40 * 1000,

clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java

@@ -285,6 +285,7 @@ public class ProducerConfig extends AbstractConfig {
                                 .define(SaslConfigs.SASL_KERBEROS_TICKET_RENEW_WINDOW_FACTOR, Type.DOUBLE, SaslConfigs.DEFAULT_KERBEROS_TICKET_RENEW_WINDOW_FACTOR, Importance.LOW, SaslConfigs.SASL_KERBEROS_TICKET_RENEW_WINDOW_FACTOR_DOC)
                                 .define(SaslConfigs.SASL_KERBEROS_TICKET_RENEW_JITTER, Type.DOUBLE, SaslConfigs.DEFAULT_KERBEROS_TICKET_RENEW_JITTER, Importance.LOW, SaslConfigs.SASL_KERBEROS_TICKET_RENEW_JITTER_DOC)
                                 .define(SaslConfigs.SASL_KERBEROS_MIN_TIME_BEFORE_RELOGIN, Type.LONG, SaslConfigs.DEFAULT_KERBEROS_MIN_TIME_BEFORE_RELOGIN, Importance.LOW, SaslConfigs.SASL_KERBEROS_MIN_TIME_BEFORE_RELOGIN_DOC)
+                                .define(SaslConfigs.AUTH_TO_LOCAL, Type.LIST, SaslConfigs.DEFAULT_AUTH_TO_LOCAL, Importance.MEDIUM, SaslConfigs.AUTH_TO_LOCAL_DOC)
                                 /* default is set to be a bit lower than the server default (10 min), to avoid both client and server closing connection at same time */
                                 .define(CONNECTIONS_MAX_IDLE_MS_CONFIG, Type.LONG, 9 * 60 * 1000, Importance.MEDIUM, CommonClientConfigs.CONNECTIONS_MAX_IDLE_MS_DOC)
                                 .define(PARTITIONER_CLASS_CONFIG, Type.CLASS, "org.apache.kafka.clients.producer.internals.DefaultPartitioner", Importance.MEDIUM, PARTITIONER_CLASS_DOC);

clients/src/main/java/org/apache/kafka/common/config/SaslConfigs.java

@@ -13,6 +13,8 @@
 
 package org.apache.kafka.common.config;
 
+import java.util.Collections;
+import java.util.List;
 
 public class SaslConfigs {
     /*
@@ -41,4 +43,8 @@ public class SaslConfigs {
     public static final String SASL_KERBEROS_MIN_TIME_BEFORE_RELOGIN_DOC = "LoginThread sleep time between refresh attempts";
     public static final long DEFAULT_KERBEROS_MIN_TIME_BEFORE_RELOGIN = 1 * 60 * 1000L;
 
+    public static final String AUTH_TO_LOCAL = "kafka.security.auth.to.local";
+    public static final String AUTH_TO_LOCAL_DOC = "Rules for the mapping between principal names and operating system user names";
+    public static final List<String> DEFAULT_AUTH_TO_LOCAL = Collections.singletonList("DEFAULT");
+
 }

clients/src/main/java/org/apache/kafka/common/network/ChannelBuilders.java

@@ -23,13 +23,15 @@ private ChannelBuilders() { }
 
     /**
      * @param securityProtocol the securityProtocol
-     * @param mode the SSL mode, it must be non-null if `securityProcol` is `SSL` and it is ignored otherwise
+     * @param mode the mode, it must be non-null if `securityProtocol` is not `PLAINTEXT`;
+     *             it is ignored otherwise
+     * @param loginType the loginType, it must be non-null if `securityProtocol` is SASL_*; it is ignored otherwise
      * @param configs client/server configs
      * @return the configured `ChannelBuilder`
      * @throws IllegalArgumentException if `mode` invariants described above is not maintained
      */
-    public static ChannelBuilder create(SecurityProtocol securityProtocol, Mode mode, Map<String, ?> configs) {
-        ChannelBuilder channelBuilder = null;
+    public static ChannelBuilder create(SecurityProtocol securityProtocol, Mode mode, LoginType loginType, Map<String, ?> configs) {
+        ChannelBuilder channelBuilder;
 
         switch (securityProtocol) {
             case SSL:
@@ -39,7 +41,9 @@ public static ChannelBuilder create(SecurityProtocol securityProtocol, Mode mode
             case SASL_SSL:
             case SASL_PLAINTEXT:
                 requireNonNullMode(mode, securityProtocol);
-                channelBuilder = new SaslChannelBuilder(mode, securityProtocol);
+                if (loginType == null)
+                    throw new IllegalArgumentException("`loginType` must be non-null if `securityProtocol` is `" + securityProtocol + "`");
+                channelBuilder = new SaslChannelBuilder(mode, loginType, securityProtocol);
                 break;
             case PLAINTEXT:
             case TRACE:

clients/src/main/java/org/apache/kafka/common/network/KafkaChannel.java

@@ -61,12 +61,12 @@ public Principal principal() throws IOException {
     }
 
     /**
-     * Does handshake of transportLayer and Authentication using configured authenticator
+     * Does handshake of transportLayer and authentication using configured authenticator
      */
     public void prepare() throws IOException {
         if (!transportLayer.ready())
             transportLayer.handshake();
-        else if (transportLayer.ready() && !authenticator.complete())
+        if (transportLayer.ready() && !authenticator.complete())
             authenticator.authenticate();
     }
 

clients/src/main/java/org/apache/kafka/common/network/LoginType.java

@@ -0,0 +1,39 @@
+/**
+ * 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.kafka.common.network;
+
+import org.apache.kafka.common.security.JaasUtils;
+
+/**
+ * The type of the login context, it should be SERVER for the broker and CLIENT for the clients (i.e. consumer and
+ * producer). It provides the the login context name which defines the section of the JAAS configuration file to be used
+ * for login.
+ */
+public enum LoginType {
+    CLIENT(JaasUtils.LOGIN_CONTEXT_CLIENT),
+    SERVER(JaasUtils.LOGIN_CONTEXT_SERVER);
+
+    private final String contextName;
+
+    LoginType(String contextName) {
+        this.contextName = contextName;
+    }
+
+    public String contextName() {
+        return contextName;
+    }
+}

clients/src/main/java/org/apache/kafka/common/network/SSLChannelBuilder.java

@@ -68,10 +68,8 @@ public void close()  {
 
     protected SSLTransportLayer buildTransportLayer(SSLFactory sslFactory, String id, SelectionKey key) throws IOException {
         SocketChannel socketChannel = (SocketChannel) key.channel();
-        SSLTransportLayer transportLayer = new SSLTransportLayer(id, key,
-                sslFactory.createSSLEngine(socketChannel.socket().getInetAddress().getHostName(),
-                                           socketChannel.socket().getPort()));
-        transportLayer.startHandshake();
-        return transportLayer;
+        return SSLTransportLayer.create(id, key,
+            sslFactory.createSSLEngine(socketChannel.socket().getInetAddress().getHostName(),
+            socketChannel.socket().getPort()));
     }
 }

clients/src/main/java/org/apache/kafka/common/network/SSLTransportLayer.java

@@ -58,7 +58,14 @@ public class SSLTransportLayer implements TransportLayer {
     private ByteBuffer appReadBuffer;
     private ByteBuffer emptyBuf = ByteBuffer.allocate(0);
 
-    public SSLTransportLayer(String channelId, SelectionKey key, SSLEngine sslEngine) throws IOException {
+    public static SSLTransportLayer create(String channelId, SelectionKey key, SSLEngine sslEngine) throws IOException {
+        SSLTransportLayer transportLayer = new SSLTransportLayer(channelId, key, sslEngine);
+        transportLayer.startHandshake();
+        return transportLayer;
+    }
+
+    // Prefer `create`, only use this in tests
+    SSLTransportLayer(String channelId, SelectionKey key, SSLEngine sslEngine) throws IOException {
         this.channelId = channelId;
         this.key = key;
         this.socketChannel = (SocketChannel) key.channel();

clients/src/main/java/org/apache/kafka/common/network/SaslChannelBuilder.java

@@ -12,11 +12,16 @@
  */
 package org.apache.kafka.common.network;
 
+import java.io.IOException;
 import java.nio.channels.SelectionKey;
 import java.nio.channels.SocketChannel;
+import java.util.List;
 import java.util.Map;
 
+import org.apache.kafka.common.config.SaslConfigs;
+import org.apache.kafka.common.security.JaasUtils;
 import org.apache.kafka.common.security.auth.PrincipalBuilder;
+import org.apache.kafka.common.security.kerberos.KerberosNameParser;
 import org.apache.kafka.common.security.kerberos.LoginManager;
 import org.apache.kafka.common.security.authenticator.SaslClientAuthenticator;
 import org.apache.kafka.common.security.authenticator.SaslServerAuthenticator;
@@ -34,23 +39,35 @@ public class SaslChannelBuilder implements ChannelBuilder {
 
     private final SecurityProtocol securityProtocol;
     private final Mode mode;
+    private final LoginType loginType;
 
     private LoginManager loginManager;
     private PrincipalBuilder principalBuilder;
     private SSLFactory sslFactory;
     private Map<String, ?> configs;
+    private KerberosNameParser kerberosNameParser;
 
-    public SaslChannelBuilder(Mode mode, SecurityProtocol securityProtocol) {
+    public SaslChannelBuilder(Mode mode, LoginType loginType, SecurityProtocol securityProtocol) {
         this.mode = mode;
+        this.loginType = loginType;
         this.securityProtocol = securityProtocol;
     }
 
     public void configure(Map<String, ?> configs) throws KafkaException {
         try {
             this.configs = configs;
-            this.loginManager = LoginManager.acquireLoginManager(mode, configs);
+            this.loginManager = LoginManager.acquireLoginManager(loginType, configs);
             this.principalBuilder = (PrincipalBuilder) Utils.newInstance((Class<?>) configs.get(SSLConfigs.PRINCIPAL_BUILDER_CLASS_CONFIG));
             this.principalBuilder.configure(configs);
+
+            String defaultRealm;
+            try {
+                defaultRealm = JaasUtils.defaultRealm();
+            } catch (Exception ke) {
+                defaultRealm = "";
+            }
+            kerberosNameParser = new KerberosNameParser(defaultRealm, (List<String>) configs.get(SaslConfigs.AUTH_TO_LOCAL));
+
             if (this.securityProtocol == SecurityProtocol.SASL_SSL) {
                 this.sslFactory = new SSLFactory(mode);
                 this.sslFactory.configure(this.configs);
@@ -63,19 +80,13 @@ public void configure(Map<String, ?> configs) throws KafkaException {
     public KafkaChannel buildChannel(String id, SelectionKey key, int maxReceiveSize) throws KafkaException {
         try {
             SocketChannel socketChannel = (SocketChannel) key.channel();
-            TransportLayer transportLayer;
-            if (this.securityProtocol == SecurityProtocol.SASL_SSL) {
-                transportLayer = new SSLTransportLayer(id, key,
-                                                       sslFactory.createSSLEngine(socketChannel.socket().getInetAddress().getHostName(),
-                                                                                  socketChannel.socket().getPort()));
-            } else {
-                transportLayer = new PlaintextTransportLayer(key);
-            }
+            TransportLayer transportLayer = buildTransportLayer(id, key, socketChannel);
             Authenticator authenticator;
             if (mode == Mode.SERVER)
-                authenticator = new SaslServerAuthenticator(id, loginManager.subject());
+                authenticator = new SaslServerAuthenticator(id, loginManager.subject(), kerberosNameParser);
             else
-                authenticator = new SaslClientAuthenticator(id, loginManager.subject(), loginManager.serviceName(), socketChannel.socket().getInetAddress().getHostName());
+                authenticator = new SaslClientAuthenticator(id, loginManager.subject(), loginManager.serviceName(),
+                        socketChannel.socket().getInetAddress().getHostName(), kerberosNameParser);
             authenticator.configure(transportLayer, this.principalBuilder, this.configs);
             return new KafkaChannel(id, transportLayer, authenticator, maxReceiveSize);
         } catch (Exception e) {
@@ -88,4 +99,15 @@ public void close()  {
         this.principalBuilder.close();
         this.loginManager.release();
     }
+
+    protected TransportLayer buildTransportLayer(String id, SelectionKey key, SocketChannel socketChannel) throws IOException {
+        if (this.securityProtocol == SecurityProtocol.SASL_SSL) {
+            return SSLTransportLayer.create(id, key,
+                sslFactory.createSSLEngine(socketChannel.socket().getInetAddress().getHostName(),
+                socketChannel.socket().getPort()));
+        } else {
+            return new PlaintextTransportLayer(key);
+        }
+    }
+
 }

clients/src/main/java/org/apache/kafka/common/security/JaasUtils.java

@@ -23,11 +23,7 @@
 import java.lang.reflect.Method;
 import java.io.IOException;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 public class JaasUtils {
-    private static final Logger LOG = LoggerFactory.getLogger(JaasUtils.class);
     public static final String LOGIN_CONTEXT_SERVER = "KafkaServer";
     public static final String LOGIN_CONTEXT_CLIENT = "KafkaClient";
     public static final String SERVICE_NAME = "serviceName";
@@ -58,6 +54,10 @@ public static String defaultRealm()
         throws ClassNotFoundException, NoSuchMethodException,
                IllegalArgumentException, IllegalAccessException,
                InvocationTargetException {
+
+        //TODO Find a way to avoid using these proprietary classes as access to Java 9 will block access by default
+        //due to the Jigsaw module system
+
         Object kerbConf;
         Class<?> classRef;
         Method getInstanceMethod;

clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslClientAuthenticator.java

@@ -46,9 +46,9 @@
 import org.apache.kafka.common.network.TransportLayer;
 import org.apache.kafka.common.security.auth.KafkaPrincipal;
 import org.apache.kafka.common.security.auth.PrincipalBuilder;
-import org.apache.kafka.common.security.kerberos.KerberosName;
 import org.apache.kafka.common.KafkaException;
 
+import org.apache.kafka.common.security.kerberos.KerberosNameParser;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -64,6 +64,7 @@ public enum SaslState {
     private final String servicePrincipal;
     private final String host;
     private final String node;
+    private final KerberosNameParser kerberosNameParser;
 
     // assigned in `configure`
     private SaslClient saslClient;
@@ -76,11 +77,12 @@ public enum SaslState {
 
     private SaslState saslState = SaslState.INITIAL;
 
-    public SaslClientAuthenticator(String node, Subject subject, String servicePrincipal, String host) throws IOException {
+    public SaslClientAuthenticator(String node, Subject subject, String servicePrincipal, String host, KerberosNameParser kerberosNameParser) throws IOException {
         this.node = node;
         this.subject = subject;
         this.host = host;
         this.servicePrincipal = servicePrincipal;
+        this.kerberosNameParser = kerberosNameParser;
     }
 
     public void configure(TransportLayer transportLayer, PrincipalBuilder principalBuilder, Map<String, ?> configs) throws KafkaException {
@@ -89,7 +91,7 @@ public void configure(TransportLayer transportLayer, PrincipalBuilder principalB
 
             // determine client principal from subject.
             Principal clientPrincipal = subject.getPrincipals().iterator().next();
-            this.clientPrincipalName = new KerberosName(clientPrincipal.getName()).toString();
+            this.clientPrincipalName = kerberosNameParser.parse(clientPrincipal.getName()).toString();
             this.saslClient = createSaslClient();
         } catch (Exception e) {
             throw new KafkaException("Failed to configure SaslClientAuthenticator", e);

clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticator.java

@@ -35,6 +35,7 @@
 
 import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.security.kerberos.KerberosName;
+import org.apache.kafka.common.security.kerberos.KerberosNameParser;
 import org.ietf.jgss.GSSContext;
 import org.ietf.jgss.GSSCredential;
 import org.ietf.jgss.GSSException;
@@ -58,6 +59,7 @@ public class SaslServerAuthenticator implements Authenticator {
     private final SaslServer saslServer;
     private final Subject subject;
     private final String node;
+    private final KerberosNameParser kerberosNameParser;
 
     // assigned in `configure`
     private TransportLayer transportLayer;
@@ -66,13 +68,14 @@ public class SaslServerAuthenticator implements Authenticator {
     private NetworkReceive netInBuffer;
     private NetworkSend netOutBuffer;
 
-    public SaslServerAuthenticator(String node, final Subject subject) throws IOException {
+    public SaslServerAuthenticator(String node, final Subject subject, KerberosNameParser kerberosNameParser) throws IOException {
         if (subject == null)
             throw new IllegalArgumentException("subject cannot be null");
         if (subject.getPrincipals().isEmpty())
             throw new IllegalArgumentException("subject must have at least one principal");
         this.node = node;
         this.subject = subject;
+        this.kerberosNameParser = kerberosNameParser;
         saslServer = createSaslServer();
     }
 
@@ -82,11 +85,12 @@ public void configure(TransportLayer transportLayer, PrincipalBuilder principalB
 
     private SaslServer createSaslServer() throws IOException {
         // server is using a JAAS-authenticated subject: determine service principal name and hostname from kafka server's subject.
-        final SaslServerCallbackHandler saslServerCallbackHandler = new SaslServerCallbackHandler(Configuration.getConfiguration());
+        final SaslServerCallbackHandler saslServerCallbackHandler = new SaslServerCallbackHandler(
+                Configuration.getConfiguration(), kerberosNameParser);
         final Principal servicePrincipal = subject.getPrincipals().iterator().next();
         KerberosName kerberosName;
         try {
-            kerberosName = new KerberosName(servicePrincipal.getName());
+            kerberosName = kerberosNameParser.parse(servicePrincipal.getName());
         } catch (IllegalArgumentException e) {
             throw new KafkaException("Principal has name with unexpected format " + servicePrincipal);
         }