merlimat closed pull request #1087: Add basic authentication plugin
URL: https://github.com/apache/incubator-pulsar/pull/1087
 
 
   

This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:

As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):

diff --git a/pom.xml b/pom.xml
index 11797a49c..5c71efe2c 100644
--- a/pom.xml
+++ b/pom.xml
@@ -622,6 +622,7 @@ flexible messaging model and an intuitive client 
API.</description>
             <exclude>logs/**</exclude>
             <exclude>**/*.versionsBackup</exclude>
             <exclude>**/circe/**</exclude>
+            
<exclude>pulsar-broker/src/test/resources/authentication/basic/.htpasswd</exclude>
             <exclude>pulsar-client-cpp/lib/checksum/int_types.h</exclude>
             <exclude>pulsar-client-cpp/lib/checksum/gf2.hpp</exclude>
             <exclude>pulsar-client-cpp/lib/checksum/crc32c_sse42.cc</exclude>
diff --git 
a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderBasic.java
 
b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderBasic.java
new file mode 100644
index 000000000..f2f2f6588
--- /dev/null
+++ 
b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderBasic.java
@@ -0,0 +1,142 @@
+/**
+ * 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 org.apache.commons.codec.digest.Crypt;
+import org.apache.commons.codec.digest.Md5Crypt;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.pulsar.broker.ServiceConfiguration;
+
+import javax.naming.AuthenticationException;
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.FileReader;
+import java.io.IOException;
+import java.util.*;
+
+public class AuthenticationProviderBasic implements AuthenticationProvider {
+    private final static String HTTP_HEADER_NAME = "Authorization";
+    private final static String CONF_SYSTEM_PROPERTY_KEY = 
"pulsar.auth.basic.conf";
+    private Map<String, String> users;
+
+    @Override
+    public void close() throws IOException {
+        // noop
+    }
+
+    @Override
+    public void initialize(ServiceConfiguration config) throws IOException {
+        File confFile = new File(System.getProperty(CONF_SYSTEM_PROPERTY_KEY));
+        if (!confFile.exists()) {
+            throw new IOException("The password auth conf file does not 
exist");
+        } else if (!confFile.isFile()) {
+            throw new IOException("The path is not a file");
+        }
+        BufferedReader reader = new BufferedReader(new FileReader(confFile));
+        users = new HashMap<>();
+        for (String line : reader.lines().toArray(s -> new String[s])) {
+            List<String> splitLine = Arrays.asList(line.split(":"));
+            if (splitLine.size() != 2) {
+                throw new IOException("The format of the password auth conf 
file is invalid");
+            }
+            users.put(splitLine.get(0), splitLine.get(1));
+        }
+        reader.close();
+    }
+
+    @Override
+    public String getAuthMethodName() {
+        return "basic";
+    }
+
+    @Override
+    public String authenticate(AuthenticationDataSource authData) throws 
AuthenticationException {
+        AuthParams authParams = new AuthParams(authData);
+        String userId = authParams.getUserId();
+        String password = authParams.getPassword();
+        String msg = "Unknown user or invalid password";
+
+        if (users.get(userId) == null) {
+            throw new AuthenticationException(msg);
+        }
+
+        String encryptedPassword = users.get(userId);
+
+        // For md5 algorithm
+        if ((users.get(userId).startsWith("$apr1"))) {
+            List<String> splitEncryptedPassword = 
Arrays.asList(encryptedPassword.split("\\$"));
+            if (splitEncryptedPassword.size() != 4 || !encryptedPassword
+                    .equals(Md5Crypt.apr1Crypt(password.getBytes(), 
splitEncryptedPassword.get(2)))) {
+                throw new AuthenticationException(msg);
+            }
+        // For crypt algorithm
+        } else if (!encryptedPassword.equals(Crypt.crypt(password.getBytes(), 
encryptedPassword.substring(0, 2)))) {
+            throw new AuthenticationException(msg);
+        }
+
+        return userId;
+    }
+
+    private class AuthParams {
+        private String userId;
+        private String password;
+
+        public AuthParams(AuthenticationDataSource authData) throws 
AuthenticationException {
+            String authParams;
+            if (authData.hasDataFromCommand()) {
+                authParams = authData.getCommandData();
+            } else if (authData.hasDataFromHttp()) {
+                String rawAuthToken = authData.getHttpHeader(HTTP_HEADER_NAME);
+                // parsing and validation
+                if (StringUtils.isBlank(rawAuthToken) || 
!rawAuthToken.toUpperCase().startsWith("BASIC ")) {
+                    throw new AuthenticationException("Authentication token 
has to be started with \"Basic \"");
+                }
+                String[] splitRawAuthToken = rawAuthToken.split(" ");
+                if (splitRawAuthToken.length != 2) {
+                    throw new AuthenticationException("Base64 encoded token is 
not found");
+                }
+
+                try {
+                    authParams = new 
String(Base64.getDecoder().decode(splitRawAuthToken[1]));
+                } catch (Exception e) {
+                    throw new AuthenticationException("Base64 decoding is 
failure: " + e.getMessage());
+                }
+            } else {
+                throw new AuthenticationException("Authentication data source 
does not have data");
+            }
+
+            String[] parsedAuthParams = authParams.split(":");
+            if (parsedAuthParams.length != 2) {
+                throw new AuthenticationException("Base64 decoded params are 
invalid");
+            }
+
+            userId = parsedAuthParams[0];
+            password = parsedAuthParams[1];
+        }
+
+        public String getUserId() {
+            return userId;
+        }
+
+        public String getPassword() {
+            return password;
+        }
+    }
+}
diff --git 
a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthenticatedProducerConsumerTest.java
 
b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthenticatedProducerConsumerTest.java
index 83b15f5a6..729294033 100644
--- 
a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthenticatedProducerConsumerTest.java
+++ 
b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthenticatedProducerConsumerTest.java
@@ -18,32 +18,17 @@
  */
 package org.apache.pulsar.client.api;
 
