1 files changed, 69 insertions, 52 deletions

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
--- 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;
+
+	}
 }