HADOOP-6632. Adds support for using different keytabs for different servers in a Hadoop cluster. In the earier implementation, all servers of a certain type \(like TaskTracker\), would have the same keytab and the same principal. Now the principal name is a pattern that has _HOST in it. Contributed by Kan Zhang & Jitendra Pandey.

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

CHANGES.txt

@@ -74,6 +74,12 @@ Trunk (unreleased changes)
     HADOOP-6905. add buildDTServiceName method to SecurityUtil 
     (as part of MAPREDUCE-1718)  (boryas)
 
+    HADOOP-6632. Adds support for using different keytabs for different
+    servers in a Hadoop cluster. In the earier implementation, all servers 
+    of a certain type (like TaskTracker), would have the same keytab and the
+    same principal. Now the principal name is a pattern that has _HOST in it.
+    (Kan Zhang & Jitendra Pandey via ddas)
+
   OPTIMIZATIONS
 
   BUG FIXES

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

@@ -54,6 +54,7 @@ import org.apache.hadoop.net.NetUtils;
 import org.apache.hadoop.security.KerberosInfo;
 import org.apache.hadoop.security.SaslRpcClient;
 import org.apache.hadoop.security.SaslRpcServer.AuthMethod;
+import org.apache.hadoop.security.SecurityUtil;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.security.token.Token;
 import org.apache.hadoop.security.token.TokenIdentifier;