-import static org.mockito.Mockito.doReturn;
-import static org.mockito.Mockito.spy;
-
-import java.net.URI;
-import java.util.*;
-import java.util.concurrent.TimeUnit;
-
-import javax.ws.rs.InternalServerErrorException;
-
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+import org.apache.pulsar.broker.authentication.AuthenticationProviderBasic;
 import org.apache.pulsar.broker.authentication.AuthenticationProviderTls;
 import org.apache.pulsar.client.admin.PulsarAdmin;
 import org.apache.pulsar.client.admin.PulsarAdminException;
-import org.apache.pulsar.client.api.Authentication;
-import org.apache.pulsar.client.api.Consumer;
-import org.apache.pulsar.client.api.ConsumerConfiguration;
-import org.apache.pulsar.client.api.Message;
-import org.apache.pulsar.client.api.Producer;
-import org.apache.pulsar.client.api.ProducerConfiguration;
-import org.apache.pulsar.client.api.PulsarClient;
-import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.client.impl.auth.AuthenticationBasic;
 import org.apache.pulsar.client.impl.auth.AuthenticationTls;
 import org.apache.pulsar.common.policies.data.AuthAction;
 import org.apache.pulsar.common.policies.data.ClusterData;
 import org.apache.pulsar.common.policies.data.PropertyAdmin;
-import org.apache.pulsar.common.policies.data.SubscriptionAuthMode;
-import org.mockito.Mockito;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.testng.Assert;
@@ -52,8 +37,13 @@
 import org.testng.annotations.DataProvider;
 import org.testng.annotations.Test;
 
