[LIVY-785] Enabled adding security related HTTP headers

## What changes were proposed in this pull request?

This change introduces a new configuration option `livy.server.security-headers.enabled`.
When this property is set to true, the following security headers are added to HTTP
responses by default:

 * X-XSS-Protection
 * X-Frame-Options
 * X-Content-Type-Options
 * Strict-Transport-Security
 * Content-Security-Policy

Also, adds content type information to all responses as required when using content type option nosniff

## How was this patch tested?

Tested manually

Author: nileshrathi345

conf/livy.conf.template

@@ -102,6 +102,20 @@
 # http-header "X-Requested-By" in request if the http method is POST/DELETE/PUT/PATCH.
 # livy.server.csrf-protection.enabled =
 
+# Whether to add security related HTTP headers to responses, by default true. If enabled,
+# Livy server adds HTTP headers to responses based on below configuration parameters starting with
+# `Livy.server.http.header.`
+# livy.server.security-headers.enabled = true
+
+# Security headers added to responses by default when
+# configuration `livy.server.security-headers.enabled` is set to true.
+# STS header is only added if TLS is enabled.
+# livy.server.http.header.X-XSS-Protection = 1; mode=block
+# livy.server.http.header.X-Frame-Options = SAMEORIGIN
+# livy.server.http.header.X-Content-Type-Options = nosniff
+# livy.server.http.header.Strict-Transport-Security = max-age=31536000; includeSubDomains
+# livy.server.http.header.Content-Security-Policy = default-src 'self'; script-src 'self' 'unsafe-inline'; img-src 'self'; frame-src 'self';
+
 # Whether to enable HiveContext in livy interpreter, if it is true hive-site.xml will be detected
 # on user request and then livy server classpath automatically.
 # livy.repl.enable-hive-context =

server/src/main/scala/org/apache/livy/LivyConf.scala

@@ -71,6 +71,21 @@ object LivyConf {
 
   val CSRF_PROTECTION = Entry("livy.server.csrf-protection.enabled", false)
 
+  val SECURITY_HEADERS_ENABLED = Entry("livy.server.security-headers.enabled", true)
+  val SECURITY_HEADERS_XSS_PROTECTION =
+    Entry("livy.server.http.header.X-XSS-Protection", "1; mode=block")
+  val SECURITY_HEADERS_FRAME_OPTIONS =
+    Entry("livy.server.http.header.X-Frame-Options", "SAMEORIGIN")
+  val SECURITY_HEADERS_CONTENT_TYPE_OPTIONS =
+    Entry("livy.server.http.header.X-Content-Type-Options", "nosniff")
+  val SECURITY_HEADERS_STRICT_TRANSPORT_SECURITY =
+    Entry("livy.server.http.header.Strict-Transport-Security",
+      "max-age=31536000; includeSubDomains")
+  val SECURITY_HEADERS_CONTENT_SECURITY_POLICY =
+    Entry("livy.server.http.header.Content-Security-Policy",
+      "default-src 'self'; script-src 'self' 'unsafe-inline'; img-src 'self'; " +
+        "frame-src 'self';")
+
   val IMPERSONATION_ENABLED = Entry("livy.impersonation.enabled", false)
   val SUPERUSERS = Entry("livy.superusers", null)
 

server/src/main/scala/org/apache/livy/server/LivyServer.scala

@@ -251,6 +251,12 @@ class LivyServer extends Logging {
 
       })
 
