LIVY-356 Enabling LDAP authentication for Client to Server.

conf/livy.conf.template

@@ -35,6 +35,9 @@
 # If livy should impersonate the requesting users when creating a new session.
 # livy.impersonation.enabled = true
 
+# Enable Logging queries
+# livy.server.query-logging.enabled = true
+
 # Comma-separated list of Livy RSC jars. By default Livy will upload jars from its installation
 # directory every time a session is started. By caching these files in HDFS, for example, startup
 # time of sessions on YARN can be reduced.

pom.xml

@@ -42,10 +42,11 @@
   </organization>
 
   <properties>
-    <hadoop.version>2.7.3</hadoop.version>
+    <hadoop.version>2.8.0</hadoop.version>
     <hadoop.scope>compile</hadoop.scope>
     <spark.version>1.6.2</spark.version>
     <commons-codec.version>1.9</commons-codec.version>
+    <grouper-ws.version>2.3.0</grouper-ws.version>
     <httpclient.version>4.5.2</httpclient.version>
     <httpcore.version>4.4.4</httpcore.version>
     <jackson.version>2.4.4</jackson.version>
@@ -230,7 +231,6 @@
       <version>${scalatra.version}</version>
       <scope>test</scope>
     </dependency>
-
   </dependencies>
 
   <dependencyManagement>
@@ -278,6 +278,18 @@
         <version>${commons-codec.version}</version>
       </dependency>
 
+      <dependency>
+        <groupId>edu.internet2.middleware.grouper</groupId>
+        <artifactId>grouper-ws</artifactId>
+        <version>${grouper-ws.version}</version>
+        <exclusions>
+          <exclusion>
+            <groupId>*</groupId>
+            <artifactId>*</artifactId>
+          </exclusion>
+        </exclusions>
+      </dependency>
+
       <dependency>
         <groupId>io.dropwizard.metrics</groupId>
         <artifactId>metrics-core</artifactId>

server/pom.xml

@@ -75,6 +75,17 @@
       <artifactId>jackson-module-scala_${scala.binary.version}</artifactId>
     </dependency>
 
+    <dependency>
+      <groupId>edu.internet2.middleware.grouper</groupId>
+      <artifactId>grouper-ws</artifactId>
+      <exclusions>
+        <exclusion>
+          <groupId>*</groupId>
+          <artifactId>*</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
+
     <dependency>
       <groupId>io.dropwizard.metrics</groupId>
       <artifactId>metrics-core</artifactId>

server/src/main/scala/com/cloudera/livy/LivyConf.scala