-import com.google.common.collect.Lists;
-import com.google.common.collect.Sets;
+import javax.ws.rs.InternalServerErrorException;
+import java.net.URI;
+import java.util.*;
+import java.util.concurrent.TimeUnit;
+
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.spy;
 
 public class AuthenticatedProducerConsumerTest extends ProducerConsumerBase {
     private static final Logger log = 
LoggerFactory.getLogger(AuthenticatedProducerConsumerTest.class);
@@ -64,6 +54,8 @@
     private final String TLS_CLIENT_CERT_FILE_PATH = 
"./src/test/resources/authentication/tls/client-cert.pem";
     private final String TLS_CLIENT_KEY_FILE_PATH = 
"./src/test/resources/authentication/tls/client-key.pem";
 
+    private final String BASIC_CONF_FILE_PATH = 
"./src/test/resources/authentication/basic/.htpasswd";
+
     @BeforeMethod
     @Override
     protected void setup() throws Exception {
@@ -83,6 +75,7 @@ protected void setup() throws Exception {
         Set<String> superUserRoles = new HashSet<>();
         superUserRoles.add("localhost");
         superUserRoles.add("superUser");
+        superUserRoles.add("superUser2");
         conf.setSuperUserRoles(superUserRoles);
 
         
conf.setBrokerClientAuthenticationPlugin(AuthenticationTls.class.getName());
@@ -91,6 +84,8 @@ protected void setup() throws Exception {
 
         Set<String> providers = new HashSet<>();
         providers.add(AuthenticationProviderTls.class.getName());
+        providers.add(AuthenticationProviderBasic.class.getName());
+        System.setProperty("pulsar.auth.basic.conf", BASIC_CONF_FILE_PATH);
         conf.setAuthenticationProviders(providers);
 
         conf.setClusterName("use");
@@ -107,7 +102,13 @@ protected final void internalSetup(Authentication auth) 
throws Exception {
         clientConf.setUseTls(true);
 
         admin = spy(new PulsarAdmin(brokerUrlTls, clientConf));
-        String lookupUrl = new URI("pulsar+ssl://localhost:" + 
BROKER_PORT_TLS).toString();
+        String lookupUrl;
+        // For http basic authentication test
+        if (methodName.equals("testBasicCryptSyncProducerAndConsumer")) {
+            lookupUrl = new URI("https://localhost:"; + 
BROKER_WEBSERVICE_PORT_TLS).toString();
+        } else {
+            lookupUrl = new URI("pulsar+ssl://localhost:" + 
BROKER_PORT_TLS).toString();
+        }
         pulsarClient = PulsarClient.create(lookupUrl, clientConf);
     }
 
@@ -167,8 +168,6 @@ public void testTlsSyncProducerAndConsumer(int 
batchMessageDelayMs) throws Excep
         authTls.configure(authParams);
         internalSetup(authTls);
 
-        admin.clusters().updateCluster("use", new 
ClusterData(brokerUrl.toString(), brokerUrlTls.toString(),
-                "pulsar://localhost:" + BROKER_PORT, "pulsar+ssl://localhost:" 
+ BROKER_PORT_TLS));
         admin.properties().createProperty("my-property",
                 new PropertyAdmin(Lists.newArrayList("appid1", "appid2"), 
Sets.newHashSet("use")));
         admin.namespaces().createNamespace("my-property/use/my-ns");
@@ -178,6 +177,39 @@ public void testTlsSyncProducerAndConsumer(int 
batchMessageDelayMs) throws Excep
         log.info("-- Exiting {} test --", methodName);
     }
 
+    @Test(dataProvider = "batch")
+    public void testBasicCryptSyncProducerAndConsumer(int batchMessageDelayMs) 
throws Exception {
+        log.info("-- Starting {} test --", methodName);
+        AuthenticationBasic authPassword = new AuthenticationBasic();
+        
authPassword.configure("{\"userId\":\"superUser\",\"password\":\"supepass\"}");
+        internalSetup(authPassword);
+
+        admin.properties()
+                .createProperty("my-property", new 
PropertyAdmin(Lists.newArrayList(), Sets.newHashSet("use")));
+        admin.namespaces().createNamespace("my-property/use/my-ns");
+
+        testSyncProducerAndConsumer(batchMessageDelayMs);
+
+        log.info("-- Exiting {} test --", methodName);
+    }
+
+    @Test(dataProvider = "batch")
+    public void testBasicArp1SyncProducerAndConsumer(int batchMessageDelayMs) 
throws Exception {
+        log.info("-- Starting {} test --", methodName);
+        AuthenticationBasic authPassword = new AuthenticationBasic();
+        
authPassword.configure("{\"userId\":\"superUser2\",\"password\":\"superpassword\"}");
+        internalSetup(authPassword);
+
+        admin.properties()
+                .createProperty("my-property", new 
PropertyAdmin(Lists.newArrayList(), Sets.newHashSet("use")));
+        admin.namespaces().createNamespace("my-property/use/my-ns");
+
+        testSyncProducerAndConsumer(batchMessageDelayMs);
+
+        log.info("-- Exiting {} test --", methodName);
+    }
+
+
     @Test(dataProvider = "batch")
     public void testAnonymousSyncProducerAndConsumer(int batchMessageDelayMs) 
throws Exception {
         log.info("-- Starting {} test --", methodName);
@@ -223,7 +255,7 @@ public void testAnonymousSyncProducerAndConsumer(int 
batchMessageDelayMs) throws
 
     /**
      * Verifies: on 500 server error, broker invalidates session and client 
receives 500 correctly.
-     * 
+     *
      * @throws Exception
      */
     @Test
@@ -254,7 +286,7 @@ public void testAuthenticationFilterNegative() throws 
Exception {
     /**
      * verifies that topicLookup/PartitionMetadataLookup gives 
InternalServerError(500) instead 401(auth_failed) on
      * unknown-exception failure
-     * 
+     *
      * @throws Exception
      */
     @Test
@@ -292,4 +324,4 @@ public void testInternalServerExceptionOnLookup() throws 
Exception {
 
     }
 
-}
\ No newline at end of file
+}
diff --git a/pulsar-broker/src/test/resources/authentication/basic/.htpasswd 
b/pulsar-broker/src/test/resources/authentication/basic/.htpasswd
new file mode 100644
index 000000000..b1a099a5f
--- /dev/null
+++ b/pulsar-broker/src/test/resources/authentication/basic/.htpasswd
@@ -0,0 +1,2 @@
+superUser:mQQQIsyvvKRtU
+superUser2:$apr1$foobarmq$kuSZlLgOITksCkRgl57ie/
diff --git a/pulsar-client/pom.xml b/pulsar-client/pom.xml
index 1f0f4417c..3c26a9e87 100644
--- a/pulsar-client/pom.xml
+++ b/pulsar-client/pom.xml
@@ -69,6 +69,12 @@
       <groupId>org.bouncycastle</groupId>
       <artifactId>bcpkix-jdk15on</artifactId>
     </dependency>
+
+    <dependency>
+      <groupId>com.google.code.gson</groupId>
+      <artifactId>gson</artifactId>
+    </dependency>
+
   </dependencies>
 
   <build>
diff --git 
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationBasic.java
 
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationBasic.java
new file mode 100644
index 000000000..37454d2f0
--- /dev/null
+++ 
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationBasic.java
@@ -0,0 +1,72 @@
+/**
+ * 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.client.impl.auth;
+
+import com.google.gson.Gson;
+import com.google.gson.JsonObject;
+import org.apache.pulsar.client.api.Authentication;
+import org.apache.pulsar.client.api.AuthenticationDataProvider;
+import org.apache.pulsar.client.api.EncodedAuthenticationParameterSupport;
+import org.apache.pulsar.client.api.PulsarClientException;
+
+import java.io.IOException;
+import java.util.Map;
+
+public class AuthenticationBasic implements Authentication, 
EncodedAuthenticationParameterSupport {
+    private String userId;
+    private String password;
+
+    @Override
+    public void close() throws IOException {
+        // noop
+    }
+
+    @Override
+    public String getAuthMethodName() {
+        return "basic";
+    }
+
+    @Override
+    public AuthenticationDataProvider getAuthData() throws 
PulsarClientException {
+        try {
+            return new AuthenticationDataBasic(userId, password);
+        } catch (Exception e) {
+            throw new PulsarClientException(e);
+        }
+    }
+
+    @Override
+    public void configure(Map<String, String> authParams) {
+        configure(new Gson().toJson(authParams));
+    }
+
+    @Override
+    public void configure(String encodedAuthParamString) {
+        JsonObject params = new Gson().fromJson(encodedAuthParamString, 
JsonObject.class);
+        userId = params.get("userId").getAsString();
+        password = params.get("password").getAsString();
+    }
+
+    @Override
+    public void start() throws PulsarClientException {
+        // noop
+    }
+
+}
diff --git 
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationDataBasic.java
 
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationDataBasic.java
new file mode 100644
index 000000000..39b7d5f39
--- /dev/null
+++ 
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationDataBasic.java
@@ -0,0 +1,61 @@
+/**
+ * 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.client.impl.auth;
+
+import org.apache.pulsar.client.api.AuthenticationDataProvider;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+import java.util.Base64;
+
+public class AuthenticationDataBasic implements AuthenticationDataProvider {
+    private final static String HTTP_HEADER_NAME = "Authorization";
+    private String httpAuthToken;
+    private String commandAuthToken;
+
+    public AuthenticationDataBasic(String userId, String password) {
+        httpAuthToken = "Basic " + Base64.getEncoder().encodeToString((userId 
+ ":" + password).getBytes());
+        commandAuthToken = userId+":"+password;
+    }
+
+    @Override
+    public boolean hasDataForHttp() {
+        return true;
+    }
+
+    @Override
+    public Set<Map.Entry<String, String>> getHttpHeaders() {
+        Map<String, String> headers = new HashMap<>();
+        headers.put(HTTP_HEADER_NAME, httpAuthToken);
+        return headers.entrySet();
+    }
+
+    @Override
+    public boolean hasDataFromCommand() {
+        return true;
+    }
+
+    @Override
+    public String getCommandData() {
+        return commandAuthToken;
+    }
+}


 

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


With regards,
Apache Git Services

Reply via email to