Added AAF plugin

Issue-ID: DMAAP-528
Change-Id: Ie24eba58e44b8df727e28003b1d67fddd8ffcd81
Signed-off-by: sunil unnava <sunil.unnava@att.com>

16 files changed, 759 insertions, 15 deletions

diff --git a/src/main/docker/Dockerfile b/src/main/docker/Dockerfile
index 39f997c..102ade0 100644
--- a/src/main/docker/Dockerfile
+++ b/src/main/docker/Dockerfile
@@ -8,6 +8,13 @@ RUN apk add --update unzip wget curl docker jq coreutils
 
 ENV KAFKA_VERSION=$kafka_version SCALA_VERSION=$scala_version
 ADD download-kafka.sh /tmp/download-kafka.sh
+ADD kafka_server_jaas.conf /tmp/kafka_server_jaas.conf
+ADD truststoreONAPall.jks /tmp/truststoreONAPall.jks
+ADD org.onap.dmaap.mr.p12 /tmp/org.onap.dmaap.mr.p12
+ADD keyfilenew /tmp/keyfilenew
+ADD cadi.properties /tmp/cadi.properties
+ADD kafka11aaf-jar-with-dependencies.jar /tmp/kafka11aaf-jar-with-dependencies.jar
+
 RUN chmod a+x /tmp/download-kafka.sh && sync && /tmp/download-kafka.sh && tar xfz /tmp/kafka_${SCALA_VERSION}-${KAFKA_VERSION}.tgz -C /opt && rm /tmp/kafka_${SCALA_VERSION}-${KAFKA_VERSION}.tgz && ln -s /opt/kafka_${SCALA_VERSION}-${KAFKA_VERSION} /opt/kafka
 
 VOLUME ["/kafka"]
