diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXMLRecordReader.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXMLRecordReader.scala index 8793eebc97202..d087fdd8f7992 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXMLRecordReader.scala +++ b/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)) } diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXmlParser.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXmlParser.scala index 340dc61fb5112..d964564fbde1f 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXmlParser.scala +++ b/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. diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/ValidatorUtil.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/ValidatorUtil.scala index dccbb40fdd985..436c81a49df4c 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/ValidatorUtil.scala +++ b/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) + } + } } diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/XmlInferSchema.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/XmlInferSchema.scala index d2a59498549a3..3c1334c4c215b 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/XmlInferSchema.scala +++ b/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)) diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/xml/XmlSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/xml/XmlSuite.scala index 8a8ab4de3a8f8..3c2000214b983 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/xml/XmlSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/xml/XmlSuite.scala @@ -16,7 +16,7 @@ */ 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} @@ -24,6 +24,7 @@ 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,6 +35,7 @@ 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 @@ -41,7 +43,7 @@ import org.apache.spark.sql.{AnalysisException, DataFrame, Dataset, Encoders, Ro 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""" + |]> + |&ext;1""".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")