HADOOP-15922. Fixed doAsUser decoding for DelegationTokenAuthenticationFilter.

Contributed by He Xiaoqiao

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/token/delegation/web/DelegationTokenAuthenticationFilter.java

@@ -51,6 +51,7 @@
 import javax.servlet.http.HttpServletResponse;
 
 import java.io.IOException;
+import java.net.URLDecoder;
 import java.nio.charset.Charset;
 import java.security.Principal;
 import java.util.Enumeration;
@@ -230,7 +231,12 @@ static String getDoAs(HttpServletRequest request) {
       for (NameValuePair nv : list) {
         if (DelegationTokenAuthenticatedURL.DO_AS.
             equalsIgnoreCase(nv.getName())) {
-          return nv.getValue();
+          String doAsUser = nv.getValue();
+          try {
+            doAsUser = URLDecoder.decode(nv.getValue(), UTF8_CHARSET.name());
+          } finally {
+            return doAsUser;
+          }
         }
       }
     }

hadoop-common-project/hadoop-kms/src/test/java/org/apache/hadoop/crypto/key/kms/server/TestKMS.java

@@ -2115,6 +2115,70 @@ public Void run() throws Exception {
     });
   }
 
+  @Test
+  public void testGetDelegationTokenByProxyUser() throws Exception {
+    Configuration conf = new Configuration();
+    conf.set("hadoop.security.authentication", "kerberos");
+    UserGroupInformation.setConfiguration(conf);
+    final File testDir = getTestDir();
+
+    conf = createBaseKMSConf(testDir, conf);
+    conf.set("hadoop.kms.authentication.type", "kerberos");
+    conf.set("hadoop.kms.authentication.kerberos.keytab",
+        keytab.getAbsolutePath());
+    conf.set("hadoop.kms.authentication.kerberos.principal", "HTTP/localhost");
+    conf.set("hadoop.kms.authentication.kerberos.name.rules", "DEFAULT");
+    conf.set("hadoop.kms.proxyuser.client.users", "foo/localhost");
+    conf.set("hadoop.kms.proxyuser.client.hosts", "localhost");
+    conf.set(KeyAuthorizationKeyProvider.KEY_ACL + "kcc.ALL",
+        "foo/localhost");
+
+    writeConf(testDir, conf);
+
+    runServer(null, null, testDir, new KMSCallable<Void>() {
+      @Override
+      public Void call() throws Exception {
+        final Configuration conf = new Configuration();
+        final URI uri = createKMSUri(getKMSUrl());
+
+        // proxyuser client using kerberos credentials
+        UserGroupInformation proxyUgi = UserGroupInformation.
+            loginUserFromKeytabAndReturnUGI("client/host", keytab.getAbsolutePath());
+        UserGroupInformation foo = UserGroupInformation.createProxyUser(
+            "foo/localhost", proxyUgi);
+        final Credentials credentials = new Credentials();
+        foo.doAs(new PrivilegedExceptionAction<Void>() {
+          @Override
+          public Void run() throws Exception {
+            final KeyProvider kp = createProvider(uri, conf);
+            KeyProviderDelegationTokenExtension keyProviderDelegationTokenExtension
+                = KeyProviderDelegationTokenExtension
+                    .createKeyProviderDelegationTokenExtension(kp);
+            keyProviderDelegationTokenExtension.addDelegationTokens("client",
+                credentials);
+            Assert.assertNotNull(kp.createKey("kcc",
+                new KeyProvider.Options(conf)));
+            return null;
+          }
+        });
+
+        // current user client using token credentials for proxy user
+        UserGroupInformation nonKerberosUgi
+            = UserGroupInformation.getCurrentUser();
+        nonKerberosUgi.addCredentials(credentials);
+        nonKerberosUgi.doAs(new PrivilegedExceptionAction<Void>() {
+          @Override
+          public Void run() throws Exception {
+            final KeyProvider kp = createProvider(uri, conf);
+            Assert.assertNotNull(kp.getMetadata("kcc"));
+            return null;
+          }
+        });
+        return null;
+      }
+    });
+  }
+
   private Configuration setupConfForKerberos(File confDir) throws Exception {
     final Configuration conf =  createBaseKMSConf(confDir, null);
     conf.set("hadoop.security.authentication", "kerberos");