Added AAF plugin

Issue-ID: DMAAP-528 Change-Id: Ie24eba58e44b8df727e28003b1d67fddd8ffcd81 Signed-off-by: sunil unnava <sunil.unnava@att.com>
author: sunil unnava <sunil.unnava@att.com> 2018-09-19 18:08:36 -0400
committer: sunil unnava <sunil.unnava@att.com> 2018-09-19 18:13:20 -0400
commit: b01a7330883cbd5bce618ea44ea0f86ce6332729 (patch)
tree: 6adeee94a55c91cdfceefa6e190ce31697e923c9
parent: ba5557be62af8b3c4f91268a169447adeca85dbc (diff)
30 files changed, 1349 insertions, 22 deletions
diff --git a/pom.xml b/pom.xml
index 4b03b49..de67dac 100644
--- a/pom.xml
+++ b/pom.xml
@@ -1,4 +1,3 @@
-<?xml version="1.0"?>
 <!-- ============LICENSE_START======================================================= 
 	org.onap.dmaap ================================================================================ 
 	Copyright © 2017 AT&T Intellectual Property. All rights reserved. ================================================================================ 
@@ -10,7 +9,8 @@
 	OF ANY KIND, either express or implied. See the License for the specific 
 	language governing permissions and limitations under the License. ============LICENSE_END========================================================= 
 	ECOMP is a trademark and service mark of AT&T Intellectual Property. -->
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
 	<modelVersion>4.0.0</modelVersion>
 
 	<parent>
@@ -21,7 +21,7 @@
 
 	<groupId>org.onap.dmaap.kafka</groupId>
 	<artifactId>kafka11aaf</artifactId>
-	<version>1.0.0-SNAPSHOT</version>
+	<version>0.0.2-SNAPSHOT</version>
 	<name>kafka11aaf</name>
 	<licenses>
 		<license>
@@ -32,7 +32,7 @@
 	<developers>
 		<developer>
 			<name>Sunil Unnava</name>
-			<email/>
+			<email></email>
 			<organization>ATT</organization>
 			<organizationUrl>www.att.com</organizationUrl>
 		</developer>
@@ -42,6 +42,28 @@
 		<plugins>
 			<plugin>
 				<groupId>org.apache.maven.plugins</groupId>
+				<artifactId>maven-surefire-plugin</artifactId>
+				<version>2.12.4</version>
+				<configuration>
+					<excludes>
+						<!-- exclude until junits updated <exclude>**/DME2*.java</exclude> -->
+					</excludes>
+					<!-- <skipTests>true</skipTests> -->
+				</configuration>
+			</plugin>
+			<plugin>
+				<groupId>org.codehaus.mojo</groupId>
+				<artifactId>cobertura-maven-plugin</artifactId>
+				<version>2.7</version>
+				<configuration>
+					<formats>
+						<format>html</format>
+						<format>xml</format>
+					</formats>
+				</configuration>
+			</plugin>
+			<plugin>
+				<groupId>org.apache.maven.plugins</groupId>
 				<artifactId>maven-site-plugin</artifactId>
 				<version>3.6</version>
 				<dependencies>
@@ -201,12 +223,36 @@
 				</executions>
 			</plugin>
 
+			<plugin>
+				<groupId>org.apache.maven.plugins</groupId>
+				<artifactId>maven-assembly-plugin</artifactId>
+				<version>2.4.1</version>
+				<configuration>
+					<!-- get all project dependencies -->
+					<descriptorRefs>
+						<descriptorRef>jar-with-dependencies</descriptorRef>
+					</descriptorRefs>
+					<!-- MainClass in mainfest make a executable jar -->
+                 <finalName>kafka11aaf</finalName>
+				</configuration>
+				<executions>
+					<execution>
+						<id>make-assembly</id>
+						<!-- bind to the packaging phase -->
+						<phase>package</phase>
+						<goals>
+							<goal>single</goal>
+						</goals>
+					</execution>
+				</executions>
+			</plugin>
+
 		</plugins>
 	</build>
 
 	<properties>
 		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-		<KafkaImg>0.0.1</KafkaImg>
