HADOOP-18795. S3A DelegationToken plugin to expand return type of binding (#5821)

Adds a class DelegationBindingInfo which contains binding info
beyond just the AWS credential list.

The binding class can be expanded when needed. Until then, all existing
implementations will work, as the new method
  DelegationBindingInfo deploy(AbstractS3ATokenIdentifier retrievedIdentifier)
falls back to the original methods.

Contributed by Steve Loughran

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/auth/delegation/AbstractDelegationTokenBinding.java

@@ -198,15 +198,37 @@ public abstract class AbstractDelegationTokenBinding extends AbstractDTService {
     return (T) identifier;
   }
 
+  /**
+   * Deploy, returning the binding information.
+   * The base implementation calls
+   *
+   * @param retrievedIdentifier any identifier -null if deployed unbonded.
+   * @return binding information
+   * @throws IOException any failure.
+   */
+  public DelegationBindingInfo deploy(AbstractS3ATokenIdentifier retrievedIdentifier)
+        throws IOException {
+    requireServiceStarted();
+    AWSCredentialProviderList credentialProviders =
+        retrievedIdentifier == null
+            ? deployUnbonded()
+            : bindToTokenIdentifier(retrievedIdentifier);
+    return new DelegationBindingInfo()
+        .withCredentialProviders(credentialProviders);
+  }
+
   /**
    * Perform any actions when deploying unbonded, and return a list
    * of credential providers.
    * @return non-empty list of AWS credential providers to use for
    * authenticating this client with AWS services.
    * @throws IOException any failure.
+   * @throws UnsupportedOperationException in the base implementation.
    */
-  public abstract AWSCredentialProviderList deployUnbonded()
-      throws IOException;
+  public AWSCredentialProviderList deployUnbonded()
+      throws IOException {
+    throw new UnsupportedOperationException("unimplemented");
+  }
 
   /**
    * Bind to the token identifier, returning the credential providers to use
@@ -214,11 +236,14 @@ public abstract class AbstractDelegationTokenBinding extends AbstractDTService {
    * @param retrievedIdentifier the unmarshalled data
    * @return non-empty list of AWS credential providers to use for
    * authenticating this client with AWS services.
-   * @throws IOException any failure.
+   * @throws IOException any failure
+   * @throws UnsupportedOperationException in the base implementation.
    */
-  public abstract AWSCredentialProviderList bindToTokenIdentifier(
+  public AWSCredentialProviderList bindToTokenIdentifier(
       AbstractS3ATokenIdentifier retrievedIdentifier)
-      throws IOException;
+      throws IOException  {
+    throw new UnsupportedOperationException("unimplemented");
+  }
 
   /**
    * Create a new subclass of {@link AbstractS3ATokenIdentifier}.

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/auth/delegation/DelegationBindingInfo.java

@@ -0,0 +1,52 @@
+/*
+ * 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.hadoop.fs.s3a.auth.delegation;
+
+import org.apache.hadoop.fs.s3a.AWSCredentialProviderList;
+
+import static java.util.Objects.requireNonNull;
+
+/**
+ * Binding information returned by the token provider.
+ */
+public final class DelegationBindingInfo {
+
+  /**
+   * List of credential providers.
+   */
+  private AWSCredentialProviderList credentialProviders;
+
+  /**
+   * Get list of credential providers.
+   * @return list of credential providers
+   */
+  public AWSCredentialProviderList getCredentialProviders() {
+    return credentialProviders;
+  }
+
+  /**
+   * Set builder value.
+   * @param value non null value
+   * @return the builder
+   */
+  public DelegationBindingInfo withCredentialProviders(final AWSCredentialProviderList value) {
+    credentialProviders = requireNonNull(value);
+    return this;
+  }
+}

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/auth/delegation/S3ADelegationTokens.java

@@ -123,7 +123,12 @@ public class S3ADelegationTokens extends AbstractDTService {
   /**
    * List of cred providers; unset until {@link #bindToDelegationToken(Token)}.
    */
-  private Optional<AWSCredentialProviderList> credentialProviders
+  //private Optional<AWSCredentialProviderList> credentialProviders = Optional.empty();
+
+  /**
+   * delegation binding information; unset until {@link #bindToDelegationToken(Token)}.
+   */
+  private Optional<DelegationBindingInfo> bindingInfo
       = Optional.empty();
 
   /**
@@ -235,10 +240,7 @@ public class S3ADelegationTokens extends AbstractDTService {
 
   /**
    * Perform the unbonded deployment operations.
-   * Create the AWS credential provider chain to use
-   * when talking to AWS when there is no delegation token to work with.
-   * authenticating this client with AWS services, and saves it
-   * to {@link #credentialProviders}
+   * Create and store the binding information.
    *
    * @throws IOException any failure.
    */
@@ -248,7 +250,7 @@ public class S3ADelegationTokens extends AbstractDTService {
     checkState(!isBoundToDT(),
         "Already Bound to a delegation token");
     LOG.debug("No delegation tokens present: using direct authentication");
-    credentialProviders = Optional.of(tokenBinding.deployUnbonded());
+    bindingInfo = Optional.of(tokenBinding.deploy(null));
   }
 
   /**
@@ -260,12 +262,12 @@ public class S3ADelegationTokens extends AbstractDTService {
    * <ol>
    *   <li>{@link #boundDT} is set to the retrieved token.</li>
    *   <li>{@link #decodedIdentifier} is set to the extracted identifier.</li>
-   *   <li>{@link #credentialProviders} is set to the credential
-   *   provider(s) returned by the token binding.</li>
+   *   <li>{@link #bindingInfo} is set to the information
+   *   returned by the token binding.</li>
    * </ol>
    * If unsuccessful, {@link #deployUnbonded()} is called for the
    * unbonded codepath instead, which will set
-   * {@link #credentialProviders} to its value.
+   * {@link #bindingInfo} to its value.
    *
    * This means after this call (and only after) the token operations
    * can be invoked.
@@ -276,14 +278,14 @@ public class S3ADelegationTokens extends AbstractDTService {
    * @throws IOException selection/extraction/validation failure.
    */
   private void bindToAnyDelegationToken() throws IOException {
-    checkState(!credentialProviders.isPresent(), E_ALREADY_DEPLOYED);
+    checkState(!bindingInfo.isPresent(), E_ALREADY_DEPLOYED);
     Token<AbstractS3ATokenIdentifier> token = selectTokenFromFSOwner();
     if (token != null) {
       bindToDelegationToken(token);
     } else {
       deployUnbonded();
     }
-    if (credentialProviders.get().size() == 0) {
+    if (getCredentialProviders().size() == 0) {
       throw new DelegationTokenIOException("No AWS credential providers"
           + " created by Delegation Token Binding "
           + tokenBinding.getName());
@@ -306,7 +308,7 @@ public class S3ADelegationTokens extends AbstractDTService {
   @VisibleForTesting
   void resetTokenBindingToDT(final Token<AbstractS3ATokenIdentifier> token)
       throws IOException{
-    credentialProviders = Optional.empty();
+    bindingInfo = Optional.empty();
     bindToDelegationToken(token);
   }
 
@@ -317,8 +319,8 @@ public class S3ADelegationTokens extends AbstractDTService {
    * <ol>
    *   <li>{@link #boundDT} is set to {@code token}.</li>
    *   <li>{@link #decodedIdentifier} is set to the extracted identifier.</li>
-   *   <li>{@link #credentialProviders} is set to the credential
-   *   provider(s) returned by the token binding.</li>
+   *   <li>{@link #bindingInfo} is set to the info
+   *    returned by the token binding.</li>
    * </ol>
    * @param token token to decode and bind to.
    * @throws IOException selection/extraction/validation failure.
@@ -327,7 +329,7 @@ public class S3ADelegationTokens extends AbstractDTService {
   public void bindToDelegationToken(
       final Token<AbstractS3ATokenIdentifier> token)
       throws IOException {
-    checkState(!credentialProviders.isPresent(), E_ALREADY_DEPLOYED);
+    checkState(!bindingInfo.isPresent(), E_ALREADY_DEPLOYED);
     boundDT = Optional.of(token);
     AbstractS3ATokenIdentifier dti = extractIdentifier(token);
     LOG.info("Using delegation token {}", dti);
@@ -335,8 +337,7 @@ public class S3ADelegationTokens extends AbstractDTService {
     try (DurationInfo ignored = new DurationInfo(LOG, DURATION_LOG_AT_INFO,
         "Creating Delegation Token")) {
       // extract the credential providers.
-      credentialProviders = Optional.of(
-          tokenBinding.bindToTokenIdentifier(dti));
+      bindingInfo = Optional.of(tokenBinding.deploy(dti));
     }
   }
 
@@ -470,8 +471,9 @@ public class S3ADelegationTokens extends AbstractDTService {
    */
   public AWSCredentialProviderList getCredentialProviders()
       throws IOException {
-    return credentialProviders.orElseThrow(
-        () -> new DelegationTokenIOException("Not yet bonded"));
+    return bindingInfo.map(DelegationBindingInfo::getCredentialProviders)
+        .orElseThrow(() ->
+            new DelegationTokenIOException("Not yet bonded"));
   }
 
   /**