kerberos: authentication between client and broker

conf/broker.conf

@@ -251,7 +251,7 @@ authenticateOriginalAuthData=false
 # Deprecated - Use webServicePortTls and brokerServicePortTls instead
 tlsEnabled=false
 
-# Tls cert refresh duration in seconds (set 0 to check on every new connection) 
+# Tls cert refresh duration in seconds (set 0 to check on every new connection)
 tlsCertRefreshCheckDurationSec=300
 
 # Path for the TLS certificate file
@@ -343,6 +343,22 @@ tokenPublicKey=
 # The token "claim" that will be interpreted as the authentication "role" or "principal" by AuthenticationProviderToken (defaults to "sub" if blank)
 tokenAuthClaim=
 
+### --- SASL Authentication Provider --- ###
+
+# Whether Use SASL Authentication or not.
+# TODO: used to bypass web resource check. will remove it after implementation the support.
+# github issue #3653 {@link: https://github.com/apache/pulsar/issues/3653}
+isSaslAuthentication=
+
+# This is a regexp, which limits the range of possible ids which can connect to the Broker using SASL.
+# Default value: `SaslConstants.JAAS_CLIENT_ALLOWED_IDS_DEFAULT`, which is ".*pulsar.*",
+# so only clients whose id contains 'pulsar' are allowed to connect.
+saslJaasClientAllowedIds=
+
+# Service Principal, for login context name.
+# Default value `SaslConstants.JAAS_DEFAULT_BROKER_SECTION_NAME`, which is "Broker".
+saslJaasBrokerSectionName=
+
 ### --- BookKeeper Client --- ###
 
 # Authentication plugin to use when connecting to bookies

pom.xml

@@ -105,6 +105,8 @@ flexible messaging model and an intuitive client API.</description>
     <module>pulsar-log4j2-appender</module>
     <module>pulsar-sql</module>
     <module>dashboard</module>
+    <module>pulsar-broker-auth-sasl</module>
+    <module>pulsar-client-auth-sasl</module>
 
     <!-- jclouds shaded for gson conflict: https://issues.apache.org/jira/browse/JCLOUDS-1166 -->
     <module>jclouds-shaded</module>
@@ -196,6 +198,7 @@ flexible messaging model and an intuitive client API.</description>
     <cassandra.version>3.6.0</cassandra.version>
     <disruptor.version>3.4.0</disruptor.version>
     <testcontainers.version>1.10.5</testcontainers.version>
+    <kerby.version>1.1.1</kerby.version>
 
     <!-- Plugin dependencies -->
     <protobuf-maven-plugin.version>0.6.1</protobuf-maven-plugin.version>

pulsar-broker-auth-sasl/pom.xml

@@ -0,0 +1,102 @@
+<?xml version="1.0"?>
+<!--
+
+    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.
+
+-->
+<project
+  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"
+  xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
+  <modelVersion>4.0.0</modelVersion>
+  <parent>
+    <groupId>org.apache.pulsar</groupId>
+    <artifactId>pulsar</artifactId>
+    <version>2.4.0-SNAPSHOT</version>
+  </parent>
+
+  <artifactId>pulsar-broker-auth-sasl</artifactId>
+  <packaging>jar</packaging>
+  <description>SASL authentication plugin for broker</description>
+
+  <dependencies>
+
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>pulsar-broker</artifactId>
+      <version>${project.version}</version>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.kerby</groupId>
+      <artifactId>kerby-config</artifactId>
+      <version>${kerby.version}</version>
+      <scope>test</scope>
+      <exclusions>
+        <exclusion>
+          <groupId>org.slf4j</groupId>
+          <artifactId>*</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.kerby</groupId>
+      <artifactId>kerb-simplekdc</artifactId>
+      <version>${kerby.version}</version>
+      <scope>test</scope>
+      <exclusions>
+        <exclusion>
+          <groupId>org.slf4j</groupId>
+          <artifactId>*</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
+
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>pulsar-broker</artifactId>
+      <version>${project.version}</version>
+      <scope>test</scope>
+      <type>test-jar</type>
+    </dependency>
+
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>managed-ledger-original</artifactId>
+      <version>${project.version}</version>
+      <type>test-jar</type>
+      <scope>test</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>pulsar-zookeeper-utils</artifactId>
+      <version>${project.version}</version>
+      <type>test-jar</type>
+      <scope>test</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>pulsar-client-auth-sasl</artifactId>
+      <version>${project.version}</version>
+      <scope>test</scope>
+    </dependency>
+
+  </dependencies>
+</project>

