peter-toth commented on code in PR #58614:
URL: https://github.com/apache/spark/pull/58614#discussion_r4003648578
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -7016,6 +7016,22 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ // Kept here with the other `spark.sql.avro.*` entries rather than in
`StaticSQLConf`, though it
+ // is static: `buildStaticConf` registers the static key wherever it is
declared.
+ val AVRO_SCHEMA_URL_ALLOWED_SCHEMES =
+ buildStaticConf("spark.sql.avro.schemaUrlAllowedSchemes")
+ .internal()
+ .doc("A comma-separated allowlist of URI schemes permitted for the
'avroSchemaUrl' Avro " +
+ "option. Empty by default, which permits any scheme and preserves the
previous behavior; " +
+ "when non-empty, an avroSchemaUrl whose scheme is not listed is
rejected before it is " +
+ "opened. This is a static configuration fixed when the SparkSession is
created and not " +
+ "modifiable at runtime, so it is an operator-level boundary that a
session cannot relax.")
Review Comment:
**Finding 11.** The `.doc` ends with "an operator-level boundary that a
session cannot relax". `buildStaticConf` pins the allowlist *value*, and that
part holds. It does not pin which file system an allowed scheme resolves to,
and that is caller-controlled:
- `SessionState.newHadoopConf` copies every SQL conf into the Hadoop conf
(`sql/core/src/main/scala/org/apache/spark/sql/internal/SessionState.scala:149`),
so `spark.conf.set("fs.s3a.impl", ...)` on a live session lands there.
- `newHadoopConfWithOptions` adds every read option except `path` and
`paths`
(`sql/core/src/main/scala/org/apache/spark/sql/internal/SessionState.scala:121`),
so `.option("fs.s3a.impl", ...)` lands there too.
- Hadoop's `FileSystem.getFileSystemClass` prefers `fs.<scheme>.impl` from
that conf over the service-loaded implementation (`FileSystem.java:3376`, 3.3.0
sources).
That is the same `conf` this `AvroOptions` holds. So the scheme string the
check compares and the class that opens the URL are two separate decisions.
Measured on this head. Allowlist `s3a`, data path local so only the schema
URL is affected, and `fs.<scheme>.impl.disable.cache=true` in every arm -
`FileSystem.CACHE` keys on scheme and authority rather than on the conf, so
without it the first arm's instance serves the rest.
| arm | `avroSchemaUrl` | remap | result |
|---|---|---|---|
| A | `s3a://127.0.0.11:2121/user.avsc` | `.option("fs.s3a.impl",
"org.apache.hadoop.fs.ftp.FTPFileSystem")` | `SocketTimeoutException: Connect
timed out` at `java.net.Socket.connect` |
| B | `s3a://127.0.0.12:2121/user.avsc` | `spark.conf.set("fs.s3a.impl",
...)` at runtime | same |
| C | `s3a://127.0.0.13:2121/user.avsc` | none | `NoAuthWithAWSException` at
`org.apache.hadoop.fs.s3a.S3AFileSystem.s3GetFileStatus` |
| D | `ftp://127.0.0.14:2121/user.avsc` | `fs.ftp.impl` set | the allowlist
error, thrown at `AvroOptions.scala:99` |
C is the control: without a remap the allowed scheme does go to the real
S3A. A and B open an FTP connection to a host the caller picked, through a
scheme the allowlist permits. B is the one the wording turns on - no read
option, one `spark.conf.set` on a running session. `FTPFileSystem` ships inside
`hadoop-client-api`, so nothing extra has to be on the classpath.
D says the scheme name itself is enforced, so I would fix the sentence
rather than the check:
```suggestion
"modifiable at runtime. It restricts the scheme an avroSchemaUrl may
name. It does not " +
"restrict which file system serves that scheme. That is decided by
fs.<scheme>.impl, " +
"which a session can still set.")
```
Making it a real boundary would mean also rejecting when `fs.<scheme>.impl`
for the resolved scheme differs from the value in
`sparkContext.hadoopConfiguration`. That is a much bigger change than this PR
wants.
##########
connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala:
##########
@@ -1255,6 +1255,91 @@ abstract class AvroSuite
assertExceptionMsg[FileNotFoundException](e, "File not_exists.avsc does
not exist")
}
+ // spark.sql.avro.schemaUrlAllowedSchemes is a static SQL config, so it
cannot be set with
+ // withSQLConf; these drive AvroOptions directly under a SQLConf provided
via withExistingConf.
+ // testFile returns a "file:" URL, so its scheme is an explicit "file"; the
scheme-less path that
+ // resolves against the default file system is covered by its own test below.
+ test("SPARK-59329: avroSchemaUrl scheme allowlist permits an allowed
scheme") {
+ val avroSchemaUrl = testFile("test_sub.avsc")
+ val hadoopConf = spark.sessionState.newHadoopConf()
+ val conf = new SQLConf()
+ conf.setConf(SQLConf.AVRO_SCHEMA_URL_ALLOWED_SCHEMES, Seq("file"))
+ SQLConf.withExistingConf(conf) {
+ val options = new AvroOptions(Map("avroSchemaUrl" -> avroSchemaUrl),
hadoopConf)
+ assert(options.schema.isDefined)
+ }
+ }
+
+ test("SPARK-59329: avroSchemaUrl allowlist rejects a 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. The URL uses an
+ // upper-case "S3A" scheme so the lower-case "s3a" in the message pins the
scheme-side case
+ // folding: dropping the fold on the scheme leaves no lower-case "s3a" in
the message.
+ val hadoopConf = spark.sessionState.newHadoopConf()
+ val conf = new SQLConf()
+ conf.setConf(SQLConf.AVRO_SCHEMA_URL_ALLOWED_SCHEMES, Seq("file"))
+ SQLConf.withExistingConf(conf) {
+ val e = intercept[AnalysisException] {
+ new AvroOptions(Map("avroSchemaUrl" -> "S3A://bucket/user.avsc"),
hadoopConf)
+ }
+ 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("The scheme 's3a'"))
Review Comment:
**Finding 12.** The echo landed with nothing pinning it. I removed
`${allowedSchemes.mkString("[", ", ", "]")}` from
`sql/core/src/main/scala/org/apache/spark/sql/avro/AvroOptions.scala:98` and
re-ran: the five `SPARK-59329` cases here and the three in
`AvroSchemaUrlAllowlistSuite` all stayed green. So the case the echo exists for
- an operator writing `file://` where `file` was meant - has no test. That case
came from my ask at
[r3995617105](https://github.com/apache/spark/pull/58614#discussion_r3995617105).
This test is the only one of the nine that fails with the echo removed. I
ran it both ways, green on this head and red under the mutation:
```scala
test("SPARK-59329: the allowlist rejection echoes the parsed allowlist") {
val hadoopConf = spark.sessionState.newHadoopConf()
// "file://" is the shape an operator is most likely to write by
mistake: it parses to one
// entry that matches nothing, so the message has to show what it parsed
to be readable.
val conf = new SQLConf()
conf.setConf(SQLConf.AVRO_SCHEMA_URL_ALLOWED_SCHEMES, Seq("file://"))
SQLConf.withExistingConf(conf) {
val e = intercept[AnalysisException] {
new AvroOptions(Map("avroSchemaUrl" -> testFile("test_sub.avsc")),
hadoopConf)
}
assert(e.getMessage.contains("The scheme 'file'"))
assert(e.getMessage.contains("not in the allowlist [file://]"))
}
}
```
--
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]
