HADOOP-6510. Adds a way for superusers to impersonate other users in a secure environment. Contributed by Jitendra Nath Pandey.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/trunk@907549 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -51,6 +51,9 @@ Trunk (unreleased changes)
     HADOOP-6419. Adds SASL based authentication to RPC.
     (Kan Zhang via ddas)
 
+    HADOOP-6510. Adds a way for superusers to impersonate other users
+    in a secure environment. (Jitendra Nath Pandey via ddas)
+
   IMPROVEMENTS
 
     HADOOP-6283. Improve the exception messages thrown by

src/java/org/apache/hadoop/ipc/Client.java

@@ -234,8 +234,6 @@ public Connection(ConnectionId remoteId) throws IOException {
       
       UserGroupInformation ticket = remoteId.getTicket();
       Class<?> protocol = remoteId.getProtocol();
-      header = 
-        new ConnectionHeader(protocol == null ? null : protocol.getName(), ticket);
       this.useSasl = UserGroupInformation.isSecurityEnabled();
       if (useSasl && protocol != null) {
         TokenInfo tokenInfo = protocol.getAnnotation(TokenInfo.class);
@@ -269,6 +267,10 @@ public Connection(ConnectionId remoteId) throws IOException {
       } else {
         authMethod = AuthMethod.KERBEROS;
       }
+      
+      header = new ConnectionHeader(protocol == null ? null : protocol
+          .getName(), ticket, authMethod);
+      
       if (LOG.isDebugEnabled())
         LOG.debug("Use " + authMethod + " authentication for protocol "
             + protocol.getSimpleName());
@@ -400,7 +402,13 @@ private synchronized void setupIOstreams() throws InterruptedException {
         if (useSasl) {
           final InputStream in2 = inStream;
           final OutputStream out2 = outStream;
-          remoteId.getTicket().doAs(new PrivilegedExceptionAction<Object>() {
+          UserGroupInformation ticket = remoteId.getTicket();
+          if (authMethod == AuthMethod.KERBEROS) {
+            if (ticket.getRealUser() != null) {
+              ticket = ticket.getRealUser();
+            }
+          }
+          ticket.doAs(new PrivilegedExceptionAction<Object>() {
             @Override
             public Object run() throws IOException {
               saslRpcClient = new SaslRpcClient(authMethod, token,

src/java/org/apache/hadoop/ipc/ConnectionHeader.java

@@ -20,12 +20,16 @@
 import java.io.DataInput;
 import java.io.DataOutput;
 import java.io.IOException;
+import java.util.Collection;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.io.Writable;
 import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.SaslRpcServer.AuthMethod;
+import org.apache.hadoop.security.token.Token;
+import org.apache.hadoop.security.token.TokenIdentifier;
 
 /**
  * The IPC connection header sent by the client to the server
@@ -36,6 +40,7 @@ class ConnectionHeader implements Writable {
   
   private String protocol;
   private UserGroupInformation ugi = null;
+  private AuthMethod authMethod;
   
   public ConnectionHeader() {}
   
@@ -47,9 +52,10 @@ public ConnectionHeader() {}
    * @param ugi {@link UserGroupInformation} of the client communicating with
    *            the server
    */
-  public ConnectionHeader(String protocol, UserGroupInformation ugi) {
+  public ConnectionHeader(String protocol, UserGroupInformation ugi, AuthMethod authMethod) {
     this.protocol = protocol;
     this.ugi = ugi;
+    this.authMethod = authMethod;
   }
 
   @Override
@@ -62,7 +68,15 @@ public void readFields(DataInput in) throws IOException {
     boolean ugiUsernamePresent = in.readBoolean();
     if (ugiUsernamePresent) {
       String username = in.readUTF();
+      boolean realUserNamePresent = in.readBoolean();
+      if (realUserNamePresent) {
+        String realUserName = in.readUTF();
+        UserGroupInformation realUserUgi = UserGroupInformation
+            .createRemoteUser(realUserName);
+        ugi = UserGroupInformation.createProxyUser(username, realUserUgi);
+      } else {
       	ugi = UserGroupInformation.createRemoteUser(username);
+      }
     } else {
       ugi = null;
     }
@@ -72,8 +86,27 @@ public void readFields(DataInput in) throws IOException {
   public void write(DataOutput out) throws IOException {
     Text.writeString(out, (protocol == null) ? "" : protocol);
     if (ugi != null) {
+      if (UserGroupInformation.isSecurityEnabled()) {
+        if (authMethod == AuthMethod.KERBEROS) {
+          //Send effective user for Kerberos auth
           out.writeBoolean(true);
           out.writeUTF(ugi.getUserName());
+          out.writeBoolean(false);
+        } else {
+          //Don't send user for token auth
+          out.writeBoolean(false);
+        }
+      } else {
+        //Send both effective user and real user for simple auth
+        out.writeBoolean(true);
+        out.writeUTF(ugi.getUserName());
+        if (ugi.getRealUser() != null) {
+          out.writeBoolean(true);
+          out.writeUTF(ugi.getRealUser().getUserName());
+        } else {
+          out.writeBoolean(false);
+        }
+      }
     } else {
       out.writeBoolean(false);
     }

src/java/org/apache/hadoop/ipc/Server.java

@@ -68,6 +68,7 @@
 import org.apache.hadoop.security.SaslRpcServer.SaslDigestCallbackHandler;
 import org.apache.hadoop.security.SaslRpcServer.SaslGssCallbackHandler;
 import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.authorize.ProxyUsers;
 import org.apache.hadoop.security.authorize.AuthorizationException;
 import org.apache.hadoop.security.authorize.ServiceAuthorizationManager;
 import org.apache.hadoop.security.token.TokenIdentifier;
@@ -820,6 +821,17 @@ private boolean timedOut(long currentTime) {
       return false;
     }
     
+    private UserGroupInformation getAuthorizedUgi(String authorizedId)
+        throws IOException {
+      if (authMethod == SaslRpcServer.AuthMethod.DIGEST) {
+        TokenIdentifier tokenId = SaslRpcServer.getIdentifier(authorizedId,
+            secretManager);
+        return tokenId.getUser();
+      } else {
+        return UserGroupInformation.createRemoteUser(authorizedId);
+      }
+    }
+
     private void saslReadAndProcess(byte[] saslToken) throws IOException,
         InterruptedException {
       if (!saslContextEstablished) {
@@ -882,8 +894,7 @@ public Object run() throws IOException {
             LOG.debug("SASL server context established. Negotiated QoP is "
                 + saslServer.getNegotiatedProperty(Sasl.QOP));
           }
-          user = UserGroupInformation.createRemoteUser(saslServer
-              .getAuthorizationID());
+          user = getAuthorizedUgi(saslServer.getAuthorizationID());
           LOG.info("SASL server successfully authenticated client: " + user);
           saslContextEstablished = true;
         }
@@ -1009,10 +1020,19 @@ private void processHeader(byte[] buf) throws IOException {
       UserGroupInformation protocolUser = header.getUgi();
       if (!useSasl) {
         user = protocolUser;
-      } else if (protocolUser != null && !protocolUser.equals(user)) {
+      } else if ((protocolUser != null)
+          && (!protocolUser.getUserName().equals(user.getUserName()))) {
+        if (authMethod == AuthMethod.DIGEST) {
+          // Not allowed to doAs if token authentication is used
           throw new AccessControlException("Authenticated user (" + user
               + ") doesn't match what the client claims to be (" + protocolUser
               + ")");
+        } else {
+          //Effective user can be different from authenticated user
+          //for simple auth or kerberos auth
+          user = UserGroupInformation.createProxyUser(protocolUser
+              .getUserName(), user);
+        }
       }
     }
     
@@ -1088,6 +1108,14 @@ private void processData(byte[] buf) throws  IOException, InterruptedException {
 
     private boolean authorizeConnection() throws IOException {
       try {
+        // If auth method is DIGEST, the token was obtained by the
+        // real user for the effective user, therefore not required to
+        // authorize real user. doAs is allowed only for simple or kerberos
+        // authentication
+        if (user != null && user.getRealUser() != null
+            && (authMethod != AuthMethod.DIGEST)) {
+          ProxyUsers.authorize(user, this.getHostAddress(), conf);
+        }
         authorize(user, header);
         if (LOG.isDebugEnabled()) {
           LOG.debug("Successfully authorized " + header);

src/java/org/apache/hadoop/security/SaslRpcServer.java

@@ -64,6 +64,15 @@ static byte[] decodeIdentifier(String identifier) {
     return Base64.decodeBase64(identifier.getBytes());
   }
 
+  public static TokenIdentifier getIdentifier(String id,
+      SecretManager<TokenIdentifier> secretManager) throws IOException {
+    byte[] tokenId = decodeIdentifier(id);
+    TokenIdentifier tokenIdentifier = secretManager.createIdentifier();
+    tokenIdentifier.readFields(new DataInputStream(new ByteArrayInputStream(
+        tokenId)));
+    return tokenIdentifier;
+  }
+
   static char[] encodePassword(byte[] password) {
     return new String(Base64.encodeBase64(password)).toCharArray();
   }
@@ -121,14 +130,6 @@ public SaslDigestCallbackHandler(
       this.secretManager = secretManager;
     }
 
-    private TokenIdentifier getIdentifier(String id) throws IOException {
-      byte[] tokenId = decodeIdentifier(id);
-      TokenIdentifier tokenIdentifier = secretManager.createIdentifier();
-      tokenIdentifier.readFields(new DataInputStream(new ByteArrayInputStream(
-          tokenId)));
-      return tokenIdentifier;
-    }
-
     private char[] getPassword(TokenIdentifier tokenid) throws IOException {
       return encodePassword(secretManager.retrievePassword(tokenid));
     }
@@ -155,11 +156,11 @@ public void handle(Callback[] callbacks) throws IOException,
         }
       }
       if (pc != null) {
-        TokenIdentifier tokenIdentifier = getIdentifier(nc.getDefaultName());
+        TokenIdentifier tokenIdentifier = getIdentifier(nc.getDefaultName(), secretManager);
         char[] password = getPassword(tokenIdentifier);
         if (LOG.isDebugEnabled()) {
           LOG.debug("SASL server DIGEST-MD5 callback: setting password "
-              + "for client: " + tokenIdentifier.getUsername());
+              + "for client: " + tokenIdentifier.getUser());
         }
         pc.setPassword(password);
       }
@@ -172,11 +173,12 @@ public void handle(Callback[] callbacks) throws IOException,
           ac.setAuthorized(false);
         }
         if (ac.isAuthorized()) {
-          String username = getIdentifier(authzid).getUsername().toString();
+          String username = getIdentifier(authzid, secretManager).getUser()
+              .getUserName().toString();
           if (LOG.isDebugEnabled())
             LOG.debug("SASL server DIGEST-MD5 callback: setting "
                 + "canonicalized client ID: " + username);
-          ac.setAuthorizedID(username);
+          ac.setAuthorizedID(authzid);
         }
       }
     }

src/java/org/apache/hadoop/security/UserGroupInformation.java

@@ -216,6 +216,43 @@ public static boolean isSecurityEnabled() {
     }
   }
   
+  private static class RealUser implements Principal {
+    private final UserGroupInformation realUser;
+    
+    RealUser(UserGroupInformation realUser) {
+      this.realUser = realUser;
+    }
+    
+    public String getName() {
+      return realUser.getUserName();
+    }
+    
+    public UserGroupInformation getRealUser() {
+      return realUser;
+    }
+    
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) {
+        return true;
+      } else if (o == null || getClass() != o.getClass()) {
+        return false;
+      } else {
+        return realUser.equals(((RealUser) o).realUser);
+      }
+    }
+    
+    @Override
+    public int hashCode() {
+      return realUser.hashCode();
+    }
+    
+    @Override
+    public String toString() {
+      return realUser.toString();
+    }
+  }
+  
   /**
    * A JAAS configuration that defines the login modules that we want
    * to use for login.
@@ -377,6 +414,36 @@ public static UserGroupInformation createRemoteUser(String user) {
     return new UserGroupInformation(subject);
   }
 
+  /* Create a proxy user using username of the effective user and the ugi of the
+   * real user.
+   *
+   * @param effective
+   *          user, UGI for real user.
+   * @return
+   */
+  public static UserGroupInformation createProxyUser(String user,
+      UserGroupInformation realUser) {
+    if (user == null || "".equals(user)) {
+      throw new IllegalArgumentException("Null user");
+    }
+    if (realUser == null) {
+      throw new IllegalArgumentException("Null real user");
+    }
+    Subject subject = new Subject();
+    subject.getPrincipals().add(new User(user));
+    subject.getPrincipals().add(new RealUser(realUser));
+    return new UserGroupInformation(subject);
+  }
+
+  public UserGroupInformation getRealUser() {
+    for (RealUser p: subject.getPrincipals(RealUser.class)) {
+      return p.getRealUser();
+    }
+    return null;
+  }
+
+
+  
   /**
    * This class is used for storing the groups for testing. It stores a local
    * map that has the translation of usernames to groups.
@@ -423,6 +490,32 @@ public static UserGroupInformation createUserForTesting(String user,
     return ugi;
   }
 
+
+  /**
+   * Create a proxy user UGI for testing HDFS and MapReduce
+   * 
+   * @param user
+   *          the full user principal name for effective user
+   * @param realUser
+   *          UGI of the real user
+   * @param userGroups
+   *          the names of the groups that the user belongs to
+   * @return a fake user for running unit tests
+   */
+  @InterfaceAudience.LimitedPrivate( { HDFS, MAPREDUCE })
+  public static UserGroupInformation createProxyUserForTesting(String user,
+      UserGroupInformation realUser, String[] userGroups) {
+    ensureInitialized();
+    UserGroupInformation ugi = createProxyUser(user, realUser);
+    // make sure that the testing object is setup
+    if (!(groups instanceof TestingGroups)) {
+      groups = new TestingGroups();
+    }
+    // add the user groups
+    ((TestingGroups) groups).setUserGroups(ugi.getShortUserName(), userGroups);
+    return ugi;
+  }
+  
   /**
    * Get the user's login name.
    * @return the user's name up to the first '/' or '@'.
@@ -493,8 +586,12 @@ public synchronized String[] getGroupNames() {
    */
   @Override
   public String toString() {
+    if (getRealUser() != null) {
+      return getUserName() + " via " +  getRealUser().toString();
+    } else {
       return getUserName();
     }
+  }
 
   /**
    * Compare the subjects to see if they are equal to each other.
@@ -599,4 +696,5 @@ public static void main(String [] args) throws Exception {
       System.out.println("Keytab: " + ugi);
     }
   }
+
 }

src/java/org/apache/hadoop/security/authorize/ProxyUsers.java

@@ -0,0 +1,106 @@
+/**
+ * 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.security.authorize;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.Collection;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.classification.InterfaceAudience;
+
+@InterfaceAudience.Private
+public class ProxyUsers {
+
+  /*
+   * Returns configuration key for effective user groups allowed for a superuser
+   * 
+   * @param userName name of the superuser
+   * @return configuration key for superuser groups
+   */
+  public static String getProxySuperuserGroupConfKey(String userName) {
+    return "hadoop.proxyuser."+userName+".users";
+  }
+  
+  /*
+   * Return configuration key for superuser ip addresses
+   * 
+   * @param userName name of the superuser
+   * @return configuration key for superuser ip-addresses
+   */
+  public static String getProxySuperuserIpConfKey(String userName) {
+    return "hadoop.proxyuser."+userName+".ip-addresses";
+  }
+  
+  /*
+   * Authorize the superuser which is doing doAs
+   * 
+   * @param user ugi of the effective or proxy user which contains a real user
+   * @param remoteAddress the ip address of client
+   * @param conf configuration
+   * @throws AuthorizationException
+   */
+  public static void authorize(UserGroupInformation user, String remoteAddress,
+      Configuration conf) throws AuthorizationException {
+
+    if (user.getRealUser() == null) {
+      return;
+    }
+    boolean groupAuthorized = false;
+    UserGroupInformation superUser = user.getRealUser();
+
+    Collection<String> allowedUserGroups = conf
+        .getStringCollection(getProxySuperuserGroupConfKey(superUser
+            .getShortUserName()));
+    if (!allowedUserGroups.isEmpty()) {
+      for (String group : user.getGroupNames()) {
+        if (allowedUserGroups.contains(group)) {
+          groupAuthorized = true;
+          break;
+        }
+      }
+    }
+
+    if (!groupAuthorized) {
+      throw new AuthorizationException("User: " + superUser.getUserName()
+          + " is not allowed to impersonate " + user.getUserName());
+    }
+    
+    Collection<String> ipList = conf
+        .getStringCollection(getProxySuperuserIpConfKey(superUser
+            .getShortUserName()));
+    if (!ipList.isEmpty()) {
+      for (String allowedHost : ipList) {
+        InetAddress hostAddr;
+        try {
+          hostAddr = InetAddress.getByName(allowedHost);
+        } catch (UnknownHostException e) {
+          continue;
+        }
+        if (hostAddr.getHostAddress().equals(remoteAddress)) {
+          // Authorization is successful
+          return;
+        }
+      }
+    }
+    throw new AuthorizationException("Unauthorized connection for super-user: "
+        + superUser.getUserName() + " from IP " + remoteAddress);
+  }
+}

src/java/org/apache/hadoop/security/token/TokenIdentifier.java

@@ -24,6 +24,7 @@
 import org.apache.hadoop.io.DataOutputBuffer;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.security.UserGroupInformation;
 
 /**
  * An identifier that identifies a token, may contain public information 
@@ -37,10 +38,12 @@ public abstract class TokenIdentifier implements Writable {
   public abstract Text getKind();
 
   /**
-   * Get the username encoded in the token identifier
-   * @return the username
+   * Get the Ugi with the username encoded in the token identifier
+   * 
+   * @return the username. null is returned if username in the identifier is
+   *         empty or null.
    */
-  public abstract Text getUsername();
+  public abstract UserGroupInformation getUser();
 
   /**
    * Get the bytes for the token identifier

src/test/core/org/apache/hadoop/ipc/TestSaslRPC.java

@@ -71,29 +71,52 @@ public class TestSaslRPC {
 
   public static class TestTokenIdentifier extends TokenIdentifier {
     private Text tokenid;
+    private Text realUser;
     final static Text KIND_NAME = new Text("test.token");
     
     public TestTokenIdentifier() {
       this.tokenid = new Text();
+      this.realUser = new Text();
     }
     public TestTokenIdentifier(Text tokenid) {
       this.tokenid = tokenid;
+      this.realUser = new Text();
+    }
+    public TestTokenIdentifier(Text tokenid, Text realUser) {
+      this.tokenid = tokenid;
+      if (realUser == null) {
+        this.realUser = new Text();
+      } else {
+        this.realUser = realUser;
+      }
     }
     @Override
     public Text getKind() {
       return KIND_NAME;
     }
     @Override
-    public Text getUsername() {
-      return tokenid;
+    public UserGroupInformation getUser() {
+      if ((realUser == null) || ("".equals(realUser.toString()))) {
+        return UserGroupInformation.createRemoteUser(tokenid.toString());
+      } else {
+        UserGroupInformation realUgi = UserGroupInformation
+            .createRemoteUser(realUser.toString());
+        return UserGroupInformation
+            .createProxyUser(tokenid.toString(), realUgi);
       }
+    }
+
     @Override
     public void readFields(DataInput in) throws IOException {
       tokenid.readFields(in);
+      realUser.readFields(in);
     }
     @Override
     public void write(DataOutput out) throws IOException {
       tokenid.write(out);
+      if (realUser != null) {
+        realUser.write(out);
+      }
     }
   }
   

src/test/core/org/apache/hadoop/security/TestDoAsEffectiveUser.java

@@ -0,0 +1,445 @@
+/**
+ * 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.security;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.security.PrivilegedExceptionAction;
+
+import junit.framework.Assert;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.CommonConfigurationKeys;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.ipc.RPC;
+import org.apache.hadoop.ipc.Server;
+import org.apache.hadoop.ipc.VersionedProtocol;
+import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.security.authorize.ProxyUsers;
+import org.apache.hadoop.security.token.Token;
+import org.apache.hadoop.security.token.TokenInfo;
+import org.junit.Test;
+import org.apache.hadoop.ipc.TestSaslRPC;
+import org.apache.hadoop.ipc.TestSaslRPC.TestTokenSecretManager;
+import org.apache.hadoop.ipc.TestSaslRPC.TestTokenIdentifier;
+import org.apache.hadoop.ipc.TestSaslRPC.TestTokenSelector;
+
+/**
+ *
+ */
+public class TestDoAsEffectiveUser {
+  final private static String REAL_USER_NAME = "realUser1@HADOOP.APACHE.ORG";
+  final private static String REAL_USER_SHORT_NAME = "realUser1";
+  final private static String PROXY_USER_NAME = "proxyUser";
+  final private static String GROUP1_NAME = "group1";
+  final private static String GROUP2_NAME = "group2";
+  final private static String[] GROUP_NAMES = new String[] { GROUP1_NAME,
+      GROUP2_NAME };
+  private static final String ADDRESS = "0.0.0.0";
+  private TestProtocol proxy;
+
+  /**
+   * Test method for
+   * {@link org.apache.hadoop.security.UserGroupInformation#createProxyUser(java.lang.String, org.apache.hadoop.security.UserGroupInformation)}
+   * .
+   */
+  @Test
+  public void testCreateProxyUser() throws Exception {
+    // ensure that doAs works correctly
+    UserGroupInformation realUserUgi = UserGroupInformation
+        .createRemoteUser(REAL_USER_NAME);
+    UserGroupInformation proxyUserUgi = UserGroupInformation.createProxyUser(
+        PROXY_USER_NAME, realUserUgi);
+    UserGroupInformation curUGI = proxyUserUgi
+        .doAs(new PrivilegedExceptionAction<UserGroupInformation>() {
+          public UserGroupInformation run() throws IOException {
+            return UserGroupInformation.getCurrentUser();
+          }
+        });
+    Assert.assertTrue(curUGI.toString().equals(
+        PROXY_USER_NAME + " via " + REAL_USER_NAME));
+  }
+
+  @TokenInfo(TestTokenSelector.class)
+  public interface TestProtocol extends VersionedProtocol {
+    public static final long versionID = 1L;
+
+    String aMethod() throws IOException;
+  }
+
+  public class TestImpl implements TestProtocol {
+
+    public String aMethod() throws IOException {
+      return UserGroupInformation.getCurrentUser().toString();
+    }
+
+    public long getProtocolVersion(String protocol, long clientVersion)
+        throws IOException {
+      // TODO Auto-generated method stub
+      return TestProtocol.versionID;
+    }
+  }
+
+  @Test
+  public void testRealUserSetup() throws IOException {
+    final Configuration conf = new Configuration();
+    conf.setStrings(ProxyUsers
+        .getProxySuperuserGroupConfKey(REAL_USER_SHORT_NAME), "group1");
+    conf.setStrings(ProxyUsers.getProxySuperuserIpConfKey(REAL_USER_SHORT_NAME),
+        "127.0.0.1","127.0.1.1", "localhost");
+    Server server = RPC.getServer(TestProtocol.class, new TestImpl(), ADDRESS,
+        0, 5, true, conf, null);
+
+    try {
+      server.start();
+
+      final InetSocketAddress addr = NetUtils.getConnectAddress(server);
+
+      UserGroupInformation realUserUgi = UserGroupInformation
+          .createRemoteUser(REAL_USER_NAME);
+      UserGroupInformation proxyUserUgi = UserGroupInformation.createProxyUserForTesting(
+          PROXY_USER_NAME, realUserUgi, GROUP_NAMES);
+      String retVal = proxyUserUgi
+          .doAs(new PrivilegedExceptionAction<String>() {
+            public String run() throws IOException {
+              proxy = (TestProtocol) RPC.getProxy(TestProtocol.class,
+                  TestProtocol.versionID, addr, conf);
+              String ret = proxy.aMethod();
+              return ret;
+            }
+          });
+
+      Assert.assertEquals(PROXY_USER_NAME + " via " + REAL_USER_NAME, retVal);
+    } catch (Exception e) {
+      e.printStackTrace();
+      Assert.fail();
+    } finally {
+      server.stop();
+      if (proxy != null) {
+        RPC.stopProxy(proxy);
+      }
+    }
+  }
+
+  @Test
+  public void testRealUserAuthorizationSuccess() throws IOException {
+    final Configuration conf = new Configuration();
+    conf.setStrings(ProxyUsers.getProxySuperuserIpConfKey(REAL_USER_SHORT_NAME),
+        "127.0.0.1","127.0.1.1", "localhost");
+    conf.setStrings(ProxyUsers.getProxySuperuserGroupConfKey(REAL_USER_SHORT_NAME),
+        "group1");
+    Server server = RPC.getServer(TestProtocol.class, new TestImpl(), ADDRESS,
+        0, 2, false, conf, null);
+
+    try {
+      server.start();
+
+      final InetSocketAddress addr = NetUtils.getConnectAddress(server);
+
+      UserGroupInformation realUserUgi = UserGroupInformation
+          .createRemoteUser(REAL_USER_NAME);
+
+      UserGroupInformation proxyUserUgi = UserGroupInformation
+          .createProxyUserForTesting(PROXY_USER_NAME, realUserUgi, GROUP_NAMES);
+      String retVal = proxyUserUgi
+          .doAs(new PrivilegedExceptionAction<String>() {
+            public String run() throws IOException {
+              proxy = (TestProtocol) RPC.getProxy(TestProtocol.class,
+                  TestProtocol.versionID, addr, conf);
+              String ret = proxy.aMethod();
+              return ret;
+            }
+          });
+
+      Assert.assertEquals(PROXY_USER_NAME + " via " + REAL_USER_NAME, retVal);
+    } catch (Exception e) {
+      e.printStackTrace();
+      Assert.fail();
+    } finally {
+      server.stop();
+      if (proxy != null) {
+        RPC.stopProxy(proxy);
+      }
+    }
+  }
+
+  /*
+   * Tests authorization of superuser's ip.
+   */
+  @Test
+  public void testRealUserIPAuthorizationFailure() throws IOException {
+    final Configuration conf = new Configuration();
+    conf.setStrings(ProxyUsers.getProxySuperuserIpConfKey(REAL_USER_SHORT_NAME),
+        "20.20.20.20"); //Authorized IP address
+    conf.setStrings(ProxyUsers.getProxySuperuserGroupConfKey(REAL_USER_SHORT_NAME),
+        "group1");
+    Server server = RPC.getServer(TestProtocol.class, new TestImpl(), ADDRESS,
+        0, 2, false, conf, null);
+
+    try {
+      server.start();
+
+      final InetSocketAddress addr = NetUtils.getConnectAddress(server);
+
+      UserGroupInformation realUserUgi = UserGroupInformation
+          .createRemoteUser(REAL_USER_NAME);
+
+      UserGroupInformation proxyUserUgi = UserGroupInformation
+          .createProxyUserForTesting(PROXY_USER_NAME, realUserUgi, GROUP_NAMES);
+      String retVal = proxyUserUgi
+          .doAs(new PrivilegedExceptionAction<String>() {
+            public String run() throws IOException {
+              proxy = (TestProtocol) RPC.getProxy(TestProtocol.class,
+                  TestProtocol.versionID, addr, conf);
+              String ret = proxy.aMethod();
+              return ret;
+            }
+          });
+
+      Assert.fail("The RPC must have failed " + retVal);
+    } catch (Exception e) {
+      e.printStackTrace();
+    } finally {
+      server.stop();
+      if (proxy != null) {
+        RPC.stopProxy(proxy);
+      }
+    }
+  }
+  
+  @Test
+  public void testRealUserIPNotSpecified() throws IOException {
+    final Configuration conf = new Configuration();
+    conf.setStrings(ProxyUsers
+        .getProxySuperuserGroupConfKey(REAL_USER_SHORT_NAME), "group1");
+    Server server = RPC.getServer(TestProtocol.class, new TestImpl(), ADDRESS,
+        0, 2, false, conf, null);
+
+    try {
+      server.start();
+
+      final InetSocketAddress addr = NetUtils.getConnectAddress(server);
+
+      UserGroupInformation realUserUgi = UserGroupInformation
+          .createRemoteUser(REAL_USER_NAME);
+
+      UserGroupInformation proxyUserUgi = UserGroupInformation
+          .createProxyUserForTesting(PROXY_USER_NAME, realUserUgi, GROUP_NAMES);
+      String retVal = proxyUserUgi
+          .doAs(new PrivilegedExceptionAction<String>() {
+            public String run() throws IOException {
+              proxy = (TestProtocol) RPC.getProxy(TestProtocol.class,
+                  TestProtocol.versionID, addr, conf);
+              String ret = proxy.aMethod();
+              return ret;
+            }
+          });
+
+      Assert.fail("The RPC must have failed " + retVal);
+    } catch (Exception e) {
+      e.printStackTrace();
+    } finally {
+      server.stop();
+      if (proxy != null) {
+        RPC.stopProxy(proxy);
+      }
+    }
+  }
+
+  @Test
+  public void testRealUserGroupNotSpecified() throws IOException {
+    final Configuration conf = new Configuration();
+    conf.setStrings(ProxyUsers.getProxySuperuserIpConfKey(REAL_USER_SHORT_NAME),
+        "127.0.0.1","127.0.1.1","localhost");
+    Server server = RPC.getServer(TestProtocol.class, new TestImpl(), ADDRESS,
+        0, 2, false, conf, null);
+
+    try {
+      server.start();
+
+      final InetSocketAddress addr = NetUtils.getConnectAddress(server);
+
+      UserGroupInformation realUserUgi = UserGroupInformation
+          .createRemoteUser(REAL_USER_NAME);
+
+      UserGroupInformation proxyUserUgi = UserGroupInformation
+          .createProxyUserForTesting(PROXY_USER_NAME, realUserUgi, GROUP_NAMES);
+      String retVal = proxyUserUgi
+          .doAs(new PrivilegedExceptionAction<String>() {
+            public String run() throws IOException {
+              proxy = (TestProtocol) RPC.getProxy(TestProtocol.class,
+                  TestProtocol.versionID, addr, conf);
+              String ret = proxy.aMethod();
+              return ret;
+            }
+          });
+
+      Assert.fail("The RPC must have failed " + retVal);
+    } catch (Exception e) {
+      e.printStackTrace();
+    } finally {
+      server.stop();
+      if (proxy != null) {
+        RPC.stopProxy(proxy);
+      }
+    }
+  }
+  
+  @Test
+  public void testRealUserGroupAuthorizationFailure() throws IOException {
+    final Configuration conf = new Configuration();
+    conf.setStrings(ProxyUsers.getProxySuperuserIpConfKey(REAL_USER_SHORT_NAME),
+        "127.0.0.1","127.0.1.1","localhost");
+    conf.setStrings(ProxyUsers.getProxySuperuserGroupConfKey(REAL_USER_SHORT_NAME),
+        "group3");
+    Server server = RPC.getServer(TestProtocol.class, new TestImpl(), ADDRESS,
+        0, 2, false, conf, null);
+
+    try {
+      server.start();
+
+      final InetSocketAddress addr = NetUtils.getConnectAddress(server);
+
+      UserGroupInformation realUserUgi = UserGroupInformation
+          .createRemoteUser(REAL_USER_NAME);
+
+      UserGroupInformation proxyUserUgi = UserGroupInformation
+          .createProxyUserForTesting(PROXY_USER_NAME, realUserUgi, GROUP_NAMES);
+      String retVal = proxyUserUgi
+          .doAs(new PrivilegedExceptionAction<String>() {
+            public String run() throws IOException {
+              proxy = (TestProtocol) RPC.getProxy(TestProtocol.class,
+                  TestProtocol.versionID, addr, conf);
+              String ret = proxy.aMethod();
+              return ret;
+            }
+          });
+
+      Assert.fail("The RPC must have failed " + retVal);
+    } catch (Exception e) {
+      e.printStackTrace();
+    } finally {
+      server.stop();
+      if (proxy != null) {
+        RPC.stopProxy(proxy);
+      }
+    }
+  }
+
+  /*
+   *  Tests the scenario when token authorization is used.
+   *  The server sees only the the owner of the token as the
+   *  user.
+   */
+  @Test
+  public void testProxyWithToken() throws Exception {
+    final Configuration conf = new Configuration();
+    TestTokenSecretManager sm = new TestTokenSecretManager();
+    conf
+        .set(CommonConfigurationKeys.HADOOP_SECURITY_AUTHENTICATION, "kerberos");
+    UserGroupInformation.setConfiguration(conf);
+    final Server server = RPC.getServer(TestProtocol.class, new TestImpl(),
+        ADDRESS, 0, 5, true, conf, sm);
+
+    server.start();
+
+    final UserGroupInformation current = UserGroupInformation
+        .createRemoteUser(REAL_USER_NAME);
+    final InetSocketAddress addr = NetUtils.getConnectAddress(server);
+    TestTokenIdentifier tokenId = new TestTokenIdentifier(new Text(current
+        .getUserName()), new Text("SomeSuperUser"));
+    Token<TestTokenIdentifier> token = new Token<TestTokenIdentifier>(tokenId,
+        sm);
+    Text host = new Text(addr.getAddress().getHostAddress() + ":"
+        + addr.getPort());
+    token.setService(host);
+    UserGroupInformation proxyUserUgi = UserGroupInformation
+        .createProxyUserForTesting(PROXY_USER_NAME, current, GROUP_NAMES);
+    proxyUserUgi.addToken(token);
+    String retVal = proxyUserUgi.doAs(new PrivilegedExceptionAction<String>() {
+      @Override
+      public String run() throws Exception {
+        try {
+          proxy = (TestProtocol) RPC.getProxy(TestProtocol.class,
+              TestProtocol.versionID, addr, conf);
+          String ret = proxy.aMethod();
+          return ret;
+        } catch (Exception e) {
+          e.printStackTrace();
+          throw e;
+        } finally {
+          server.stop();
+          if (proxy != null) {
+            RPC.stopProxy(proxy);
+          }
+        }
+      }
+    });
+    //The user returned by server must be the one in the token.
+    Assert.assertEquals(REAL_USER_NAME + " via SomeSuperUser", retVal);
+  }
+
+  /*
+   * The user gets the token via a superuser. Server should authenticate
+   * this user. 
+   */
+  @Test
+  public void testTokenBySuperUser() throws Exception {
+    TestTokenSecretManager sm = new TestTokenSecretManager();
+    final Configuration newConf = new Configuration();
+    newConf.set(CommonConfigurationKeys.HADOOP_SECURITY_AUTHENTICATION,
+        "kerberos");
+    UserGroupInformation.setConfiguration(newConf);
+    final Server server = RPC.getServer(TestProtocol.class, new TestImpl(),
+        ADDRESS, 0, 5, true, newConf, sm);
+
+    server.start();
+
+    final UserGroupInformation current = UserGroupInformation
+        .createUserForTesting(REAL_USER_NAME, GROUP_NAMES);
+    final InetSocketAddress addr = NetUtils.getConnectAddress(server);
+    TestTokenIdentifier tokenId = new TestTokenIdentifier(new Text(current
+        .getUserName()), new Text("SomeSuperUser"));
+    Token<TestTokenIdentifier> token = new Token<TestTokenIdentifier>(tokenId,
+        sm);
+    Text host = new Text(addr.getAddress().getHostAddress() + ":"
+        + addr.getPort());
+    token.setService(host);
+    current.addToken(token);
+    String retVal = current.doAs(new PrivilegedExceptionAction<String>() {
+      @Override
+      public String run() throws Exception {
+        try {
+          proxy = (TestProtocol) RPC.getProxy(TestProtocol.class,
+              TestProtocol.versionID, addr, newConf);
+          String ret = proxy.aMethod();
+          return ret;
+        } catch (Exception e) {
+          e.printStackTrace();
+          throw e;
+        } finally {
+          server.stop();
+          if (proxy != null) {
+            RPC.stopProxy(proxy);
+          }
+        }
+      }
+    });
+    Assert.assertEquals(REAL_USER_NAME + " via SomeSuperUser", retVal);
+  }
+}

src/test/core/org/apache/hadoop/security/TestUserGroupInformation.java

@@ -154,6 +154,21 @@ public void testEquals() throws Exception {
     assertEquals(uugi, new UserGroupInformation(uugi.getSubject()));
   }
   
+  @Test
+  public void testEqualsWithRealUser() throws Exception {
+    UserGroupInformation realUgi1 = UserGroupInformation.createUserForTesting(
+        "RealUser", GROUP_NAMES);
+    UserGroupInformation realUgi2 = UserGroupInformation.createUserForTesting(
+        "RealUser", GROUP_NAMES);
+    UserGroupInformation proxyUgi1 = UserGroupInformation.createProxyUser(
+        USER_NAME, realUgi1);
+    UserGroupInformation proxyUgi2 = UserGroupInformation.createProxyUser(
+        USER_NAME, realUgi2);
+    UserGroupInformation remoteUgi = UserGroupInformation.createRemoteUser(USER_NAME);
+    assertEquals(proxyUgi1, proxyUgi2);
+    assertFalse(remoteUgi.equals(proxyUgi1));
+  }
+  
   @Test
   public void testGettingGroups() throws Exception {
     UserGroupInformation uugi =