HADOOP-6600. mechanism for authorization check for inter-server protocols

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

CHANGES.txt

@@ -4,6 +4,9 @@ Trunk (unreleased changes)
 
   IMPROVEMENTS
 
+    HADOOP-6600. mechanism for authorization check for inter-server 
+    protocols. (boryas)
+
     HADOOP-6623. Add StringUtils.split for non-escaped single-character
     separator. (Todd Lipcon via tomwhite)
 

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

@@ -253,7 +253,7 @@ public Connection(ConnectionId remoteId) throws IOException {
         }
         KerberosInfo krbInfo = protocol.getAnnotation(KerberosInfo.class);
         if (krbInfo != null) {
-          String serverKey = krbInfo.value();
+          String serverKey = krbInfo.serverPrincipal();
           if (serverKey != null) {
             serverPrincipal = conf.get(serverKey);
           }

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

@@ -1615,7 +1615,7 @@ public void authorize(UserGroupInformation user,
         throw new AuthorizationException("Unknown protocol: " + 
                                          connection.getProtocol());
       }
-      ServiceAuthorizationManager.authorize(user, protocol);
+      ServiceAuthorizationManager.authorize(user, protocol, getConf());
     }
   }
   

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

@@ -27,5 +27,6 @@
 @Target(ElementType.TYPE)
 public @interface KerberosInfo {
   /** Key for getting server's Kerberos principal name from Configuration */
-  String value();
+  String serverPrincipal();
+  String clientPrincipal() default "";
 }

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

@@ -24,6 +24,7 @@
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.CommonConfigurationKeys;
+import org.apache.hadoop.security.KerberosInfo;
 import org.apache.hadoop.security.UserGroupInformation;
 
 /**
@@ -62,18 +63,30 @@ public class ServiceAuthorizationManager {
    * @throws AuthorizationException on authorization failure
    */
   public static void authorize(UserGroupInformation user, 
-                               Class<?> protocol
+                               Class<?> protocol,
+                               Configuration conf
                                ) throws AuthorizationException {
     AccessControlList acl = protocolToAcl.get(protocol);
     if (acl == null) {
       throw new AuthorizationException("Protocol " + protocol + 
                                        " is not known.");
     }
-    if (!acl.isUserAllowed(user)) {
+    
+    // get client principal key to verify (if available)
+    KerberosInfo krbInfo = protocol.getAnnotation(KerberosInfo.class);
+    String clientPrincipal = null; 
+    if (krbInfo != null) {
+      String clientKey = krbInfo.clientPrincipal();
+      if (clientKey != null && !clientKey.equals("")) {
+        clientPrincipal = conf.get(clientKey);
+      }
+    }
+    if((clientPrincipal != null && !clientPrincipal.equals(user.getUserName())) || 
+        !acl.isUserAllowed(user)) {
       auditLOG.warn(AUTHZ_FAILED_FOR + user + " for protocol="+protocol);
       throw new AuthorizationException("User " + user + 
-                                       " is not authorized for protocol " + 
+          " is not authorized for protocol " + 
-                                       protocol);
+          protocol);
     }
     auditLOG.info(AUTHZ_SUCCESSFULL_FOR + user + " for protocol="+protocol);
   }

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

@@ -101,7 +101,8 @@ protected TestDelegationTokenSelector() {
     }    
   }
   
-  @KerberosInfo(USER_NAME_KEY)
+  @KerberosInfo(
+      serverPrincipal=USER_NAME_KEY)
   @TokenInfo(TestDelegationTokenSelector.class)
   public static interface MiniProtocol extends VersionedProtocol {
     public static final long versionID = 1L;

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

@@ -162,7 +162,8 @@ public Token<TestTokenIdentifier> selectToken(Text service,
     }
   }
   
-  @KerberosInfo(SERVER_PRINCIPAL_KEY)
+  @KerberosInfo(
+      serverPrincipal = SERVER_PRINCIPAL_KEY)
   @TokenInfo(TestTokenSelector.class)
   public interface TestSaslProtocol extends TestRPC.TestProtocol {
     public AuthenticationMethod getAuthMethod() throws IOException;