[SPARK-59405][SQL] XSD Validation improvements

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXMLRecordReader.scala

@@ -41,7 +41,7 @@ case class StaxXMLRecordReader(inputStream: () => InputStream, options: XmlOptio
   private lazy val primaryEventReader = StaxXmlParserUtils.filteredReader(in1, options)
 
   private val xsdSchemaValidator = Option(options.rowValidationXSDPath)
-    .map(path => ValidatorUtil.getSchema(path).newValidator())
+    .map(path => ValidatorUtil.newValidator(ValidatorUtil.getSchema(path)))
   // Reader for the XSD validation, if an XSD schema is provided.
   private lazy val in2 = xsdSchemaValidator.map(_ => inputStream())
   // An XMLStreamReader used by StAXSource for XSD validation.
@@ -103,7 +103,9 @@ case class StaxXMLRecordReader(inputStream: () => InputStream, options: XmlOptio
     while (!rowTagStarted && streamReader.hasNext) {
       streamReader.next()
     }
-    xsdSchemaValidator.get.reset()
+    // Reuse the Validator across records: Validator.reset() drops the secure-processing
+    // configuration applied at construction, so re-apply it on every record.
+    ValidatorUtil.reset(xsdSchemaValidator.get)
     xsdSchemaValidator.get.validate(new StAXSource(streamReader))
   }
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXmlParser.scala

@@ -146,7 +146,7 @@ class StaxXmlParser(
     lazy val xmlRecord = UTF8String.fromString(xml)
     try {
       xsdSchema.foreach { schema =>
-        schema.newValidator().validate(new StreamSource(new StringReader(xml)))
+        ValidatorUtil.newValidator(schema).validate(new StreamSource(new StringReader(xml)))
       }
       if (options.singleVariantColumn.isDefined || options.rootVariantType) {
         // If the singleVariantColumn is specified or the requested output is a root Variant,
@@ -163,6 +163,9 @@ class StaxXmlParser(
       }
     } catch {
       case e: SparkUpgradeException => throw e
+      // ValidatorUtil.newValidator throws this when the JAXP implementation cannot
+      // disable external access; that is an environment error, not a bad record.
+      case e: UnsupportedOperationException => throw e
       case e@(_: RuntimeException | _: XMLStreamException | _: MalformedInputException
               | _: SAXException) =>
         // XML parser currently doesn't support partial results for corrupted records.

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/ValidatorUtil.scala

@@ -19,10 +19,11 @@ package org.apache.spark.sql.catalyst.xml
 import java.io.{File, FileInputStream, InputStream}
 import javax.xml.XMLConstants
 import javax.xml.transform.stream.StreamSource
-import javax.xml.validation.{Schema, SchemaFactory}
+import javax.xml.validation.{Schema, SchemaFactory, Validator}
 
 import com.google.common.cache.{CacheBuilder, CacheLoader}
 import org.apache.hadoop.fs.Path
+import org.xml.sax.{SAXNotRecognizedException, SAXNotSupportedException}
 
 import org.apache.spark.SparkFiles
 import org.apache.spark.deploy.SparkHadoopUtil
@@ -75,4 +76,43 @@ object ValidatorUtil extends Logging {
    * @return Schema for the file at that path
    */
   def getSchema(path: String): Schema = cache.get(path)
+
+  /**
+   * Creates a [[Validator]] for the given schema that does not resolve external DTDs,
+   * entities, or schema references found in the document being validated. The record
+   * parser does not process DTDs, so validation applies the same restrictions to keep
+   * the two consistent. All validation of record data must use a Validator returned by
+   * this method, or one reused through [[reset]].
+   */
+  def newValidator(schema: Schema): Validator = {
+    val validator = schema.newValidator()
+    configureForRecordValidation(validator)
+    validator
+  }
+
+  /**
+   * Resets a [[Validator]] created by [[newValidator]] and re-applies the
+   * record-validation configuration. [[Validator.reset]] does not retain configuration
+   * applied after construction, so the secure-processing settings must be re-applied on
+   * every reset. Reusing one Validator across records avoids the per-record allocation
+   * cost of [[newValidator]].
+   */
+  def reset(validator: Validator): Unit = {
+    validator.reset()
+    configureForRecordValidation(validator)
+  }
+
+  private def configureForRecordValidation(validator: Validator): Unit = {
+    try {
+      validator.setProperty(XMLConstants.ACCESS_EXTERNAL_DTD, "")
+      validator.setProperty(XMLConstants.ACCESS_EXTERNAL_SCHEMA, "")
+      validator.setFeature(XMLConstants.FEATURE_SECURE_PROCESSING, true)
+    } catch {
+      case e @ (_: SAXNotRecognizedException | _: SAXNotSupportedException) =>
+        // Throw rather than validate with a Validator that cannot disable external access.
+        throw new UnsupportedOperationException(
+          "The JAXP Validator implementation in use does not support disabling external " +
+            "DTD/schema access; refusing to validate with it.", e)
+    }
+  }
 }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/XmlInferSchema.scala

@@ -178,7 +178,7 @@ class XmlInferSchema(private val options: XmlOptions, private val caseSensitive:
     try {
       val xsd = xsdSchema.orElse(Option(options.rowValidationXSDPath).map(ValidatorUtil.getSchema))
       xsd.foreach { schema =>
-        schema.newValidator().validate(new StreamSource(new StringReader(xml)))
+        ValidatorUtil.newValidator(schema).validate(new StreamSource(new StringReader(xml)))
       }
       parser = StaxXmlParserUtils.filteredReader(xml)
       val rootAttributes = StaxXmlParserUtils.gatherRootAttributes(parser)
@@ -205,6 +205,9 @@ class XmlInferSchema(private val options: XmlOptions, private val caseSensitive:
         Some(StructType(Nil))
       case e: FileNotFoundException if !options.ignoreMissingFiles => throw e
       case e @ (_ : AccessControlException | _ : BlockMissingException) => throw e
+      // ValidatorUtil.newValidator throws this when the JAXP implementation cannot
+      // disable external access; that is an environment error, not a bad record.
+      case e: UnsupportedOperationException => throw e
       case e @ (_: IOException | _: RuntimeException) if options.ignoreCorruptFiles =>
         logWarning("Skipped the rest of the content in the corrupted file", e)
         Some(StructType(Nil))

sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/xml/XmlSuite.scala

@@ -16,14 +16,15 @@
  */
 package org.apache.spark.sql.execution.datasources.xml
 
-import java.io.{EOFException, File, FileOutputStream, StringWriter}
+import java.io.{EOFException, File, FileOutputStream, StringReader, StringWriter}
 import java.nio.charset.{StandardCharsets, UnsupportedCharsetException}
 import java.nio.file.{Files, Path, Paths}
 import java.sql.{Date, Timestamp}
 import java.time.{Instant, LocalDateTime, Year}
 import java.util.TimeZone
 import java.util.concurrent.ConcurrentHashMap
 import javax.xml.stream.{XMLOutputFactory, XMLStreamException}
+import javax.xml.transform.stream.StreamSource
 
 import scala.collection.immutable.ArraySeq
 import scala.collection.mutable
@@ -34,14 +35,15 @@ import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs.FSDataInputStream
 import org.apache.hadoop.io.{LongWritable, Text}
 import org.apache.hadoop.io.compress.{CompressionCodecFactory, GzipCodec}
+import org.xml.sax.SAXException
 
 import org.apache.spark.{DebugFilesystem, SparkConf, SparkException}
 import org.apache.spark.io.ZStdCompressionCodec
 import org.apache.spark.sql.{AnalysisException, DataFrame, Dataset, Encoders, Row, SaveMode, YearUDT}
 import org.apache.spark.sql.catalyst.encoders.AgnosticEncoders.UDTEncoder
 import org.apache.spark.sql.catalyst.util._
 import org.apache.spark.sql.catalyst.util.TypeUtils.ordinalNumber
-import org.apache.spark.sql.catalyst.xml.{IndentingXMLStreamWriter, XmlOptions}
+import org.apache.spark.sql.catalyst.xml.{IndentingXMLStreamWriter, ValidatorUtil, XmlOptions}
 import org.apache.spark.sql.catalyst.xml.XmlOptions._
 import org.apache.spark.sql.errors.QueryCompilationErrors
 import org.apache.spark.sql.execution.datasources.CommonFileDataSourceSuite
@@ -1274,6 +1276,25 @@ class XmlSuite
     }
   }
 
+  test("XSD validation does not resolve external references in records") {
+    val schema = ValidatorUtil.getSchema(
+      getTestResourcePath(resDir + "basket.xsd").replace("file:/", "/"))
+
+    withTempDir { dir =>
+      val dataFile = new File(dir, "data.txt")
+      Files.write(dataFile.toPath, "9027".getBytes(StandardCharsets.UTF_8))
+      // The validator sees the raw record before the record parser does, so it must
+      // reject DOCTYPE references the same way the parser does.
+      val record =
+        s"""<?xml version="1.0"?>
+           |<!DOCTYPE basket [<!ENTITY ext SYSTEM "${dataFile.toURI}">]>
+           |<basket><entry><key>&ext;</key><value>1</value></entry></basket>""".stripMargin
+      intercept[SAXException] {
+        ValidatorUtil.newValidator(schema).validate(new StreamSource(new StringReader(record)))
+      }
+    }
+  }
+
   test("test XSD validation with validation error") {
     val basketDF = spark.read
       .option("rowTag", "basket")