[MR] Add support for configuring jaas.sasl.config at runtime

[dmaap/messagerouter/messageservice.git] / src / main / java / org / onap / dmaap / dmf / mr / utils / Utils.java

diff --git a/src/main/java/org/onap/dmaap/dmf/mr/utils/Utils.java b/src/main/java/org/onap/dmaap/dmf/mr/utils/Utils.java
index 662f0f7..c420072 100644 (file)
--- a/src/main/java/org/onap/dmaap/dmf/mr/utils/Utils.java
+++ b/src/main/java/org/onap/dmaap/dmf/mr/utils/Utils.java
@@ -8,30 +8,30 @@
  *  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=========================================================
- *  
+ *
  *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
- *  
+ *
  *******************************************************************************/
 package org.onap.dmaap.dmf.mr.utils;
 
 import com.att.eelf.configuration.EELFLogger;
 import com.att.eelf.configuration.EELFManager;
-import org.onap.dmaap.dmf.mr.beans.DMaaPContext;
-
-import javax.servlet.http.HttpServletRequest;
 import java.io.IOException;
 import java.io.InputStream;
 import java.security.Principal;
 import java.text.DecimalFormat;
 import java.text.SimpleDateFormat;
 import java.util.*;
+import javax.servlet.http.HttpServletRequest;
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.onap.dmaap.dmf.mr.beans.DMaaPContext;
 
 /**
  * This is an utility class for various operations for formatting
@@ -46,13 +46,14 @@ public class Utils {
        private static final String BATCH_ID_FORMAT = "000000";
        private static final String X509_ATTR = "javax.servlet.request.X509Certificate";
        private static final EELFLogger log = EELFManager.getInstance().getLogger(Utils.class);
+       public static final String SASL_MECH = "sasl.mechanism";
 
        private Utils() {
                super();
        }
 
        /**
-        * Formatting the date 
+        * Formatting the date
         * @param date
         * @return date or null
         */
@@ -128,55 +129,71 @@ public class Utils {
         */
        public static long getSleepMsForRate ( double ratePerMinute )
        {
-               if ( ratePerMinute <= 0.0 ) 
+               if ( ratePerMinute <= 0.0 )
                {
                        return 0;
                }
                return Math.max ( 1000, Math.round ( 60 * 1000 / ratePerMinute ) );
        }
 
-         public static String getRemoteAddress(DMaaPContext ctx)
-         {
-           String reqAddr = ctx.getRequest().getRemoteAddr();
-           String fwdHeader = getFirstHeader("X-Forwarded-For",ctx);
-           return ((fwdHeader != null) ? fwdHeader : reqAddr);
-         }
-         public static String getFirstHeader(String h,DMaaPContext ctx)
-         {
-           List l = getHeader(h,ctx);
-           return ((l.size() > 0) ? (String)l.iterator().next() : null);
-         }
-         public static List<String> getHeader(String h,DMaaPContext ctx)
-         {
-           LinkedList list = new LinkedList();
-           Enumeration e = ctx.getRequest().getHeaders(h);
-           while (e.hasMoreElements())
-           {
-             list.add(e.nextElement().toString());
-           }
-           return list;
-         }
-         
-         public static String getKafkaproperty(){
-                 InputStream input = new Utils().getClass().getResourceAsStream("/kafka.properties");
-                       Properties props = new Properties();
-                       try {
-                               props.load(input);
-                       } catch (IOException e) {
-                               log.error("failed to read kafka.properties");
-                       }
-                       return props.getProperty("key");
-                       
-                 
-         }
-         
-         public static boolean isCadiEnabled(){
-                 boolean enableCadi=false;
-                 if(System.getenv("enableCadi")!=null&&System.getenv("enableCadi").equals("true")){
-                         enableCadi=true;
-                       }
-                 
-                 return enableCadi;
-         }
-                 
+       public static String getRemoteAddress(DMaaPContext ctx)
+       {
+               String reqAddr = ctx.getRequest().getRemoteAddr();
+               String fwdHeader = getFirstHeader("X-Forwarded-For",ctx);
+               return ((fwdHeader != null) ? fwdHeader : reqAddr);
+       }
+       public static String getFirstHeader(String h,DMaaPContext ctx)
+       {
+               List l = getHeader(h,ctx);
+               return ((l.size() > 0) ? (String)l.iterator().next() : null);
+       }
+       public static List<String> getHeader(String h,DMaaPContext ctx)
+       {
+               LinkedList list = new LinkedList();
+               Enumeration e = ctx.getRequest().getHeaders(h);
+               while (e.hasMoreElements())
+               {
+                       list.add(e.nextElement().toString());
+               }
+               return list;
+       }
+
+       public static String getKafkaproperty(){
+               InputStream input = new Utils().getClass().getResourceAsStream("/kafka.properties");
+               Properties props = new Properties();
+               try {
+                       props.load(input);
+               } catch (IOException e) {
+                       log.error("failed to read kafka.properties");
+               }
+               return props.getProperty("key");
+
+
+       }
+
+       public static boolean isCadiEnabled(){
+               boolean enableCadi=false;
+               if(System.getenv("enableCadi")!=null&&System.getenv("enableCadi").equals("true")){
+                       enableCadi=true;
+               }
+
+               return enableCadi;
+       }
+
+       public static Properties addSaslProps(){
+               Properties props = new Properties();
+               String saslMech = System.getenv("SASLMECH");
+               if (saslMech != null && saslMech.equals("scram-sha-512")) {
+                       props.put("sasl.jaas.config", System.getenv("JAASLOGIN"));
+                       props.put(SASL_MECH, saslMech.toUpperCase());
+               }
+               else {
+                       props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username='admin' password='" + getKafkaproperty() + "';");
+                       props.put(SASL_MECH, "PLAIN");
+               }
+               props.put(AdminClientConfig.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
+               log.info("KafkaAdmin sasl.mechanism set to " + props.getProperty(SASL_MECH));
+               return props;
+
+       }
 }