[SPARK-59353][UI] Avoid pre-fetch and crawlers triggering actions in UI

core/src/main/scala/org/apache/spark/deploy/master/ui/MasterPage.scala

@@ -338,6 +338,7 @@ private[ui] class MasterPage(parent: MasterWebUI) extends WebUIPage("") {
       <form action="app/kill/" method="POST" class="d-inline float-end">
         <input type="hidden" name="id" value={app.id}/>
         <input type="hidden" name="terminate" value="true"/>
+        <input type="hidden" name="csrfToken" value={parent.csrfToken}/>
         <button type="submit"
                 data-kill-message={s"Are you sure you want to kill application ${app.id} ?"}
                 class="btn btn-sm btn-outline-danger kill-link">Kill</button>
@@ -358,6 +359,7 @@ private[ui] class MasterPage(parent: MasterWebUI) extends WebUIPage("") {
       val label = action.capitalize
       <form action={s"app/$action/"} method="POST" class="d-inline float-end">
         <input type="hidden" name="id" value={app.id}/>
+        <input type="hidden" name="csrfToken" value={parent.csrfToken}/>
         <button type="submit" data-confirm-message={message}
                 class="btn btn-sm btn-outline-secondary confirm-link">{label}</button>
       </form>
@@ -407,6 +409,7 @@ private[ui] class MasterPage(parent: MasterWebUI) extends WebUIPage("") {
       <form action="driver/kill/" method="POST" class="d-inline float-end">
         <input type="hidden" name="id" value={driver.id}/>
         <input type="hidden" name="terminate" value="true"/>
+        <input type="hidden" name="csrfToken" value={parent.csrfToken}/>
         <button type="submit"
                 data-kill-message={s"Are you sure you want to kill driver ${driver.id} ?"}
                 class="btn btn-sm btn-outline-danger kill-link">Kill</button>

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 {

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")

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

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.
    */

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

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
               <li>
                 <strong>Application:</strong>
                 {status}
-                <a href={s"$basePathUri/jobs/$action/"} role="button"
+                <a href={s"$basePathUri/jobs/$action/?csrfToken=${parent.csrfToken}"}
+                   role="button"
                    data-confirm-message={confirm}
                    class="btn btn-sm btn-outline-secondary confirm-link">{label}</a>
                 {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}"
-      <a href={killLinkUri} role="button"
-         data-kill-message={s"Are you sure you want to kill job ${job.jobId} ?"}
-         class="btn btn-sm btn-outline-danger kill-link float-end">Kill</a>
+      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.
+        <a href={s"$basePath/jobs/job/kill/?id=${job.jobId}&csrfToken=$csrfToken"}
+           role="button"
+           data-kill-message={killMessage}
+           class="btn btn-sm btn-outline-danger kill-link float-end">Kill</a>
+      } 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.
+        <form action={s"$basePath/jobs/job/kill/"} method="POST" class="d-inline float-end">
+          <input type="hidden" name="id" value={job.jobId.toString}/>
+          <input type="hidden" name="csrfToken" value={csrfToken}/>
+          <button type="submit"
+                  data-kill-message={killMessage}
+                  class="btn btn-sm btn-outline-danger kill-link">Kill</button>
+        </form>
+      }
     } else {
       Seq.empty
     }

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)))

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

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.

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 = <h4>Summary </h4> ++ poolTable.toNodeSeq(request)