diff --git a/core/src/main/scala/org/apache/spark/deploy/master/ui/MasterPage.scala b/core/src/main/scala/org/apache/spark/deploy/master/ui/MasterPage.scala index cd77876234238..c89f586f7e6de 100644 --- a/core/src/main/scala/org/apache/spark/deploy/master/ui/MasterPage.scala +++ b/core/src/main/scala/org/apache/spark/deploy/master/ui/MasterPage.scala @@ -338,6 +338,7 @@ private[ui] class MasterPage(parent: MasterWebUI) extends WebUIPage("") {
+ @@ -358,6 +359,7 @@ private[ui] class MasterPage(parent: MasterWebUI) extends WebUIPage("") { val label = action.capitalize +
@@ -407,6 +409,7 @@ private[ui] class MasterPage(parent: MasterWebUI) extends WebUIPage("") {
+ diff --git a/core/src/main/scala/org/apache/spark/deploy/master/ui/MasterWebUI.scala b/core/src/main/scala/org/apache/spark/deploy/master/ui/MasterWebUI.scala index 343cfc1b6da00..92e79fd33ebc7 100644 --- a/core/src/main/scala/org/apache/spark/deploy/master/ui/MasterWebUI.scala +++ b/core/src/main/scala/org/apache/spark/deploy/master/ui/MasterWebUI.scala @@ -65,17 +65,26 @@ class MasterWebUI( attachPage(masterPage) addStaticHandler(MasterWebUI.STATIC_RESOURCE_DIR) addRenderLogHandler(this, master.conf) + // These endpoints change state, so they require the per-UI CSRF token that MasterPage + // embeds in the forms it renders. /workers/kill is deliberately not guarded this way: + // it has no form, it is a documented endpoint operators call directly, and a caller + // with no token could not send one. It stays gated by POST, modify ACLs and + // spark.master.ui.decommission.allow.mode. if (killEnabled) { attachHandler(createRedirectHandler( - "/app/kill", "/", masterPage.handleAppKillRequest, httpMethods = Set("POST"))) + "/app/kill", "/", masterPage.handleAppKillRequest, httpMethods = Set("POST"), + csrfToken = Some(csrfToken))) attachHandler(createRedirectHandler( - "/driver/kill", "/", masterPage.handleDriverKillRequest, httpMethods = Set("POST"))) + "/driver/kill", "/", masterPage.handleDriverKillRequest, httpMethods = Set("POST"), + csrfToken = Some(csrfToken))) } if (holdEnabled) { attachHandler(createRedirectHandler( - "/app/hold", "/", masterPage.handleAppHoldRequest, httpMethods = Set("POST"))) + "/app/hold", "/", masterPage.handleAppHoldRequest, httpMethods = Set("POST"), + csrfToken = Some(csrfToken))) attachHandler(createRedirectHandler( - "/app/resume", "/", masterPage.handleAppResumeRequest, httpMethods = Set("POST"))) + "/app/resume", "/", masterPage.handleAppResumeRequest, httpMethods = Set("POST"), + csrfToken = Some(csrfToken))) } if (decommissionEnabled) { attachHandler(createServletHandler("/workers/kill", new HttpServlet { diff --git a/core/src/main/scala/org/apache/spark/internal/config/UI.scala b/core/src/main/scala/org/apache/spark/internal/config/UI.scala index e70743fa9df0e..3fb5e00ee0320 100644 --- a/core/src/main/scala/org/apache/spark/internal/config/UI.scala +++ b/core/src/main/scala/org/apache/spark/internal/config/UI.scala @@ -105,6 +105,21 @@ private[spark] object UI { .booleanConf .createWithDefault(true) + val UI_KILL_VIA_GET_ENABLED = ConfigBuilder("spark.ui.killViaGetEnabled") + .doc("Whether the job/stage kill endpoints of the web UI accept HTTP GET requests in " + + "addition to POST. Unset, this defaults to true when spark.master is yarn, because " + + "the YARN ResourceManager/AM proxy does not forward POST requests (SPARK-6846), and " + + "to false everywhere else. Either way the state-changing endpoints require the " + + "random per-UI CSRF token embedded in the links and forms the UI renders, and " + + "reject prefetch requests (Purpose/Sec-Purpose/X-Moz headers) and HEAD requests, so " + + "forged cross-site requests and incidental link fetches cannot trigger them; " + + "prefetch rejection relies on the prefetcher identifying itself via those headers. " + + "Introduced in 4.3.0; also available in 3.5.10, 4.0.5, 4.1.4 and 4.2.1; and in all " + + "versions after 4.3.0.") + .version("4.3.0") + .booleanConf + .createOptional + val UI_THREAD_DUMPS_ENABLED = ConfigBuilder("spark.ui.threadDumpsEnabled") .doc("Whether to show a link for executor thread dumps in Stages and Executor pages.") .version("1.2.0") 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 164a3bccf2f39..163af8d546646 100644 --- a/core/src/main/scala/org/apache/spark/ui/JettyUtils.scala +++ b/core/src/main/scala/org/apache/spark/ui/JettyUtils.scala @@ -18,7 +18,9 @@ package org.apache.spark.ui import java.net.{URI, URL, URLDecoder} -import java.util.{EnumSet, List => JList} +import java.nio.charset.StandardCharsets.UTF_8 +import java.security.{MessageDigest, SecureRandom} +import java.util.{EnumSet, HexFormat, List => JList, Locale} import scala.jdk.CollectionConverters._ import scala.language.implicitConversions @@ -127,13 +129,25 @@ private[spark] object JettyUtils extends Logging { contextHandler } - /** Create a handler that always redirects the user to the given path */ + /** + * Create a handler that always redirects the user to the given path. + * + * @param csrfToken when defined, mark the handler as state-changing and guard it: + * requests must carry the token as the "csrfToken" parameter or they + * are rejected with 403 (see isValidCsrfToken), prefetch requests are + * rejected with 403 (see isPrefetchRequest), and HEAD is refused with + * 405 instead of being delegated to doGet. A cross-site page can + * neither read the token out of the UI's pages (same-origin policy) + * nor guess it, so it cannot forge a request. Set this for handlers + * that change state (e.g. the job/stage kill endpoints). + */ def createRedirectHandler( srcPath: String, destPath: String, beforeRedirect: HttpServletRequest => Unit = x => (), basePath: String = "", - httpMethods: Set[String] = Set("GET")): ServletContextHandler = { + httpMethods: Set[String] = Set("GET"), + csrfToken: Option[String] = None): ServletContextHandler = { val prefixedDestPath = basePath + destPath val servlet = new HttpServlet { override def doGet(request: HttpServletRequest, response: HttpServletResponse): Unit = { @@ -150,7 +164,26 @@ private[spark] object JettyUtils extends Logging { response.sendError(HttpServletResponse.SC_METHOD_NOT_ALLOWED) } } + // HEAD must be safe (RFC 9110) but HttpServlet.doHead delegates to doGet, which would + // run beforeRedirect; refuse to route HEAD to a state-changing handler at all. + protected override def doHead(req: HttpServletRequest, res: HttpServletResponse): Unit = { + if (csrfToken.isDefined) { + res.sendError(HttpServletResponse.SC_METHOD_NOT_ALLOWED) + } else { + super.doHead(req, res) + } + } private def doRequest(request: HttpServletRequest, response: HttpServletResponse): Unit = { + if (csrfToken.isDefined && isPrefetchRequest(request)) { + response.sendError(HttpServletResponse.SC_FORBIDDEN, + "Prefetch request rejected on a state-changing endpoint.") + return + } + if (!csrfToken.forall(isValidCsrfToken(request, _))) { + response.sendError(HttpServletResponse.SC_FORBIDDEN, + "Missing or invalid CSRF token on a state-changing endpoint.") + return + } beforeRedirect(request) // Make sure we don't end up with "//" in the middle val requestURL = new URI(request.getRequestURL.toString).toURL @@ -167,6 +200,44 @@ private[spark] object JettyUtils extends Logging { createServletHandler(srcPath, servlet, basePath) } + /** + * True when the request identifies itself as a prefetch rather than a deliberate user + * navigation: Chrome and derivatives send "Sec-Purpose: prefetch" (or the older + * "Purpose: prefetch", and compound values such as "prefetch;prerender"), and Firefox + * sends "X-Moz: prefetch". State-changing endpoints reject these so that a link + * prefetcher cannot trigger the action: a prefetch of a kill link must not kill. + */ + private[spark] def isPrefetchRequest(request: HttpServletRequest): Boolean = { + val purpose = Option(request.getHeader("Sec-Purpose")) + .orElse(Option(request.getHeader("Purpose"))) + purpose.exists(_.toLowerCase(Locale.ROOT).contains("prefetch")) || + request.getHeader("X-Moz") != null + } + + /** + * A fresh token for guarding a UI's state-changing endpoints. 128 random bits, which a + * cross-site page cannot guess; see createRedirectHandler and isValidCsrfToken. + */ + private[spark] def newCsrfToken(): String = { + val bytes = new Array[Byte](16) + new SecureRandom().nextBytes(bytes) + HexFormat.of().formatHex(bytes) + } + + /** + * Synchronizer-token check for state-changing endpoints: the request must carry the + * per-UI random token (embedded in the links and forms the UI renders) as the + * "csrfToken" parameter. A cross-site page cannot read the token out of the UI's pages + * (same-origin policy) and cannot guess it, so forged kill/hold requests fail here + * regardless of which headers the browser sends or suppresses. Constant-time + * comparison, though the token is not exactly timing-sensitive at 128 bits. + */ + private[spark] def isValidCsrfToken(request: HttpServletRequest, expected: String): Boolean = { + Option(request.getParameter("csrfToken")).exists { provided => + MessageDigest.isEqual(provided.getBytes(UTF_8), expected.getBytes(UTF_8)) + } + } + /** Create a handler for serving files from a static directory */ def createStaticHandler(resourceBase: String, path: String): ServletContextHandler = { val contextHandler = new ServletContextHandler diff --git a/core/src/main/scala/org/apache/spark/ui/SparkUI.scala b/core/src/main/scala/org/apache/spark/ui/SparkUI.scala index 97c5d16873445..e29f576937308 100644 --- a/core/src/main/scala/org/apache/spark/ui/SparkUI.scala +++ b/core/src/main/scala/org/apache/spark/ui/SparkUI.scala @@ -27,6 +27,7 @@ import org.apache.spark.internal.Logging import org.apache.spark.internal.LogKeys.{CLASS_NAME, WEB_URL} import org.apache.spark.internal.config.DRIVER_LOG_LOCAL_DIR import org.apache.spark.internal.config.UI._ +import org.apache.spark.launcher.SparkLauncher import org.apache.spark.scheduler._ import org.apache.spark.status.AppStatusStore import org.apache.spark.status.api.v1._ @@ -55,6 +56,8 @@ private[spark] class SparkUI private ( val killEnabled = sc.map(_.conf.get(UI_KILL_ENABLED)).getOrElse(false) + val killViaGetEnabled = SparkUI.killViaGetEnabled(conf) + val holdEnabled = sc.map(_.conf.get(UI_HOLD_ENABLED)).getOrElse(false) var appId: String = _ @@ -121,16 +124,24 @@ private[spark] class SparkUI private ( attachHandler(PrometheusResource.getServletHandler(this)) } - // These should be POST only, but, the YARN AM proxy won't proxy POSTs + // These endpoints change state, so they require the per-UI CSRF token and reject + // prefetch requests (see JettyUtils.createRedirectHandler). Kill also accepts GET + // where a proxy in front of the UI cannot forward POST -- the token rides in the + // request parameters, which such proxies do forward. See SparkUI.killViaGetEnabled. + val killHttpMethods: Set[String] = + if (killViaGetEnabled) Set("GET", "POST") else Set("POST") attachHandler(createRedirectHandler( - "/jobs/job/kill", "/jobs/", jobsTab.handleKillRequest, httpMethods = Set("GET", "POST"))) + "/jobs/job/kill", "/jobs/", jobsTab.handleKillRequest, httpMethods = killHttpMethods, + csrfToken = Some(csrfToken))) attachHandler(createRedirectHandler( "/stages/stage/kill", "/stages/", stagesTab.handleKillRequest, - httpMethods = Set("GET", "POST"))) + httpMethods = killHttpMethods, csrfToken = Some(csrfToken))) attachHandler(createRedirectHandler( - "/jobs/hold", "/jobs/", jobsTab.handleHoldRequest, httpMethods = Set("GET", "POST"))) + "/jobs/hold", "/jobs/", jobsTab.handleHoldRequest, httpMethods = Set("GET", "POST"), + csrfToken = Some(csrfToken))) attachHandler(createRedirectHandler( - "/jobs/resume", "/jobs/", jobsTab.handleResumeRequest, httpMethods = Set("GET", "POST"))) + "/jobs/resume", "/jobs/", jobsTab.handleResumeRequest, httpMethods = Set("GET", "POST"), + csrfToken = Some(csrfToken))) } initialize() @@ -256,6 +267,16 @@ private[spark] object SparkUI { conf.get(UI_PORT) } + /** + * Whether the kill endpoints accept GET as well as POST. Unset, this follows the cluster + * manager: on for YARN, whose ResourceManager/AM proxy does not forward POST + * (SPARK-6846), and off everywhere else. + */ + def killViaGetEnabled(conf: SparkConf): Boolean = { + conf.get(UI_KILL_VIA_GET_ENABLED) + .getOrElse(conf.get(SparkLauncher.SPARK_MASTER, null) == "yarn") + } + /** * Create a new UI backed by an AppStatusStore. */ 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 7f8f2556dd088..70c3239320a02 100644 --- a/core/src/main/scala/org/apache/spark/ui/WebUI.scala +++ b/core/src/main/scala/org/apache/spark/ui/WebUI.scala @@ -59,6 +59,11 @@ private[spark] abstract class WebUI( conf.get(DRIVER_HOST_ADDRESS)) protected val className = Utils.getFormattedClassName(this) + // Random per-UI token embedded in the links and forms that hit this UI's state-changing + // endpoints. A cross-site page can neither read it (same-origin policy) nor guess it, so + // forged requests fail the check in JettyUtils.createRedirectHandler. + private[spark] val csrfToken: String = newCsrfToken() + def getBasePath: String = basePath def getTabs: Seq[WebUITab] = tabs.toSeq def getHandlers: Seq[ServletContextHandler] = handlers.toSeq diff --git a/core/src/main/scala/org/apache/spark/ui/jobs/AllJobsPage.scala b/core/src/main/scala/org/apache/spark/ui/jobs/AllJobsPage.scala index daf04085fc022..4d5ebf92ef054 100644 --- a/core/src/main/scala/org/apache/spark/ui/jobs/AllJobsPage.scala +++ b/core/src/main/scala/org/apache/spark/ui/jobs/AllJobsPage.scala @@ -260,7 +260,9 @@ private[ui] class AllJobsPage(parent: JobsTab, store: AppStatusStore) extends We UIUtils.prependBaseUri(request, parent.basePath), "jobs", // subPath killEnabled, - jobIdTitle + jobIdTitle, + parent.killViaGetEnabled, + parent.csrfToken ).table(jobPage) } catch { case e @ (_ : IllegalArgumentException | _ : IndexOutOfBoundsException) => @@ -373,7 +375,8 @@ private[ui] class AllJobsPage(parent: JobsTab, store: AppStatusStore) extends We
  • Application: {status} - {label} {parent.lastHoldRequestStatus.getOrElse("")} @@ -554,7 +557,9 @@ private[ui] class JobPagedTable( basePath: String, subPath: String, killEnabled: Boolean, - jobIdTitle: String + jobIdTitle: String, + killViaGetEnabled: Boolean, + csrfToken: String ) extends PagedTable[JobTableRowData] { private val (sortColumn, desc, pageSize) = getTableParameters(request, jobTag, jobIdTitle) @@ -613,11 +618,27 @@ private[ui] class JobPagedTable( val job = jobTableRow.jobData val killLink = if (killEnabled) { - // SPARK-6846 this should be POST-only but YARN AM won't proxy POST - val killLinkUri = s"$basePath/jobs/job/kill/?id=${job.jobId}" - Kill + val killMessage = s"Are you sure you want to kill job ${job.jobId} ?" + if (killViaGetEnabled) { + // Default: a plain GET link, which also works through proxies that do not forward + // POST, such as the YARN ResourceManager/AM proxy (SPARK-6846). The endpoint + // requires the CSRF token and rejects prefetch requests (see SparkUI.initialize), + // and webui.js gates the click on the confirmation dialog. + Kill + } else { + // POST-only mode (spark.ui.killViaGetEnabled=false): submit the kill as a form, + // the same pattern the master UI uses for killing applications and drivers. + + + + +
  • + } } else { Seq.empty } diff --git a/core/src/main/scala/org/apache/spark/ui/jobs/AllStagesPage.scala b/core/src/main/scala/org/apache/spark/ui/jobs/AllStagesPage.scala index 6895db261330b..04df1c6722ba6 100644 --- a/core/src/main/scala/org/apache/spark/ui/jobs/AllStagesPage.scala +++ b/core/src/main/scala/org/apache/spark/ui/jobs/AllStagesPage.scala @@ -97,7 +97,8 @@ private[ui] class AllStagesPage(parent: StagesTab) extends WebUIPage("") { val stagesTable = new StageTableBase(parent.store, request, stages, statusName(status), stageTag(status), - parent.basePath, subPath, parent.isFairScheduler, killEnabled, isFailedStage) + parent.basePath, subPath, parent.isFairScheduler, killEnabled, isFailedStage, + parent.killViaGetEnabled, parent.csrfToken) val stagesSize = stages.size (Some(summary(appSummary, status, stagesSize)), Some(table(appSummary, status, stagesTable, stagesSize))) diff --git a/core/src/main/scala/org/apache/spark/ui/jobs/JobPage.scala b/core/src/main/scala/org/apache/spark/ui/jobs/JobPage.scala index da82afe34aedc..0c39dbf2dba5b 100644 --- a/core/src/main/scala/org/apache/spark/ui/jobs/JobPage.scala +++ b/core/src/main/scala/org/apache/spark/ui/jobs/JobPage.scala @@ -341,19 +341,23 @@ private[ui] class JobPage(parent: JobsTab, store: AppStatusStore) extends WebUIP val activeStagesTable = new StageTableBase(store, request, activeStages.toSeq, "active", "activeStage", parent.basePath, basePath, parent.isFairScheduler, - killEnabled = parent.killEnabled, isFailedStage = false) + killEnabled = parent.killEnabled, isFailedStage = false, + killViaGetEnabled = parent.killViaGetEnabled, csrfToken = parent.csrfToken) val pendingOrSkippedStagesTable = new StageTableBase(store, request, pendingOrSkippedStages.toSeq, pendingOrSkippedTableId, "pendingStage", parent.basePath, basePath, parent.isFairScheduler, - killEnabled = false, isFailedStage = false) + killEnabled = false, isFailedStage = false, + killViaGetEnabled = parent.killViaGetEnabled, csrfToken = parent.csrfToken) val completedStagesTable = new StageTableBase(store, request, completedStages.toSeq, "completed", "completedStage", parent.basePath, basePath, parent.isFairScheduler, - killEnabled = false, isFailedStage = false) + killEnabled = false, isFailedStage = false, + killViaGetEnabled = parent.killViaGetEnabled, csrfToken = parent.csrfToken) val failedStagesTable = new StageTableBase(store, request, failedStages.toSeq, "failed", "failedStage", parent.basePath, basePath, parent.isFairScheduler, - killEnabled = false, isFailedStage = true) + killEnabled = false, isFailedStage = true, + killViaGetEnabled = parent.killViaGetEnabled, csrfToken = parent.csrfToken) val shouldShowActiveStages = activeStages.nonEmpty val shouldShowPendingStages = !isComplete && pendingOrSkippedStages.nonEmpty diff --git a/core/src/main/scala/org/apache/spark/ui/jobs/JobsTab.scala b/core/src/main/scala/org/apache/spark/ui/jobs/JobsTab.scala index 98ae0de22fb3f..9665f063587e1 100644 --- a/core/src/main/scala/org/apache/spark/ui/jobs/JobsTab.scala +++ b/core/src/main/scala/org/apache/spark/ui/jobs/JobsTab.scala @@ -35,6 +35,8 @@ private[ui] class JobsTab(parent: SparkUI, store: AppStatusStore) val sc = parent.sc val conf = parent.conf val killEnabled = parent.killEnabled + val killViaGetEnabled = parent.killViaGetEnabled + val csrfToken = parent.csrfToken val holdEnabled = parent.holdEnabled // Show pool information for only live UI. diff --git a/core/src/main/scala/org/apache/spark/ui/jobs/PoolPage.scala b/core/src/main/scala/org/apache/spark/ui/jobs/PoolPage.scala index f5c4a2bf21e4e..50514407a9a2e 100644 --- a/core/src/main/scala/org/apache/spark/ui/jobs/PoolPage.scala +++ b/core/src/main/scala/org/apache/spark/ui/jobs/PoolPage.scala @@ -44,7 +44,8 @@ private[ui] class PoolPage(parent: StagesTab) extends WebUIPage("pool") { val activeStages = uiPool.stageIds.toSeq.map(parent.store.lastStageAttempt(_)) val activeStagesTable = new StageTableBase(parent.store, request, activeStages, "", "activeStage", parent.basePath, - "stages/pool", parent.isFairScheduler, parent.killEnabled, false) + "stages/pool", parent.isFairScheduler, parent.killEnabled, false, + parent.killViaGetEnabled, parent.csrfToken) val poolTable = new PoolTable(Map(pool -> uiPool), parent) var content =

    Summary

    ++ poolTable.toNodeSeq(request) diff --git a/core/src/main/scala/org/apache/spark/ui/jobs/StageTable.scala b/core/src/main/scala/org/apache/spark/ui/jobs/StageTable.scala index aa5720bbf240b..f47f7f85fc394 100644 --- a/core/src/main/scala/org/apache/spark/ui/jobs/StageTable.scala +++ b/core/src/main/scala/org/apache/spark/ui/jobs/StageTable.scala @@ -40,7 +40,9 @@ private[ui] class StageTableBase( subPath: String, isFairScheduler: Boolean, killEnabled: Boolean, - isFailedStage: Boolean) { + isFailedStage: Boolean, + killViaGetEnabled: Boolean, + csrfToken: String) { val stagePage = Option(request.getParameter(stageTag + ".page")).map(_.toInt).getOrElse(1) @@ -56,6 +58,8 @@ private[ui] class StageTableBase( subPath, isFairScheduler, killEnabled, + killViaGetEnabled, + csrfToken, currentTime, isFailedStage, request @@ -101,6 +105,8 @@ private[ui] class StagePagedTable( subPath: String, isFairScheduler: Boolean, killEnabled: Boolean, + killViaGetEnabled: Boolean, + csrfToken: String, currentTime: Long, isFailedStage: Boolean, request: HttpServletRequest) extends PagedTable[StageTableRowData] { @@ -221,11 +227,27 @@ private[ui] class StagePagedTable( val basePathUri = UIUtils.prependBaseUri(request, basePath) val killLink = if (killEnabled) { - // SPARK-6846 this should be POST-only but YARN AM won't proxy POST - val killLinkUri = s"$basePathUri/stages/stage/kill/?id=${s.stageId}" - Kill + val killMessage = s"Are you sure you want to kill stage ${s.stageId} ?" + if (killViaGetEnabled) { + // Default: a plain GET link, which also works through proxies that do not forward + // POST, such as the YARN ResourceManager/AM proxy (SPARK-6846). The endpoint + // requires the CSRF token and rejects prefetch requests (see SparkUI.initialize), + // and webui.js gates the click on the confirmation dialog. + Kill + } else { + // POST-only mode (spark.ui.killViaGetEnabled=false): submit the kill as a form, + // the same pattern the master UI uses for killing applications and drivers. +
    + + + +
    + } } else { Seq.empty } diff --git a/core/src/main/scala/org/apache/spark/ui/jobs/StagesTab.scala b/core/src/main/scala/org/apache/spark/ui/jobs/StagesTab.scala index 8be4998827aca..268c1e80d69c3 100644 --- a/core/src/main/scala/org/apache/spark/ui/jobs/StagesTab.scala +++ b/core/src/main/scala/org/apache/spark/ui/jobs/StagesTab.scala @@ -33,6 +33,8 @@ private[ui] class StagesTab(val parent: SparkUI, val store: AppStatusStore) val sc = parent.sc val conf = parent.conf val killEnabled = parent.killEnabled + val killViaGetEnabled = parent.killViaGetEnabled + val csrfToken = parent.csrfToken val threadDumpEnabled = parent.sc.isDefined && parent.conf.get(UI_THREAD_DUMPS_ENABLED) diff --git a/core/src/test/scala/org/apache/spark/deploy/master/ui/MasterWebUISuite.scala b/core/src/test/scala/org/apache/spark/deploy/master/ui/MasterWebUISuite.scala index 53716e980eacc..4cca6e394037b 100644 --- a/core/src/test/scala/org/apache/spark/deploy/master/ui/MasterWebUISuite.scala +++ b/core/src/test/scala/org/apache/spark/deploy/master/ui/MasterWebUISuite.scala @@ -26,7 +26,7 @@ import scala.collection.mutable.HashMap import scala.io.Source import jakarta.servlet.http.HttpServletResponse.SC_FORBIDDEN -import org.mockito.Mockito.{mock, times, verify, when} +import org.mockito.Mockito.{mock, never, times, verify, when} import org.apache.spark.{SecurityManager, SparkConf, SparkFunSuite} import org.apache.spark.deploy.DeployMessages.{DecommissionWorkersOnHosts, KillDriverResponse, MasterStateResponse, RequestApplicationHold, RequestKillDriver, RequestMasterState} @@ -73,7 +73,8 @@ class MasterWebUISuite extends SparkFunSuite { when(master.idToApp).thenReturn(HashMap[String, ApplicationInfo]((activeApp.id, activeApp))) val url = s"http://${Utils.localHostNameForURI()}:${masterWebUI.boundPort}/app/kill/" - val body = convPostDataToString(Map(("id", activeApp.id), ("terminate", "true"))) + val body = convPostDataToString( + Map(("id", activeApp.id), ("terminate", "true"), ("csrfToken", masterWebUI.csrfToken))) val conn = sendHttpRequest(url, "POST", body) conn.getResponseCode @@ -84,7 +85,8 @@ class MasterWebUISuite extends SparkFunSuite { test("kill driver") { val activeDriverId = "driver-0" val url = s"http://${Utils.localHostNameForURI()}:${masterWebUI.boundPort}/driver/kill/" - val body = convPostDataToString(Map(("id", activeDriverId), ("terminate", "true"))) + val body = convPostDataToString( + Map(("id", activeDriverId), ("terminate", "true"), ("csrfToken", masterWebUI.csrfToken))) val conn = sendHttpRequest(url, "POST", body) conn.getResponseCode @@ -92,10 +94,53 @@ class MasterWebUISuite extends SparkFunSuite { verify(masterEndpointRef, times(1)).ask[KillDriverResponse](RequestKillDriver(activeDriverId)) } + test("state-changing master UI endpoints reject a request without the CSRF token") { + val appDesc = createAppDesc() + val activeApp = new ApplicationInfo( + new Date().getTime, "app-notoken", appDesc, new Date(), null, Int.MaxValue) + when(master.idToApp).thenReturn(HashMap[String, ApplicationInfo]((activeApp.id, activeApp))) + + // Without the token the request is refused before the handler runs, so the master is + // never asked to remove the application. + Seq("app/kill", "driver/kill", "app/hold", "app/resume").foreach { path => + val url = s"http://${Utils.localHostNameForURI()}:${masterWebUI.boundPort}/$path/" + val body = convPostDataToString(Map(("id", activeApp.id), ("terminate", "true"))) + assert(sendHttpRequest(url, "POST", body).getResponseCode === SC_FORBIDDEN, + s"$path accepted a request with no CSRF token") + } + verify(master, never()).removeApplication(activeApp, ApplicationState.KILLED) + } + + test("state-changing master UI endpoints reject a wrong CSRF token") { + val url = s"http://${Utils.localHostNameForURI()}:${masterWebUI.boundPort}/app/kill/" + val body = convPostDataToString( + Map(("id", "app-wrongtoken"), ("terminate", "true"), ("csrfToken", "not-the-token"))) + assert(sendHttpRequest(url, "POST", body).getResponseCode === SC_FORBIDDEN) + } + + test("the master page embeds the CSRF token in the forms it renders") { + val app = new ApplicationInfo( + new Date().getTime, "app-render", createAppDesc(), new Date(), null, Int.MaxValue) + app.holdSupported = true + val state = new MasterStateResponse( + "host", 8080, None, Array.empty, Array(app), Array.empty, + Array.empty, Array.empty, RecoveryState.ALIVE) + when(masterEndpointRef.askSync[MasterStateResponse](RequestMasterState)).thenReturn(state) + val url = s"http://${Utils.localHostNameForURI()}:${masterWebUI.boundPort}/" + val rendered = + Source.fromInputStream(sendHttpRequest(url, "GET", "").getInputStream).mkString + + // Both the kill and the hold form carry it, so the browser can drive them. + assert(rendered.contains("app/kill/") && rendered.contains("app/hold/")) + assert(rendered.split("csrfToken").length - 1 >= 2, "expected a token in each form") + assert(rendered.contains(masterWebUI.csrfToken)) + } + private def testHoldApplication(action: String, hold: Boolean): Unit = { val appId = s"app-$action" val url = s"http://${Utils.localHostNameForURI()}:${masterWebUI.boundPort}/app/$action/" - val conn = sendHttpRequest(url, "POST", convPostDataToString(Map(("id", appId)))) + val conn = sendHttpRequest(url, "POST", + convPostDataToString(Map(("id", appId), ("csrfToken", masterWebUI.csrfToken)))) conn.getResponseCode // Verify that the master was asked to forward the request to the driver of that application diff --git a/core/src/test/scala/org/apache/spark/ui/JettyUtilsSuite.scala b/core/src/test/scala/org/apache/spark/ui/JettyUtilsSuite.scala new file mode 100644 index 0000000000000..4eb610a559191 --- /dev/null +++ b/core/src/test/scala/org/apache/spark/ui/JettyUtilsSuite.scala @@ -0,0 +1,64 @@ +/* + * 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.spark.ui + +import jakarta.servlet.http.HttpServletRequest +import org.mockito.Mockito.{mock, when} + +import org.apache.spark.SparkFunSuite + +class JettyUtilsSuite extends SparkFunSuite { + + private def newRequest( + secPurpose: String = null, + purpose: String = null, + xMoz: String = null, + csrfTokenParam: String = null): HttpServletRequest = { + val req = mock(classOf[HttpServletRequest]) + when(req.getHeader("Sec-Purpose")).thenReturn(secPurpose) + when(req.getHeader("Purpose")).thenReturn(purpose) + when(req.getHeader("X-Moz")).thenReturn(xMoz) + when(req.getParameter("csrfToken")).thenReturn(csrfTokenParam) + req + } + + test("isPrefetchRequest is false for ordinary requests") { + assert(!JettyUtils.isPrefetchRequest(newRequest())) + } + + test("isPrefetchRequest detects browser prefetch headers") { + assert(JettyUtils.isPrefetchRequest(newRequest(secPurpose = "prefetch"))) + assert(JettyUtils.isPrefetchRequest(newRequest(secPurpose = "prefetch;prerender"))) + assert(JettyUtils.isPrefetchRequest(newRequest(purpose = "prefetch"))) + assert(JettyUtils.isPrefetchRequest(newRequest(xMoz = "prefetch"))) + } + + test("isValidCsrfToken accepts only the matching token") { + val token = "0123456789abcdef0123456789abcdef" + assert(JettyUtils.isValidCsrfToken(newRequest(csrfTokenParam = token), token)) + assert(!JettyUtils.isValidCsrfToken(newRequest(csrfTokenParam = "bogus"), token)) + assert(!JettyUtils.isValidCsrfToken( + newRequest(csrfTokenParam = token + "00"), token)) + assert(!JettyUtils.isValidCsrfToken(newRequest(), token)) + } + + test("isValidCsrfToken rejects without throwing on empty or malformed input") { + assert(!JettyUtils.isValidCsrfToken(newRequest(csrfTokenParam = ""), "abc")) + assert(!JettyUtils.isValidCsrfToken(newRequest(csrfTokenParam = "abc"), "")) + } +} 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 fde52ed57a3d0..406b9bb20a77a 100644 --- a/core/src/test/scala/org/apache/spark/ui/UISeleniumSuite.scala +++ b/core/src/test/scala/org/apache/spark/ui/UISeleniumSuite.scala @@ -24,6 +24,7 @@ import scala.io.Source import scala.xml.Node import jakarta.servlet.http.{HttpServletRequest, HttpServletResponse} +import org.eclipse.jetty.client.{HttpClient, StringRequestContent} import org.glassfish.jersey.internal.util.collection.MultivaluedStringMap import org.htmlunit.DefaultCssErrorHandler import org.htmlunit.cssparser.parser.CSSParseException @@ -593,28 +594,151 @@ class UISeleniumSuite extends SparkFunSuite with WebBrowser with Matchers { } } - test("kill stage POST/GET response is correct") { + // The state-changing endpoints require the per-UI token rendered into the kill links + // (or the hidden form field in POST-only mode); scrape it the way a scripted client + // would. + private def scrapeCsrfToken(sc: SparkContext): String = { + val html = Utils.tryWithResource( + Source.fromURL(sc.ui.get.webUrl.stripSuffix("/") + "/jobs/"))(_.mkString) + """(?:csrfToken=|name="csrfToken" value=")([0-9a-f]+)""".r + .findFirstMatchIn(html) + .map(_.group(1)) + .getOrElse(fail("no CSRF token found on the jobs page")) + } + + test("kill stage requires the CSRF token, rejecting prefetch and HEAD requests") { + // GET mode is off by default outside YARN, and this test is about the token rather than + // the default, so ask for GET explicitly. + withSpark(newSparkContext(killEnabled = true, + additionalConfs = Map(UI_KILL_VIA_GET_ENABLED.key -> "true"))) { sc => + sc.parallelize(1 to 10).map{x => Thread.sleep(10000); x}.countAsync() + // java.net.HttpURLConnection silently drops some headers, so use Jetty's client + // for requests that must carry specific prefetch headers. + val client = new HttpClient() + client.start() + try { + val base = sc.ui.get.webUrl.stripSuffix("/") + // Retry only until the kill link appears. Everything after this runs once: a request + // the endpoint accepts kills the job, and the link is then gone, so retrying the whole + // block could never succeed a second time -- it would fail in scrapeCsrfToken and + // report a missing token rather than whatever actually went wrong. + val token = eventually(timeout(5.seconds), interval(50.milliseconds))(scrapeCsrfToken(sc)) + val noToken = new URI(base + "/stages/stage/kill/?id=0").toURL + val withToken = new URI( + base + s"/stages/stage/kill/?id=0&csrfToken=$token").toURL + // Forged or scripted requests without the token, or with a wrong one, fail. + TestUtils.httpResponseCode(noToken, "GET") should be (403) + TestUtils.httpResponseCode( + new URI(base + "/stages/stage/kill/?id=0&csrfToken=bogus").toURL, + "GET") should be (403) + // HEAD must be safe (RFC 9110), so it is refused rather than delegated to doGet. + TestUtils.httpResponseCode(withToken, "HEAD") should be (405) + // ...but HEAD still works on unguarded redirect handlers ("/" -> "/jobs/"). + TestUtils.httpResponseCode(new URI(base + "/").toURL, "HEAD") should be (200) + // Browser link prefetchers identify themselves; a valid token must not + // save them, since the token rides in the link they prefetch. + client.newRequest(withToken.toURI).headers( + _.add("Sec-Purpose", "prefetch")).send().getStatus should be (403) + client.newRequest(withToken.toURI).headers( + _.add("Purpose", "prefetch")).send().getStatus should be (403) + client.newRequest(withToken.toURI).headers( + _.add("X-Moz", "prefetch")).send().getStatus should be (403) + // Last, because a deliberate click from the UI carries the token, goes through, and + // kills the stage. + TestUtils.httpResponseCode(withToken, "GET") should be (200) + TestUtils.httpResponseCode(withToken, "POST") should be (200) + } finally { + client.stop() + } + } + } + + test("kill job requires the CSRF token") { + withSpark(newSparkContext(killEnabled = true, + additionalConfs = Map(UI_KILL_VIA_GET_ENABLED.key -> "true"))) { sc => + sc.parallelize(1 to 10).map{x => Thread.sleep(10000); x}.countAsync() + val base = sc.ui.get.webUrl.stripSuffix("/") + // Retry only until the kill link appears; the accepted requests below kill the job. + val token = eventually(timeout(5.seconds), interval(50.milliseconds))(scrapeCsrfToken(sc)) + TestUtils.httpResponseCode( + new URI(base + "/jobs/job/kill/?id=0").toURL, "GET") should be (403) + TestUtils.httpResponseCode( + new URI(base + s"/jobs/job/kill/?id=0&csrfToken=$token").toURL, + "GET") should be (200) + TestUtils.httpResponseCode( + new URI(base + s"/jobs/job/kill/?id=0&csrfToken=$token").toURL, + "POST") should be (200) + } + } + + test("kill stage is POST-only when spark.ui.killViaGetEnabled is disabled") { + withSpark(newSparkContext(killEnabled = true, + additionalConfs = Map(UI_KILL_VIA_GET_ENABLED.key -> "false"))) { sc => + sc.parallelize(1 to 10).map{x => Thread.sleep(10000); x}.countAsync() + val client = new HttpClient() + client.start() + try { + eventually(timeout(5.seconds), interval(50.milliseconds)) { + val base = sc.ui.get.webUrl.stripSuffix("/") + val token = scrapeCsrfToken(sc) + val withToken = new URI( + base + s"/stages/stage/kill/?id=0&csrfToken=$token").toURL + TestUtils.httpResponseCode(withToken, "HEAD") should be (405) + TestUtils.httpResponseCode(withToken, "GET") should be (405) + TestUtils.httpResponseCode(withToken, "POST") should be (200) + // The browser path in this mode: the form posts id and token in the body. + client.POST(new URI(base + "/stages/stage/kill/")).body( + new StringRequestContent("application/x-www-form-urlencoded", + s"id=0&csrfToken=$token")).send().getStatus should be (200) + } + } finally { + client.stop() + } + } + } + + test("hold and resume endpoints require the CSRF token") { withSpark(newSparkContext(killEnabled = true)) { sc => sc.parallelize(1 to 10).map{x => Thread.sleep(10000); x}.countAsync() eventually(timeout(5.seconds), interval(50.milliseconds)) { - val url = new URI( - sc.ui.get.webUrl.stripSuffix("/") + "/stages/stage/kill/?id=0").toURL - // SPARK-6846: should be POST only but YARN AM doesn't proxy POST - TestUtils.httpResponseCode(url, "GET") should be (200) - TestUtils.httpResponseCode(url, "POST") should be (200) + val base = sc.ui.get.webUrl.stripSuffix("/") + val token = scrapeCsrfToken(sc) + // holdEnabled is off in this context, so the actions themselves no-op; what + // is exercised here is the guard wiring on the endpoints. + Seq("hold", "resume").foreach { action => + TestUtils.httpResponseCode( + new URI(base + s"/jobs/$action/").toURL, "GET") should be (403) + TestUtils.httpResponseCode( + new URI(base + s"/jobs/$action/?csrfToken=$token").toURL, + "GET") should be (200) + } } } } - test("kill job POST/GET response is correct") { + test("kill link rendering follows spark.ui.killViaGetEnabled") { + withSpark(newSparkContext(killEnabled = true, + additionalConfs = Map(UI_KILL_VIA_GET_ENABLED.key -> "true"))) { sc => + sc.parallelize(1 to 10).map{x => Thread.sleep(10000); x}.countAsync() + eventually(timeout(5.seconds), interval(50.milliseconds)) { + val html = Utils.tryWithResource( + Source.fromURL(sc.ui.get.webUrl.stripSuffix("/") + "/jobs/"))(_.mkString) + // GET mode: a link carrying the token, which also works through proxies that + // block POST. ("&" is the XML-escaped "&" in the rendered href.) + html should include regex ("/jobs/job/kill/\\?id=\\d+&csrfToken=[0-9a-f]+") + html should not include ("""action="/jobs/job/kill/"""") + } + } + // No explicit setting and spark.master is local, so the default resolves to POST-only. withSpark(newSparkContext(killEnabled = true)) { sc => sc.parallelize(1 to 10).map{x => Thread.sleep(10000); x}.countAsync() eventually(timeout(5.seconds), interval(50.milliseconds)) { - val url = new URI( - sc.ui.get.webUrl.stripSuffix("/") + "/jobs/job/kill/?id=0").toURL - // SPARK-6846: should be POST only but YARN AM doesn't proxy POST - TestUtils.httpResponseCode(url, "GET") should be (200) - TestUtils.httpResponseCode(url, "POST") should be (200) + val html = Utils.tryWithResource( + Source.fromURL(sc.ui.get.webUrl.stripSuffix("/") + "/jobs/"))(_.mkString) + // POST-only mode: the kill control is a form posting the job id and token. + html should not include ("/jobs/job/kill/?id=") + html should include ("""action="/jobs/job/kill/"""") + html should include ("""name="csrfToken"""") } } } 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 7a6f409c04efc..d73dcd6a4e591 100644 --- a/core/src/test/scala/org/apache/spark/ui/UISuite.scala +++ b/core/src/test/scala/org/apache/spark/ui/UISuite.scala @@ -567,6 +567,20 @@ class UISuite extends SparkFunSuite { stopServer(targetServer) } } + + test("kill via GET follows the cluster manager unless set explicitly") { + // The YARN ResourceManager/AM proxy does not forward POST (SPARK-6846), so GET is + // accepted there and refused everywhere else. + assert(SparkUI.killViaGetEnabled(new SparkConf().set("spark.master", "yarn"))) + assert(!SparkUI.killViaGetEnabled(new SparkConf().set("spark.master", "local[*]"))) + assert(!SparkUI.killViaGetEnabled(new SparkConf())) + + // An explicit setting wins in both directions. + assert(!SparkUI.killViaGetEnabled( + new SparkConf().set("spark.master", "yarn").set(UI.UI_KILL_VIA_GET_ENABLED, false))) + assert(SparkUI.killViaGetEnabled( + new SparkConf().set("spark.master", "local[*]").set(UI.UI_KILL_VIA_GET_ENABLED, true))) + } } // Filter for testing; returns a configurable code for every request. diff --git a/docs/configuration.md b/docs/configuration.md index e758c22bac5d3..f09e308fde6d6 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -1666,6 +1666,30 @@ Apart from these, the following properties are also available, and may be useful 1.0.0 + + spark.ui.killViaGetEnabled + true on YARN, false otherwise + + Whether the job/stage kill endpoints of the web UI accept HTTP GET requests in + addition to POST. Left unset, this follows the cluster manager: GET is accepted when + spark.master is yarn, because the YARN ResourceManager/AM + proxy does not forward POST requests, and refused everywhere else. + Either way the state-changing endpoints require the random per-UI CSRF token embedded + in the links and forms the UI renders, and reject prefetch requests (identified by + the Purpose, Sec-Purpose, or X-Moz headers) and HEAD requests, so forged cross-site + requests and incidental link fetches cannot trigger them. Scripted clients can read + the token from the jobs page before calling the endpoint. In GET mode the + token appears in the kill link URL, so it can be recorded in browser history and + server or proxy access logs; it is random per UI instance and grants nothing beyond + the UI's own state-changing endpoints. Prefetch rejection relies on the prefetcher + identifying itself via those headers; one that sends none of them is not detected. + In POST-only mode the kill links on the jobs and stages pages are rendered as POST + forms. + Introduced in 4.3.0; also available in 3.5.10, 4.0.5, 4.1.4 and 4.2.1; and in all + versions after 4.3.0. + + 4.3.0 + spark.ui.holdEnabled true