diff --git a/src/main/docker/cadi.properties b/src/main/docker/cadi.properties
new file mode 100644
index 0000000..01aae97
--- /dev/null
+++ b/src/main/docker/cadi.properties
@@ -0,0 +1,20 @@
+#aaf_locate_url=https://aaf-onap-test.osaaf.org:8095
+aaf_url=https://AAF_LOCATE_URL/AAF_NS.service:2.1
+aaf_env=DEV
+aaf_lur=org.onap.aaf.cadi.aaf.v2_0.AAFLurPerm
+
+cadi_truststore=/opt/config/truststoreONAPall.jks
+cadi_truststore_password=changeit
+
+cadi_keyfile=/opt/config/keyfilenew
+
+cadi_alias=dmaapmr@mr.dmaap.onap.org
+cadi_keystore=/opt/kafka/org.onap.dmaap.mr.p12
+cadi_keystore_password=Messaging for All
+cadi_x509_issuers=CN=intermediateCA_1, OU=OSAAF, O=ONAP, C=US
+
+
+cadi_loglevel=INFO
+cadi_protocols=TLSv1.1,TLSv1.2
+cadi_latitude=37.78187
+cadi_longitude=-122.26147
\ No newline at end of file
diff --git a/src/main/docker/docker-compose.yml b/src/main/docker/docker-compose.yml
deleted file mode 100644
index 04b82c3..0000000
--- a/src/main/docker/docker-compose.yml
+++ /dev/null
@@ -1,15 +0,0 @@
-version: '2'
-services:
-  zookeeper:
-    image: wurstmeister/zookeeper
-    ports:
-      - "2181:2181"
-  kafka:
-    build: .
-    ports:
-      - "9092"
-    environment:
-      KAFKA_ADVERTISED_HOST_NAME: 192.168.99.100
-      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
-    volumes:
-      - /var/run/docker.sock:/var/run/docker.sock
diff --git a/src/main/docker/kafka_server_jaas.conf b/src/main/docker/kafka_server_jaas.conf
new file mode 100644
index 0000000..163041b
--- /dev/null
+++ b/src/main/docker/kafka_server_jaas.conf
@@ -0,0 +1,5 @@
+KafkaServer {
+  org.onap.dmaap.kafkaAuthorize.PlainLoginModule1 required
+  username="admin"
+  password="admin_secret";
+};
diff --git a/src/main/docker/keyfilenew b/src/main/docker/keyfilenew
new file mode 100644
index 0000000..884375f
--- /dev/null
+++ b/src/main/docker/keyfilenew
@@ -0,0 +1,27 @@
+Riwh4gx5yeqp3KFVdmuREXNlB2ie9JSWKRBR08cNhaubYzsoAlCgOYu8g1OuA735u59jaRwAtLxt
+5m3aMD5MJZ1ItS4x6CeGCKQ0X3F3OzDRsIv-6iDBhlKdOX9pdR8UF7CBqgqbDmvhg3D-h2JcoYJ4
+uzCPI0ZMXeUELkB3l1ZyhsiDrI892AL_VOxQhhsZk1E3P4UFmhfy_579OCVRVhC38xvL0vrtWkHK
+5-1wO3enzrt_p2Jrv-LTgNHTwLF7djyesb55FC9VlTqCrvIomBXvG6NaFuy9_tNJ507ees1_KfTh
+4_BVWfZwoXx8ZXWG9_Pu-S8qKn-f8HtgbJnvAW9wze0H7jpRmOQ1nattTqq7sUTgBT-gzzMsFFH9
+61Mwf_OZc41PneLK9ajy8AzvffPVbW_KNssUC96X6DEkzjrk--fN6uE1VMJVK515smSV0bpcbD6e
+o5GRC2xaa6t3IpZ6Z4f08Dxgob5oyWPKNYKSdcvIgp_HT6oJ7m4TovOQm23ZuuLsGAz9My1pJn42
+fcug_tR2sVSzTYTO9mEAEfRRhPQAWYpAFxclb08Frd-ZOy9V9epsJwLE1tFxjNX31lkFb5G-i0MP
+ZHhtDpIlHM_CvX3tlKrJWMSA91JIfZ0E1mXEkrG9Tzz8jifoijzM_rTvAQf5RQqqAhiuEMSjZeVV
+UoKhEp9duhsJCwNelgpjbAvthYa-InQhC9b4FmMWN0QnhUddb8dw_cNOIfuQu8i38qm9MmkGBSD3
+6dS4Ly6XXqAfz7j7TjrqDJfYWaYRa3OkE2I1jxwo-3IUkKLah9gYKX_FkaNlObHN1c1A5uQ4wJVK
+FAkd98e4vr3UiY6wuKBgKeE-wwU0mUK1lRVmb5fwrsVmCUPOXO8wZZxtjmJddB08jkACyLbHEMg2
+U5fKBpaqq_9DQxnLvd0-ydNcVxYgiTCB9vsmIJ38maLROARmUtfiuuZD-cwOLnDzRkTmARwwxPks
+6ea0cpx-SckhwZHuavq4DLGYbsk-pXToia-M6pPf9rW8qQqeMyUBg4c3--unHBSajT0UxPSbiFrL
+9pxwVeUBulB4j1BtLOPhQaAXHTWpD-85n6ecPEYfpIK73_S7fLBfUD1gyQ6tZj4VYjoSfGKAFStE
+zCUMvryARBNVFJ0ENq-xKyst_M4V3WjcIeiLW3LmjByk-aymys-e8mUL_tcn_MO9pCktAr1xu5Yx
+wBcBOrFlB2UP8Im7vBHbGgf77ssqyxy5_cJhaO9MBKUx5KZQw9eE9ePf6UvELTev7Urhla4QKUm9
+AMemzy2RvC2ghZeh7fzBahbZpRM3vDDm4IhbcZavA2d2DEgq8c0AUhlPYE-LCv2BOKBeUEkGULxU
+29uIc8LkcLHh37WHmJOjVbH9gB9enHH0sBf9cnv7A70R1evSWeHn0ty9vVXPOLODSQGqbB40qAhQ
+MEsRt-13WUAlHjosA7yj2zHTLMeuSqqPuPeiyGPtblkWUC-gpEJxgK8hTb1LzoZVZeteqgdMKlde
+Q0gmI_0CX5RtCjITSlHaKxzw6ly9qqv52GZVpAYlu2SWeFdlCg9txh2ke0x3rTMKsM8i0ccCdmLq
+E60akH2bPa4vB7zRiu3im-IVli9V8zz8U2roQrfN08IJCAatSQRVfUiyAAJkOEcghuHmaErA-kD8
+fu0sWuAHsEgKBtfaeOu5OFeyeLmNRiPKpVotMyDHrEjjBW-TVTppWwgN5Utmx80RghSmzwUjglyG
+3aaM3iJqp8xvgtlLtoJkq2A8rMbw0eAQ7I33hAn-jfBkmjsVkzsVgffe5xqGA1DDYm1lTkv4OjFX
+_tTzYfN2V1BtYNUN_edhQRMsNh5-mpZwOeb5JpdJQFZUXaFtwDedx_sqe54NEJ4jV7w4up7H0MXp
+WTazMQmwRzsLTs1U8zeJ0Ib0LAb4EsX6DML3Ue87nmYCj450KE9DM0tYGWn13jiGWoDEhW4noi2X
+gkcjwIcM-87wwvsb-rMIOdo2DXQee8zKzB51N4YAn4VBUfjXVMhjjSwg40yHlzKQE0hAOuJN
\ No newline at end of file
diff --git a/src/main/docker/org.onap.dmaap.mr.p12 b/src/main/docker/org.onap.dmaap.mr.p12
new file mode 100644
index 0000000..79549ed
--- /dev/null
+++ b/src/main/docker/org.onap.dmaap.mr.p12
diff --git a/src/main/docker/start-kafka.sh b/src/main/docker/start-kafka.sh
index 90eaaea..2a89b03 100644
--- a/src/main/docker/start-kafka.sh
+++ b/src/main/docker/start-kafka.sh
@@ -134,5 +134,17 @@ done
 if [[ -n "$CUSTOM_INIT_SCRIPT" ]] ; then
   eval $CUSTOM_INIT_SCRIPT
 fi
