You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by va...@apache.org on 2017/01/27 00:53:34 UTC
spark git commit: [SPARK-19220][UI] Make redirection to HTTPS apply
to all URIs. (branch-2.1)
Repository: spark
Updated Branches:
refs/heads/branch-2.1 b12a76a41 -> 59502bbcf
[SPARK-19220][UI] Make redirection to HTTPS apply to all URIs. (branch-2.1)
The redirect handler was installed only for the root of the server;
any other context ended up being served directly through the HTTP
port. Since every sub page (e.g. application UIs in the history
server) is a separate servlet context, this meant that everything
but the root was accessible via HTTP still.
The change adds separate names to each connector, and binds contexts
to specific connectors so that content is only served through the
HTTPS connector when it's enabled. In that case, the only thing that
binds to the HTTP connector is the redirect handler.
Tested with new unit tests and by checking a live history server.
(cherry picked from commit d3dcb63b9709a34337327be9b7d3705698716077)
Author: Marcelo Vanzin <va...@cloudera.com>
Closes #16711 from vanzin/SPARK-19220_2.1.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/59502bbc
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/59502bbc
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/59502bbc
Branch: refs/heads/branch-2.1
Commit: 59502bbcf6e64e5b5e3dda080441054afaf58c53
Parents: b12a76a
Author: Marcelo Vanzin <va...@cloudera.com>
Authored: Thu Jan 26 16:53:28 2017 -0800
Committer: Marcelo Vanzin <va...@cloudera.com>
Committed: Thu Jan 26 16:53:28 2017 -0800
----------------------------------------------------------------------
.../main/scala/org/apache/spark/TestUtils.scala | 38 ++++++++-
.../scala/org/apache/spark/ui/JettyUtils.scala | 87 ++++++++++++++------
.../main/scala/org/apache/spark/ui/WebUI.scala | 14 +---
.../org/apache/spark/ui/UISeleniumSuite.scala | 19 +----
.../scala/org/apache/spark/ui/UISuite.scala | 56 ++++++++++++-
5 files changed, 155 insertions(+), 59 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/59502bbc/core/src/main/scala/org/apache/spark/TestUtils.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/TestUtils.scala b/core/src/main/scala/org/apache/spark/TestUtils.scala
index 2909191..7d866f8 100644
--- a/core/src/main/scala/org/apache/spark/TestUtils.scala
+++ b/core/src/main/scala/org/apache/spark/TestUtils.scala
@@ -18,19 +18,22 @@
package org.apache.spark
import java.io.{ByteArrayInputStream, File, FileInputStream, FileOutputStream}
-import java.net.{URI, URL}
+import java.net.{HttpURLConnection, URI, URL}
import java.nio.charset.StandardCharsets
import java.nio.file.Paths
+import java.security.SecureRandom
+import java.security.cert.X509Certificate
import java.util.Arrays
import java.util.concurrent.{CountDownLatch, TimeUnit}
import java.util.jar.{JarEntry, JarOutputStream}
+import javax.net.ssl._
+import javax.tools.{JavaFileObject, SimpleJavaFileObject, ToolProvider}
import scala.collection.JavaConverters._
import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer
import com.google.common.io.{ByteStreams, Files}
-import javax.tools.{JavaFileObject, SimpleJavaFileObject, ToolProvider}
import org.apache.spark.executor.TaskMetrics
import org.apache.spark.scheduler._
@@ -182,6 +185,37 @@ private[spark] object TestUtils {
assert(spillListener.numSpilledStages == 0, s"expected $identifier to not spill, but did")
}
+ /**
+ * Returns the response code from an HTTP(S) URL.
+ */
+ def httpResponseCode(url: URL, method: String = "GET"): Int = {
+ val connection = url.openConnection().asInstanceOf[HttpURLConnection]
+ connection.setRequestMethod(method)
+
+ // Disable cert and host name validation for HTTPS tests.
+ if (connection.isInstanceOf[HttpsURLConnection]) {
+ val sslCtx = SSLContext.getInstance("SSL")
+ val trustManager = new X509TrustManager {
+ override def getAcceptedIssuers(): Array[X509Certificate] = null
+ override def checkClientTrusted(x509Certificates: Array[X509Certificate], s: String) {}
+ override def checkServerTrusted(x509Certificates: Array[X509Certificate], s: String) {}
+ }
+ val verifier = new HostnameVerifier() {
+ override def verify(hostname: String, session: SSLSession): Boolean = true
+ }
+ sslCtx.init(null, Array(trustManager), new SecureRandom())
+ connection.asInstanceOf[HttpsURLConnection].setSSLSocketFactory(sslCtx.getSocketFactory())
+ connection.asInstanceOf[HttpsURLConnection].setHostnameVerifier(verifier)
+ }
+
+ try {
+ connection.connect()
+ connection.getResponseCode()
+ } finally {
+ connection.disconnect()
+ }
+ }
+
}
http://git-wip-us.apache.org/repos/asf/spark/blob/59502bbc/core/src/main/scala/org/apache/spark/ui/JettyUtils.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/ui/JettyUtils.scala b/core/src/main/scala/org/apache/spark/ui/JettyUtils.scala
index 35c3c8d..f713619 100644
--- a/core/src/main/scala/org/apache/spark/ui/JettyUtils.scala
+++ b/core/src/main/scala/org/apache/spark/ui/JettyUtils.scala
@@ -45,6 +45,9 @@ import org.apache.spark.util.Utils
*/
private[spark] object JettyUtils extends Logging {
+ val SPARK_CONNECTOR_NAME = "Spark"
+ val REDIRECT_CONNECTOR_NAME = "HttpsRedirect"
+
// Base type for a function that returns something based on an HTTP request. Allows for
// implicit conversion from many types of functions to jetty Handlers.
type Responder[T] = HttpServletRequest => T
@@ -274,17 +277,18 @@ private[spark] object JettyUtils extends Logging {
conf: SparkConf,
serverName: String = ""): ServerInfo = {
- val collection = new ContextHandlerCollection
addFilters(handlers, conf)
val gzipHandlers = handlers.map { h =>
+ h.setVirtualHosts(Array("@" + SPARK_CONNECTOR_NAME))
+
val gzipHandler = new GzipHandler
gzipHandler.setHandler(h)
gzipHandler
}
// Bind to the given port, or throw a java.net.BindException if the port is occupied
- def connect(currentPort: Int): (Server, Int) = {
+ def connect(currentPort: Int): ((Server, Option[Int]), Int) = {
val pool = new QueuedThreadPool
if (serverName.nonEmpty) {
pool.setName(serverName)
@@ -292,7 +296,9 @@ private[spark] object JettyUtils extends Logging {
pool.setDaemon(true)
val server = new Server(pool)
- val connectors = new ArrayBuffer[ServerConnector]
+ val connectors = new ArrayBuffer[ServerConnector]()
+ val collection = new ContextHandlerCollection
+
// Create a connector on port currentPort to listen for HTTP requests
val httpConnector = new ServerConnector(
server,
@@ -306,26 +312,33 @@ private[spark] object JettyUtils extends Logging {
httpConnector.setPort(currentPort)
connectors += httpConnector
- sslOptions.createJettySslContextFactory().foreach { factory =>
- // If the new port wraps around, do not try a privileged port.
- val securePort =
- if (currentPort != 0) {
- (currentPort + 400 - 1024) % (65536 - 1024) + 1024
- } else {
- 0
- }
- val scheme = "https"
- // Create a connector on port securePort to listen for HTTPS requests
- val connector = new ServerConnector(server, factory)
- connector.setPort(securePort)
-
- connectors += connector
-
- // redirect the HTTP requests to HTTPS port
- collection.addHandler(createRedirectHttpsHandler(securePort, scheme))
+ val httpsConnector = sslOptions.createJettySslContextFactory() match {
+ case Some(factory) =>
+ // If the new port wraps around, do not try a privileged port.
+ val securePort =
+ if (currentPort != 0) {
+ (currentPort + 400 - 1024) % (65536 - 1024) + 1024
+ } else {
+ 0
+ }
+ val scheme = "https"
+ // Create a connector on port securePort to listen for HTTPS requests
+ val connector = new ServerConnector(server, factory)
+ connector.setPort(securePort)
+ connector.setName(SPARK_CONNECTOR_NAME)
+ connectors += connector
+
+ // redirect the HTTP requests to HTTPS port
+ httpConnector.setName(REDIRECT_CONNECTOR_NAME)
+ collection.addHandler(createRedirectHttpsHandler(securePort, scheme))
+ Some(connector)
+
+ case None =>
+ // No SSL, so the HTTP connector becomes the official one where all contexts bind.
+ httpConnector.setName(SPARK_CONNECTOR_NAME)
+ None
}
- gzipHandlers.foreach(collection.addHandler)
// As each acceptor and each selector will use one thread, the number of threads should at
// least be the number of acceptors and selectors plus 1. (See SPARK-13776)
var minThreads = 1
@@ -337,17 +350,20 @@ private[spark] object JettyUtils extends Logging {
// The number of selectors always equals to the number of acceptors
minThreads += connector.getAcceptors * 2
}
- server.setConnectors(connectors.toArray)
pool.setMaxThreads(math.max(pool.getMaxThreads, minThreads))
val errorHandler = new ErrorHandler()
errorHandler.setShowStacks(true)
errorHandler.setServer(server)
server.addBean(errorHandler)
+
+ gzipHandlers.foreach(collection.addHandler)
server.setHandler(collection)
+
+ server.setConnectors(connectors.toArray)
try {
server.start()
- (server, httpConnector.getLocalPort)
+ ((server, httpsConnector.map(_.getLocalPort())), httpConnector.getLocalPort)
} catch {
case e: Exception =>
server.stop()
@@ -356,13 +372,16 @@ private[spark] object JettyUtils extends Logging {
}
}
- val (server, boundPort) = Utils.startServiceOnPort[Server](port, connect, conf, serverName)
- ServerInfo(server, boundPort, collection)
+ val ((server, securePort), boundPort) = Utils.startServiceOnPort(port, connect, conf,
+ serverName)
+ ServerInfo(server, boundPort, securePort,
+ server.getHandler().asInstanceOf[ContextHandlerCollection])
}
private def createRedirectHttpsHandler(securePort: Int, scheme: String): ContextHandler = {
val redirectHandler: ContextHandler = new ContextHandler
redirectHandler.setContextPath("/")
+ redirectHandler.setVirtualHosts(Array("@" + REDIRECT_CONNECTOR_NAME))
redirectHandler.setHandler(new AbstractHandler {
override def handle(
target: String,
@@ -442,7 +461,23 @@ private[spark] object JettyUtils extends Logging {
private[spark] case class ServerInfo(
server: Server,
boundPort: Int,
- rootHandler: ContextHandlerCollection) {
+ securePort: Option[Int],
+ private val rootHandler: ContextHandlerCollection) {
+
+ def addHandler(handler: ContextHandler): Unit = {
+ handler.setVirtualHosts(Array("@" + JettyUtils.SPARK_CONNECTOR_NAME))
+ rootHandler.addHandler(handler)
+ if (!handler.isStarted()) {
+ handler.start()
+ }
+ }
+
+ def removeHandler(handler: ContextHandler): Unit = {
+ rootHandler.removeHandler(handler)
+ if (handler.isStarted) {
+ handler.stop()
+ }
+ }
def stop(): Unit = {
server.stop()
http://git-wip-us.apache.org/repos/asf/spark/blob/59502bbc/core/src/main/scala/org/apache/spark/ui/WebUI.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/ui/WebUI.scala b/core/src/main/scala/org/apache/spark/ui/WebUI.scala
index 4118fcf..b1a6ef2 100644
--- a/core/src/main/scala/org/apache/spark/ui/WebUI.scala
+++ b/core/src/main/scala/org/apache/spark/ui/WebUI.scala
@@ -91,23 +91,13 @@ private[spark] abstract class WebUI(
/** Attach a handler to this UI. */
def attachHandler(handler: ServletContextHandler) {
handlers += handler
- serverInfo.foreach { info =>
- info.rootHandler.addHandler(handler)
- if (!handler.isStarted) {
- handler.start()
- }
- }
+ serverInfo.foreach(_.addHandler(handler))
}
/** Detach a handler from this UI. */
def detachHandler(handler: ServletContextHandler) {
handlers -= handler
- serverInfo.foreach { info =>
- info.rootHandler.removeHandler(handler)
- if (handler.isStarted) {
- handler.stop()
- }
- }
+ serverInfo.foreach(_.removeHandler(handler))
}
/**
http://git-wip-us.apache.org/repos/asf/spark/blob/59502bbc/core/src/test/scala/org/apache/spark/ui/UISeleniumSuite.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/org/apache/spark/ui/UISeleniumSuite.scala b/core/src/test/scala/org/apache/spark/ui/UISeleniumSuite.scala
index e5d408a..6e734d7 100644
--- a/core/src/test/scala/org/apache/spark/ui/UISeleniumSuite.scala
+++ b/core/src/test/scala/org/apache/spark/ui/UISeleniumSuite.scala
@@ -475,8 +475,8 @@ class UISeleniumSuite extends SparkFunSuite with WebBrowser with Matchers with B
val url = new URL(
sc.ui.get.appUIAddress.stripSuffix("/") + "/stages/stage/kill/?id=0")
// SPARK-6846: should be POST only but YARN AM doesn't proxy POST
- getResponseCode(url, "GET") should be (200)
- getResponseCode(url, "POST") should be (200)
+ TestUtils.httpResponseCode(url, "GET") should be (200)
+ TestUtils.httpResponseCode(url, "POST") should be (200)
}
}
}
@@ -488,8 +488,8 @@ class UISeleniumSuite extends SparkFunSuite with WebBrowser with Matchers with B
val url = new URL(
sc.ui.get.appUIAddress.stripSuffix("/") + "/jobs/job/kill/?id=0")
// SPARK-6846: should be POST only but YARN AM doesn't proxy POST
- getResponseCode(url, "GET") should be (200)
- getResponseCode(url, "POST") should be (200)
+ TestUtils.httpResponseCode(url, "GET") should be (200)
+ TestUtils.httpResponseCode(url, "POST") should be (200)
}
}
}
@@ -671,17 +671,6 @@ class UISeleniumSuite extends SparkFunSuite with WebBrowser with Matchers with B
}
}
- def getResponseCode(url: URL, method: String): Int = {
- val connection = url.openConnection().asInstanceOf[HttpURLConnection]
- connection.setRequestMethod(method)
- try {
- connection.connect()
- connection.getResponseCode()
- } finally {
- connection.disconnect()
- }
- }
-
def goToUi(sc: SparkContext, path: String): Unit = {
goToUi(sc.ui.get, path)
}
http://git-wip-us.apache.org/repos/asf/spark/blob/59502bbc/core/src/test/scala/org/apache/spark/ui/UISuite.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/org/apache/spark/ui/UISuite.scala b/core/src/test/scala/org/apache/spark/ui/UISuite.scala
index 4abcfb7..7c3d891 100644
--- a/core/src/test/scala/org/apache/spark/ui/UISuite.scala
+++ b/core/src/test/scala/org/apache/spark/ui/UISuite.scala
@@ -18,12 +18,12 @@
package org.apache.spark.ui
import java.net.{BindException, ServerSocket}
-import java.net.URI
-import javax.servlet.http.HttpServletRequest
+import java.net.{URI, URL}
+import javax.servlet.http.{HttpServlet, HttpServletRequest, HttpServletResponse}
import scala.io.Source
-import org.eclipse.jetty.servlet.ServletContextHandler
+import org.eclipse.jetty.servlet.{ServletContextHandler, ServletHolder}
import org.mockito.Mockito.{mock, when}
import org.scalatest.concurrent.Eventually._
import org.scalatest.time.SpanSugar._
@@ -167,6 +167,7 @@ class UISuite extends SparkFunSuite {
val boundPort = serverInfo.boundPort
assert(server.getState === "STARTED")
assert(boundPort != 0)
+ assert(serverInfo.securePort.isDefined)
intercept[BindException] {
socket = new ServerSocket(boundPort)
}
@@ -228,8 +229,55 @@ class UISuite extends SparkFunSuite {
assert(newHeader === null)
}
+ test("http -> https redirect applies to all URIs") {
+ var serverInfo: ServerInfo = null
+ try {
+ val servlet = new HttpServlet() {
+ override def doGet(req: HttpServletRequest, res: HttpServletResponse): Unit = {
+ res.sendError(HttpServletResponse.SC_OK)
+ }
+ }
+
+ def newContext(path: String): ServletContextHandler = {
+ val ctx = new ServletContextHandler()
+ ctx.setContextPath(path)
+ ctx.addServlet(new ServletHolder(servlet), "/root")
+ ctx
+ }
+
+ val (conf, sslOptions) = sslEnabledConf()
+ serverInfo = JettyUtils.startJettyServer("0.0.0.0", 0, sslOptions,
+ Seq[ServletContextHandler](newContext("/"), newContext("/test1")),
+ conf)
+ assert(serverInfo.server.getState === "STARTED")
+
+ val testContext = newContext("/test2")
+ serverInfo.addHandler(testContext)
+ testContext.start()
+
+ val httpPort = serverInfo.boundPort
+
+ val tests = Seq(
+ ("http", serverInfo.boundPort, HttpServletResponse.SC_FOUND),
+ ("https", serverInfo.securePort.get, HttpServletResponse.SC_OK))
+
+ tests.foreach { case (scheme, port, expected) =>
+ val urls = Seq(
+ s"$scheme://localhost:$port/root",
+ s"$scheme://localhost:$port/test1/root",
+ s"$scheme://localhost:$port/test2/root")
+ urls.foreach { url =>
+ val rc = TestUtils.httpResponseCode(new URL(url))
+ assert(rc === expected, s"Unexpected status $rc for $url")
+ }
+ }
+ } finally {
+ stopServer(serverInfo)
+ }
+ }
+
def stopServer(info: ServerInfo): Unit = {
- if (info != null && info.server != null) info.server.stop
+ if (info != null) info.stop()
}
def closeSocket(socket: ServerSocket): Unit = {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org