@@ -64,6 +64,7 @@ object LivyConf {
   val SERVER_PORT = Entry("livy.server.port", 8998)
 
   val UI_ENABLED = Entry("livy.ui.enabled", true)
+  val BATCH_ENABLED = Entry("livy.batch.enabled", true)
 
   val REQUEST_HEADER_SIZE = Entry("livy.server.request-header.size", 131072)
   val RESPONSE_HEADER_SIZE = Entry("livy.server.response-header.size", 131072)
@@ -76,6 +77,8 @@ object LivyConf {
   val ACCESS_CONTROL_ENABLED = Entry("livy.server.access-control.enabled", false)
   val ACCESS_CONTROL_USERS = Entry("livy.server.access-control.users", null)
 
+  val QUERY_LOGGER_ENABLED = Entry("livy.server.query-logging.enabled", false)
+
   val SSL_KEYSTORE = Entry("livy.keystore", null)
   val SSL_KEYSTORE_PASSWORD = Entry("livy.keystore.password", null)
   val SSL_KEY_PASSWORD = Entry("livy.key-password", null)
@@ -92,6 +95,13 @@ object LivyConf {
   val LAUNCH_KERBEROS_REFRESH_INTERVAL = Entry("livy.server.launch.kerberos.refresh-interval", "1h")
   val KINIT_FAIL_THRESHOLD = Entry("livy.server.launch.kerberos.kinit-fail-threshold", 5)
 
+  // Ldap properties
+  val AUTH_LDAP_URL = Entry("livy.server.auth.ldap.url", null)
+  val AUTH_LDAP_BASE_DN = Entry("livy.server.auth.ldap.base-dn", null)
+  val AUTH_LDAP_USERNAME_DOMAIN = Entry("livy.server.auth.ldap.username-domain", null)
+  val AUTH_LDAP_ENABLE_START_TLS = Entry("livy.server.auth.ldap.enable-start-tls", "false")
+  val AUTH_LDAP_SECURITY_AUTH = Entry("livy.server.auth.ldap.security-authentication", "simple")
+
   /**
    * Recovery mode of Livy. Possible values:
    * off: Default. Turn off recovery. Every time Livy shuts down, it stops and forgets all sessions.
@@ -185,6 +195,7 @@ object LivyConf {
     CSRF_PROTECTION.key -> DepConf("livy.server.csrf_protection.enabled", "0.4"),
     ACCESS_CONTROL_ENABLED.key -> DepConf("livy.server.access_control.enabled", "0.4"),
     ACCESS_CONTROL_USERS.key -> DepConf("livy.server.access_control.users", "0.4"),
+    QUERY_LOGGER_ENABLED.key -> DepConf("livy.server.query_logging.enabled", "0.4"),
     AUTH_KERBEROS_NAME_RULES.key -> DepConf("livy.server.auth.kerberos.name_rules", "0.4"),
     LAUNCH_KERBEROS_REFRESH_INTERVAL.key ->
       DepConf("livy.server.launch.kerberos.refresh_interval", "0.4"),

server/src/main/scala/com/cloudera/livy/server/LivyServer.scala

@@ -18,31 +18,32 @@
 
 package com.cloudera.livy.server
 
-import java.util.concurrent._
 import java.util.EnumSet
+import java.util.concurrent._
 import javax.servlet._
 
-import scala.concurrent.ExecutionContext.Implicits.global
-import scala.concurrent.Future
-
-import org.apache.hadoop.security.{SecurityUtil, UserGroupInformation}
-import org.apache.hadoop.security.authentication.server._
-import org.eclipse.jetty.servlet.FilterHolder
-import org.scalatra.metrics.MetricsBootstrap
-import org.scalatra.metrics.MetricsSupportExtensions._
-import org.scalatra.ScalatraServlet
-import org.scalatra.servlet.{MultipartConfig, ServletApiImplicits}
-
 import com.cloudera.livy._
+import com.cloudera.livy.server.auth.LdapAuthenticationHandlerImpl
 import com.cloudera.livy.server.batch.BatchSessionServlet
 import com.cloudera.livy.server.interactive.InteractiveSessionServlet
 import com.cloudera.livy.server.recovery.{SessionStore, StateStore}
 import com.cloudera.livy.server.ui.UIServlet
-import com.cloudera.livy.sessions.{BatchSessionManager, InteractiveSessionManager}
 import com.cloudera.livy.sessions.SessionManager.SESSION_RECOVERY_MODE_OFF
+import com.cloudera.livy.sessions.{BatchSessionManager, InteractiveSessionManager}
 import com.cloudera.livy.utils.LivySparkUtils._
 import com.cloudera.livy.utils.SparkYarnApp
 
+import org.apache.hadoop.security.authentication.server.{AuthenticationFilter, KerberosAuthenticationHandler}
+import org.apache.hadoop.security.{SecurityUtil, UserGroupInformation}
+import org.eclipse.jetty.servlet.FilterHolder
+import org.scalatra.ScalatraServlet
+import org.scalatra.metrics.MetricsBootstrap
+import org.scalatra.metrics.MetricsSupportExtensions._
+import org.scalatra.servlet.{MultipartConfig, ServletApiImplicits}
+
+import scala.concurrent.ExecutionContext.Implicits.global
+import scala.concurrent.Future
+
 class LivyServer extends Logging {
 
   import LivyConf._
@@ -180,9 +181,11 @@ class LivyServer extends Logging {
               new InteractiveSessionServlet(interactiveSessionManager, sessionStore, livyConf)
             mount(context, interactiveServlet, "/sessions/*")
 
-            val batchServlet = new BatchSessionServlet(batchSessionManager, sessionStore, livyConf)
-            mount(context, batchServlet, "/batches/*")
-
+            if(livyConf.getBoolean(BATCH_ENABLED)) {
+              val batchServlet = new BatchSessionServlet(batchSessionManager,
+                sessionStore, livyConf)
+              mount(context, batchServlet, "/batches/*")
+            }
             if (livyConf.getBoolean(UI_ENABLED)) {
               val uiServlet = new UIServlet
               mount(context, uiServlet, "/ui/*")
@@ -223,6 +226,21 @@ class LivyServer extends Logging {
         server.context.addFilter(holder, "/*", EnumSet.allOf(classOf[DispatcherType]))
         info(s"SPNEGO auth enabled (principal = $principal)")
 
+      case authType @ LdapAuthenticationHandlerImpl.TYPE =>
+        val holder = new FilterHolder(new AuthenticationFilter())
+        holder.setInitParameter(AuthenticationFilter.AUTH_TYPE, authType)
+        Option(livyConf.get(LivyConf.AUTH_LDAP_URL)).foreach(url =>
+          holder.setInitParameter(LdapAuthenticationHandlerImpl.PROVIDER_URL, url))
+        Option(livyConf.get(LivyConf.AUTH_LDAP_USERNAME_DOMAIN)).foreach(domain =>
+          holder.setInitParameter(LdapAuthenticationHandlerImpl.LDAP_BIND_DOMAIN, domain))
+        Option(livyConf.get(LivyConf.AUTH_LDAP_BASE_DN)).foreach(baseDN =>
+          holder.setInitParameter(LdapAuthenticationHandlerImpl.BASE_DN, baseDN))
+        holder.setInitParameter(LdapAuthenticationHandlerImpl.SECURITY_AUTHENTICATION,
+          livyConf.get(LivyConf.AUTH_LDAP_SECURITY_AUTH))
+        holder.setInitParameter(LdapAuthenticationHandlerImpl.ENABLE_START_TLS,
+          livyConf.get(LivyConf.AUTH_LDAP_ENABLE_START_TLS))
+        server.context.addFilter(holder, "/*", EnumSet.allOf(classOf[DispatcherType]))
+        info("LDAP auth enabled.")
      case null =>
         // Nothing to do.
 
@@ -247,6 +265,12 @@ class LivyServer extends Logging {
       }
     }
 
+    if (livyConf.getBoolean(QUERY_LOGGER_ENABLED)) {
+      info("Query logging is enabled.")
+      val servletLoggingHolder = new FilterHolder(new ServletLoggerFilter(livyConf))
+      server.context.addFilter(servletLoggingHolder, "/*", EnumSet.allOf(classOf[DispatcherType]))
+    }
+
     server.start()
 
     Runtime.getRuntime().addShutdownHook(new Thread("Livy Server Shutdown") {

server/src/main/scala/com/cloudera/livy/server/ServletLoggerFilter.scala

@@ -0,0 +1,103 @@
+/*
+ * Licensed to Cloudera, Inc. under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  Cloudera, Inc. licenses this file
+ * to you 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.
+ */
+
+package com.cloudera.livy.server
+
+import java.io.IOException
+import java.util.Enumeration
+import javax.servlet._
+import javax.servlet.http.HttpServletRequest
+
+import edu.internet2.middleware.grouper.ws.j2ee.{HttpServletRequestCopier, ServletFilterLogger}
+
+import com.cloudera.livy.{LivyConf, Logging}
+
+/**
+ * Logs Requests Body
+ */
+class ServletLoggerFilter(livyConf: LivyConf) extends ServletFilterLogger with Logging {
+
+  import LivyConf._
+
+  override def init(filterConfig: FilterConfig): Unit = {}
+  override def destroy(): Unit = {}
+
+  @throws(classOf[IOException])
+  @throws(classOf[ServletException])
+  override def doFilter(servletRequest: ServletRequest,
+                        servletResponse: ServletResponse,
+                        chain: FilterChain): Unit = {
+    var requestCopier: HttpServletRequestCopier = new HttpServletRequestCopier(
+      servletRequest.asInstanceOf[HttpServletRequest])
+
+    try {
+      chain.doFilter(requestCopier, servletResponse)
+    } finally {
+      logRequestHandler(requestCopier)
+    }
+  }
+
+  /**
+   * Method that handles copying request
+   */
+  def logRequestHandler(servletRequest: HttpServletRequestCopier): Unit = {
+    try {
+      val requestCopier: HttpServletRequestCopier = servletRequest
+      requestCopier.finishReading()
+
+      var requestParams = new StringBuilder()
+      val enumeration = servletRequest.getParameterNames()
+      while (enumeration.hasMoreElements()) {
+        val name: String = enumeration.nextElement()
+        requestParams.append(name + " = " + servletRequest.getParameter(name) + ", ")
+      }
+
+      val httpRequest = servletRequest.asInstanceOf[HttpServletRequest]
+      logHttpRequest(httpRequest, requestCopier)
+    } catch {
+      case e: Exception => sys.error("Error in handling Request to be logged")
+    }
+  }
+
+  /**
+   * Method that logs request
+   */
+  def logHttpRequest(httpRequest: HttpServletRequest,
+                    requestCopier: HttpServletRequestCopier): Unit = {
+    try {
+      val ip = httpRequest.getRemoteAddr
+      val method = httpRequest.getMethod
+      val requestCopy = requestCopier.getCopy
+
+      if (method.contains("POST")) {
+        if (livyConf.get(AUTH_TYPE) != null) {
+          val userid = httpRequest.getRemoteUser
+          info("\nUser: " + userid.toString()
+                + "\nIP : " + ip.toString()
+                + "\n" + new String(requestCopy))
+        } else {
+          info("\nIP : " + ip.toString()
+                + "\n" + new String(requestCopy))
+        }
+      }
+    } catch {
+      case e: Exception => error("Error logging request", e)
+    }
+  }
+
+}

server/src/main/scala/com/cloudera/livy/server/SessionServlet.scala

@@ -23,6 +23,7 @@ import javax.servlet.http.HttpServletRequest
 import org.scalatra._
 import scala.concurrent._
 import scala.concurrent.duration._
+import scala.sys.process._
 
 import com.cloudera.livy.{LivyConf, Logging}
 import com.cloudera.livy.sessions.{Session, SessionManager}
@@ -171,7 +172,9 @@ abstract class SessionServlet[S <: Session, R <: RecoveryMetadata](
    */
   protected def hasAccess(target: String, req: HttpServletRequest): Boolean = {
     val user = remoteUser(req)
-    user == null || user == target || livyConf.superusers().contains(user)
+    val cmdResult = Process("sh", Seq("-c", "getent netgroup " + target)).!!
+    val headlessGroup = cmdResult.split(",").zipWithIndex.filter(_._2 % 2 == 1).map(_._1).map(_.trim)
+    user == null || user == target || livyConf.superusers().contains(user) || headlessGroup.contains(user)
   }
 
   /**