+cp /tmp/kafka11aaf-jar-with-dependencies.jar $KAFKA_HOME/libs
+cp /tmp/keyfilenew  $KAFKA_HOME/config
+cp /tmp/truststoreONAPall.jks $KAFKA_HOME/config
+cp /tmp/org.onap.dmaap.mr.p12 $KAFKA_HOME/config
+cp /tmp/kafka_server_jaas.conf $KAFKA_HOME/config
+cp /tmp/cadi.properties $KAFKA_HOME/config
+export KAFKA_OPTS="-Djava.security.auth.login.config=$KAFKA_HOME/config/kafka_server_jaas.conf"
+
+echo "authorizer.class.name=org.onap.dmaap.kafkaAuthorize.KafkaCustomAuthorizer" >> $KAFKA_HOME/config/server.properties
+echo "security.inter.broker.protocol=SASL_PLAINTEXT" >> $KAFKA_HOME/config/server.properties
+echo "sasl.enabled.mechanisms=PLAIN" >> $KAFKA_HOME/config/server.properties
+echo "sasl.mechanism.inter.broker.protocol=PLAIN" >> $KAFKA_HOME/config/server.properties
 
 exec $KAFKA_HOME/bin/kafka-server-start.sh $KAFKA_HOME/config/server.properties
diff --git a/src/main/docker/truststoreONAPall.jks b/src/main/docker/truststoreONAPall.jks
new file mode 100644
index 0000000..2da1dcc
--- /dev/null
+++ b/src/main/docker/truststoreONAPall.jks
diff --git a/src/main/java/org/onap/dmaap/commonauth/kafka/base/authorization/AuthorizationProvider.java b/src/main/java/org/onap/dmaap/commonauth/kafka/base/authorization/AuthorizationProvider.java
new file mode 100644
index 0000000..da96929
--- /dev/null
+++ b/src/main/java/org/onap/dmaap/commonauth/kafka/base/authorization/AuthorizationProvider.java
@@ -0,0 +1,30 @@
+/*******************************************************************************
+ *  ============LICENSE_START=======================================================
+ *  org.onap.dmaap
+ *  ================================================================================
+ *  Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ *  ================================================================================
+ *  Licensed 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.
+ *  ============LICENSE_END=========================================================
+ *  
+ *  
+ *******************************************************************************/
+package org.onap.dmaap.commonauth.kafka.base.authorization;
+
+public interface AuthorizationProvider {
+
+	public boolean hasPermission(String userId, String permission, String instance, String action);
+
+	public String getId();
+
+	public String authenticate(String userId, String password) throws Exception;
+}
diff --git a/src/main/java/org/onap/dmaap/commonauth/kafka/base/authorization/AuthorizationProviderFactory.java b/src/main/java/org/onap/dmaap/commonauth/kafka/base/authorization/AuthorizationProviderFactory.java
new file mode 100644
index 0000000..6b872af
--- /dev/null
+++ b/src/main/java/org/onap/dmaap/commonauth/kafka/base/authorization/AuthorizationProviderFactory.java
@@ -0,0 +1,51 @@
+/*******************************************************************************
+ *  ============LICENSE_START=======================================================
+ *  org.onap.dmaap
+ *  ================================================================================
+ *  Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ *  ================================================================================
+ *  Licensed 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.
+ *  ============LICENSE_END=========================================================
+ *  
+ *  
+ *******************************************************************************/
+package org.onap.dmaap.commonauth.kafka.base.authorization;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.ServiceLoader;
+
+public class AuthorizationProviderFactory<K, V> {
+	private static final Map<String, AuthorizationProvider> AUTHORIZATION_PROVIDER_MAP = new HashMap<String, AuthorizationProvider>();
+	private static final AuthorizationProviderFactory AUTHORIZATION_PROVIDER_FACTORY = new AuthorizationProviderFactory();
+
+	private AuthorizationProviderFactory() {
+		try {
+			ServiceLoader<AuthorizationProvider> serviceLoader = ServiceLoader.load(AuthorizationProvider.class);
+			for (AuthorizationProvider authzProvider : serviceLoader) {
+				AUTHORIZATION_PROVIDER_MAP.put(authzProvider.getId(), authzProvider);
+
+			}
+		} catch (Exception ee) {
+			System.out.println(ee);
+			System.exit(0);
+		}
+	}
+
+	public static AuthorizationProviderFactory getProviderFactory() {
+		return AUTHORIZATION_PROVIDER_FACTORY;
+	}
+
+	public AuthorizationProvider getProvider() {
+		return AUTHORIZATION_PROVIDER_MAP.get(System.getProperty("kafka.authorization.provider", "CADI_AAF_PROVIDER"));
+	}
+}
diff --git a/src/main/java/org/onap/dmaap/commonauth/kafka/base/authorization/Cadi3AAFProvider.java b/src/main/java/org/onap/dmaap/commonauth/kafka/base/authorization/Cadi3AAFProvider.java
new file mode 100644
index 0000000..60c1868
--- /dev/null
+++ b/src/main/java/org/onap/dmaap/commonauth/kafka/base/authorization/Cadi3AAFProvider.java
@@ -0,0 +1,158 @@
+/*******************************************************************************
+ *  ============LICENSE_START=======================================================
+ *  org.onap.dmaap
+ *  ================================================================================
+ *  Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ *  ================================================================================
+ *  Licensed 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.
+ *  ============LICENSE_END=========================================================
+ *  
+ *  
+ *******************************************************************************/
+package org.onap.dmaap.commonauth.kafka.base.authorization;
+
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.util.Properties;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.onap.aaf.cadi.CadiException;
+import org.onap.aaf.cadi.PropAccess;
+import org.onap.aaf.cadi.aaf.AAFPermission;
+import org.onap.aaf.cadi.aaf.v2_0.AAFAuthn;
+import org.onap.aaf.cadi.aaf.v2_0.AAFCon;
+import org.onap.aaf.cadi.aaf.v2_0.AAFConHttp;
+import org.onap.aaf.cadi.aaf.v2_0.AbsAAFLur;
+import org.onap.aaf.cadi.principal.UnAuthPrincipal;
+
+public class Cadi3AAFProvider implements AuthorizationProvider {
+
+	private static PropAccess access;
+	private static AAFCon<?> aafcon;
+	private static final String CADI_PROPERTIES = "/opt/kafka/config/cadi.properties";
+	private static final String AAF_LOCATOR_ENV = "aaf_locate_url";
+	private static final String MR_NAMESPACE = "	org.onap.dmaap.mr";
+
+	public static AAFAuthn<?> getAafAuthn() throws CadiException {
+		if (aafAuthn == null) {
+			throw new CadiException("Cadi is uninitialized in Cadi3AAFProvider.getAafAuthn()");
+		}
+		return aafAuthn;
+	}
+
+	private static AAFAuthn<?> aafAuthn;
+	private static AbsAAFLur<AAFPermission> aafLur;
+
+	private static boolean props_ok = false;
+
+	private static final Logger logger = LoggerFactory.getLogger(Cadi3AAFProvider.class);
+
+	public Cadi3AAFProvider() {
+		setup();
+	}
+
+	private synchronized void setup() {
+		if (access == null) {
+
+			Properties props = new Properties();
+			FileInputStream fis = null;
+			try {
+				if (System.getProperty("CADI_PROPERTIES") != null) {
+					fis = new FileInputStream(System.getProperty("CADI_PROPERTIES"));
+				} else {
+					fis = new FileInputStream(CADI_PROPERTIES);
+				}
+				try {
+					props.load(fis);
+					if (System.getenv(AAF_LOCATOR_ENV) != null)
+						props.setProperty(AAF_LOCATOR_ENV, System.getenv(AAF_LOCATOR_ENV));
+					access = new PropAccess(props);
+				} finally {
+					fis.close();
+				}
+			} catch (IOException e) {
+				logger.error("Unable to load " + CADI_PROPERTIES);
+				logger.error("Error", e);
+			}
+
+			props_ok = true;
+			if (props_ok == false) {
+				return;
+			}
+		}
+
+		if (aafAuthn == null) {
+			try {
+				aafcon = new AAFConHttp(access);
+				aafAuthn = aafcon.newAuthn();
+				aafLur = aafcon.newLur(aafAuthn);
+			} catch (final Exception e) {
+				aafAuthn = null;
+				if (access != null)
+					access.log(e, "Failed to initialize AAF");
+				props_ok = false;
+			}
+		}
+
+	}
+
+	/**
+	 * Checks if a user has a particular permission
+	 * <p/>
+	 * Returns true if the permission in found
+	 */
+	public boolean hasPermission(String userId, String permission, String instance, String action) {
+		boolean hasPermission = false;
+		try {
+			logger.info("^ Event at hasPermission to validate userid " + userId + " with " + permission + " " + instance
+					+ " " + action);
+			// AAF Style permissions are in the form
+			// Resource Name, Resource Type, Action
+			if (userId.equals("admin")) {
+				hasPermission = true;
+				return hasPermission;
+			}
+			AAFPermission perm = new AAFPermission(MR_NAMESPACE, permission, instance, action);
+			if (aafLur != null) {
+				hasPermission = aafLur.fish(new UnAuthPrincipal(userId), perm);
+				logger.trace("Permission: " + perm.getKey() + " for user :" + userId + " found: " + hasPermission);
+			} else {
+				logger.error("AAF client not initialized. Not able to find permissions.");
+			}
+		} catch (Exception e) {
+			logger.error("AAF client not initialized", e);
+		}
+		return hasPermission;
+	}
+
+	public String getId() {
+		return "CADI_AAF_PROVIDER";
+	}
+
+	public String authenticate(String userId, String password) throws Exception {
+		logger.info("^Event received  with   username " + userId);
+		if (userId.equals("admin")) {
+			logger.info("User Admin by passess AAF call ....");
+			return null;
+		}
+		String aafResponse = aafAuthn.validate(userId, password);
+		logger.info("aafResponse=" + aafResponse + " for " + userId);
+
+		if (aafResponse != null) {
+			logger.error("Authentication failed for user ." + userId);
+		}
+		return aafResponse;
+	}
+
+}
diff --git a/src/main/java/org/onap/dmaap/kafkaAuthorize/KafkaCustomAuthorizer.java b/src/main/java/org/onap/dmaap/kafkaAuthorize/KafkaCustomAuthorizer.java
new file mode 100644
index 0000000..cb33e29
--- /dev/null
+++ b/src/main/java/org/onap/dmaap/kafkaAuthorize/KafkaCustomAuthorizer.java
@@ -0,0 +1,153 @@
+/*******************************************************************************
+ *  ============LICENSE_START=======================================================
+ *  org.onap.dmaap
+ *  ================================================================================
+ *  Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ *  ================================================================================
+ *  Licensed 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.
+ *  ============LICENSE_END=========================================================
+ *  
+ *  
+ *******************************************************************************/
+package org.onap.dmaap.kafkaAuthorize;
+
+import java.util.Map;
+
+import org.apache.kafka.common.security.auth.KafkaPrincipal;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.onap.aaf.cadi.PropAccess;
+import org.onap.dmaap.commonauth.kafka.base.authorization.AuthorizationProviderFactory;
+import kafka.network.RequestChannel.Session;
+import kafka.security.auth.Acl;
+import kafka.security.auth.Authorizer;
+import kafka.security.auth.Operation;
+import kafka.security.auth.Resource;
+import scala.collection.immutable.Set;
+
+/**
+ * A trivial Kafka Authorizer for use with SSL and AAF
+ * Authentication/Authorization.
+ * 
+ */
+public class KafkaCustomAuthorizer implements Authorizer {
+	private PropAccess access;
+	private static final Logger logger = LoggerFactory.getLogger(KafkaCustomAuthorizer.class);
+
+	// I'm assuming this is called BEFORE any usage...
+	@Override
+	public void configure(final Map<String, ?> arg0) {
+		// TODO Auto-generate method stub
+	}
+
+	@Override
+	public void addAcls(final Set<Acl> arg0, final Resource arg1) {
+		// TODO Auto-generated method stub
+
+	}
+
+	@Override
+	public boolean authorize(final Session arg0, final Operation arg1, final Resource arg2) {
+		if (arg0.principal() == null) {
+			return false;
+		}
+
+		String fullName = arg0.principal().getName();
+		fullName = fullName != null ? fullName.trim() : fullName;
+		String topicName = null;
+		String namspace = null;
+		String ins = null;
+		String type = null;
+		String action = null;
+
+		String kafkaactivity = arg1.name();
+
+		if (kafkaactivity.equals("Read")) {
+			action = "sub";
+		} else if (kafkaactivity.equals("Write")) {
+			action = "pub";
+		} else if (kafkaactivity.equals("Describe")) {
+			return true;
+		}
+		if (arg2.resourceType().name().equals("Topic")) {
+			topicName = arg2.name();
+		} else {
+			return true;
+		}
+
+		try {
+
+			if (null != topicName && topicName.indexOf(".") > 0) {
+				namspace = topicName.substring(0, topicName.lastIndexOf("."));
+				ins = namspace + ".topic";
+				type = ":topic." + topicName;
+				logger.info("^Event Received for topic " + topicName + " , User " + fullName + " , action = " + action);
+			}
+
+			if (fullName.equals("admin")) {
+				return true;
+			}
+
+			if (null != topicName) {
+				boolean hasResp = AuthorizationProviderFactory.getProviderFactory().getProvider()
+						.hasPermission(fullName, ins, type, action);
+				if (hasResp) {
+					logger.info("Successful Authorization for " + fullName + " on " + topicName + " for " + ins + "|"
+							+ type + "|" + action);
+				}
+				if (!hasResp) {
+					logger.info(fullName + " is not allowed in " + ins + "|" + type + "|" + action);
+					throw new Exception(fullName + " is not allowed in " + ins + "|" + type + "|" + action);
+				}
+			}
+		} catch (final Exception e) {
+			return false;
+		}
+		return true;
+	}
+
+	@Override
+	public void close() {
+		// TODO Auto-generated method stub
+
+	}
+
+	@Override
+	public scala.collection.immutable.Map<Resource, Set<Acl>> getAcls() {
+		// TODO Auto-generated method stub
+		return null;
+	}
+
+	@Override
+	public scala.collection.immutable.Map<Resource, Set<Acl>> getAcls(final KafkaPrincipal arg0) {
+		// TODO Auto-generated method stub
+		return null;
+	}
+
+	@Override
+	public boolean removeAcls(final Resource arg0) {
+		// TODO Auto-generated method stub
+		return false;
+	}
+
+	@Override
+	public boolean removeAcls(final Set<Acl> arg0, final Resource arg1) {
+		// TODO Auto-generated method stub
+		return false;
+	}
+
+	public Set<Acl> getAcls(Resource arg0) {
+		// TODO Auto-generated method stub
+		return null;
+	}
+}
diff --git a/src/main/java/org/onap/dmaap/kafkaAuthorize/PlainLoginModule1.java b/src/main/java/org/onap/dmaap/kafkaAuthorize/PlainLoginModule1.java
new file mode 100644
index 0000000..dd21682
--- /dev/null
+++ b/src/main/java/org/onap/dmaap/kafkaAuthorize/PlainLoginModule1.java
@@ -0,0 +1,70 @@
+/*******************************************************************************
+ *  ============LICENSE_START=======================================================
+ *  org.onap.dmaap
+ *  ================================================================================
+ *  Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ *  ================================================================================
+ *  Licensed 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.
+ *  ============LICENSE_END=========================================================
+ *  
+ *  
+ *******************************************************************************/
+package org.onap.dmaap.kafkaAuthorize;
+
+import java.util.Map;
+
+import javax.security.auth.Subject;
+import javax.security.auth.callback.CallbackHandler;
+import javax.security.auth.login.LoginException;
+import javax.security.auth.spi.LoginModule;
+
+public class PlainLoginModule1 implements LoginModule {
+
+	private static final String USERNAME_CONFIG = "username";
+	private static final String PASSWORD_CONFIG = "password";
+
+	static {
+		PlainSaslServerProvider1.initialize();
+	}
+
+	@Override
+	public void initialize(Subject subject, CallbackHandler callbackHandler, Map<String, ?> sharedState,
+			Map<String, ?> options) {
+		String username = (String) options.get(USERNAME_CONFIG);
+		if (username != null)
+			subject.getPublicCredentials().add(username);
+		String password = (String) options.get(PASSWORD_CONFIG);
+		if (password != null)
+			subject.getPrivateCredentials().add(password);
+
+	}
+
+	@Override
+	public boolean login() throws LoginException {
+		return true;
+	}
+
+	@Override
+	public boolean logout() throws LoginException {
+		return true;
+	}
+
+	@Override
+	public boolean commit() throws LoginException {
+		return true;
+	}
+
+	@Override
+	public boolean abort() throws LoginException {
+		return false;
+	}
+}
diff --git a/src/main/java/org/onap/dmaap/kafkaAuthorize/PlainSaslServer1.java b/src/main/java/org/onap/dmaap/kafkaAuthorize/PlainSaslServer1.java
new file mode 100644
index 0000000..f28671b
--- /dev/null
+++ b/src/main/java/org/onap/dmaap/kafkaAuthorize/PlainSaslServer1.java
@@ -0,0 +1,185 @@
+/*******************************************************************************
+ *  ============LICENSE_START=======================================================
+ *  org.onap.dmaap
+ *  ================================================================================
+ *  Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ *  ================================================================================
+ *  Licensed 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.
+ *  ============LICENSE_END=========================================================
+ *  
+ *  
+ *******************************************************************************/
+package org.onap.dmaap.kafkaAuthorize;
+
+import java.io.UnsupportedEncodingException;
+import java.util.Arrays;
+import java.util.Map;
+
+import javax.security.auth.callback.CallbackHandler;
+import javax.security.sasl.Sasl;
+import javax.security.sasl.SaslException;
+import javax.security.sasl.SaslServer;
+import javax.security.sasl.SaslServerFactory;
+
+import org.apache.kafka.common.security.JaasContext;
+import org.apache.kafka.common.security.authenticator.SaslServerCallbackHandler;
+
+import org.onap.dmaap.commonauth.kafka.base.authorization.AuthorizationProviderFactory;
+
+/**
+ * Simple SaslServer implementation for SASL/PLAIN. In order to make this
+ * implementation fully pluggable, authentication of username/password is fully
+ * contained within the server implementation.
+ * <p>
+ * Valid users with passwords are specified in the Jaas configuration file. Each
+ * user is specified with user_<username> as key and <password> as value. This
+ * is consistent with Zookeeper Digest-MD5 implementation.
+ * <p>
+ * To avoid storing clear passwords on disk or to integrate with external
+ * authentication servers in production systems, this module can be replaced
+ * with a different implementation.
+ *
+ */
+public class PlainSaslServer1 implements SaslServer {
+
+	public static final String PLAIN_MECHANISM = "PLAIN";
+
+	private final JaasContext jaasContext;
+
+	private boolean complete;
+	private String authorizationID;
+
+	public PlainSaslServer1(JaasContext jaasContext) {
+		this.jaasContext = jaasContext;
+	}
+
+	@Override
+	public byte[] evaluateResponse(byte[] response) throws SaslException {
+		/*
+		 * Message format (from https://tools.ietf.org/html/rfc4616):
+		 *
+		 * message = [authzid] UTF8NUL authcid UTF8NUL passwd authcid = 1*SAFE ;
+		 * MUST accept up to 255 octets authzid = 1*SAFE ; MUST accept up to 255
+		 * octets passwd = 1*SAFE ; MUST accept up to 255 octets UTF8NUL = %x00
+		 * ; UTF-8 encoded NUL character
+		 *
+		 * SAFE = UTF1 / UTF2 / UTF3 / UTF4 ;; any UTF-8 encoded Unicode
+		 * character except NUL
+		 */
+
+		String[] tokens;
+		try {
+			tokens = new String(response, "UTF-8").split("\u0000");
+		} catch (UnsupportedEncodingException e) {
+			throw new SaslException("UTF-8 encoding not supported", e);
+		}
+		if (tokens.length != 3)
+			throw new SaslException("Invalid SASL/PLAIN response: expected 3 tokens, got " + tokens.length);
+		authorizationID = tokens[0];
+		String username = tokens[1];
+		String password = tokens[2];
+
+		if (username.isEmpty()) {
+			throw new SaslException("Authentication failed: username not specified");
+		}
+		if (password.isEmpty()) {
+			throw new SaslException("Authentication failed: password not specified");
+		}
+		if (authorizationID.isEmpty())
+			authorizationID = username;
+
+		String aafResponse = "Not Verified";
+		try {
+			aafResponse = AuthorizationProviderFactory.getProviderFactory().getProvider().authenticate(username,
+					password);
+		} catch (Exception e) {
+		}
+
+		if (null != aafResponse) {
+			throw new SaslException("Authentication failed: " + aafResponse + " User " + username);
+		}
+
+		complete = true;
+		return new byte[0];
+	}
+
+	@Override
+	public String getAuthorizationID() {
+		if (!complete)
+			throw new IllegalStateException("Authentication exchange has not completed");
+		return authorizationID;
+	}
+
+	@Override
+	public String getMechanismName() {
+		return PLAIN_MECHANISM;
+	}
+
+	@Override
+	public Object getNegotiatedProperty(String propName) {
+		if (!complete)
+			throw new IllegalStateException("Authentication exchange has not completed");
+		return null;
+	}
+
+	@Override
+	public boolean isComplete() {
+		return complete;
+	}
+
+	@Override
+	public byte[] unwrap(byte[] incoming, int offset, int len) throws SaslException {
+		if (!complete)
+			throw new IllegalStateException("Authentication exchange has not completed");
+		return Arrays.copyOfRange(incoming, offset, offset + len);
+	}
+
+	@Override
+	public byte[] wrap(byte[] outgoing, int offset, int len) throws SaslException {
+		if (!complete)
+			throw new IllegalStateException("Authentication exchange has not completed");
+		return Arrays.copyOfRange(outgoing, offset, offset + len);
+	}
+
+	@Override
+	public void dispose() throws SaslException {
+	}
+
+	public static class PlainSaslServerFactory1 implements SaslServerFactory {
+
+		@Override
+		public SaslServer createSaslServer(String mechanism, String protocol, String serverName, Map<String, ?> props,
+				CallbackHandler cbh) throws SaslException {
+
+			if (!PLAIN_MECHANISM.equals(mechanism))
+				throw new SaslException(
+						String.format("Mechanism \'%s\' is not supported. Only PLAIN is supported.", mechanism));
+
+			if (!(cbh instanceof SaslServerCallbackHandler))
+				throw new SaslException(
+						"CallbackHandler must be of type SaslServerCallbackHandler, but it is: " + cbh.getClass());
+
+			return new PlainSaslServer1(((SaslServerCallbackHandler) cbh).jaasContext());
+		}
+
+		@Override
+		public String[] getMechanismNames(Map<String, ?> props) {
+			if (props == null)
+				return new String[] { PLAIN_MECHANISM };
+			String noPlainText = (String) props.get(Sasl.POLICY_NOPLAINTEXT);
+			if ("true".equals(noPlainText))
+				return new String[] {};
+			else
+				return new String[] { PLAIN_MECHANISM };
+		}
+	}
+}
diff --git a/src/main/java/org/onap/dmaap/kafkaAuthorize/PlainSaslServerProvider1.java b/src/main/java/org/onap/dmaap/kafkaAuthorize/PlainSaslServerProvider1.java
new file mode 100644
index 0000000..16a11f4
--- /dev/null
+++ b/src/main/java/org/onap/dmaap/kafkaAuthorize/PlainSaslServerProvider1.java
@@ -0,0 +1,40 @@
+/*******************************************************************************
+ *  ============LICENSE_START=======================================================
+ *  org.onap.dmaap
+ *  ================================================================================
+ *  Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ *  ================================================================================
+ *  Licensed 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.
+ *  ============LICENSE_END=========================================================
+ *  
+ *  
+ *******************************************************************************/
+package org.onap.dmaap.kafkaAuthorize;
+
+import java.security.Provider;
+import java.security.Security;
+
+import org.onap.dmaap.kafkaAuthorize.PlainSaslServer1.PlainSaslServerFactory1;
+
+public class PlainSaslServerProvider1 extends Provider {
+
+	private static final long serialVersionUID = 1L;
+
+	protected PlainSaslServerProvider1() {
+		super("Simple SASL/PLAIN Server Provider", 1.0, "Simple SASL/PLAIN Server Provider for Kafka");
+		super.put("SaslServerFactory." + PlainSaslServer1.PLAIN_MECHANISM, PlainSaslServerFactory1.class.getName());
+	}
+
+	public static void initialize() {
+		Security.addProvider(new PlainSaslServerProvider1());
+	}
+}
diff --git a/src/main/resources/META-INF/services/org.onap.dmaap.commonauth.kafka.base.authorization.AuthorizationProvider b/src/main/resources/META-INF/services/org.onap.dmaap.commonauth.kafka.base.authorization.AuthorizationProvider
new file mode 100644
index 0000000..0388ce7
--- /dev/null
+++ b/src/main/resources/META-INF/services/org.onap.dmaap.commonauth.kafka.base.authorization.AuthorizationProvider
@@ -0,0 +1 @@
+org.onap.dmaap.commonauth.kafka.base.authorization.Cadi3AAFProvider
\ No newline at end of file