+    if (livyConf.getBoolean(SECURITY_HEADERS_ENABLED)) {
+      info("Adding security headers is enabled.")
+      val securityHeadersHolder = new FilterHolder(new SecurityHeadersFilter(livyConf))
+      server.context.addFilter(securityHeadersHolder, "/*", EnumSet.allOf(classOf[DispatcherType]))
+    }
+
     livyConf.get(AUTH_TYPE) match {
       case authType @ KerberosAuthenticationHandler.TYPE =>
         val principal = SecurityUtil.getServerPrincipal(livyConf.get(AUTH_KERBEROS_PRINCIPAL),

server/src/main/scala/org/apache/livy/server/SecurityHeadersFilter.scala

@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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 org.apache.livy.server
+
+import javax.servlet.{Filter, FilterChain, FilterConfig, ServletRequest, ServletResponse}
+import javax.servlet.http.HttpServletResponse
+
+import org.apache.livy.LivyConf
+
+/**
+ * Adds security related headers to HTTP responses.
+ */
+class SecurityHeadersFilter(livyConf: LivyConf) extends Filter {
+
+  private def isSslEnabled: Boolean = {
+    Option(livyConf.get(LivyConf.SSL_KEYSTORE)).exists(_.length > 0)
+  }
+
+  val headers : Map[String, String] = Map(
+    "X-Content-Type-Options" -> livyConf.get(LivyConf.SECURITY_HEADERS_CONTENT_TYPE_OPTIONS),
+    "X-Frame-Options" -> livyConf.get(LivyConf.SECURITY_HEADERS_FRAME_OPTIONS),
+    "X-XSS-Protection" -> livyConf.get(LivyConf.SECURITY_HEADERS_XSS_PROTECTION),
+    "Strict-Transport-Security" ->
+      (if (isSslEnabled) livyConf.get(LivyConf.SECURITY_HEADERS_STRICT_TRANSPORT_SECURITY) else ""),
+    "Content-Security-Policy" -> livyConf.get(LivyConf.SECURITY_HEADERS_CONTENT_SECURITY_POLICY))
+    .filter(e => e._2 != null && e._2.trim().length > 0)
+
+  override def init(filterConfig: FilterConfig): Unit = {}
+
+  override def doFilter(request: ServletRequest,
+                        response: ServletResponse,
+                        chain: FilterChain): Unit = {
+    val servletResponse = response.asInstanceOf[HttpServletResponse]
+    for ((k, v) <- headers) {
+      servletResponse.addHeader(k, v)
+    }
+    chain.doFilter(request, response)
+  }
+
+  override def destroy(): Unit = {}
+}

server/src/test/scala/org/apache/livy/server/SecurityHeadersFilterSpec.scala

@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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 org.apache.livy.server
+
+import javax.servlet.FilterChain
+import javax.servlet.http.{HttpServletRequest, HttpServletResponse}
+
+import org.mockito.ArgumentCaptor
+import org.mockito.Mockito.{atLeastOnce, verify}
+import org.scalatest.{FunSpec, Matchers}
+import org.scalatestplus.mockito.MockitoSugar.mock
+
+import org.apache.livy.{LivyBaseUnitTestSuite, LivyConf}
+
+class SecurityHeadersFilterSpec extends FunSpec with Matchers with LivyBaseUnitTestSuite {
+
+  val requiredHeaders = Set("X-Content-Type-Options", "X-Frame-Options", "X-XSS-Protection",
+    "Content-Security-Policy")
+
+  private def runFilterAndGetResponseHeaders(configEntries: Set[(LivyConf.Entry, String)]):
+  List[(String, String)] = {
+    import scala.collection.JavaConverters._
+    val livyConf = createLivyConf(configEntries)
+    val securityHeadersFilter = new SecurityHeadersFilter(livyConf)
+    val response = mock[HttpServletResponse]
+    val request = mock[HttpServletRequest]
+    val chain = mock[FilterChain]
+    val keyCaptor = ArgumentCaptor.forClass(classOf[String])
+    val valueCaptor = ArgumentCaptor.forClass(classOf[String])
+
+    securityHeadersFilter.doFilter(request, response, chain)
+    verify(response, atLeastOnce()).addHeader(keyCaptor.capture(), valueCaptor.capture())
+    keyCaptor.getAllValues.asScala.toList.zip(valueCaptor.getAllValues.asScala.toList)
+  }
+
+  private def createLivyConf(entries: Set[(LivyConf.Entry, String)]) = {
+    val livyConf = new LivyConf(false)
+    entries.foreach({ case (key, value) => livyConf.set(key, value) })
+    livyConf
+  }
+
+  describe("SecurityHeadersFilter") {
+
+    it("should add security headers with overrides") {
+      val responseHeaders = runFilterAndGetResponseHeaders(Set(
+        (LivyConf.SECURITY_HEADERS_XSS_PROTECTION, "xss"),
+        (LivyConf.SECURITY_HEADERS_CONTENT_SECURITY_POLICY, "csp")))
+      assert(requiredHeaders.subsetOf(responseHeaders.map(_._1).toSet))
+      assert(responseHeaders.contains(("X-XSS-Protection", "xss")))
+      assert(responseHeaders.contains(("Content-Security-Policy", "csp")))
+    }
+
+    it("should not add headers that are overridden with empty values") {
+      val responseHeaders = runFilterAndGetResponseHeaders(Set(
+        (LivyConf.SECURITY_HEADERS_XSS_PROTECTION, ""),
+        (LivyConf.SECURITY_HEADERS_CONTENT_SECURITY_POLICY, "")))
+      assert(!responseHeaders.exists(_._1 == "X-XSS-Protection"))
+      assert(!responseHeaders.exists(_._1 == "Content-Security-Policy"))
+    }
+
+    it("should not set HSTS header if TLS is not enabled") {
+      val responseHeaders = runFilterAndGetResponseHeaders(Set.empty)
+      assert(!responseHeaders.exists(_._1 == "Strict-Transport-Security"))
+    }
+
+    it("should set HSTS header if TLS is enabled, value not overridden") {
+      val responseHeaders = runFilterAndGetResponseHeaders(Set(
+        (LivyConf.SSL_KEYSTORE, "/tmp")))
+      assert(responseHeaders.exists(_._1 == "Strict-Transport-Security"))
+    }
+
+    it("should set HSTS header if TLS is enabled, value overridden") {
+      val responseHeaders = runFilterAndGetResponseHeaders(Set(
+        (LivyConf.SSL_KEYSTORE, "/tmp"),
+        (LivyConf.SECURITY_HEADERS_STRICT_TRANSPORT_SECURITY, "sts")))
+      assert(responseHeaders.contains(("Strict-Transport-Security", "sts")))
+    }
+
+    it("should not set HSTS header if TLS is enabled, value overridden with empty string") {
+      val responseHeaders = runFilterAndGetResponseHeaders(Set(
+        (LivyConf.SSL_KEYSTORE, "/tmp"),
+        (LivyConf.SECURITY_HEADERS_STRICT_TRANSPORT_SECURITY, "")))
+      assert(!responseHeaders.exists(_._1 == "Strict-Transport-Security"))
+    }
+  }
+}