@@ -254,13 +255,15 @@ public class Client {
         KerberosInfo krbInfo = protocol.getAnnotation(KerberosInfo.class);
         if (krbInfo != null) {
           String serverKey = krbInfo.serverPrincipal();
-          if (serverKey != null) {
-            if(LOG.isDebugEnabled()) {
-              LOG.info("server principal key for protocol="
-                  + protocol.getCanonicalName() + " is " + serverKey + 
-                  " and val =" + conf.get(serverKey));
-            }
-            serverPrincipal = conf.get(serverKey);
+          if (serverKey == null) {
+            throw new IOException(
+                "Can't obtain server Kerberos config key from KerberosInfo");
+          }
+          serverPrincipal = SecurityUtil.getServerPrincipal(
+              conf.get(serverKey), server.getAddress().getCanonicalHostName());
+          if (LOG.isDebugEnabled()) {
+            LOG.debug("RPC Server Kerberos principal name for protocol="
+                + protocol.getCanonicalName() + " is " + serverPrincipal);
           }
         }
       }

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

@@ -827,6 +827,7 @@ public abstract class Server {
     // Cache the remote host & port info so that even if the socket is 
     // disconnected, we can say where it used to connect to.
     private String hostAddress;
+    private String hostName;
     private int remotePort;
     
     ConnectionHeader header = new ConnectionHeader();
@@ -869,6 +870,7 @@ public abstract class Server {
         this.hostAddress = "*Unknown*";
       } else {
         this.hostAddress = addr.getHostAddress();
+        this.hostName = addr.getCanonicalHostName();
       }
       this.remotePort = socket.getPort();
       this.responseQueue = new LinkedList<Call>();
@@ -891,6 +893,10 @@ public abstract class Server {
       return hostAddress;
     }
 
+    public String getHostName() {
+      return hostName;
+    }
+    
     public void setLastContact(long lastContact) {
       this.lastContact = lastContact;
     }
@@ -1296,7 +1302,7 @@ public abstract class Server {
             && (authMethod != AuthMethod.DIGEST)) {
           ProxyUsers.authorize(user, this.getHostAddress(), conf);
         }
-        authorize(user, header);
+        authorize(user, header, getHostName());
         if (LOG.isDebugEnabled()) {
           LOG.debug("Successfully authorized " + header);
         }
@@ -1626,10 +1632,12 @@ public abstract class Server {
    * 
    * @param user client user
    * @param connection incoming connection
+   * @param hostname fully-qualified domain name of incoming connection
    * @throws AuthorizationException when the client isn't authorized to talk the protocol
    */
   public void authorize(UserGroupInformation user, 
-                        ConnectionHeader connection
+                        ConnectionHeader connection,
+                        String hostname
                         ) throws AuthorizationException {
     if (authorize) {
       Class<?> protocol = null;
@@ -1639,7 +1647,7 @@ public abstract class Server {
         throw new AuthorizationException("Unknown protocol: " + 
                                          connection.getProtocol());
       }
-      ServiceAuthorizationManager.authorize(user, protocol, getConf());
+      ServiceAuthorizationManager.authorize(user, protocol, getConf(), hostname);
     }
   }
   

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

@@ -17,8 +17,10 @@
 package org.apache.hadoop.security;
 
 import java.io.IOException;
+import java.net.InetAddress;
 import java.net.URI;
 import java.net.URL;
+import java.net.UnknownHostException;
 import java.security.AccessController;
 import java.util.Set;
 
@@ -29,6 +31,8 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.net.NetUtils;
 
 import sun.security.jgss.krb5.Krb5Util;
@@ -38,7 +42,8 @@ import sun.security.krb5.PrincipalName;
 @InterfaceAudience.LimitedPrivate({"HDFS", "MapReduce"})
 @InterfaceStability.Evolving
 public class SecurityUtil {
-  private static final Log LOG = LogFactory.getLog(SecurityUtil.class);
+  public static final Log LOG = LogFactory.getLog(SecurityUtil.class);
+  public static final String HOSTNAME_PATTERN = "_HOST";
 
   /**
    * Find the original TGT within the current subject's credentials. Cross-realm
@@ -49,9 +54,13 @@ public class SecurityUtil {
    *           if TGT can't be found
    */
   private static KerberosTicket getTgtFromSubject() throws IOException {
-    Set<KerberosTicket> tickets = Subject.getSubject(
-        AccessController.getContext()).getPrivateCredentials(
-        KerberosTicket.class);
+    Subject current = Subject.getSubject(AccessController.getContext());
+    if (current == null) {
+      throw new IOException(
+          "Can't get TGT from current Subject, because it is null");
+    }
+    Set<KerberosTicket> tickets = current
+        .getPrivateCredentials(KerberosTicket.class);
     for (KerberosTicket t : tickets) {
       if (isOriginalTGT(t.getServer().getName()))
         return t;
@@ -90,7 +99,8 @@ public class SecurityUtil {
       return;
     
     String serviceName = "host/" + remoteHost.getHost();
-    LOG.debug("Fetching service ticket for host at: " + serviceName);
+    if (LOG.isDebugEnabled())
+      LOG.debug("Fetching service ticket for host at: " + serviceName);
     Credentials serviceCred = null;
     try {
       PrincipalName principal = new PrincipalName(serviceName,
@@ -98,7 +108,7 @@ public class SecurityUtil {
       serviceCred = Credentials.acquireServiceCreds(principal
           .toString(), Krb5Util.ticketToCreds(getTgtFromSubject()));
     } catch (Exception e) {
-      throw new IOException("Invalid service principal name: "
+      throw new IOException("Can't get service ticket for: "
           + serviceName, e);
     }
     if (serviceCred == null) {
@@ -107,6 +117,91 @@ public class SecurityUtil {
     Subject.getSubject(AccessController.getContext()).getPrivateCredentials()
         .add(Krb5Util.credsToTicket(serviceCred));
   }
+  
+  /**
+   * Convert Kerberos principal name conf values to valid Kerberos principal
+   * names. It replaces $host in the conf values with hostname, which should be
+   * fully-qualified domain name. If hostname is null or "0.0.0.0", it uses
+   * dynamically looked-up fqdn of the current host instead.
+   * 
+   * @param principalConfig
+   *          the Kerberos principal name conf value to convert
+   * @param hostname
+   *          the fully-qualified domain name used for substitution
+   * @return converted Kerberos principal name
+   * @throws IOException
+   */
+  public static String getServerPrincipal(String principalConfig,
+      String hostname) throws IOException {
+    if (principalConfig == null)
+      return null;
+    String[] components = principalConfig.split("[/@]");
+    if (components.length != 3) {
+      throw new IOException(
+          "Kerberos service principal name isn't configured properly "
+              + "(should have 3 parts): " + principalConfig);
+    }
+
+    if (components[1].equals(HOSTNAME_PATTERN)) {
+      String fqdn = hostname;
+      if (fqdn == null || fqdn.equals("") || fqdn.equals("0.0.0.0")) {
+        fqdn = getLocalHostName();
+      }
+      return components[0] + "/" + fqdn + "@" + components[2];
+    } else {
+      return principalConfig;
+    }
+  }
+  
+  static String getLocalHostName() throws UnknownHostException {
+    return InetAddress.getLocalHost().getCanonicalHostName();
+  }
+
+  /**
+   * If a keytab has been provided, login as that user. Substitute $host in
+   * user's Kerberos principal name with a dynamically looked-up fully-qualified
+   * domain name of the current host.
+   * 
+   * @param conf
+   *          conf to use
+   * @param keytabFileKey
+   *          the key to look for keytab file in conf
+   * @param userNameKey
+   *          the key to look for user's Kerberos principal name in conf
+   * @throws IOException
+   */
+  public static void login(final Configuration conf,
+      final String keytabFileKey, final String userNameKey) throws IOException {
+    login(conf, keytabFileKey, userNameKey, getLocalHostName());
+  }
+
+  /**
+   * If a keytab has been provided, login as that user. Substitute $host in
+   * user's Kerberos principal name with hostname.
+   * 
+   * @param conf
+   *          conf to use
+   * @param keytabFileKey
+   *          the key to look for keytab file in conf
+   * @param userNameKey
+   *          the key to look for user's Kerberos principal name in conf
+   * @param hostname
+   *          hostname to use for substitution
+   * @throws IOException
+   */
+  public static void login(final Configuration conf,
+      final String keytabFileKey, final String userNameKey, String hostname)
+      throws IOException {
+    String keytabFilename = conf.get(keytabFileKey);
+    if (keytabFilename == null)
+      return;
+
+    String principalConfig = conf.get(userNameKey, System
+        .getProperty("user.name"));
+    String principalName = SecurityUtil.getServerPrincipal(principalConfig,
+        hostname);
+    UserGroupInformation.loginUserFromKeytab(principalName, keytabFilename);
+  }
 
   /**
    * create service name for Delegation token ip:port

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

@@ -437,6 +437,8 @@ public class UserGroupInformation {
       throw new IOException("Login failure for " + user + " from keytab " + 
                             path, le);
     }
+    LOG.info("Login successful for user " + keytabPrincipal
+        + " using keytab file " + keytabFile);
   }
   
   /**

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

@@ -28,6 +28,7 @@ import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.CommonConfigurationKeys;
 import org.apache.hadoop.security.KerberosInfo;
+import org.apache.hadoop.security.SecurityUtil;
 import org.apache.hadoop.security.KerberosName;
 import org.apache.hadoop.security.UserGroupInformation;
 
@@ -68,11 +69,14 @@ public class ServiceAuthorizationManager {
    * 
    * @param user user accessing the service 
    * @param protocol service being accessed
+   * @param conf configuration to use
+   * @param hostname fully qualified domain name of the client
    * @throws AuthorizationException on authorization failure
    */
   public static void authorize(UserGroupInformation user, 
                                Class<?> protocol,
-                               Configuration conf
+                               Configuration conf,
+                               String hostname
                                ) throws AuthorizationException {
     AccessControlList acl = protocolToAcl.get(protocol);
     if (acl == null) {
@@ -86,7 +90,19 @@ public class ServiceAuthorizationManager {
     if (krbInfo != null) {
       String clientKey = krbInfo.clientPrincipal();
       if (clientKey != null && !clientKey.equals("")) {
-        clientPrincipal = conf.get(clientKey);
+        if (hostname == null) {
+          throw new AuthorizationException(
+              "Can't authorize client when client hostname is null");
+        }
+        try {
+          clientPrincipal = SecurityUtil.getServerPrincipal(
+              conf.get(clientKey), hostname);
+        } catch (IOException e) {
+          throw (AuthorizationException) new AuthorizationException(
+              "Can't figure out Kerberos principal name for connection from "
+                  + hostname + " for user=" + user + " protocol=" + protocol)
+              .initCause(e);
+        }
       }
     }
     // when authorizing use the short name only

src/java/org/apache/hadoop/security/token/delegation/AbstractDelegationTokenIdentifier.java

@@ -27,6 +27,7 @@ import java.io.IOException;
 
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.io.WritableUtils;
+import org.apache.hadoop.security.KerberosName;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.security.token.TokenIdentifier;
 
@@ -57,7 +58,12 @@ extends TokenIdentifier {
     if (renewer == null) {
       this.renewer = new Text();
     } else {
-      this.renewer = renewer;
+      KerberosName renewerKrbName = new KerberosName(renewer.toString());
+      try {
+        this.renewer = new Text(renewerKrbName.getShortName());
+      } catch (IOException e) {
+        throw new RuntimeException(e);
+      }
     }
     if (realUser == null) {
       this.realUser = new Text();

src/java/org/apache/hadoop/security/token/delegation/AbstractDelegationTokenSecretManager.java

@@ -35,6 +35,7 @@ import javax.crypto.SecretKey;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.security.AccessControlException;
+import org.apache.hadoop.security.KerberosName;
 import org.apache.hadoop.security.token.Token;
 import org.apache.hadoop.security.token.SecretManager;
 import org.apache.hadoop.util.Daemon;
@@ -280,8 +281,10 @@ extends AbstractDelegationTokenIdentifier>
     }
     String owner = id.getUser().getUserName();
     Text renewer = id.getRenewer();
+    KerberosName cancelerKrbName = new KerberosName(canceller);
+    String cancelerShortName = cancelerKrbName.getShortName();
     if (!canceller.equals(owner)
-        && (renewer == null || "".equals(renewer.toString()) || !canceller
+        && (renewer == null || "".equals(renewer.toString()) || !cancelerShortName
             .equals(renewer.toString()))) {
       throw new AccessControlException(canceller
           + " is not authorized to cancel the token");

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

@@ -48,6 +48,7 @@ import org.apache.hadoop.security.token.SecretManager.InvalidToken;
 import org.apache.hadoop.security.SaslInputStream;
 import org.apache.hadoop.security.SaslRpcClient;
 import org.apache.hadoop.security.SaslRpcServer;
+import org.apache.hadoop.security.SecurityUtil;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.security.UserGroupInformation.AuthenticationMethod;
 
@@ -63,6 +64,7 @@ public class TestSaslRPC {
   
   static final String ERROR_MESSAGE = "Token is invalid";
   static final String SERVER_PRINCIPAL_KEY = "test.ipc.server.principal";
+  static final String SERVER_KEYTAB_KEY = "test.ipc.server.keytab";
   private static Configuration conf;
   static {
     conf = new Configuration();
@@ -76,6 +78,7 @@ public class TestSaslRPC {
     ((Log4JLogger) SaslRpcClient.LOG).getLogger().setLevel(Level.ALL);
     ((Log4JLogger) SaslRpcServer.LOG).getLogger().setLevel(Level.ALL);
     ((Log4JLogger) SaslInputStream.LOG).getLogger().setLevel(Level.ALL);
+    ((Log4JLogger) SecurityUtil.LOG).getLogger().setLevel(Level.ALL);
   }
 
   public static class TestTokenIdentifier extends TokenIdentifier {
@@ -248,7 +251,8 @@ public class TestSaslRPC {
   static void testKerberosRpc(String principal, String keytab) throws Exception {
     final Configuration newConf = new Configuration(conf);
     newConf.set(SERVER_PRINCIPAL_KEY, principal);
-    UserGroupInformation.loginUserFromKeytab(principal, keytab);
+    newConf.set(SERVER_KEYTAB_KEY, keytab);
+    SecurityUtil.login(newConf, SERVER_KEYTAB_KEY, SERVER_PRINCIPAL_KEY);
     UserGroupInformation current = UserGroupInformation.getCurrentUser();
     System.out.println("UGI: " + current);
 
@@ -269,6 +273,7 @@ public class TestSaslRPC {
         RPC.stopProxy(proxy);
       }
     }
+    System.out.println("Test is successful.");
   }
   
   @Test

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

@@ -17,6 +17,9 @@
 package org.apache.hadoop.security;
 
 import static org.junit.Assert.*;
+
+import java.io.IOException;
+
 import org.junit.Test;
 
 public class TestSecurityUtil {
@@ -32,4 +35,30 @@ public class TestSecurityUtil {
     assertFalse(SecurityUtil.isOriginalTGT("this@is/notright"));
     assertFalse(SecurityUtil.isOriginalTGT("krbtgt/foo@FOO"));
   }
+  
+  private void verify(String original, String hostname, String expected)
+      throws IOException {
+    assertTrue(SecurityUtil.getServerPrincipal(original, hostname).equals(
+        expected));
+    assertTrue(SecurityUtil.getServerPrincipal(original, null).equals(
+        expected));
+    assertTrue(SecurityUtil.getServerPrincipal(original, "").equals(
+        expected));
+    assertTrue(SecurityUtil.getServerPrincipal(original, "0.0.0.0").equals(
+        expected));
+  }
+
+  @Test
+  public void testGetServerPrincipal() throws IOException {
+    String service = "hdfs/";
+    String realm = "@REALM";
+    String hostname = SecurityUtil.getLocalHostName();
+    String shouldReplace = service + SecurityUtil.HOSTNAME_PATTERN + realm;
+    String replaced = service + hostname + realm;
+    verify(shouldReplace, hostname, replaced);
+    String shouldNotReplace = service + SecurityUtil.HOSTNAME_PATTERN + "NAME"
+        + realm;
+    verify(shouldNotReplace, hostname, shouldNotReplace);
+    verify(shouldNotReplace, shouldNotReplace, shouldNotReplace);
+  }
 }