+		<KafkaImg>0.0.2</KafkaImg>
 		<sitePath>/content/sites/site/org/onap/dmaap/kafka0111/${project.artifactId}/${project.version}</sitePath>
 		<skip.docker.build>true</skip.docker.build>
 		<skip.docker.push>true</skip.docker.push>
@@ -223,6 +269,53 @@
 	</distributionManagement>
 
 	<dependencies>
+		<dependency>
+			<groupId>org.onap.aaf.authz</groupId>
+			<artifactId>aaf-cadi-aaf</artifactId>
+			<version>2.1.2-SNAPSHOT</version>
+		</dependency>
+		<dependency>
+			<groupId>org.slf4j</groupId>
+			<artifactId>slf4j-api</artifactId>
+			<version>1.7.2</version>
+		</dependency>
+		<dependency>
+			<groupId>org.slf4j</groupId>
+			<artifactId>slf4j-simple</artifactId>
+			<version>1.7.2</version>
+			<scope>runtime</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.apache.kafka</groupId>
+			<artifactId>kafka_2.11</artifactId>
+			<version>0.11.0.3</version>
+		</dependency>
+		<dependency>
+			<groupId>org.powermock</groupId>
+			<artifactId>powermock-api-mockito</artifactId>
+			<version>1.6.4</version>
+			<scope>test</scope>
+		</dependency>
+
+		<dependency>
+			<groupId>org.powermock</groupId>
+			<artifactId>powermock-module-junit4</artifactId>
+			<version>1.6.4</version>
+			<scope>test</scope>
+		</dependency>
+
+		<dependency>
+			<groupId>org.powermock</groupId>
+			<artifactId>powermock-module-junit4-rule</artifactId>
+			<version>1.6.4</version>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.mockito</groupId>
+			<artifactId>mockito-core</artifactId>
+			<version>1.10.19</version>
+			<scope>test</scope>
+		</dependency>
 	</dependencies>
 
 	<profiles>
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
diff --git a/src/test/java/org/onap/dmaap/commonauth/kafka/base/authorization/AuthorizationProviderFactoryTest.java b/src/test/java/org/onap/dmaap/commonauth/kafka/base/authorization/AuthorizationProviderFactoryTest.java
new file mode 100644
index 0000000..4ac81f3
--- /dev/null
+++ b/src/test/java/org/onap/dmaap/commonauth/kafka/base/authorization/AuthorizationProviderFactoryTest.java
@@ -0,0 +1,36 @@
+/*******************************************************************************
+ *  ============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 org.junit.Test;
+import org.junit.runner.RunWith;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+@RunWith(PowerMockRunner.class)
+public class AuthorizationProviderFactoryTest {
+
+	@Test
+	public void testFactory() {
+		AuthorizationProviderFactory.getProviderFactory().getProvider();
+
+	}
+
+}
diff --git a/src/test/java/org/onap/dmaap/commonauth/kafka/base/authorization/Cadi3AAFProviderTest.java b/src/test/java/org/onap/dmaap/commonauth/kafka/base/authorization/Cadi3AAFProviderTest.java
new file mode 100644
index 0000000..fbe1e59
--- /dev/null
+++ b/src/test/java/org/onap/dmaap/commonauth/kafka/base/authorization/Cadi3AAFProviderTest.java
@@ -0,0 +1,76 @@
+/*******************************************************************************
+ *  ============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 static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.mockito.Mockito.when;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+import org.onap.aaf.cadi.PropAccess;
+import org.onap.aaf.cadi.aaf.v2_0.AAFAuthn;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+@RunWith(PowerMockRunner.class)
+public class Cadi3AAFProviderTest {
+
+	public Cadi3AAFProvider cadi3AAFProvider;
+
+	@Mock
+	private static AAFAuthn<?> aafAuthn;
+
+	@Mock
+	private static PropAccess access;
+
+	@Before
+	public void setUp() throws Exception {
+		MockitoAnnotations.initMocks(this);
+	}
+
+	@Test
+	public void testHasPermission() {
+		System.setProperty("CADI_PROPERTIES", "src/test/resources/cadi.properties");
+		cadi3AAFProvider = new Cadi3AAFProvider();
+		assertFalse(cadi3AAFProvider.hasPermission("userID", "permission", "instance", "action"));
+	}
+
+	@Test(expected = NullPointerException.class)
+	public void tesAuthenticate() throws Exception {
+		System.setProperty("CADI_PROPERTIES", "src/test/resources/cadi.properties");
+		cadi3AAFProvider = new Cadi3AAFProvider();
+		when(aafAuthn.validate("userId", "password")).thenReturn("valid");
+		assertEquals(cadi3AAFProvider.authenticate("userId", "password"), "valid");
+	}
+
+	@Test
+	public void tesAuthenticateadmin() throws Exception {
+		System.setProperty("CADI_PROPERTIES", "src/test/resources/cadi.properties");
+		cadi3AAFProvider = new Cadi3AAFProvider();
+		when(aafAuthn.validate("admin", "password")).thenReturn("valid");
+		assertNull(cadi3AAFProvider.authenticate("admin", "password"));
+	}
+
+}
diff --git a/src/test/java/org/onap/dmaap/commonauth/kafka/base/authorization/JUnitTestSuite.java b/src/test/java/org/onap/dmaap/commonauth/kafka/base/authorization/JUnitTestSuite.java
new file mode 100644
index 0000000..e9b52d1
--- /dev/null
+++ b/src/test/java/org/onap/dmaap/commonauth/kafka/base/authorization/JUnitTestSuite.java
@@ -0,0 +1,41 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * ONAP Policy Engine
+ * ================================================================================
+ * Copyright (C) 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 junit.framework.TestSuite;
+import org.junit.runner.RunWith;
+import org.junit.runners.Suite;
+import org.junit.runners.Suite.SuiteClasses;
+import org.apache.log4j.Logger;
+
+@RunWith(Suite.class)
+@SuiteClasses({ AuthorizationProviderFactoryTest.class, Cadi3AAFProviderTest.class })
+public class JUnitTestSuite {
+	private static final Logger LOGGER = Logger.getLogger(JUnitTestSuite.class);
+
+	public static void main(String[] args) {
+		LOGGER.info("Running the test suite");
+
+		TestSuite tstSuite = new TestSuite();
+		LOGGER.info("Total Test Counts " + tstSuite.countTestCases());
+	}
+
+}
diff --git a/src/test/java/org/onap/dmaap/commonauth/kafka/base/authorization/TestRunner.java b/src/test/java/org/onap/dmaap/commonauth/kafka/base/authorization/TestRunner.java
new file mode 100644
index 0000000..7f7ea11
--- /dev/null
+++ b/src/test/java/org/onap/dmaap/commonauth/kafka/base/authorization/TestRunner.java
@@ -0,0 +1,41 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * ONAP Policy Engine
+ * ================================================================================
+ * Copyright (C) 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 org.junit.runner.JUnitCore;
+import org.junit.runner.Result;
+import org.junit.runner.notification.Failure;
+import org.apache.log4j.Logger;
+
+public class TestRunner {
+	private static final Logger LOGGER = Logger.getLogger(TestRunner.class);
+
+	public static void main(String[] args) {
+		// TODO Auto-generated method stub
+		Result result = JUnitCore.runClasses(JUnitTestSuite.class);
+		for (Failure failure : result.getFailures()) {
+			LOGGER.info(failure.toString());
+
+		}
+		LOGGER.info(result.wasSuccessful());
+	}
+
+}
diff --git a/src/test/java/org/onap/dmaap/kafkaAuthorize/JUnitTestSuite.java b/src/test/java/org/onap/dmaap/kafkaAuthorize/JUnitTestSuite.java
new file mode 100644
index 0000000..d52c8bc
--- /dev/null
+++ b/src/test/java/org/onap/dmaap/kafkaAuthorize/JUnitTestSuite.java
@@ -0,0 +1,41 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * ONAP Policy Engine
+ * ================================================================================
+ * Copyright (C) 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 junit.framework.TestSuite;
+import org.junit.runner.RunWith;
+import org.junit.runners.Suite;
+import org.junit.runners.Suite.SuiteClasses;
+import org.apache.log4j.Logger;
+
+@RunWith(Suite.class)
+@SuiteClasses({ KafkaCustomAuthorizerTest.class, PlainSaslServer1Test.class })
+public class JUnitTestSuite {
+	private static final Logger LOGGER = Logger.getLogger(JUnitTestSuite.class);
+
+	public static void main(String[] args) {
+		LOGGER.info("Running the test suite");
+
+		TestSuite tstSuite = new TestSuite();
+		LOGGER.info("Total Test Counts " + tstSuite.countTestCases());
+	}
+
+}
diff --git a/src/test/java/org/onap/dmaap/kafkaAuthorize/KafkaCustomAuthorizerTest.java b/src/test/java/org/onap/dmaap/kafkaAuthorize/KafkaCustomAuthorizerTest.java
new file mode 100644
index 0000000..3075327
--- /dev/null
+++ b/src/test/java/org/onap/dmaap/kafkaAuthorize/KafkaCustomAuthorizerTest.java
@@ -0,0 +1,99 @@
+/*******************************************************************************
+ *  ============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 static org.junit.Assert.assertTrue;
+
+import org.apache.kafka.common.security.auth.KafkaPrincipal;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+import org.onap.dmaap.commonauth.kafka.base.authorization.AuthorizationProvider;
+import org.onap.dmaap.commonauth.kafka.base.authorization.AuthorizationProviderFactory;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+import kafka.network.RequestChannel.Session;
+import kafka.security.auth.Operation;
+import kafka.security.auth.Resource;
+import kafka.security.auth.ResourceType;
+
+@RunWith(PowerMockRunner.class)
+@PrepareForTest({ AuthorizationProviderFactory.class })
+public class KafkaCustomAuthorizerTest {
+	@Mock
+	Session arg0;
+	@Mock
+	Operation arg1;
+	@Mock
+	Resource arg2;
+	@Mock
+	KafkaPrincipal principal;
+	@Mock
+	ResourceType resourceType;
+	@Mock
+	AuthorizationProviderFactory factory;
+	@Mock
+	AuthorizationProvider provider;
+
+	KafkaCustomAuthorizer authorizer = new KafkaCustomAuthorizer();
+
+	@Before
+	public void setUp() throws Exception {
+
+		MockitoAnnotations.initMocks(this);
+		PowerMockito.when(principal.getName()).thenReturn("fullName");
+		PowerMockito.when(arg0.principal()).thenReturn(principal);
+		PowerMockito.when(arg1.name()).thenReturn("Write");
+		PowerMockito.when(resourceType.name()).thenReturn("Topic");
+		PowerMockito.when(arg2.resourceType()).thenReturn(resourceType);
+		PowerMockito.when(arg2.name()).thenReturn("namespace.Topic");
+		PowerMockito.mockStatic(AuthorizationProviderFactory.class);
+		PowerMockito.when(AuthorizationProviderFactory.getProviderFactory()).thenReturn(factory);
+		PowerMockito.when(factory.getProvider()).thenReturn(provider);
+
+	}
+
+	@Test
+	public void testAuthorizerSuccess() {
+		PowerMockito.when(provider.hasPermission("fullName", "namespace.topic", ":topic.namespace.Topic", "pub"))
+				.thenReturn(true);
+		assertTrue(authorizer.authorize(arg0, arg1, arg2));
+
+	}
+
+	@Test
+	public void testAuthorizerFailure() {
+
+		PowerMockito.when(provider.hasPermission("fullName", "namespace.topic", ":topic.namespace.Topic", "pub"))
+				.thenReturn(false);
+		try {
+			authorizer.authorize(arg0, arg1, arg2);
+		} catch (Exception e) {
+			assertTrue(true);
+		}
+
+	}
+
+}
diff --git a/src/test/java/org/onap/dmaap/kafkaAuthorize/PlainSaslServer1Test.java b/src/test/java/org/onap/dmaap/kafkaAuthorize/PlainSaslServer1Test.java
new file mode 100644
index 0000000..8826f17
--- /dev/null
+++ b/src/test/java/org/onap/dmaap/kafkaAuthorize/PlainSaslServer1Test.java
@@ -0,0 +1,68 @@
+/*******************************************************************************
+ *  ============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 static org.junit.Assert.assertNotNull;
+
+import javax.security.sasl.SaslException;
+
+import org.apache.kafka.common.security.JaasContext;
+import org.apache.kafka.common.security.plain.PlainSaslServer;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+import org.onap.dmaap.commonauth.kafka.base.authorization.AuthorizationProvider;
+import org.onap.dmaap.commonauth.kafka.base.authorization.AuthorizationProviderFactory;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+@RunWith(PowerMockRunner.class)
+@PrepareForTest({ AuthorizationProviderFactory.class })
+public class PlainSaslServer1Test {
+
+	PlainSaslServer1 sslServer = new PlainSaslServer1(null);
+	@Mock
+	JaasContext jaasContext;
+	@Mock
+	AuthorizationProviderFactory factory;
+	@Mock
+	AuthorizationProvider provider;
+
+	@Before
+	public void setUp() throws Exception {
+
+		MockitoAnnotations.initMocks(this);
+		PowerMockito.mockStatic(AuthorizationProviderFactory.class);
+		PowerMockito.when(AuthorizationProviderFactory.getProviderFactory()).thenReturn(factory);
+		PowerMockito.when(factory.getProvider()).thenReturn(provider);
+	}
+
+	@Test
+	public void testAuthentication() throws Exception {
+		String response = "authorizationID\u0000username\u0000password";
+		PowerMockito.when(provider.authenticate("username", "password")).thenReturn(null);
+		assertNotNull(sslServer.evaluateResponse(response.getBytes()));
+
+	}
+}
diff --git a/src/test/java/org/onap/dmaap/kafkaAuthorize/TestRunner.java b/src/test/java/org/onap/dmaap/kafkaAuthorize/TestRunner.java
new file mode 100644
index 0000000..829d18d
--- /dev/null
+++ b/src/test/java/org/onap/dmaap/kafkaAuthorize/TestRunner.java
@@ -0,0 +1,41 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * ONAP Policy Engine
+ * ================================================================================
+ * Copyright (C) 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 org.junit.runner.JUnitCore;
+import org.junit.runner.Result;
+import org.junit.runner.notification.Failure;
+import org.apache.log4j.Logger;
+
+public class TestRunner {
+	private static final Logger LOGGER = Logger.getLogger(TestRunner.class);
+
+	public static void main(String[] args) {
+		// TODO Auto-generated method stub
+		Result result = JUnitCore.runClasses(JUnitTestSuite.class);
+		for (Failure failure : result.getFailures()) {
+			LOGGER.info(failure.toString());
+
+		}
+		LOGGER.info(result.wasSuccessful());
+	}
+
+}
diff --git a/src/test/resources/cadi.properties b/src/test/resources/cadi.properties
new file mode 100644
index 0000000..4ce7689
--- /dev/null
+++ b/src/test/resources/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=src/test/resources/truststoreONAPall.jks
+cadi_truststore_password=changeit
+
+cadi_keyfile=src/test/resources/keyfilenew
+
+cadi_alias=dmaapmr@mr.dmaap.onap.org
+cadi_keystore=src/test/resources/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/test/resources/keyfilenew b/src/test/resources/keyfilenew
new file mode 100644
index 0000000..884375f
--- /dev/null
+++ b/src/test/resources/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/test/resources/org.onap.dmaap.mr.p12 b/src/test/resources/org.onap.dmaap.mr.p12
new file mode 100644
index 0000000..79549ed
--- /dev/null
+++ b/src/test/resources/org.onap.dmaap.mr.p12
diff --git a/src/test/resources/truststoreONAPall.jks b/src/test/resources/truststoreONAPall.jks
new file mode 100644
index 0000000..2da1dcc
--- /dev/null
+++ b/src/test/resources/truststoreONAPall.jks
diff --git a/version.properties b/version.properties
index 65e7bbc..96247dc 100644
--- a/version.properties
+++ b/version.properties
@@ -24,9 +24,9 @@
 # Note that these variables cannot be structured (e.g. : version.release or version.snapshot etc... )
 # because they are used in Jenkins, whose plug-in doesn't support
 
-major=1
+major=0
 minor=0
-patch=0
+patch=2
 
 base_version=${major}.${minor}.${patch}
author	sunil unnava <sunil.unnava@att.com>	2018-09-19 18:08:36 -0400
committer	sunil unnava <sunil.unnava@att.com>	2018-09-19 18:13:20 -0400
commit	b01a7330883cbd5bce618ea44ea0f86ce6332729 (patch)
tree	6adeee94a55c91cdfceefa6e190ce31697e923c9
parent	ba5557be62af8b3c4f91268a169447adeca85dbc (diff)