pulsar-broker-auth-sasl/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderSasl.java

@@ -0,0 +1,113 @@
+/**
+ * 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.pulsar.broker.authentication;
+
+import static org.apache.pulsar.common.sasl.SaslConstants.JAAS_BROKER_SECTION_NAME;
+import static org.apache.pulsar.common.sasl.SaslConstants.JAAS_CLIENT_ALLOWED_IDS;
+import static org.apache.pulsar.common.sasl.SaslConstants.KINIT_COMMAND;
+
+import java.io.IOException;
+import java.net.SocketAddress;
+import java.util.Map;
+import java.util.regex.Pattern;
+import java.util.regex.PatternSyntaxException;
+
+import javax.naming.AuthenticationException;
+import javax.net.ssl.SSLSession;
+import javax.security.auth.login.LoginException;
+
+import com.google.common.collect.Maps;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.common.api.AuthData;
+import org.apache.pulsar.common.sasl.JAASCredentialsContainer;
+import org.apache.pulsar.common.sasl.SaslConstants;
+
+@Slf4j
+public class AuthenticationProviderSasl implements AuthenticationProvider {
+
+    private Pattern allowedIdsPattern;
+    private Map<String, String> configuration;
+
+    private JAASCredentialsContainer jaasCredentialsContainer;
+    private String loginContextName;
+
+    @Override
+    public void initialize(ServiceConfiguration config) throws IOException {
+        this.configuration = Maps.newHashMap();
+        final String allowedIdsPatternRegExp = config.getSaslJaasClientAllowedIds();
+        configuration.put(JAAS_CLIENT_ALLOWED_IDS, allowedIdsPatternRegExp);
+        configuration.put(JAAS_BROKER_SECTION_NAME, config.getSaslJaasBrokerSectionName());
+        configuration.put(KINIT_COMMAND, config.getKinitCommand());
+
+        try {
+            this.allowedIdsPattern = Pattern.compile(allowedIdsPatternRegExp);
+        } catch (PatternSyntaxException error) {
+            log.error("Invalid regular expression for id " + allowedIdsPatternRegExp, error);
+            throw new IOException(error);
+        }
+
+        loginContextName = config.getSaslJaasBrokerSectionName();
+        if (jaasCredentialsContainer == null) {
+            log.info("JAAS loginContext is: {}." , loginContextName);
+            try {
+                jaasCredentialsContainer = new JAASCredentialsContainer(
+                    loginContextName,
+                    new PulsarSaslServer.SaslServerCallbackHandler(allowedIdsPattern),
+                    configuration);
+            } catch (LoginException e) {
+                log.error("JAAS login in broker failed: {}" , e);
+                throw new IOException(e);
+            }
+        }
+    }
+
+    @Override
+    public String authenticate(AuthenticationDataSource authData) throws AuthenticationException {
+        if (authData instanceof SaslAuthenticationDataSource) {
+            return ((SaslAuthenticationDataSource)authData).getAuthorizationID();
+        } else {
+            throw new AuthenticationException("Not support authDataSource type, expect sasl.");
+        }
+    }
+
+    @Override
+    public String getAuthMethodName() {
+        return SaslConstants.AUTH_METHOD_NAME;
+    }
+
+    @Override
+    public void close() throws IOException {
+    }
+
+    @Override
+    public AuthenticationState newAuthState(AuthData authData,
+                                            SocketAddress remoteAddress,
+                                            SSLSession sslSession) throws AuthenticationException {
+        try {
+            new PulsarSaslServer(jaasCredentialsContainer.getSubject(), allowedIdsPattern);
+            return new SaslAuthenticationState(
+                new SaslAuthenticationDataSource(
+                    new PulsarSaslServer(jaasCredentialsContainer.getSubject(), allowedIdsPattern)));
+        } catch (Throwable t) {
+            log.error("Failed create sasl auth state: {}" , t);
+            throw new AuthenticationException(t.getMessage());
+        }
+    }
+}