peter-toth commented on code in PR #58614:
URL: https://github.com/apache/spark/pull/58614#discussion_r3978941730
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -7016,6 +7016,19 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val AVRO_SCHEMA_URL_ALLOWED_SCHEMES =
+ buildConf("spark.sql.avro.schemaUrlAllowedSchemes")
Review Comment:
**Finding 1.** This is `buildConf`, so the allowlist is a runtime SQL conf.
Any session can turn it off with `SET spark.sql.avro.schemaUrlAllowedSchemes=`
or `spark.conf.set(...)`, and the PR's own tests set it exactly that way
through `withSQLConf`. So it does not constrain a user who can set the
`avroSchemaUrl` option in the first place, which is the user that "allowlist" /
"permitted" / "rejected" implies it is protecting against.
Two ways out, and which one is right depends on the threat model you have in
mind:
- If the point is to constrain what users may reference, this needs
`buildStaticConf`, so it can only be set when the `SparkSession` is created.
The cost is that the tests can no longer use `withSQLConf` and would have to
build a session per case.
- If the point is a guardrail against accidental misconfiguration rather
than a boundary, the `.doc` should say that outright. An operator reading
"rejected before it is opened" will deploy it as a boundary.
##########
sql/core/src/main/scala/org/apache/spark/sql/avro/AvroOptions.scala:
##########
@@ -76,6 +76,25 @@ private[sql] class AvroOptions(
parameters.get(AVRO_SCHEMA).map(AvroUtils.parseAvroSchema).orElse({
val avroUrlSchema = parameters.get(AVRO_SCHEMA_URL).map(url => {
log.debug("loading avro schema from url: " + url)
+ // Optional operator-configured allowlist of URI schemes for
avroSchemaUrl. Empty by
+ // default, which permits any scheme and leaves the file-system
resolution below unchanged.
+ // When set, the scheme is resolved and checked before the file system
for the URL is
+ // instantiated, so a disallowed scheme is rejected with a clear error
rather than a
+ // lower-level failure while opening it. A scheme-less URL takes the
default file system's
+ // scheme, so it can be permitted by allowing that scheme.
+ val allowedSchemes =
SQLConf.get.getConf(SQLConf.AVRO_SCHEMA_URL_ALLOWED_SCHEMES)
Review Comment:
**Finding 3.** The control is per-option, so an operator who sets it still
has the same shape open in XML. `rowValidationXSDPath`
(`sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/XmlOptions.scala:128`)
is a user-supplied path that `StaxXmlParser` and `XmlInferSchema` hand to
`ValidatorUtil.openSchemaFile`, which does
`xsdPath.getFileSystem(SparkHadoopUtil.get.conf)` and `fs.open(xsdPath)` with
no scheme check
(`sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/ValidatorUtil.scala:51`).
I am not asking you to fix XML here. But if the intent is an operator-level
restriction on which URIs a data source option may open, an Avro-only internal
conf is a partial answer, and the next source will want its own conf under its
own key. Worth a line in the description saying whether a shared helper is the
direction, or whether Avro is deliberately the only case in scope.
##########
sql/core/src/main/scala/org/apache/spark/sql/avro/AvroOptions.scala:
##########
@@ -76,6 +76,25 @@ private[sql] class AvroOptions(
parameters.get(AVRO_SCHEMA).map(AvroUtils.parseAvroSchema).orElse({
val avroUrlSchema = parameters.get(AVRO_SCHEMA_URL).map(url => {
log.debug("loading avro schema from url: " + url)
+ // Optional operator-configured allowlist of URI schemes for
avroSchemaUrl. Empty by
+ // default, which permits any scheme and leaves the file-system
resolution below unchanged.
+ // When set, the scheme is resolved and checked before the file system
for the URL is
+ // instantiated, so a disallowed scheme is rejected with a clear error
rather than a
+ // lower-level failure while opening it. A scheme-less URL takes the
default file system's
+ // scheme, so it can be permitted by allowing that scheme.
+ val allowedSchemes =
SQLConf.get.getConf(SQLConf.AVRO_SCHEMA_URL_ALLOWED_SCHEMES)
+ .map(_.toLowerCase(Locale.ROOT))
+ if (allowedSchemes.nonEmpty) {
+ val scheme = Option(new URI(url).getScheme)
Review Comment:
**Finding 4.** The `getOrElse("")` is unreachable.
`FileSystem.getDefaultUri` throws `IllegalArgumentException("No scheme in
default FS: ...")` rather than returning a scheme-less URI, so the
`Option(...)` around it is never `None` (checked against the Hadoop 3.3.0
`FileSystem.java` sources; the method reads the same in later releases). While
you are here, `new URI(url)` is built again at line 98 for `FileSystem.get`.
```scala
val uri = new URI(url)
...
val scheme = Option(uri.getScheme)
.getOrElse(FileSystem.getDefaultUri(conf).getScheme)
.toLowerCase(Locale.ROOT)
...
val fs = FileSystem.get(uri, conf)
```
##########
connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala:
##########
@@ -1255,6 +1255,65 @@ abstract class AvroSuite
assertExceptionMsg[FileNotFoundException](e, "File not_exists.avsc does
not exist")
}
+ test("SPARK-59329: avroSchemaUrl scheme allowlist permits an allowed
scheme") {
+ val avroSchemaUrl = testFile("test_sub.avsc")
+ // A scheme-less local path resolves to the default file system ("file"),
so allowing
+ // "file" lets it through.
+ withSQLConf(SQLConf.AVRO_SCHEMA_URL_ALLOWED_SCHEMES.key -> "file") {
+ val result = spark.read.option("avroSchemaUrl", avroSchemaUrl)
+ .format("avro")
+ .load(testAvro)
+ .collect()
+ val expected =
spark.read.format("avro").load(testAvro).select("string").collect()
+ assert(result.sameElements(expected))
+ }
+ }
+
+ test("SPARK-59329: avroSchemaUrl scheme allowlist rejects a scheme not
listed") {
+ val avroSchemaUrl = testFile("test_sub.avsc")
+ withSQLConf(SQLConf.AVRO_SCHEMA_URL_ALLOWED_SCHEMES.key -> "s3a") {
+ val e = intercept[AnalysisException] {
+ spark.read.option("avroSchemaUrl", avroSchemaUrl)
+ .format("avro")
+ .load(testAvro)
+ .collect()
+ }
+ assert(e.getCondition == "STDS_INVALID_OPTION_VALUE.WITH_MESSAGE")
+ assert(e.getMessage.contains("avroSchemaUrl"))
+ assert(e.getMessage.contains("not in the allowlist"))
+ }
+ }
+
+ test("SPARK-59329: avroSchemaUrl allowlist rejects an explicit disallowed
scheme " +
+ "before opening the file system") {
+ // An explicit non-"file" scheme is rejected by the allowlist check, which
runs before the
+ // file system for the URL is instantiated -- so this surfaces the clean
allowlist error
+ // rather than a lower-level failure from trying to load the s3a file
system.
+ withSQLConf(SQLConf.AVRO_SCHEMA_URL_ALLOWED_SCHEMES.key -> "file") {
+ val e = intercept[AnalysisException] {
+ spark.read.option("avroSchemaUrl", "s3a://bucket/user.avsc")
+ .format("avro")
+ .load(testAvro)
+ .collect()
+ }
+ assert(e.getCondition == "STDS_INVALID_OPTION_VALUE.WITH_MESSAGE")
+ assert(e.getMessage.contains("avroSchemaUrl"))
+ assert(e.getMessage.contains("not in the allowlist"))
+ assert(e.getMessage.contains("s3a"))
+ }
+ }
+
+ test("SPARK-59329: avroSchemaUrl scheme allowlist is disabled by default") {
Review Comment:
**Finding 2.** This is the existing `SPARK-34416: support user provided avro
schema url` at
`connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala:1212`
with a comment added. Same `testFile("test_sub.avsc")`, same read, same
`expected`, same assertion, and it passes on base for the same reason that one
does.
The added code that nothing covers is the case folding of the conf value at
`sql/core/src/main/scala/org/apache/spark/sql/avro/AvroOptions.scala:86`. Drop
the `.map(_.toLowerCase(Locale.ROOT))` there and every test in this PR still
passes. Swapping this test for that case covers the new code instead of
repeating an old test:
```scala
test("SPARK-59329: avroSchemaUrl scheme allowlist is case-insensitive") {
val avroSchemaUrl = testFile("test_sub.avsc")
withSQLConf(SQLConf.AVRO_SCHEMA_URL_ALLOWED_SCHEMES.key -> "FILE") {
val result = spark.read.option("avroSchemaUrl", avroSchemaUrl)
.format("avro").load(testAvro).collect()
val expected =
spark.read.format("avro").load(testAvro).select("string").collect()
assert(result.sameElements(expected))
}
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
