authorization check for more Kafka operations

[dmaap/kafka11aaf.git] / src / main / java / org / onap / dmaap / commonauth / kafka / base / authorization / Cadi3AAFProvider.java

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
index 5a3be57..da01829 100644 (file)
--- 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
@@ -22,8 +22,12 @@ package org.onap.dmaap.commonauth.kafka.base.authorization;
 
 import java.io.FileInputStream;
 import java.io.IOException;
+import java.util.Map;
 import java.util.Properties;
 
+import javax.security.auth.login.AppConfigurationEntry;
+import javax.security.auth.login.Configuration;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -42,7 +46,45 @@ public class Cadi3AAFProvider implements AuthorizationProvider {
        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";
+       private static String apiKey = null;
+       private static String kafkaUsername = null;
+       private static AAFAuthn<?> aafAuthn;
+       private static AbsAAFLur<AAFPermission> aafLur;
+
+       private static final Logger logger = LoggerFactory.getLogger(Cadi3AAFProvider.class);
+
+       static {
+
+               Configuration config = Configuration.getConfiguration();
+               try {
+                       if (config == null) {
+                               logger.error("CRITICAL ERROR|Check java.security.auth.login.config VM argument|");
+                       } else {
+                               // read the section for KafkaServer
+                               AppConfigurationEntry[] entries = config.getAppConfigurationEntry("KafkaServer");
+                               if (entries == null) {
+                                       logger.error(
+                                                       "CRITICAL ERROR|Check config contents passed in java.security.auth.login.config VM argument|");
+                                       kafkaUsername = "kafkaUsername";
+                                       apiKey = "apiKey";
+
+                               } else {
+                                       for (int i = 0; i < entries.length; i++) {
+                                               AppConfigurationEntry entry = entries[i];
+                                               Map<String, ?> optionsMap = entry.getOptions();
+                                               kafkaUsername = (String) optionsMap.get("username");
+                                               apiKey = (String) optionsMap.get("password");
+                                       }
+                               }
+                       }
+               } catch (Exception e) {
+                       logger.error("CRITICAL ERROR: JAAS configuration incorrectly set: " + e.getMessage());
+               }
+       }
+
+       public static String getKafkaUsername() {
+               return kafkaUsername;
+       }
 
        public static AAFAuthn<?> getAafAuthn() throws CadiException {
                if (aafAuthn == null) {
@@ -51,13 +93,6 @@ public class Cadi3AAFProvider implements AuthorizationProvider {
                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();
        }
@@ -85,11 +120,6 @@ public class Cadi3AAFProvider implements AuthorizationProvider {
                                logger.error("Unable to load " + CADI_PROPERTIES);
                                logger.error("Error", e);
                        }
-
-                       props_ok = true;
-                       if (props_ok == false) {
-                               return;
-                       }
                }
 
                if (aafAuthn == null) {
@@ -101,7 +131,6 @@ public class Cadi3AAFProvider implements AuthorizationProvider {
                                aafAuthn = null;
                                if (access != null)
                                        access.log(e, "Failed to initialize AAF");
-                               props_ok = false;
                        }
                }
 
@@ -123,7 +152,7 @@ public class Cadi3AAFProvider implements AuthorizationProvider {
                                hasPermission = true;
                                return hasPermission;
                        }
-                       AAFPermission perm = new AAFPermission(MR_NAMESPACE, permission, instance, action);
+                       AAFPermission perm = new AAFPermission(null, permission, instance, action);
                        if (aafLur != null) {
                                hasPermission = aafLur.fish(new UnAuthPrincipal(userId), perm);
                                logger.trace("Permission: " + perm.getKey() + " for user :" + userId + " found: " + hasPermission);
@@ -141,11 +170,20 @@ public class Cadi3AAFProvider implements AuthorizationProvider {
        }
 
        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;
+               if (userId.equals(kafkaUsername)) {
+                       if (password.equals(apiKey)) {
+                               logger.info("by passes the authentication for the admin " + kafkaUsername);
+                               return null;
+                       } else {
+                               String errorMessage = "Authentication failed for user " + kafkaUsername;
+                               logger.error(errorMessage);
+                               return errorMessage;
+                       }
+
                }
+
                String aafResponse = aafAuthn.validate(userId, password);
                logger.info("aafResponse=" + aafResponse + " for " + userId);