HADOOP-16658. S3A connector does not support including the token renewer in the token identifier.

Contributed by Phil Zampino.

Change-Id: Iea9d5028dcf58bda4da985604f5cd3ac283619bd

Author: pzampino

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java

@@ -110,6 +110,7 @@
 import org.apache.hadoop.fs.s3a.s3guard.BulkOperationState;
 import org.apache.hadoop.fs.s3a.select.InternalSelectConstants;
 import org.apache.hadoop.io.IOUtils;
+import org.apache.hadoop.io.Text;
 import org.apache.hadoop.security.token.TokenIdentifier;
 import org.apache.hadoop.util.DurationInfo;
 import org.apache.hadoop.util.LambdaUtils;
@@ -3186,7 +3187,8 @@ public Token<AbstractS3ATokenIdentifier> getDelegationToken(String renewer)
     entryPoint(Statistic.INVOCATION_GET_DELEGATION_TOKEN);
     LOG.debug("Delegation token requested");
     if (delegationTokens.isPresent()) {
-      return delegationTokens.get().getBoundOrNewDT(encryptionSecrets);
+      return delegationTokens.get().getBoundOrNewDT(encryptionSecrets,
+          (renewer != null ? new Text(renewer) : new Text()));
     } else {
       // Delegation token support is not set up
       LOG.debug("Token support is not enabled");

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

@@ -129,15 +129,17 @@ public S3ADelegationTokens.TokenIssuingPolicy getTokenIssuingPolicy() {
    * filesystem has been deployed unbonded.
    * @param policy minimum policy to use, if known.
    * @param encryptionSecrets encryption secrets for the token.
+   * @param renewer the principal permitted to renew the token.
    * @return the token or null if the back end does not want to issue one.
    * @throws IOException if one cannot be created
    */
   public Token<AbstractS3ATokenIdentifier> createDelegationToken(
       final Optional<RoleModel.Policy> policy,
-      final EncryptionSecrets encryptionSecrets) throws IOException {
+      final EncryptionSecrets encryptionSecrets,
+      final Text renewer) throws IOException {
     requireServiceStarted();
     final AbstractS3ATokenIdentifier tokenIdentifier =
-            createTokenIdentifier(policy, encryptionSecrets);
+            createTokenIdentifier(policy, encryptionSecrets, renewer);
     if (tokenIdentifier != null) {
       Token<AbstractS3ATokenIdentifier> token =
           new Token<>(tokenIdentifier, secretManager);
@@ -157,17 +159,19 @@ public Token<AbstractS3ATokenIdentifier> createDelegationToken(
    * This will only be called if a new DT is needed, that is: the
    * filesystem has been deployed unbonded.
    *
-   * If {@link #createDelegationToken(Optional, EncryptionSecrets)}
+   * If {@link #createDelegationToken(Optional, EncryptionSecrets, Text)}
    * is overridden, this method can be replaced with a stub.
    *
    * @param policy minimum policy to use, if known.
    * @param encryptionSecrets encryption secrets for the token.
+   * @param renewer the principal permitted to renew the token.
    * @return the token data to include in the token identifier.
    * @throws IOException failure creating the token data.
    */
   public abstract AbstractS3ATokenIdentifier createTokenIdentifier(
       Optional<RoleModel.Policy> policy,
-      EncryptionSecrets encryptionSecrets) throws IOException;
+      EncryptionSecrets encryptionSecrets,
+      Text renewer) throws IOException;
 
   /**
    * Verify that a token identifier is of a specific class.

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

@@ -103,17 +103,23 @@ public abstract class AbstractS3ATokenIdentifier
    * Constructor.
    * @param kind token kind.
    * @param uri filesystem URI.
-   * @param owner token owner
+   * @param owner token owner.
+   * @param renewer token renewer.
    * @param origin origin text for diagnostics.
    * @param encryptionSecrets encryption secrets to set.
    */
   protected AbstractS3ATokenIdentifier(
       final Text kind,
       final URI uri,
       final Text owner,
+      final Text renewer,
       final String origin,
       final EncryptionSecrets encryptionSecrets) {
-    this(kind, owner, new Text(), new Text(), uri);
+    this(kind,
+         owner,
+         (renewer != null ? renewer : new Text()),
+         new Text(),
+         uri);
     this.origin = requireNonNull(origin);
     this.encryptionSecrets = requireNonNull(encryptionSecrets);
   }
@@ -237,6 +243,7 @@ public String toString() {
     sb.append(getKind());
     sb.append("; uri=").append(uri);
     sb.append("; timestamp=").append(created);
+    sb.append("; renewer=").append(getRenewer());
     sb.append("; encryption=").append(encryptionSecrets.toString());
     sb.append("; ").append(uuid);
     sb.append("; ").append(origin);

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

@@ -30,6 +30,7 @@
 import org.apache.hadoop.fs.s3a.auth.MarshalledCredentials;
 import org.apache.hadoop.fs.s3a.auth.RoleModel;
 import org.apache.hadoop.fs.s3native.S3xLoginHelper;
+import org.apache.hadoop.io.Text;
 
 import static org.apache.hadoop.fs.s3a.auth.delegation.DelegationConstants.FULL_TOKEN_KIND;
 
@@ -61,7 +62,7 @@ public class FullCredentialsTokenBinding extends
   private String credentialOrigin;
 
   /**
-   * Constructor, uses name of {@link #name} and token kind of
+   * Constructor, uses name of {@link #NAME} and token kind of
    * {@link DelegationConstants#FULL_TOKEN_KIND}.
    *
    */
@@ -138,11 +139,13 @@ public AWSCredentialProviderList deployUnbonded() throws IOException {
   @Override
   public AbstractS3ATokenIdentifier createTokenIdentifier(
       final Optional<RoleModel.Policy> policy,
-      final EncryptionSecrets encryptionSecrets) throws IOException {
+      final EncryptionSecrets encryptionSecrets,
+      final Text renewer) throws IOException {
     requireServiceStarted();
 
     return new FullCredentialsTokenIdentifier(getCanonicalUri(),
         getOwnerText(),
+        renewer,
         awsCredentials,
         encryptionSecrets,
         credentialOrigin);

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

@@ -37,11 +37,13 @@ public FullCredentialsTokenIdentifier() {
 
   public FullCredentialsTokenIdentifier(final URI uri,
       final Text owner,
+      final Text renewer,
       final MarshalledCredentials marshalledCredentials,
       final EncryptionSecrets encryptionSecrets,
       String origin) {
     super(DelegationConstants.FULL_TOKEN_KIND,
         owner,
+        renewer,
         uri,
         marshalledCredentials,
         encryptionSecrets,

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

@@ -36,6 +36,7 @@
 import org.apache.hadoop.fs.s3a.auth.MarshalledCredentials;
 import org.apache.hadoop.fs.s3a.auth.RoleModel;
 import org.apache.hadoop.fs.s3a.auth.STSClientFactory;
+import org.apache.hadoop.io.Text;
 
 import static org.apache.hadoop.fs.s3a.auth.MarshalledCredentialBinding.fromSTSCredentials;
 import static org.apache.hadoop.fs.s3a.auth.delegation.DelegationConstants.DELEGATION_TOKEN_CREDENTIALS_PROVIDER;
@@ -75,7 +76,7 @@ public class RoleTokenBinding extends SessionTokenBinding {
 
   /**
    * Constructor.
-   * Name is {@link #name}; token kind is
+   * Name is {@link #NAME}; token kind is
    * {@link DelegationConstants#ROLE_TOKEN_KIND}.
    */
   public RoleTokenBinding() {
@@ -129,7 +130,8 @@ COMPONENT, getFileSystem().getUri(),
   @Retries.RetryTranslated
   public RoleTokenIdentifier createTokenIdentifier(
       final Optional<RoleModel.Policy> policy,
-      final EncryptionSecrets encryptionSecrets) throws IOException {
+      final EncryptionSecrets encryptionSecrets,
+      final Text renewer) throws IOException {
     requireServiceStarted();
     Preconditions.checkState(!roleArn.isEmpty(), E_NO_ARN);
     String policyJson = policy.isPresent() ?
@@ -152,6 +154,7 @@ public RoleTokenIdentifier createTokenIdentifier(
     return new RoleTokenIdentifier(
         getCanonicalUri(),
         getOwnerText(),
+        renewer,
         fromSTSCredentials(credentials),
         encryptionSecrets,
         AbstractS3ATokenIdentifier.createDefaultOriginMessage()

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

@@ -35,11 +35,13 @@ public RoleTokenIdentifier() {
 
   public RoleTokenIdentifier(final URI uri,
       final Text owner,
+      final Text renewer,
       final MarshalledCredentials marshalledCredentials,
       final EncryptionSecrets encryptionSecrets,
       final String origin) {
     super(DelegationConstants.ROLE_TOKEN_KIND,
         owner,
+        renewer,
         uri,
         marshalledCredentials,
         encryptionSecrets,

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

@@ -352,7 +352,7 @@ public Optional<Token<AbstractS3ATokenIdentifier>> getBoundDT() {
 
   /**
    * Predicate: will this binding issue a DT if requested
-   * in a call to {@link #getBoundOrNewDT(EncryptionSecrets)}?
+   * in a call to {@link #getBoundOrNewDT(EncryptionSecrets, Text)}?
    * That is: should the filesystem declare that it is issuing
    * delegation tokens?
    * @return a declaration of what will happen when asked for a token.
@@ -368,10 +368,12 @@ public TokenIssuingPolicy getTokenIssuingPolicy() {
    * @return a delegation token.
    * @throws IOException if one cannot be created
    * @param encryptionSecrets encryption secrets for any new token.
+   * @param renewer the token renewer.
    */
   @SuppressWarnings("OptionalGetWithoutIsPresent")
   public Token<AbstractS3ATokenIdentifier> getBoundOrNewDT(
-      final EncryptionSecrets encryptionSecrets)
+      final EncryptionSecrets encryptionSecrets,
+      final Text renewer)
       throws IOException {
     LOG.debug("Delegation token requested");
     if (isBoundToDT()) {
@@ -382,13 +384,13 @@ public Token<AbstractS3ATokenIdentifier> getBoundOrNewDT(
       // not bound to a token, so create a new one.
       // issued DTs are not cached so that long-lived filesystems can
       // reliably issue session/role tokens.
-      return createDelegationToken(encryptionSecrets);
+      return createDelegationToken(encryptionSecrets, renewer);
     }
   }
 
   /**
    * How many delegation tokens have been issued?
-   * @return the number times {@link #createDelegationToken(EncryptionSecrets)}
+   * @return the number times {@link #createDelegationToken(EncryptionSecrets, Text)}
    * returned a token.
    */
   public int getCreationCount() {
@@ -400,12 +402,14 @@ public int getCreationCount() {
    * This will only be called if a new DT is needed, that is: the
    * filesystem has been deployed unbonded.
    * @param encryptionSecrets encryption secrets for the token.
+   * @param renewer the token renewer
    * @return the token
    * @throws IOException if one cannot be created
    */
   @VisibleForTesting
   public Token<AbstractS3ATokenIdentifier> createDelegationToken(
-      final EncryptionSecrets encryptionSecrets) throws IOException {
+      final EncryptionSecrets encryptionSecrets,
+      final Text renewer) throws IOException {
     requireServiceStarted();
     checkArgument(encryptionSecrets != null,
         "Null encryption secrets");
@@ -420,7 +424,7 @@ public Token<AbstractS3ATokenIdentifier> createDelegationToken(
     try(DurationInfo ignored = new DurationInfo(LOG, DURATION_LOG_AT_INFO,
         "Creating New Delegation Token", tokenBinding.getKind())) {
       Token<AbstractS3ATokenIdentifier> token
-          = tokenBinding.createDelegationToken(rolePolicy, encryptionSecrets);
+          = tokenBinding.createDelegationToken(rolePolicy, encryptionSecrets, renewer);
       if (token != null) {
         token.setService(service);
         noteTokenCreated(token);

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

@@ -353,7 +353,8 @@ public long getDuration() {
   @Retries.RetryTranslated
   public SessionTokenIdentifier createTokenIdentifier(
       final Optional<RoleModel.Policy> policy,
-      final EncryptionSecrets encryptionSecrets) throws IOException {
+      final EncryptionSecrets encryptionSecrets,
+      final Text renewer) throws IOException {
     requireServiceStarted();
 
     final MarshalledCredentials marshalledCredentials;
@@ -384,11 +385,12 @@ public SessionTokenIdentifier createTokenIdentifier(
       }
     }
     return new SessionTokenIdentifier(getKind(),
-        getOwnerText(),
-        getCanonicalUri(),
-        marshalledCredentials,
-        encryptionSecrets,
-        origin);
+         getOwnerText(),
+         renewer,
+         getCanonicalUri(),
+         marshalledCredentials,
+         encryptionSecrets,
+         origin);
   }
 
   @Override

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

@@ -73,7 +73,8 @@ protected SessionTokenIdentifier(final Text kind) {
   /**
    * Constructor.
    * @param kind token kind.
-   * @param owner token owner
+   * @param owner token owner.
+   * @param renewer token renewer.
    * @param uri filesystem URI.
    * @param marshalledCredentials credentials to marshall
    * @param encryptionSecrets encryption secrets
@@ -82,11 +83,12 @@ protected SessionTokenIdentifier(final Text kind) {
   public SessionTokenIdentifier(
       final Text kind,
       final Text owner,
+      final Text renewer,
       final URI uri,
       final MarshalledCredentials marshalledCredentials,
       final EncryptionSecrets encryptionSecrets,
       final String origin) {
-    super(kind, uri, owner, origin, encryptionSecrets);
+    super(kind, uri, owner, renewer, origin, encryptionSecrets);
     this.marshalledCredentials = marshalledCredentials;
   }