diff --git a/core/src/main/java/org/apache/spark/security/ServiceCredential.java b/core/src/main/java/org/apache/spark/security/ServiceCredential.java new file mode 100644 index 000000000000..b0502cf922a4 --- /dev/null +++ b/core/src/main/java/org/apache/spark/security/ServiceCredential.java @@ -0,0 +1,122 @@ +/* + * 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.spark.security; + +import java.io.Serializable; +import java.time.Instant; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.Objects; + +import org.apache.spark.annotation.DeveloperApi; + +/** + * :: DeveloperApi :: + * A short-lived, service-specific credential derived from the user's identity token. + *

+ * Instances are produced by credential providers on the driver and transmitted + * to executors via {@link UserCredentials}. The {@code properties} map holds + * service-specific key-value pairs (e.g., temporary AWS credentials). + *

+ * This class is immutable and {@link Serializable}. + * + * @since 4.3.0 + */ +@DeveloperApi +public final class ServiceCredential implements Serializable { + + private static final long serialVersionUID = 1L; + + private final Map properties; + private final Instant expiresAt; + + /** + * Constructs a new {@code ServiceCredential}. + * + * @param properties service-specific credential properties (must not be null; defensively copied) + * @param expiresAt credential expiry time (may be null) + */ + public ServiceCredential(Map properties, Instant expiresAt) { + Objects.requireNonNull(properties, "properties must not be null"); + this.properties = new HashMap<>(properties); + this.expiresAt = expiresAt; + } + + /** + * Returns an unmodifiable view of the credential properties. + */ + public Map getProperties() { + return Collections.unmodifiableMap(properties); + } + + /** Returns the credential expiry time, or {@code null} if not set. */ + public Instant getExpiresAt() { + return expiresAt; + } + + /** + * Returns {@code true} if this credential has expired relative to the given instant. + * If {@code expiresAt} is {@code null}, this method returns {@code false}. + * + * @param now the current time to compare against (must not be null) + * @return whether the credential is expired + */ + public boolean isExpired(Instant now) { + Objects.requireNonNull(now, "now must not be null"); + return expiresAt != null && !now.isBefore(expiresAt); + } + + @Override + public String toString() { + String redactedProps; + if (properties.isEmpty()) { + redactedProps = "{}"; + } else { + StringBuilder sb = new StringBuilder("{"); + boolean first = true; + for (String key : properties.keySet()) { + if (!first) { + sb.append(", "); + } + sb.append(key).append("=[REDACTED]"); + first = false; + } + sb.append("}"); + redactedProps = sb.toString(); + } + return "ServiceCredential{" + + "properties=" + redactedProps + + ", expiresAt=" + expiresAt + + '}'; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + ServiceCredential that = (ServiceCredential) o; + return properties.equals(that.properties) + && Objects.equals(expiresAt, that.expiresAt); + } + + @Override + public int hashCode() { + return Objects.hash(properties, expiresAt); + } +} diff --git a/core/src/main/java/org/apache/spark/security/UserContext.java b/core/src/main/java/org/apache/spark/security/UserContext.java new file mode 100644 index 000000000000..9d9f372dad34 --- /dev/null +++ b/core/src/main/java/org/apache/spark/security/UserContext.java @@ -0,0 +1,139 @@ +/* + * 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.spark.security; + +import java.time.Instant; +import java.util.Objects; + +import org.apache.spark.annotation.DeveloperApi; + +/** + * :: DeveloperApi :: + * Represents the authenticated user's identity context on the driver side. + *

+ * This class holds the OIDC token information used to derive short-lived + * {@link ServiceCredential} instances via credential providers. It is intentionally + * not {@link java.io.Serializable} and must never be transmitted to executors. + * The {@code rawToken} field is always redacted in {@link #toString()}. + * + * @since 4.3.0 + */ +@DeveloperApi +public final class UserContext { + + private final String principal; + private final String issuer; + private final String rawToken; + private final Instant issuedAt; + private final Instant expiresAt; + + /** + * Constructs a new {@code UserContext}. + * + * @param principal the {@code sub} claim from the JWT (must not be null) + * @param issuer the {@code iss} claim from the JWT (must not be null) + * @param rawToken the raw OIDC JWT string (must not be null) + * @param issuedAt token issue time (may be null) + * @param expiresAt token expiry time (may be null) + */ + public UserContext( + String principal, + String issuer, + String rawToken, + Instant issuedAt, + Instant expiresAt) { + this.principal = Objects.requireNonNull(principal, "principal must not be null"); + this.issuer = Objects.requireNonNull(issuer, "issuer must not be null"); + this.rawToken = Objects.requireNonNull(rawToken, "rawToken must not be null"); + this.issuedAt = issuedAt; + this.expiresAt = expiresAt; + } + + /** Returns the {@code sub} claim (principal identifier). */ + public String getPrincipal() { + return principal; + } + + /** Returns the {@code iss} claim (token issuer). */ + public String getIssuer() { + return issuer; + } + + /** Returns the raw OIDC JWT. This value must never be logged or transmitted to executors. */ + public String getRawToken() { + return rawToken; + } + + /** Returns the token issue time, or {@code null} if not set. */ + public Instant getIssuedAt() { + return issuedAt; + } + + /** Returns the token expiry time, or {@code null} if not set. */ + public Instant getExpiresAt() { + return expiresAt; + } + + /** + * Returns {@code true} if this context's token has expired relative to the given instant. + * If {@code expiresAt} is {@code null}, this method returns {@code false}. + * + * @param now the current time to compare against (must not be null) + * @return whether the token is expired + */ + public boolean isExpired(Instant now) { + Objects.requireNonNull(now, "now must not be null"); + return expiresAt != null && !now.isBefore(expiresAt); + } + + /** + * Returns a string representation with the {@code rawToken} redacted as {@code [REDACTED]}. + */ + @Override + public String toString() { + return "UserContext{" + + "principal='" + principal + '\'' + + ", issuer='" + issuer + '\'' + + ", rawToken='[REDACTED]'" + + ", issuedAt=" + issuedAt + + ", expiresAt=" + expiresAt + + '}'; + } + + /** + * Equality is based on identity fields ({@code principal}, {@code issuer}) and token + * validity window ({@code issuedAt}, {@code expiresAt}). The secret {@code rawToken} is + * intentionally excluded: {@code UserContext} is driver-only and never used as a map key, + * so there is no need to compare secret material. + */ + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + UserContext that = (UserContext) o; + return principal.equals(that.principal) + && issuer.equals(that.issuer) + && Objects.equals(issuedAt, that.issuedAt) + && Objects.equals(expiresAt, that.expiresAt); + } + + @Override + public int hashCode() { + return Objects.hash(principal, issuer, issuedAt, expiresAt); + } +} diff --git a/core/src/main/java/org/apache/spark/security/UserCredentials.java b/core/src/main/java/org/apache/spark/security/UserCredentials.java new file mode 100644 index 000000000000..3f5e1d54b923 --- /dev/null +++ b/core/src/main/java/org/apache/spark/security/UserCredentials.java @@ -0,0 +1,113 @@ +/* + * 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.spark.security; + +import java.io.Serializable; +import java.util.Collections; +import java.util.HashMap; +import java.util.Locale; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; + +import org.apache.spark.annotation.DeveloperApi; + +/** + * :: DeveloperApi :: + * A bundle of {@link ServiceCredential} instances keyed by scheme (e.g., "s3a", "abfss"). + *

+ * Scheme keys are normalized to lowercase ({@link Locale#ROOT}) at construction time, and + * lookups via {@link #forScheme(String)} are case-insensitive. If the supplied map contains + * keys that differ only by case, an {@link IllegalArgumentException} is thrown. + *

+ * This class is transmitted to executors and does not contain any reference + * to {@link UserContext} or raw identity tokens. It is immutable and {@link Serializable}. + * + * @since 4.3.0 + */ +@DeveloperApi +public final class UserCredentials implements Serializable { + + private static final long serialVersionUID = 1L; + + private final Map credentials; + + /** + * Constructs a new {@code UserCredentials} bundle. + *

+ * Scheme keys are normalized to lowercase using {@link Locale#ROOT}. If multiple keys + * collide after lowercasing (e.g. {@code "s3a"} and {@code "S3A"}), an + * {@link IllegalArgumentException} is thrown rather than silently keeping one of them. + * + * @param credentials per-scheme map of service credentials (must not be null; defensively copied) + * @throws IllegalArgumentException if two keys collide after case normalization + */ + public UserCredentials(Map credentials) { + Objects.requireNonNull(credentials, "credentials must not be null"); + Map normalized = new HashMap<>(credentials.size()); + for (Map.Entry entry : credentials.entrySet()) { + Objects.requireNonNull(entry.getKey(), "scheme key must not be null"); + String scheme = entry.getKey().toLowerCase(Locale.ROOT); + if (normalized.containsKey(scheme)) { + throw new IllegalArgumentException( + "Duplicate scheme after case normalization: " + entry.getKey()); + } + normalized.put(scheme, entry.getValue()); + } + this.credentials = normalized; + } + + /** + * Returns an unmodifiable view of all credentials keyed by scheme. + */ + public Map getCredentials() { + return Collections.unmodifiableMap(credentials); + } + + /** + * Looks up the {@link ServiceCredential} for the given scheme. The lookup is + * case-insensitive (the argument is lowercased with {@link Locale#ROOT}). + * + * @param scheme the target scheme (e.g., "s3a", "S3A"); must not be null + * @return an {@link Optional} containing the credential, or empty if no credential is registered + */ + public Optional forScheme(String scheme) { + Objects.requireNonNull(scheme, "scheme must not be null"); + return Optional.ofNullable(credentials.get(scheme.toLowerCase(Locale.ROOT))); + } + + @Override + public String toString() { + return "UserCredentials{" + + "credentials=" + credentials + + '}'; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + UserCredentials that = (UserCredentials) o; + return credentials.equals(that.credentials); + } + + @Override + public int hashCode() { + return Objects.hash(credentials); + } +} diff --git a/core/src/test/java/org/apache/spark/security/CredentialTypesSuite.java b/core/src/test/java/org/apache/spark/security/CredentialTypesSuite.java new file mode 100644 index 000000000000..59d8ed66135d --- /dev/null +++ b/core/src/test/java/org/apache/spark/security/CredentialTypesSuite.java @@ -0,0 +1,317 @@ +/* + * 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.spark.security; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.NotSerializableException; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; +import java.time.Instant; +import java.util.HashMap; +import java.util.Map; +import java.util.Optional; + +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class CredentialTypesSuite { + + private static final String TOKEN = "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.secret"; + private static final Instant NOW = Instant.parse("2026-01-15T12:00:00Z"); + private static final Instant PAST = Instant.parse("2026-01-15T11:00:00Z"); + private static final Instant FUTURE = Instant.parse("2026-01-15T13:00:00Z"); + + // --- UserContext tests --- + + @Test + public void testUserContextGettersRoundTrip() { + UserContext ctx = new UserContext("user1", "https://idp.example.com", TOKEN, NOW, FUTURE); + assertEquals("user1", ctx.getPrincipal()); + assertEquals("https://idp.example.com", ctx.getIssuer()); + assertEquals(TOKEN, ctx.getRawToken()); + assertEquals(NOW, ctx.getIssuedAt()); + assertEquals(FUTURE, ctx.getExpiresAt()); + } + + @Test + public void testUserContextToStringRedactsToken() { + UserContext ctx = new UserContext("user1", "https://idp.example.com", TOKEN, NOW, FUTURE); + String str = ctx.toString(); + assertFalse(str.contains(TOKEN), "toString must not contain the raw token value"); + assertTrue(str.contains("[REDACTED]"), "toString must contain [REDACTED]"); + assertTrue(str.contains("user1")); + assertTrue(str.contains("https://idp.example.com")); + } + + @Test + public void testUserContextIsExpiredBoundary() { + // expiresAt == NOW -> expired (now >= expiresAt) + UserContext ctx = new UserContext("u", "iss", TOKEN, PAST, NOW); + assertTrue(ctx.isExpired(NOW), "Should be expired when now == expiresAt"); + assertFalse(ctx.isExpired(PAST), "Should not be expired when now < expiresAt"); + assertTrue(ctx.isExpired(FUTURE), "Should be expired when now > expiresAt"); + } + + @Test + public void testUserContextIsExpiredWithNullExpiry() { + UserContext ctx = new UserContext("u", "iss", TOKEN, NOW, null); + assertFalse(ctx.isExpired(NOW), "Should not be expired when expiresAt is null"); + } + + @Test + public void testUserContextEqualsAndHashCode() { + UserContext a = new UserContext("u", "iss", TOKEN, NOW, FUTURE); + UserContext b = new UserContext("u", "iss", TOKEN, NOW, FUTURE); + UserContext c = new UserContext("other", "iss", TOKEN, NOW, FUTURE); + assertEquals(a, b); + assertEquals(a.hashCode(), b.hashCode()); + assertNotEquals(a, c); + } + + @Test + public void testUserContextEqualsExcludesRawToken() { + // rawToken is secret material and intentionally excluded from equals/hashCode: + // two contexts that differ only by rawToken are considered equal. + UserContext a = new UserContext("u", "iss", "token-A", NOW, FUTURE); + UserContext b = new UserContext("u", "iss", "token-B", NOW, FUTURE); + assertEquals(a, b, "equals must not depend on rawToken"); + assertEquals(a.hashCode(), b.hashCode(), "hashCode must not depend on rawToken"); + // A differing identity field still breaks equality. + UserContext c = new UserContext("u", "other-iss", "token-A", NOW, FUTURE); + assertNotEquals(a, c); + } + + @Test + public void testUserContextNotSerializable() { + UserContext ctx = new UserContext("user1", "https://idp.example.com", TOKEN, NOW, FUTURE); + ByteArrayOutputStream bos = new ByteArrayOutputStream(); + assertThrows(NotSerializableException.class, () -> { + try (ObjectOutputStream oos = new ObjectOutputStream(bos)) { + oos.writeObject(ctx); + } + }); + } + + // --- ServiceCredential tests --- + + @Test + public void testServiceCredentialDefensiveCopy() { + Map props = new HashMap<>(); + props.put("key", "value"); + ServiceCredential cred = new ServiceCredential(props, FUTURE); + // Mutating the original map must not affect the credential + props.put("key2", "value2"); + assertFalse(cred.getProperties().containsKey("key2")); + assertEquals(1, cred.getProperties().size()); + } + + @Test + public void testServiceCredentialIsExpired() { + ServiceCredential cred = new ServiceCredential(Map.of("k", "v"), NOW); + assertTrue(cred.isExpired(NOW), "Should be expired when now == expiresAt"); + assertFalse(cred.isExpired(PAST), "Should not be expired when now < expiresAt"); + assertTrue(cred.isExpired(FUTURE), "Should be expired when now > expiresAt"); + } + + @Test + public void testServiceCredentialIsExpiredWithNullExpiry() { + ServiceCredential cred = new ServiceCredential(Map.of("k", "v"), null); + assertFalse(cred.isExpired(NOW)); + } + + @Test + public void testServiceCredentialSerializationRoundTrip() throws Exception { + Map props = new HashMap<>(); + props.put("fs.s3a.access.key", "AKIAIOSFODNN7EXAMPLE"); + props.put("fs.s3a.secret.key", "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLE"); + ServiceCredential original = new ServiceCredential(props, FUTURE); + + byte[] bytes = serialize(original); + ServiceCredential deserialized = (ServiceCredential) deserialize(bytes); + + assertEquals(original, deserialized); + assertEquals(original.getProperties(), deserialized.getProperties()); + assertEquals(original.getExpiresAt(), deserialized.getExpiresAt()); + } + + @Test + public void testServiceCredentialEqualsAndHashCode() { + ServiceCredential a = new ServiceCredential(Map.of("k", "v"), FUTURE); + ServiceCredential b = new ServiceCredential(Map.of("k", "v"), FUTURE); + ServiceCredential c = new ServiceCredential(Map.of("k", "other"), FUTURE); + assertEquals(a, b); + assertEquals(a.hashCode(), b.hashCode()); + assertNotEquals(a, c); + } + + @Test + public void testServiceCredentialToStringRedactsValues() { + Map props = new HashMap<>(); + props.put("accessKey", "AKIAIOSFODNN7EXAMPLE"); + props.put("secretKey", "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLE"); + ServiceCredential cred = new ServiceCredential(props, FUTURE); + String str = cred.toString(); + // Secret values must NOT appear + assertFalse(str.contains("AKIAIOSFODNN7EXAMPLE"), + "toString must not contain secret property values"); + assertFalse(str.contains("wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLE"), + "toString must not contain secret property values"); + // Keys SHOULD appear for debuggability + assertTrue(str.contains("accessKey"), "toString should contain property key names"); + assertTrue(str.contains("secretKey"), "toString should contain property key names"); + // Values are replaced with [REDACTED] + assertTrue(str.contains("[REDACTED]"), "toString must contain [REDACTED] for values"); + // expiresAt should still appear + assertTrue(str.contains("expiresAt"), "toString should contain expiresAt"); + } + + @Test + public void testServiceCredentialToStringEmptyProperties() { + ServiceCredential cred = new ServiceCredential(Map.of(), null); + String str = cred.toString(); + assertTrue(str.contains("properties={}"), + "toString should handle empty properties map"); + assertTrue(str.contains("expiresAt=null")); + } + + // --- UserCredentials tests --- + + @Test + public void testUserCredentialsForSchemePresent() { + ServiceCredential s3Cred = new ServiceCredential(Map.of("k", "v"), FUTURE); + UserCredentials uc = new UserCredentials(Map.of("s3a", s3Cred)); + Optional result = uc.forScheme("s3a"); + assertTrue(result.isPresent()); + assertEquals(s3Cred, result.get()); + } + + @Test + public void testUserCredentialsForSchemeAbsent() { + ServiceCredential s3Cred = new ServiceCredential(Map.of("k", "v"), FUTURE); + UserCredentials uc = new UserCredentials(Map.of("s3a", s3Cred)); + Optional result = uc.forScheme("abfss"); + assertFalse(result.isPresent()); + } + + @Test + public void testUserCredentialsDoesNotExposeToken() { + // UserCredentials only contains ServiceCredentials, not UserContext or tokens + ServiceCredential cred = new ServiceCredential(Map.of("token", "short-lived"), FUTURE); + UserCredentials uc = new UserCredentials(Map.of("s3a", cred)); + String str = uc.toString(); + assertFalse(str.contains(TOKEN), "UserCredentials must not contain raw identity token"); + // ServiceCredential values should also be redacted in nested toString + assertFalse(str.contains("short-lived"), + "UserCredentials must not expose ServiceCredential property values"); + } + + @Test + public void testUserCredentialsSerializationRoundTrip() throws Exception { + Map creds = new HashMap<>(); + creds.put("s3a", new ServiceCredential(Map.of("key", "val1"), FUTURE)); + creds.put("abfss", new ServiceCredential(Map.of("token", "val2"), NOW)); + UserCredentials original = new UserCredentials(creds); + + byte[] bytes = serialize(original); + UserCredentials deserialized = (UserCredentials) deserialize(bytes); + + assertEquals(original, deserialized); + assertEquals(original.getCredentials(), deserialized.getCredentials()); + } + + @Test + public void testUserCredentialsDefensiveCopy() { + Map creds = new HashMap<>(); + creds.put("s3a", new ServiceCredential(Map.of("k", "v"), FUTURE)); + UserCredentials uc = new UserCredentials(creds); + // Mutating the original map must not affect the credentials + creds.put("extra", new ServiceCredential(Map.of("x", "y"), NOW)); + assertFalse(uc.forScheme("extra").isPresent()); + } + + @Test + public void testUserCredentialsEqualsAndHashCode() { + ServiceCredential cred = new ServiceCredential(Map.of("k", "v"), FUTURE); + UserCredentials a = new UserCredentials(Map.of("s3a", cred)); + UserCredentials b = new UserCredentials(Map.of("s3a", cred)); + UserCredentials c = new UserCredentials(Map.of("abfss", cred)); + assertEquals(a, b); + assertEquals(a.hashCode(), b.hashCode()); + assertNotEquals(a, c); + } + + @Test + public void testUserCredentialsForSchemeCaseInsensitive() { + ServiceCredential cred = new ServiceCredential(Map.of("key", "val"), FUTURE); + // Store with lowercase key, lookup with uppercase + UserCredentials uc1 = new UserCredentials(Map.of("s3a", cred)); + assertTrue(uc1.forScheme("S3A").isPresent(), "Uppercase lookup should find lowercase key"); + assertEquals(cred, uc1.forScheme("S3A").get()); + assertTrue(uc1.forScheme("s3a").isPresent(), "Exact case lookup should still work"); + assertTrue(uc1.forScheme("S3a").isPresent(), "Mixed case lookup should work"); + + // Store with uppercase key, lookup with lowercase + UserCredentials uc2 = new UserCredentials(Map.of("ABFSS", cred)); + assertTrue(uc2.forScheme("abfss").isPresent(), "Lowercase lookup should find uppercase key"); + assertEquals(cred, uc2.forScheme("abfss").get()); + + // Absent scheme still returns empty + assertFalse(uc2.forScheme("hdfs").isPresent(), "Absent scheme should return empty"); + } + + @Test + public void testUserCredentialsRejectsCaseCollidingSchemes() { + Map creds = new HashMap<>(); + creds.put("s3a", new ServiceCredential(Map.of("k", "v1"), FUTURE)); + creds.put("S3A", new ServiceCredential(Map.of("k", "v2"), FUTURE)); + IllegalArgumentException e = assertThrows(IllegalArgumentException.class, + () -> new UserCredentials(creds)); + assertTrue(e.getMessage().contains("Duplicate scheme"), + "Exception message should identify the collision"); + } + + @Test + public void testUserCredentialsRejectsNullSchemeKey() { + Map creds = new HashMap<>(); + creds.put(null, new ServiceCredential(Map.of("k", "v"), FUTURE)); + assertThrows(NullPointerException.class, () -> new UserCredentials(creds)); + } + + // --- Helpers --- + + private static byte[] serialize(Object obj) throws Exception { + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + try (ObjectOutputStream oos = new ObjectOutputStream(baos)) { + oos.writeObject(obj); + } + return baos.toByteArray(); + } + + private static Object deserialize(byte[] bytes) throws Exception { + ByteArrayInputStream bais = new ByteArrayInputStream(bytes); + try (ObjectInputStream ois = new ObjectInputStream(bais)) { + return ois.readObject(); + } + } +}