sunchao commented on code in PR #7026: URL: https://github.com/apache/datafusion-comet/pull/7026#discussion_r4340772487
########## spark/src/main/scala/org/apache/comet/objectstore/S3ConfigGate.scala: ########## @@ -0,0 +1,1390 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.comet.objectstore + +import java.io.IOException +import java.net.{URI, URISyntaxException} +import java.util.Locale + +import scala.collection.mutable.{Map => MutableMap} +import scala.jdk.CollectionConverters._ + +import org.apache.hadoop.conf.Configuration + +/** + * Decides whether Comet's native S3 client would resolve the same S3 configuration as Hadoop's + * `S3AFileSystem` for the buckets a scan touches. The native client builds its configuration from + * the options [[NativeConfig.extractObjectStoreOptions]] forwards, and resolves them differently + * from Hadoop in several ways (`lookupPassword` bucket precedence, JCEKS credential provider + * aliases, the clear text fallback, assumed role session policies, provider class semantics, + * scheme-less endpoints). A scan that went native over such a table would read under a different + * effective identity or endpoint than Spark would, silently. Each gate here returns a decline + * reason for the scan to fall back to Spark, or `None` when native and Hadoop agree. + * + * Shared by every native scan that forwards `fs.s3a.*` options (see + * [[NativeConfig.extractScanObjectStoreOptions]]) and by the Delta contrib's own claim gates. + * Correctness rule: when in doubt, decline. + */ +object S3ConfigGate { + + /** + * First reason the native S3 client would not resolve the same configuration as Hadoop for any + * bucket among `uris`, or `None` when it would, or when no URI is an `s3`/`s3a` one. The gates + * are ordered cheapest and most specific first. They share one propagated-conf cache so each + * bucket's full `Configuration` deep copy is made at most once. + */ + def declineReason(hadoopConf: Configuration, uris: Seq[URI]): Option[String] = { + if (!uris.exists(uri => s3Bucket(uri).isDefined)) { + return None + } + val propagatedConfCache = MutableMap.empty[String, Configuration] + unsupportedEncryptionAlgorithmReason(hadoopConf, uris, propagatedConfCache) + .orElse(proxyGateReason(hadoopConf, uris, propagatedConfCache)) + .orElse(assumedRolePolicyGateReason(hadoopConf, uris, propagatedConfCache)) + .orElse(hadoopOnlyEndpointGateReason(hadoopConf, uris, propagatedConfCache)) + .orElse(s3ConfigDivergenceReason(hadoopConf, uris, propagatedConfCache)) + .orElse(providerClassGateReason(hadoopConf, uris)) + } + + /** + * String-literal Hadoop conf keys consulted below. `hadoop-aws` is NOT on this module's runtime + * classpath, so `org.apache.hadoop.fs.s3a.Constants` must never be referenced here (would raise + * `NoClassDefFoundError` for sessions with no S3 dependency). + */ + private val HadoopCredentialProviderPathKey = "hadoop.security.credential.provider.path" + private val S3aCredentialProviderPathKey = "fs.s3a.security.credential.provider.path" + + /** + * `CommonConfigurationKeysPublic.HADOOP_SECURITY_CREDENTIAL_CLEAR_TEXT_FALLBACK`, default + * `true`, verified via `javap` against `hadoop-common` 3.3.4's + * `Configuration#getPasswordFromConfig`: `getPassword` only falls back to reading a plaintext + * conf value once `getBoolean(<this key>, true)` holds -- with the flag off, a plaintext value + * is invisible to every `getPassword`-based resolver, even when no credential provider is + * configured at all. + */ + private val ClearTextFallbackKey = "hadoop.security.credential.clear-text-fallback" + + private def s3aBucketProviderPathKey(bucket: String): String = + s"fs.s3a.bucket.$bucket.security.credential.provider.path" + + /** + * The LONG form of [[s3aBucketProviderPathKey]]: `S3AUtils#lookupPassword` resolves per-bucket + * overrides through both a long key (`fs.s3a.bucket.B.<full base key>`) and a short key; both + * must be covered here too. + */ + private def s3aBucketLongProviderPathKey(bucket: String): String = + s"fs.s3a.bucket.$bucket.fs.s3a.security.credential.provider.path" + + private def nonEmptyConf(hadoopConf: Configuration, key: String): Boolean = + Option(hadoopConf.get(key)).exists(_.nonEmpty) + + /** + * The lowercase-scheme-checked S3/S3A bucket name from `uri`'s authority, or `None` when + * `uri`'s scheme is not `s3`/`s3a`. Parses the raw authority manually rather than + * `URI#getHost`, avoiding the RFC 3986 `reg-name` pitfall (`getHost` returns `null` for the + * whole authority). + */ + private[comet] def s3Bucket(uri: URI): Option[String] = { + val scheme = Option(uri.getScheme).map(_.toLowerCase(Locale.ROOT)) + if (scheme.contains("s3") || scheme.contains("s3a")) { + val authority = Option(uri.getAuthority).getOrElse("") + val at = authority.lastIndexOf('@') + val hostAndPort = if (at >= 0) authority.substring(at + 1) else authority + val colon = hostAndPort.lastIndexOf(':') + val host = if (colon >= 0) hostAndPort.substring(0, colon) else hostAndPort + if (host.isEmpty) None else Some(host) + } else { + None + } + } + + private def plainValue(hadoopConf: Configuration, key: String): Option[String] = + Option(hadoopConf.get(key)).filter(_.nonEmpty) + + /** + * How Hadoop's OWN consumer reads one of the keys compared by [[s3ConfigDivergenceReason]], + * which decides how [[s3KeyDivergenceReason]] computes the Hadoop-effective side of its + * equality check. Exactly two consumer families exist among [[AllS3ConfigKeys]] in `hadoop-aws` + * 3.3.4, each verified via `javap`/CFR against the real call sites (cited per key on + * [[S3ConfigKeyConsumers]]). The tier must mirror the key's ACTUAL consumer: resolving a + * [[PropagatedOptionConsumer]] key through the wider `lookupPassword` cascade is NOT fail-safe + * for a value-EQUALITY comparator -- a long-form alias value Hadoop itself never reads can + * EQUAL native's resolution while Hadoop's true propagate-then-plain-get value differs, turning + * a real divergence into a wrongly-claimed scan (the endpoint `${...}`-redirect shape pinned in + * `S3ConfigGateSuite`). + */ + private[comet] sealed trait S3ConfigConsumer + + /** + * Read via `S3AUtils#lookupPassword(bucket, conf, baseKey)`, verified via `javap` against + * `hadoop-aws` 3.3.4: builds `longBucketKey = "fs.s3a.bucket." + bucket + "." + baseKey` (the + * FULL, already-`fs.s3a`-prefixed base key appended after the bucket segment) and reads it via + * `Configuration#getPassword` BEFORE the short-bucket key, keeping the long value whenever + * `getPassword` returns non-empty and only falling through to short-then-global otherwise. + * `getPassword` is Hadoop-credential-provider-aware and skips plaintext conf entirely when + * [[ClearTextFallbackKey]] is false. Modeled by [[hadoopLookupPasswordEffective]]. + */ + private[comet] case object LookupPasswordConsumer extends S3ConfigConsumer + + /** + * Read via `S3AUtils#propagateBucketOptions` followed by a plain `Configuration#get`-family + * call (`getTrimmed`/`getBoolean`/`getClasses`) against the propagated view: the short bucket + * form wins only by having overwritten the global key during propagation, the long bucket form + * folds into an unread `fs.s3a.fs.s3a.*` key, and neither a credential provider nor + * [[ClearTextFallbackKey]] is ever consulted. Modeled as a plain `Configuration#get` on the + * [[propagateBucketOptions]] result, which also expands `${...}` references under that + * propagated view exactly like the real consumer. + */ + private[comet] case object PropagatedOptionConsumer extends S3ConfigConsumer + + /** + * Every `fs.s3a.*` base key that governs whether a claimed native scan actually behaves like + * Hadoop's own reader would, paired with the consumer family Hadoop resolves it through -- ONE + * list, with each key's resolution tier declared beside it, so a key can never sit in the + * comparator without a deliberate classification (adding one without picking a tier does not + * compile). The entries are every per-bucket `fs.s3a.*` base key native's S3 client's + * `get_config` (s3.rs) resolves, verified directly against its call sites: + * `extract_s3_config_options` (endpoint.region, path.style.access, endpoint, + * requester.pays.enabled), `lookup_provider_class` (the Comet-specific + * credential-provider-class activation key), and + * `build_credential_provider`/`build_aws_credential_provider_metadata`/ + * `build_assume_role_credential_provider_metadata` (aws.credentials.provider, + * assumed.role.credentials.provider, assumed.role.arn, assumed.role.session.name). + * + * Tier assignments, each verified via `javap`/CFR against `hadoop-aws` 3.3.4: + * - access.key/secret.key/session.token: `S3AUtils#getAWSAccessKeys` and + * `MarshalledCredentialBinding#fromFileSystem` (reached from + * `TemporaryAWSCredentialsProvider`) resolve all three via `S3AUtils#lookupPassword` -- + * [[LookupPasswordConsumer]]. + * - aws.credentials.provider and assumed.role.credentials.provider: + * `S3AUtils#buildAWSProviderList` -> `loadAWSProviderClasses` -> plain + * `Configuration#getClasses` -- [[PropagatedOptionConsumer]]. + * - assumed.role.arn/session.name: `AssumedRoleCredentialProvider`'s constructor reads both + * via plain `Configuration#getTrimmed` -- [[PropagatedOptionConsumer]]. + * - endpoint (`S3AFileSystem`: `getTrimmed`), endpoint.region (`DefaultS3ClientFactory`: + * `getTrimmed`), path.style.access (`S3AFileSystem`: `getBoolean`) -- + * [[PropagatedOptionConsumer]]. + * - requester.pays.enabled: not read anywhere in `hadoop-aws` 3.3.4 (the constant does not + * even exist in its `Constants` class); later releases read it via plain `getBoolean` + * against the propagated conf, so the plain tier is both the faithful forward model and + * inert on 3.3.4 -- [[PropagatedOptionConsumer]]. + * - comet.credential.provider.class: Comet's own activation key, plain conf read on both + * sides, never a Hadoop key at all -- [[PropagatedOptionConsumer]]. + * + * SYNC NOTE: the key list must stay a superset of native's `NATIVE_S3A_CONFIG_PROPERTIES` + * constant (`native/core/src/parquet/objectstore/s3.rs`, property suffixes without the + * `fs.s3a.` prefix) -- `S3ConfigGateSuite`'s discovery-harness test asserts this mechanically + * against [[AllS3ConfigKeys]]. A Rust test in `s3.rs` also parses this `Seq(` block as text, so + * each entry must stay a single `"fs.s3a.<key>" -> <Tier>` literal pair. Literal strings, not + * the [[AwsCredentialsProviderKey]] / [[AssumedRoleCredentialsProviderKey]] vals declared + * below, purely to avoid a forward reference inside this `object` body; kept textually + * identical to those two constants. + */ + private[comet] val S3ConfigKeyConsumers: Seq[(String, S3ConfigConsumer)] = Seq( + "fs.s3a.access.key" -> LookupPasswordConsumer, + "fs.s3a.secret.key" -> LookupPasswordConsumer, + "fs.s3a.session.token" -> LookupPasswordConsumer, + "fs.s3a.aws.credentials.provider" -> PropagatedOptionConsumer, + "fs.s3a.assumed.role.arn" -> PropagatedOptionConsumer, + "fs.s3a.assumed.role.session.name" -> PropagatedOptionConsumer, + "fs.s3a.assumed.role.credentials.provider" -> PropagatedOptionConsumer, + "fs.s3a.endpoint" -> PropagatedOptionConsumer, + "fs.s3a.endpoint.region" -> PropagatedOptionConsumer, + "fs.s3a.path.style.access" -> PropagatedOptionConsumer, + "fs.s3a.requester.pays.enabled" -> PropagatedOptionConsumer, + "fs.s3a.comet.credential.provider.class" -> PropagatedOptionConsumer) + + /** The compared keys alone, in [[S3ConfigKeyConsumers]] order (discovery-harness surface). */ + private[comet] val AllS3ConfigKeys: Seq[String] = S3ConfigKeyConsumers.map(_._1) + + /** + * The short-bucket-then-global value resolved for `baseKey` under `bucket` from `hadoopConf`, + * skipping an empty value at either alias exactly like [[plainValue]]. NOT used by + * [[s3ConfigDivergenceReason]]/[[s3KeyDivergenceReason]] any more -- every key checked there + * resolves per its declared [[S3ConfigKeyConsumers]] tier (see [[s3KeyDivergenceReason]]), + * neither of which matches this function's read. This function's one remaining caller is + * [[shortThenGlobalOrReason]], which reads provider-CLASS strings (from the ORIGINAL, + * unpropagated conf) for name-support validation in [[providerClassReason]]/ + * [[assumedRoleProviderClassReason]] -- by the time those run, [[s3ConfigDivergenceReason]] has + * already proven Hadoop's and native's effective values agree for the same key, so whichever of + * the two (equal) values this narrower read returns does not affect correctness there. NEVER + * used to compute native's own effective value -- see [[nativeShortThenGlobal]] for that. + */ + private def shortThenGlobal( + hadoopConf: Configuration, + bucket: String, + baseKey: String): Option[String] = { + val shortKey = s"fs.s3a.bucket.$bucket." + baseKey.stripPrefix("fs.s3a.") + plainValue(hadoopConf, shortKey).orElse(plainValue(hadoopConf, baseKey)) + } + + /** + * The short-bucket-then-global value native's `get_config` (s3.rs) resolves for `baseKey` under + * `bucket` from the ORIGINAL, unpropagated `hadoopConf` -- `NativeConfig + * .extractObjectStoreOptions` forwards `Configuration#get`'s substituted value for every + * `fs.s3a.*` entry with no bucket-option propagation step of its own, so the original conf is + * the right input here. Unlike [[shortThenGlobal]]/[[plainValue]], this mirrors `get_config` + * faithfully: PRESENCE of the short-bucket key alone -- never its emptiness -- decides whether + * native falls back to the global key (`get_config` is a plain `HashMap::get`, which returns + * `Some` for a key explicitly set to `""`), so an explicitly empty or whitespace-only + * short-bucket value resolves to `Some("")` here and never falls through to global -- the + * OPPOSITE of Hadoop's own `getPassword`/`lookupPassword` semantics (see + * [[hadoopLookupPasswordEffective]]), which treat empty as absent and keep trying the next + * alias. The ONLY function used to compute native's effective value in + * [[s3KeyDivergenceReason]]. + * + * Deliberately does NOT apply `get_config_trimmed`'s `.trim()` here: [[s3KeyDivergenceReason]] + * trims both this value and Hadoop's effective value together, symmetrically, at the point they + * are compared, rather than one-sidedly here -- trimming only the native side would flag a + * spurious divergence for a value neither side's whitespace actually changes the behavior of + * once each side's own downstream parsing normalizes it (e.g. Hadoop's own multi-line + * `fs.s3a.aws.credentials.provider` default, which both Hadoop and native additionally trim per + * comma-separated entry after splitting), while a one-sided trim would make an + * otherwise-identical default value look diverged for every bucket, never claiming natively at + * all. + */ + private def nativeShortThenGlobal( + hadoopConf: Configuration, + bucket: String, + baseKey: String): Option[String] = { + val shortKey = s"fs.s3a.bucket.$bucket." + baseKey.stripPrefix("fs.s3a.") + Option(hadoopConf.get(shortKey)).orElse(Option(hadoopConf.get(baseKey))) + } + + /** + * Faithful in-memory replica of `S3AUtils#propagateBucketOptions` (`hadoop-aws`), which + * `S3AFileSystem#initialize` calls FIRST, before any option or credential is read: + * `Configuration conf = propagateBucketOptions(originalConf, bucket); ...; setConf(conf);` -- + * every subsequent `conf.get`/`getPassword` call in that filesystem instance, including + * `${...}` variable substitution, resolves against this propagated view, not the original conf. + * `hadoop-aws` is not on this module's runtime classpath (see the string-literal-keys note + * above), so `S3AUtils#propagateBucketOptions` cannot be called directly; this reproduces its + * logic verbatim using only `hadoop-common`'s `Configuration`: + * {{{ + * public static Configuration propagateBucketOptions(Configuration source, String bucket) { + * final String bucketPrefix = FS_S3A_BUCKET_PREFIX + bucket + '.'; + * final Configuration dest = new Configuration(source); + * for (Map.Entry<String, String> entry : source) { + * final String key = entry.getKey(); + * final String value = entry.getValue(); // the (unexpanded) value + * if (!key.startsWith(bucketPrefix) || bucketPrefix.equals(key)) continue; + * final String stripped = key.substring(bucketPrefix.length()); + * if (stripped.startsWith("bucket.") || "impl".equals(stripped)) { + * // ignored + * } else { + * final String generic = FS_S3A_PREFIX + stripped; + * dest.set(generic, value, ...); // overwrites any existing global value + * } + * } + * return dest; + * } + * }}} + * Note the LONG bucket form (`fs.s3a.bucket.B.fs.s3a.<key>`) folds to an unread + * `fs.s3a.fs.s3a.<key>` key here too, exactly like the real method -- `stripped` already starts + * with `fs.s3a.` in that case, so prepending `fs.s3a.` again produces a key nothing ever reads. + */ + private def propagateBucketOptions(hadoopConf: Configuration, bucket: String): Configuration = { + val bucketPrefix = s"fs.s3a.bucket.$bucket." + val dest = new Configuration(hadoopConf) + hadoopConf.iterator().asScala.foreach { entry => + val key = entry.getKey + if (key.startsWith(bucketPrefix) && key != bucketPrefix) { + val stripped = key.substring(bucketPrefix.length) + if (!stripped.startsWith("bucket.") && stripped != "impl") { + dest.set(s"fs.s3a.$stripped", entry.getValue) + } + } + } + dest + } + + /** + * Canonical and deprecated Hadoop S3A encryption-algorithm config keys, verified via `javap` + * against `hadoop-aws` 3.3.4's `org.apache.hadoop.fs.s3a.Constants`: `S3_ENCRYPTION_ALGORITHM = + * "fs.s3a.encryption.algorithm"` (canonical) and `SERVER_SIDE_ENCRYPTION_ALGORITHM = + * "fs.s3a.server-side-encryption-algorithm"` (DEPRECATED -- note the hyphen before "algorithm", + * unlike the corresponding `*.key` constants below, which both use a `.key` suffix). + * `hadoop-aws` is NOT on this module's runtime classpath, so these stay string literals, same + * rationale as [[HadoopCredentialProviderPathKey]]. + */ + private val S3EncryptionAlgorithmKey = "fs.s3a.encryption.algorithm" + private val DeprecatedS3EncryptionAlgorithmKey = "fs.s3a.server-side-encryption-algorithm" + + /** + * The exact strings `S3AEncryptionMethods#getMethod` accepts, verified via `javap`/CFR against + * `hadoop-aws` 3.3.4's `S3AEncryptionMethods` enum: `NONE("")`, `SSE_S3("AES256", serverSide = + * true, requiresSecret = false)`, `SSE_KMS("SSE-KMS", serverSide = true, requiresSecret = + * false)`, `SSE_C("SSE-C", serverSide = true, requiresSecret = true)`, `CSE_KMS("CSE-KMS", + * serverSide = false, requiresSecret = true)`, `CSE_CUSTOM("CSE-CUSTOM", serverSide = false, + * requiresSecret = true)`. `getMethod` parses case-insensitively + * (`values().find(_.getMethod.equalsIgnoreCase(algorithm))`), matched below the same way. + * + * ALLOWLIST, not a blocklist (replaces the former SSE-C-only blocklist): only the algorithms S3 + * decrypts transparently on GET/HEAD given read permission alone, with NO extra request header + * and NO client-side step, are safe for a native scan that forwards none of Hadoop's + * `fs.s3a.encryption.*`/`fs.s3a.server-side-encryption*` options -- + * - `AES256` (SSE_S3, `serverSide = true`): plain server-side encryption, transparent on GET. + * - `SSE-KMS` (SSE_KMS, `serverSide = true`): server-side, KMS-managed key, transparent on + * GET given KMS decrypt permission (no header). + * - `DSSE-KMS`: NOT present in this enum on `hadoop-aws` 3.3.4 (confirmed by the six values + * listed above) -- `S3AEncryptionMethods.getMethod("DSSE-KMS")` throws + * `IOException("Unknown encryption algorithm DSSE-KMS")` on this version, so + * `S3AUtils#buildEncryptionSecrets` (and therefore Hadoop's own reader) already fails + * before ever reading such a table under 3.3.4, meaning this string can never actually be + * the resolved value on the declared target version -- admitting it here is inert there. + * Included anyway, forward-compatible, for a newer `hadoop-aws` on the runtime classpath (a + * later Hadoop release; this module has no compile-time `hadoop-aws` dependency, see the + * string-literal-keys note above) where DSSE-KMS is a real, dual-layer, server-side + * algorithm decrypted transparently on GET the same way SSE-KMS is. Every other value + * declines: `SSE-C` (SSE_C is `serverSide = true` in Hadoop's own enum, but `requiresSecret + * \= true` -- S3 rejects a GET/HEAD for an SSE-C object outright (400 Bad Request) unless + * the customer key is resent as a request header on every call, so a native scan that never + * learns the key cannot succeed at all, where Hadoop's own reader -- whose request factory + * attaches the key -- would), `CSE-KMS`/`CSE-CUSTOM` (`serverSide = false`: client-side + * encryption decrypts object bytes locally in the SDK layer, which the native Parquet + * reader has no equivalent of -- it would read raw ciphertext), and any future/unknown + * value (a value `S3AEncryptionMethods.getMethod` itself would reject is certainly not one + * of the three confirmed-transparent algorithms above; declining is the only safe default + * for anything this gate cannot positively confirm). + */ + private val AllowedEncryptionAlgorithms: Set[String] = Set("AES256", "SSE-KMS", "DSSE-KMS") + + /** + * `bucket`'s effective encryption-algorithm key and value under `hadoopConf`, or `None` when + * neither the canonical nor deprecated key is set anywhere consulted. Mirrors + * `S3AUtils#buildEncryptionSecrets`'s real resolution order, verified via `javap`/CFR + * decompilation of `hadoop-aws` 3.3.4's `S3AUtils.class`: + * {{{ + * String algorithm = lookupBucketSecret(bucket, conf, "fs.s3a.encryption.algorithm"); + * if (algorithm == null) + * algorithm = lookupBucketSecret(bucket, conf, "fs.s3a.server-side-encryption-algorithm"); + * if (algorithm == null) + * algorithm = lookupPassword(null, conf, "fs.s3a.encryption.algorithm"); + * if (algorithm == null) + * algorithm = lookupPassword(null, conf, "fs.s3a.server-side-encryption-algorithm"); + * }}} + * i.e. bucket-tier (canonical, then deprecated), THEN global-tier (canonical, then deprecated) + * -- the two tiers are never interleaved key-by-key, so this must stay two explicit bucket-tier + * lookups followed by two explicit global-tier lookups, not a single + * [[hadoopLookupPasswordEffective]] call per key (which would let an unset canonical bucket key + * fall through straight to the canonical GLOBAL value ahead of a SET deprecated bucket key, the + * wrong answer). + * + * The three-argument `lookupPassword` defaults to `""`, not `null`, so the last call in the + * block above never runs: Hadoop reads the deprecated global key only through `Configuration`'s + * deprecation alias of the canonical one. This function still reads it as a fourth lookup, + * which can only add a decline. + * + * THE FIX for the SSE-C long-bucket-alias gap is entirely inside the bucket tier: + * `lookupBucketSecret` itself is long-then-short, decompiled from `hadoop-aws` 3.3.4's + * `S3AUtils.class`: + * {{{ + * // longBucketKey = fs.s3a.bucket.B.fs.s3a.<key> + * String longBucketKey = String.format(BUCKET_PATTERN, bucket, baseKey); + * String initialVal = getPassword(conf, longBucketKey, null, null); + * // shortBucketKey = fs.s3a.bucket.B.<key> + * String shortBucketKey = String.format(BUCKET_PATTERN, bucket, subkey); + * // keeps initialVal (the LONG value) if non-empty + * return getPassword(conf, shortBucketKey, initialVal, null); + * }}} + * i.e. the SAME long-bucket-key construction and long-wins-if-nonempty semantics as + * `S3AUtils#lookupPassword` (see [[LookupPasswordConsumer]]/[[hadoopLookupPasswordEffective]]). + * `buildEncryptionSecrets` runs on the [[propagateBucketOptions]] view, so + * [[unsupportedEncryptionAlgorithmReason]] calls this on that view (and on the raw conf, which + * can only add a decline). An earlier version of this function modeled the bucket tier as + * SHORT-only, documented as "the LONG bucket form is genuinely never consulted for this key" -- + * that documentation was wrong (this decompilation supersedes it): a bucket configured only via + * `fs.s3a.bucket.B.fs.s3a.encryption.algorithm=SSE-C` bypassed the SSE-C gate entirely, because + * Hadoop's own reader DOES read that long form (and picks SSE-C), while this function reported + * `None` (nothing set) and the allowlist check below never even ran. + * + * The canonical-vs-deprecated distinction below is frequently moot in practice: `hadoop-aws`'s + * `S3AFileSystem.addDeprecatedKeys()` statically registers `fs.s3a.server-side-encryption-*` as + * `Configuration`-level deprecated aliases of `fs.s3a.encryption.*` (verified via `javap`), a + * registration that lives in a static field on Hadoop's `Configuration` class -- process-wide + * once `S3AFileSystem`'s class has loaded anywhere in the JVM, which a real scan has always + * already done by the time this gate runs, since reading the S3 table at all requires loading + * that class. Once active, `Configuration#get` resolves either literal key to the identical + * value transparently, making the two-key cascade below redundant (but harmless) for that case; + * it remains the operative path only when nothing else in the process has loaded + * `S3AFileSystem` yet. + * + * ALSO walks the Hadoop-credential-provider (JCEKS) path via [[resolveViaCredentialAliases]] + * for each of the four lookups below, matching `lookupBucketSecret`/`lookupPassword`'s real + * per-alias `getPassword` calls (quoted above) exactly: both are `getPassword`, not plain + * `Configuration#get`, so a bucket storing the algorithm name ONLY in a JCEKS keystore is + * exactly as real a Hadoop deployment shape for this key as it is for the credential keys + * [[hadoopLookupPasswordEffective]] already covers -- there is nothing algorithm-specific that + * makes JCEKS storage implausible here, so an earlier version of this function skipping it + * (documented at the time as "the algorithm NAME is not credential-sensitive data, so storing + * it in a keystore is not a realistic Hadoop deployment pattern") was an unjustified, narrower + * read than Hadoop's own resolver actually performs, under-declining a bucket whose algorithm + * is keystore-only. [[resolveViaCredentialAliases]]'s Arm B/C split still means this is zero + * extra I/O for the common case: keystore I/O only happens when a Hadoop credential-provider + * path is actually configured for the bucket, contained in that function's own try/catch. + * `bucketTier`/`globalTier` return `Left` (propagated straight through by [[orElseTier]]) when + * [[resolveViaCredentialAliases]] cannot safely verify a tier at all (an S3A-scoped provider + * path, or a corrupt/unreadable global keystore) -- correctly short-circuiting the whole + * cascade with a decline rather than silently falling through to a later tier that might look + * unset only because the true value was unverifiable. + */ + private def effectiveEncryptionAlgorithm( + hadoopConf: Configuration, + bucket: String): Either[String, Option[(String, String)]] = { + def bucketTier(baseKey: String): Either[String, Option[(String, String)]] = { + val longKey = s"fs.s3a.bucket.$bucket.$baseKey" + val shortKey = s"fs.s3a.bucket.$bucket." + baseKey.stripPrefix("fs.s3a.") + resolveViaCredentialAliases(hadoopConf, bucket, Seq(longKey, shortKey)) + .map(_.map(baseKey -> _)) + } + def globalTier(baseKey: String): Either[String, Option[(String, String)]] = + resolveViaCredentialAliases(hadoopConf, bucket, Seq(baseKey)) + .map(_.map(baseKey -> _)) + + // Short-circuits on Left (unverifiable tier) or Right(Some(_)) (resolved); only Right(None) + // (tier definitively unset) falls through to `next`, mirroring buildEncryptionSecrets's + // sequential `if (algorithm == null) algorithm = ...` cascade exactly. + def orElseTier( + current: Either[String, Option[(String, String)]], + next: => Either[String, Option[(String, String)]]) + : Either[String, Option[(String, String)]] = + current match { + case Left(reason) => Left(reason) + case Right(Some(value)) => Right(Some(value)) + case Right(None) => next + } + + orElseTier( + bucketTier(S3EncryptionAlgorithmKey), + orElseTier( + bucketTier(DeprecatedS3EncryptionAlgorithmKey), + orElseTier( + globalTier(S3EncryptionAlgorithmKey), + // Hadoop reaches this key only through the deprecation alias (see above); reading + // it directly can only over-decline. + globalTier(DeprecatedS3EncryptionAlgorithmKey)))) + } + + /** + * The only algorithm values a decline reason may print. Any other resolved value is withheld, + * since a `${...}` reference can expand the algorithm key to a customer key or token. + */ + private val PrintableEncryptionAlgorithms: Seq[String] = + Seq("AES256", "SSE-S3", "SSE-KMS", "DSSE-KMS", "SSE-C", "CSE-KMS", "CSE-CUSTOM") + + private def unsupportedEncryptionAlgorithmDeclineReason( + bucket: String, + algorithmKey: String, + algorithm: String): String = { + val setting = PrintableEncryptionAlgorithms + .find(_.equalsIgnoreCase(algorithm)) + .map(name => s"$algorithmKey=$name") + .getOrElse(s"$algorithmKey set to a value that is not a recognised S3 encryption algorithm") + s"Native scan does not support $setting for $bucket " + + "(the native S3 client only supports unencrypted objects and S3's transparent " + + "server-side algorithms -- AES256/SSE-S3, SSE-KMS, and DSSE-KMS decrypt on GET/HEAD given " + + "read permission alone, with no extra request header; SSE-C additionally requires the " + + "customer-provided key resent as a header on every GET/HEAD request, which the native S3 " + + "client's extract_s3_config_options never forwards, and CSE-KMS/CSE-CUSTOM decrypt object " + + "bytes client-side, a layer the native Parquet reader does not have -- any of these would " + + "fail outright or silently read ciphertext where Hadoop's own reader succeeds)" + } + + /** + * First reason any bucket among `uris` is configured for an encryption algorithm the native S3 + * client cannot safely read, or `None` when claimable. Allowlist-based (see + * [[AllowedEncryptionAlgorithms]]): only `AES256`/`SSE-KMS`/`DSSE-KMS` (and unset/empty) pass; + * every other resolved value -- `SSE-C`, `CSE-KMS`, `CSE-CUSTOM`, or any unrecognized future + * algorithm string -- declines. Deliberately NOT a blocklist keyed on `SSE-C` alone: an + * allowlist is safe by construction against a Hadoop release adding a new encryption method + * this gate has never heard of, where a blocklist would silently admit it. Never interpolates a + * resolved key value, only key names, the bucket, and the (non-secret) algorithm name. Declines + * on a `Left` from [[effectiveEncryptionAlgorithm]] too (an unverifiable credential-provider + * arm, e.g. an S3A-scoped provider path or a corrupt/unreadable global keystore) -- the + * algorithm cannot be ruled safe when it cannot be read at all. + */ + private[comet] def unsupportedEncryptionAlgorithmReason( + hadoopConf: Configuration, + uris: Seq[URI], + propagatedConfCache: MutableMap[String, Configuration] = MutableMap.empty) + : Option[String] = { + val buckets = uris.flatMap(s3Bucket).distinct + buckets.foldLeft(Option.empty[String]) { (declined, bucket) => + if (declined.isDefined) { + declined + } else { + try { + // Hadoop resolves the algorithm on the propagated view, where a bucket override can + // change what a `${...}` reference expands to. The raw view is checked as well, so a + // disagreement between the two can only decline. + val propagatedConf = + propagatedConfCache.getOrElseUpdate( + bucket, + propagateBucketOptions(hadoopConf, bucket)) + // A bucket on the SPI whose S3A keystore path the adapters promote like Hadoop does is + // read through that keystore, so an algorithm held only there is still seen. + val spi = spiViews(hadoopConf, propagatedConf, bucket) + encryptionAlgorithmReason(spi.map(_.hadoopPropagated).getOrElse(propagatedConf), bucket) + .orElse(encryptionAlgorithmReason(spi.map(_.hadoopRaw).getOrElse(hadoopConf), bucket)) + } catch { + case e @ (_: IOException | _: RuntimeException) => + Some(unverifiableValueReason(bucket, S3EncryptionAlgorithmKey, e)) + } + } + } + } + + private def encryptionAlgorithmReason(conf: Configuration, bucket: String): Option[String] = + effectiveEncryptionAlgorithm(conf, bucket) match { + case Left(reason) => Some(reason) + case Right(None) => None + case Right(Some((key, value))) => + if (!AllowedEncryptionAlgorithms.exists(_.equalsIgnoreCase(value))) { + Some(unsupportedEncryptionAlgorithmDeclineReason(bucket, key, value)) + } else { + None + } + } + + /** + * Canonical Hadoop S3A HTTP-proxy host config key, verified via CFR decompilation of + * `hadoop-aws` 3.3.4's `S3AUtils.class` (`initProxySupport`): + * {{{ + * String proxyHost = conf.getTrimmed("fs.s3a.proxy.host", ""); + * int proxyPort = conf.getInt("fs.s3a.proxy.port", -1); + * if (!proxyHost.isEmpty()) { + * ... + * String proxyUsername = + * S3AUtils.lookupPassword(bucket, conf, "fs.s3a.proxy.username", null, null); + * String proxyPassword = + * S3AUtils.lookupPassword(bucket, conf, "fs.s3a.proxy.password", null, null); + * ... + * } + * }}} + * `fs.s3a.proxy.host`/`fs.s3a.proxy.port` resolve via a PLAIN, non-bucket-scoped, non-JCEKS + * `Configuration#getTrimmed`/`getInt` call -- NOT `lookupPassword` -- against whatever conf + * `S3AFileSystem#initialize` already ran through `propagateBucketOptions` before + * `createAwsConf`/`initProxySupport` ever runs; only the SIBLING `fs.s3a.proxy.username`/ + * `fs.s3a.proxy.password` keys go through `lookupPassword` (bucket long/short/global, + * JCEKS-aware). So the host is bucket-aware only via `propagateBucketOptions`'s short-bucket- + * form folding, never the long-bucket form, and never a credential-provider read -- the SAME + * shape as `endpoint`/`path.style.access` ([[PropagatedOptionConsumer]], see + * [[S3ConfigKeyConsumers]]'s doc), not the credential family. `hadoop-aws` 3.4.x (the Spark 4.x + * profiles' version) moves this code to `AWSClientConfig#createProxyConfiguration`/ + * `#createAsyncProxyConfiguration` but keeps the exact same reads, verified via `javap` against + * 3.4.2: `conf.getTrimmed("fs.s3a.proxy.host", "")` for the host, `S3AUtils.lookupPassword` for + * username/password only. + * + * [[proxyGateReason]] therefore resolves the host EXACTLY like its real consumer -- plain + * `Configuration#getTrimmed` on the [[propagateBucketOptions]] result, no provider arms, no + * [[ClearTextFallbackKey]] handling -- rather than through the wider `lookupPassword` cascade + * an earlier version reused here. The wider cascade was wrong in BOTH directions for this key: + * with only the GLOBAL Hadoop provider path set and [[ClearTextFallbackKey]] false, + * `getPassword` hides a plaintext host that `getTrimmed` serves to Hadoop anyway (a missed + * decline, the exact bypass this gate exists to close -- native has NO HTTP-proxy support of + * any kind, no `fs.s3a.proxy.*` key is read anywhere in `s3.rs`); and an S3A-scoped provider + * path or a lone long-form bucket alias declined a bucket whose real consumer can never see a + * host from either source (pure over-refusal -- no keystore can supply the host to a plain + * `getTrimmed`, and the long form folds into the unread `fs.s3a.fs.s3a.proxy.host`). + */ + private val S3ProxyHostKey = "fs.s3a.proxy.host" + + private def unsupportedProxyReason(bucket: String, key: String): String = + s"Native scan does not support $key configured for $bucket (the native S3 client has " + + "no HTTP proxy support at all -- no fs.s3a.proxy.* key is read anywhere in its object " + + "store layer -- so a claimed scan would connect to S3 directly instead of routing through " + + "the configured proxy, either bypassing an egress/network-segmentation policy or simply " + + "failing to reach the endpoint)" + + /** + * First reason any bucket among `uris` has an HTTP proxy configured via [[S3ProxyHostKey]], or + * `None` when claimable. Reads the host EXACTLY like its real consumer (see + * [[S3ProxyHostKey]]'s doc): plain `Configuration#getTrimmed` on the [[propagateBucketOptions]] + * result, so this gate is always zero-I/O -- no credential provider is ever consulted for the + * host, because none ever supplies it to Hadoop either. Ordered alongside + * [[unsupportedEncryptionAlgorithmReason]] among the conf-only gates, ahead of + * [[s3ConfigDivergenceReason]]. The try/catch guards `Configuration#get`'s + * `IllegalStateException` on a `${...}` substitution cycle, same as [[s3KeyDivergenceReason]]. + * Never interpolates a resolved value: the proxy HOST is not secret, but naming it here would + * be a strange place to first surface it, and proxy CREDENTIALS + * (`fs.s3a.proxy.username`/`fs.s3a.proxy.password`, not read by this gate at all -- the whole + * point of gating on the host is that a non-empty host declines before any proxy credential + * would ever need to be forwarded) must never appear in a decline reason regardless. + */ + private[comet] def proxyGateReason( + hadoopConf: Configuration, + uris: Seq[URI], + propagatedConfCache: MutableMap[String, Configuration] = MutableMap.empty) + : Option[String] = { + val buckets = uris.flatMap(s3Bucket).distinct + buckets.foldLeft(Option.empty[String]) { (declined, bucket) => + if (declined.isDefined) { + declined + } else { + try { + val propagatedConf = + propagatedConfCache.getOrElseUpdate( + bucket, + propagateBucketOptions(hadoopConf, bucket)) + if (propagatedConf.getTrimmed(S3ProxyHostKey, "").nonEmpty) { + Some(unsupportedProxyReason(bucket, S3ProxyHostKey)) + } else { + None + } + } catch { + case e @ (_: IOException | _: RuntimeException) => + Some(unverifiableValueReason(bucket, S3ProxyHostKey, e)) + } + } + } + } + + /** + * Canonical Hadoop S3A assumed-role session-policy key. `AssumedRoleCredentialProvider`'s + * constructor reads it via a plain `Configuration#getTrimmed` against the + * [[propagateBucketOptions]]-propagated conf (verified via `javap` against `hadoop-aws` 3.3.4 + * and 3.4.1: `conf.getTrimmed("fs.s3a.assumed.role.policy", "")` -- the same consumer shape as + * `assumed.role.arn`/`session.name`, see [[S3ConfigKeyConsumers]]) and, when non-empty, + * attaches it as the session policy of its STS AssumeRole request. + */ + private val S3AssumedRolePolicyKey = "fs.s3a.assumed.role.policy" + + private def assumedRolePolicyReason(bucket: String, key: String): String = + s"Native scan does not support $key configured for $bucket (Hadoop sends the " + + "configured session policy in its STS AssumeRole request, but the native S3 client's " + + "assumed-role provider never reads or forwards this key, so a claimed scan would " + + "assume the role WITHOUT the configured session restriction -- silently widening the " + + "effective permissions instead of failing)" + + /** + * First reason any bucket among `uris` configures an assumed-role session policy via + * [[S3AssumedRolePolicyKey]], or `None` when claimable. Hadoop's + * `AssumedRoleCredentialProvider` includes the policy in its AssumeRole request; native's + * assumed-role provider does not, so a configured policy must decline until native supports it. + * Resolved exactly like the key's real consumer (plain `getTrimmed` on the propagated conf, + * mirroring [[proxyGateReason]]); declined whenever set, whether or not the current provider + * chain names the assumed-role provider -- a policy that is dead config today can become live + * through a provider-chain change native never re-validates. Never interpolates the policy + * document itself, only the key and bucket. + */ + private[comet] def assumedRolePolicyGateReason( + hadoopConf: Configuration, + uris: Seq[URI], + propagatedConfCache: MutableMap[String, Configuration] = MutableMap.empty) + : Option[String] = { + val buckets = uris.flatMap(s3Bucket).distinct + buckets.foldLeft(Option.empty[String]) { (declined, bucket) => + if (declined.isDefined || cometProviderClassConfigured(hadoopConf, bucket)) { Review Comment: [P2] Compare adapter-resolved session policies before bypassing this gate. With `HadoopS3ACredentialProviderAdapter` and `AssumedRoleCredentialProvider`, set `fs.s3a.assumed.role.policy=${fs.s3a.custom.policy}` and give `fs.s3a.custom.policy` and `fs.s3a.bucket.B.custom.policy` different policies. Hadoop propagates the bucket override before expansion, while Comet forwards the already-expanded global policy. This exemption admits the scan, and neither comparison-key list checks the policy afterward. For a read under `B/bucket/*`, a bucket policy allowing that prefix and a global policy allowing only `B/global/*` therefore select different permissions, potentially failing where Spark succeeds. The base Delta gate declined this configuration, so the PR removes a working fallback. Could this exemption require policy equality against the reconstructed adapter view, as the STS endpoint exemption now does? Evidence: Freshly compiled exact-head `S3ConfigGate`, the real Spark-4.x Hadoop adapter and `AdapterSupport`, plus the policy gate extracted from base `0b04b38002bc18e615a84c688df3ad36a25e4f33`. `/tmp/comet7026-d492-review/AdapterConfigProbe.java` uses real `S3AUtils.propagateBucketOptions` and `AdapterSupport.toConfiguration` with synthetic configuration for `s3a://review-test-bucket/bucket/test.parquet`. Assertions confirm Hadoop selects the `bucket/*` policy, the adapter reconstruction selects `global/*`, head returns `None`, and base declines. An equal-policy control is admitted. The probe exited 0 and logged these results in `/tmp/comet7026-d492-review/AdapterConfigProbe.log`. Hadoop 3.3.4 and 3.4.2 `AssumedRoleCredentialProvider` sources both pass the resolved policy directly into the AssumeRole request. No S3/STS client was invoked for this comparison. -- 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]
