[SPARK-57890][CORE] Add core credential types (UserContext, ServiceCredential, UserCredentials)

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.
+ * <p>
+ * 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).
+ * <p>
+ * 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<String, String> 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<String, String> 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<String, String> 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);
+  }
+}

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.
+ * <p>
+ * This class holds the OIDC token information used to derive short-lived
+ * {@link ServiceCredential} instances via credential providers. It is intentionally
+ * <b>not</b> {@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);
+  }
+}

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").
+ * <p>
+ * 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.
+ * <p>
+ * This class is transmitted to executors and does <b>not</b> 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<String, ServiceCredential> credentials;
+
+  /**
+   * Constructs a new {@code UserCredentials} bundle.
+   * <p>
+   * 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<String, ServiceCredential> credentials) {
+    Objects.requireNonNull(credentials, "credentials must not be null");
+    Map<String, ServiceCredential> normalized = new HashMap<>(credentials.size());
+    for (Map.Entry<String, ServiceCredential> 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<String, ServiceCredential> 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<ServiceCredential> 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);
+  }
+}