HADOOP-7510. Add configurable option to use original hostname in token instead of IP to allow server IP change. Contributed by Daryn Sharp.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-0.20-security-205@1176645 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -315,6 +315,9 @@ Release 0.20.205.0 - 2011.09.27
     HDFS-2356.  Support case insensitive query parameter names in webhdfs.
     (szetszwo)
 
+    HADOOP-7510. Add configurable option to use original hostname in token
+    instead of IP to allow server IP change. (Daryn Sharp via suresh)
+
 Release 0.20.204.0 - 2011-8-25
 
   NEW FEATURES

src/core/core-default.xml

@@ -51,6 +51,18 @@
   </description>
 </property>
 
+<property>
+  <name>hadoop.security.token.service.use_ip</name>
+  <value>true</value>
+  <description>Controls whether tokens always use IP addresses.  DNS changes
+  will not be detected if this option is enabled.  Existing client connections
+  that break will always reconnect to the IP of the original host.  New clients
+  will connect to the host's new IP but fail to locate a token.  Disabling
+  this option will allow existing and new clients to detect an IP change and
+  continue to locate the new host's token.
+  </description>
+</property>
+
 <!--
 <property>
   <name>hadoop.security.service.user.name.key</name>

src/core/org/apache/hadoop/fs/CommonConfigurationKeys.java

@@ -42,6 +42,12 @@ public class CommonConfigurationKeys {
   /** See src/core/core-default.xml */
   public static final String  HADOOP_SECURITY_SERVICE_USER_NAME_KEY = 
     "hadoop.security.service.user.name.key";
+  /** See src/core/core-default.xml */
+  public static final String HADOOP_SECURITY_TOKEN_SERVICE_USE_IP =
+    "hadoop.security.token.service.use_ip";
+  public static final boolean HADOOP_SECURITY_TOKEN_SERVICE_USE_IP_DEFAULT =
+      true;
+  
   public static final String IPC_SERVER_RPC_READ_THREADS_KEY =
                                         "ipc.server.read.threadpool.size";
   public static final int IPC_SERVER_RPC_READ_THREADS_DEFAULT = 1;

src/core/org/apache/hadoop/fs/FileSystem.java

@@ -167,14 +167,16 @@ public abstract class FileSystem extends Configured implements Closeable {
   }
 
   /**
-   * Get a canonical name for this file system. It returns the uri of the file
-   * system unless overridden by a FileSystem implementation. File Systems with
-   * a valid authority can choose to return host:port or ip:port.
-   * 
-   * @return A string that uniquely identifies this file system
+   * Get a canonical service name for this file system.  The token cache is
+   * the only user of this value, and uses it to lookup this filesystem's
+   * service tokens.  The token cache will not attempt to acquire tokens if the
+   * service is null.
+   * @return a service string that uniquely identifies this file system, null
+   *         if the filesystem does not implement tokens
+   * @see SecurityUtil#buildDTServiceName(URI, int) 
    */
   public String getCanonicalServiceName() {
-    return getUri().toString();
+    return SecurityUtil.buildDTServiceName(getUri(), getDefaultPort());
   }
   
   /** @deprecated call #getUri() instead.*/

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

@@ -19,10 +19,8 @@
 package org.apache.hadoop.ipc;
 
 import java.net.InetAddress;
-import java.net.NetworkInterface;
 import java.net.Socket;
 import java.net.InetSocketAddress;
-import java.net.SocketException;
 import java.net.SocketTimeoutException;
 import java.net.UnknownHostException;
 import java.net.ConnectException;
@@ -51,7 +49,6 @@ import org.apache.commons.logging.*;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.io.IOUtils;
-import org.apache.hadoop.io.Text;
 import org.apache.hadoop.io.Writable;
 import org.apache.hadoop.io.WritableUtils;
 import org.apache.hadoop.io.DataOutputBuffer;
@@ -393,7 +390,7 @@ public class Client {
      */
     private synchronized boolean updateAddress() throws IOException {
       // Do a fresh lookup with the old host name.
-      InetSocketAddress currentAddr =  new InetSocketAddress(
+      InetSocketAddress currentAddr = NetUtils.makeSocketAddr(
                                server.getHostName(), server.getPort());
 
       if (!server.equals(currentAddr)) {
@@ -1069,7 +1066,9 @@ public class Client {
           call.error.fillInStackTrace();
           throw call.error;
         } else { // local exception
-          throw wrapException(remoteId.getAddress(), call.error);
+          // use the connection because it will reflect an ip change, unlike
+          // the remoteId
+          throw wrapException(connection.getRemoteAddress(), call.error);
         }
       } else {
         return call.value;

src/core/org/apache/hadoop/net/NetUtils.java

@@ -39,17 +39,40 @@ import javax.net.SocketFactory;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.ipc.Server;
 import org.apache.hadoop.ipc.VersionedProtocol;
+import org.apache.hadoop.security.SecurityUtil;
 import org.apache.hadoop.util.ReflectionUtils;
 
+// this will need to be replaced someday when there is a suitable replacement
+import sun.net.dns.ResolverConfiguration;
+import sun.net.util.IPAddressUtil;
+
 public class NetUtils {
   private static final Log LOG = LogFactory.getLog(NetUtils.class);
-  
+    
   private static Map<String, String> hostToResolved = 
                                      new HashMap<String, String>();
 
+  private static HostResolver hostResolver;
+  
+  static {
+    // SecurityUtils requires a more secure host resolver if tokens are
+    // using hostnames
+    setUseQualifiedHostResolver(!SecurityUtil.getTokenServiceUseIp());
+  }
+
+  /**
+   * This method is intended for use only by SecurityUtils!
+   * @param flag where the qualified or standard host resolver is used
+   *             to create socket addresses
+   */
+  public static void setUseQualifiedHostResolver(boolean flag) {
+      hostResolver = flag
+          ? new QualifiedHostResolver()
+          : new StandardHostResolver();
+  }
+  
   /**
    * Get the socket factory for the given class according to its
    * configuration parameter
@@ -135,37 +158,210 @@ public class NetUtils {
    */
   public static InetSocketAddress createSocketAddr(String target,
                                                    int defaultPort) {
-    int colonIndex = target.indexOf(':');
-    if (colonIndex < 0 && defaultPort == -1) {
-      throw new RuntimeException("Not a host:port pair: " + target);
+    if (target == null) {
+      throw new IllegalArgumentException("Socket address is null");
     }
-    String hostname;
-    int port = -1;
-    if (!target.contains("/")) {
-      if (colonIndex == -1) {
-        hostname = target;
-      } else {
-        // must be the old style <host>:<port>
-        hostname = target.substring(0, colonIndex);
-        port = Integer.parseInt(target.substring(colonIndex + 1));
-      }
-    } else {
-      // a new uri
-      URI addr = new Path(target).toUri();
-      hostname = addr.getHost();
-      port = addr.getPort();
+    boolean hasScheme = target.contains("://");    
+    URI uri = null;
+    try {
+      uri = hasScheme ? URI.create(target) : URI.create("dummyscheme://"+target);
+    } catch (IllegalArgumentException e) {
+      throw new IllegalArgumentException(
+          "Does not contain a valid host:port authority: " + target
+      );
     }
 
+    String host = uri.getHost();
+    int port = uri.getPort();
     if (port == -1) {
       port = defaultPort;
     }
+    String path = uri.getPath();
+    
+    if ((host == null) || (port < 0) ||
+        (!hasScheme && path != null && !path.isEmpty()))
+    {
+      throw new IllegalArgumentException(
+          "Does not contain a valid host:port authority: " + target
+      );
+    }
+    return makeSocketAddr(host, port);
+  }
   
-    if (getStaticResolution(hostname) != null) {
-      hostname = getStaticResolution(hostname);
+  /**
+   * Create a socket address with the given host and port.  The hostname
+   * might be replaced with another host that was set via
+   * {@link #addStaticResolution(String, String)}.  The value of
+   * hadoop.security.token.service.use_ip will determine whether the
+   * standard java host resolver is used, or if the fully qualified resolver
+   * is used.
+   * @param host the hostname or IP use to instantiate the object
+   * @param port the port number
+   * @return InetSocketAddress
+   */
+  public static InetSocketAddress makeSocketAddr(String host, int port) {
+    String staticHost = getStaticResolution(host);
+    String resolveHost = (staticHost != null) ? staticHost : host;
+    
+    InetSocketAddress addr;
+    try {
+      InetAddress iaddr = hostResolver.getByName(resolveHost);
+      // if there is a static entry for the host, make the returned
+      // address look like the original given host
+      if (staticHost != null) {
+        iaddr = InetAddress.getByAddress(host, iaddr.getAddress());
+      }
+      addr = new InetSocketAddress(iaddr, port);
+    } catch (UnknownHostException e) {
+      addr = InetSocketAddress.createUnresolved(host, port);
+    }
+    return addr;
+  }
+  
+  protected interface HostResolver {
+    InetAddress getByName(String host) throws UnknownHostException;    
+  }
+  
+  /**
+   * Uses standard java host resolution
+   */
+  protected static class StandardHostResolver implements HostResolver {
+    public InetAddress getByName(String host) throws UnknownHostException {
+      return InetAddress.getByName(host);
     }
-    return new InetSocketAddress(hostname, port);
   }
+  
+  /**
+   * This an alternate resolver with important properties that the standard
+   * java resolver lacks:
+   * 1) The hostname is fully qualified.  This avoids security issues if not
+   *    all hosts in the cluster do not share the same search domains.  It
+   *    also prevents other hosts from performing unnecessary dns searches.
+   *    In contrast, InetAddress simply returns the host as given.
+   * 2) The InetAddress is instantiated with an exact host and IP to prevent
+   *    further unnecessary lookups.  InetAddress may perform an unnecessary
+   *    reverse lookup for an IP.
+   * 3) A call to getHostName() will always return the qualified hostname, or
+   *    more importantly, the IP if instantiated with an IP.  This avoids
+   *    unnecessary dns timeouts if the host is not resolvable.
+   * 4) Point 3 also ensures that if the host is re-resolved, ex. during a
+   *    connection re-attempt, that a reverse lookup to host and forward
+   *    lookup to IP is not performed since the reverse/forward mappings may
+   *    not always return the same IP.  If the client initiated a connection
+   *    with an IP, then that IP is all that should ever be contacted.
+   *    
+   * NOTE: this resolver is only used if:
+   *       hadoop.security.token.service.use_ip=false 
+   */
+  protected static class QualifiedHostResolver implements HostResolver {
+    @SuppressWarnings("unchecked")
+    private List<String> searchDomains =
+        ResolverConfiguration.open().searchlist();
+    
+    /**
+     * Create an InetAddress with a fully qualified hostname of the given
+     * hostname.  InetAddress does not qualify an incomplete hostname that
+     * is resolved via the domain search list.
+     * {@link InetAddress#getCanonicalHostName()} will fully qualify the
+     * hostname, but it always return the A record whereas the given hostname
+     * may be a CNAME.
+     * 
+     * @param host a hostname or ip address
+     * @return InetAddress with the fully qualified hostname or ip
+     * @throws UnknownHostException if host does not exist
+     */
+    public InetAddress getByName(String host) throws UnknownHostException {
+      InetAddress addr = null;
+
+      if (IPAddressUtil.isIPv4LiteralAddress(host)) {
+        // use ipv4 address as-is
+        byte[] ip = IPAddressUtil.textToNumericFormatV4(host);
+        addr = InetAddress.getByAddress(host, ip);
+      } else if (IPAddressUtil.isIPv6LiteralAddress(host)) {
+        // use ipv6 address as-is
+        byte[] ip = IPAddressUtil.textToNumericFormatV6(host);
+        addr = InetAddress.getByAddress(host, ip);
+      } else if (host.endsWith(".")) {
+        // a rooted host ends with a dot, ex. "host."
+        // rooted hosts never use the search path, so only try an exact lookup
+        addr = getByExactName(host);
+      } else if (host.contains(".")) {
+        // the host contains a dot (domain), ex. "host.domain"
+        // try an exact host lookup, then fallback to search list
+        addr = getByExactName(host);
+        if (addr == null) {
+          addr = getByNameWithSearch(host);
+        }
+      } else {
+        // it's a simple host with no dots, ex. "host"
+        // try the search list, then fallback to exact host
+        InetAddress loopback = InetAddress.getByName(null);
+        if (host.equalsIgnoreCase(loopback.getHostName())) {
+          addr = InetAddress.getByAddress(host, loopback.getAddress());
+        } else {
+          addr = getByNameWithSearch(host);
+          if (addr == null) {
+            addr = getByExactName(host);
+          }
+        }
+      }
+      // unresolvable!
+      if (addr == null) {
+        throw new UnknownHostException(host);
+      }
+      return addr;
+    }
+
+    InetAddress getByExactName(String host) {
+      InetAddress addr = null;
+      // InetAddress will use the search list unless the host is rooted
+      // with a trailing dot.  The trailing dot will disable any use of the
+      // search path in a lower level resolver.  See RFC 1535.
+      String fqHost = host;
+      if (!fqHost.endsWith(".")) fqHost += ".";
+      try {
+        addr = getInetAddressByName(fqHost);
+        // can't leave the hostname as rooted or other parts of the system
+        // malfunction, ex. kerberos principals are lacking proper host
+        // equivalence for rooted/non-rooted hostnames
+        addr = InetAddress.getByAddress(host, addr.getAddress());
+      } catch (UnknownHostException e) {
+        // ignore, caller will throw if necessary
+      }
+      return addr;
+    }
+
+    InetAddress getByNameWithSearch(String host) {
+      InetAddress addr = null;
+      if (host.endsWith(".")) { // already qualified?
+        addr = getByExactName(host); 
+      } else {
+        for (String domain : searchDomains) {
+          String dot = !domain.startsWith(".") ? "." : "";
+          addr = getByExactName(host + dot + domain);
+          if (addr != null) break;
+        }
+      }
+      return addr;
+    }
+
+    // implemented as a separate method to facilitate unit testing
+    InetAddress getInetAddressByName(String host) throws UnknownHostException {
+      return InetAddress.getByName(host);
+    }
 
+    void setSearchDomains(String ... domains) {
+      searchDomains = Arrays.asList(domains);
+    }
+  }
+  
+  /**
+   * This is for testing only!
+   */
+  static void setHostResolver(QualifiedHostResolver newResolver) {
+    hostResolver = newResolver;
+  }
+    
   /**
    * Handle the transition from pairs of attributes specifying a host and port
    * to a single colon separated one.
@@ -268,8 +464,8 @@ public class NetUtils {
    */
   public static InetSocketAddress getConnectAddress(Server server) {
     InetSocketAddress addr = server.getListenerAddress();
-    if (addr.getAddress().getHostAddress().equals("0.0.0.0")) {
-      addr = new InetSocketAddress("127.0.0.1", addr.getPort());
+    if (addr.getAddress().isAnyLocalAddress()) {
+      addr = makeSocketAddr("127.0.0.1", addr.getPort());
     }
     return addr;
   }

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

@@ -31,6 +31,7 @@ import javax.security.auth.kerberos.KerberosTicket;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.CommonConfigurationKeys;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.net.NetUtils;
 import org.apache.hadoop.security.authorize.AccessControlList;
@@ -44,6 +45,33 @@ public class SecurityUtil {
   public static final Log LOG = LogFactory.getLog(SecurityUtil.class);
   public static final String HOSTNAME_PATTERN = "_HOST";
 
+  // controls whether buildTokenService will use an ip or host/ip as given
+  // by the user
+  private static boolean useIpForTokenService;
+  
+  static {
+    boolean useIp = new Configuration().getBoolean(
+      CommonConfigurationKeys.HADOOP_SECURITY_TOKEN_SERVICE_USE_IP,
+      CommonConfigurationKeys.HADOOP_SECURITY_TOKEN_SERVICE_USE_IP_DEFAULT);
+    setTokenServiceUseIp(useIp);
+  }
+  
+  /**
+   * For use only by tests!
+   */
+  static void setTokenServiceUseIp(boolean flag) {
+    useIpForTokenService = flag;
+    NetUtils.setUseQualifiedHostResolver(!flag);
+  }
+  
+  /**
+   * Intended only for temporary use by NetUtils.  Do not use.
+   * @return whether tokens use an IP address
+   */
+ public static boolean getTokenServiceUseIp() {
+    return useIpForTokenService;
+  }
+  
   /**
    * Find the original TGT within the current subject's credentials. Cross-realm
    * TGT's of the form "krbtgt/TWO.COM@ONE.COM" may be present.
@@ -233,6 +261,15 @@ public class SecurityUtil {
         hostname);
     UserGroupInformation.loginUserFromKeytab(principalName, keytabFilename);
   }
+
+  /**
+   * Decode the given token's service field into an InetAddress
+   * @param token from which to obtain the service
+   * @return InetAddress for the service
+   */
+  public static InetSocketAddress getTokenServiceAddr(Token<?> token) {
+    return NetUtils.createSocketAddr(token.getService().toString());
+  }
   
   /**
    * Set the given token's service to the format expected by the RPC client 
@@ -245,47 +282,41 @@ public class SecurityUtil {
   
   /**
    * Construct the service key for a token
-   * @param addr the socket for the rpc connection
-   * @return Text formatted for the service field in a token 
+   * @param addr InetSocketAddress of remote connection with a token
+   * @return "ip:port" or "host:port" depending on the value of
+   *          hadoop.security.token.service.use_ip
    */
   public static Text buildTokenService(InetSocketAddress addr) {
-    return new Text(buildDTAuthority(addr));
+    String host = null;
+    if (useIpForTokenService) {
+      if (addr.isUnresolved()) { // host has no ip address
+        throw new IllegalArgumentException(
+            new UnknownHostException(addr.getHostName())
+        );
+      }
+      host = addr.getAddress().getHostAddress();
+    } else {
+      host = addr.getHostName().toLowerCase();
+    }
+    return new Text(host + ":" + addr.getPort());
   }
   
   /**
-   * create service name for Delegation token ip:port
-   * @param uri
-   * @return "ip:port"
+   * create the service name for a Delegation token
+   * @param uri of the service
+   * @param defPort is used if the uri lacks a port
+   * @return the token service, or null if no authority
+   * @see #buildTokenService(InetSocketAddress)
    */
   public static String buildDTServiceName(URI uri, int defPort) {
-    InetSocketAddress addr = NetUtils.createSocketAddr(uri.getAuthority(),
-                                                       defPort);
-    return buildDTAuthority(addr);
+    String authority = uri.getAuthority();
+    if (authority == null || authority.isEmpty()) {
+      return null;
+    }
+    InetSocketAddress addr = NetUtils.createSocketAddr(authority, defPort);
+    return buildTokenService(addr).toString();
    }
   
-  /**
-   * create an authority name for looking up a Delegation token based
-   * on a socket
-   * @param addr InetSocketAddress of remote connection with a token
-   * @return "ip:port"
-   */
-  static String buildDTAuthority(InetSocketAddress addr) {
-    String host= addr.getAddress().getHostAddress();
-    return buildDTAuthority(host, addr.getPort());
-  }
-  
-  /**
-   * create an authority name for looking up a Delegation token based
-   * on a host/ip pair
-   * @param host the remote host
-   * @param port the remote port
-   * @return "ip:port"
-   */
-  static String buildDTAuthority(String host, int port) {
-    host = (host != null) ? NetUtils.normalizeHostName(host) : "";
-    return host + ":" + port;
-  }
-  
   /**
    * Get the ACL object representing the cluster administrators
    * The user who starts the daemon is automatically added as an admin

src/hdfs/org/apache/hadoop/hdfs/DFSClient.java

@@ -147,8 +147,8 @@ public class DFSClient implements FSConstants, java.io.Closeable {
   static ClientDatanodeProtocol createClientDatanodeProtocolProxy (
       DatanodeID datanodeid, Configuration conf, 
       Block block, Token<BlockTokenIdentifier> token, int socketTimeout) throws IOException {
-    InetSocketAddress addr = NetUtils.createSocketAddr(
-      datanodeid.getHost() + ":" + datanodeid.getIpcPort());
+    InetSocketAddress addr = NetUtils.makeSocketAddr(
+      datanodeid.getHost(), datanodeid.getIpcPort());
     if (ClientDatanodeProtocol.LOG.isDebugEnabled()) {
       ClientDatanodeProtocol.LOG.info("ClientDatanodeProtocol addr=" + addr);
     }
@@ -356,9 +356,9 @@ public class DFSClient implements FSConstants, java.io.Closeable {
       Token<DelegationTokenIdentifier> delToken = 
           (Token<DelegationTokenIdentifier>) token;
       LOG.info("Renewing " + stringifyToken(delToken));
+      InetSocketAddress addr = SecurityUtil.getTokenServiceAddr(token);
       ClientProtocol nn = 
-        createRPCNamenode(NameNode.getAddress(token.getService().toString()),
-                          conf, UserGroupInformation.getCurrentUser());
+          createRPCNamenode(addr, conf, UserGroupInformation.getCurrentUser());
       try {
         return nn.renewDelegationToken(delToken);
       } catch (RemoteException re) {
@@ -373,9 +373,9 @@ public class DFSClient implements FSConstants, java.io.Closeable {
       Token<DelegationTokenIdentifier> delToken = 
           (Token<DelegationTokenIdentifier>) token;
       LOG.info("Cancelling " + stringifyToken(delToken));
+      InetSocketAddress addr = SecurityUtil.getTokenServiceAddr(token);
       ClientProtocol nn = 
-          createRPCNamenode(NameNode.getAddress(token.getService().toString()),
-                            conf, UserGroupInformation.getCurrentUser());
+          createRPCNamenode(addr, conf, UserGroupInformation.getCurrentUser());
         try {
           nn.cancelDelegationToken(delToken);
         } catch (RemoteException re) {

src/hdfs/org/apache/hadoop/hdfs/DistributedFileSystem.java

@@ -39,7 +39,6 @@ import org.apache.hadoop.hdfs.server.namenode.NameNode;
 import org.apache.hadoop.hdfs.DFSClient.DFSOutputStream;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.security.AccessControlException;
-import org.apache.hadoop.security.SecurityUtil;
 import org.apache.hadoop.security.token.Token;
 import org.apache.hadoop.security.token.SecretManager.InvalidToken;
 import org.apache.hadoop.util.Progressable;
@@ -66,11 +65,6 @@ public class DistributedFileSystem extends FileSystem {
   public DistributedFileSystem() {
   }
 
-  @Override
-  public String getCanonicalServiceName() {
-    return SecurityUtil.buildDTServiceName(getUri(), getDefaultPort());
-  }
-  
   /** @deprecated */
   public DistributedFileSystem(InetSocketAddress namenode,
     Configuration conf) throws IOException {

src/hdfs/org/apache/hadoop/hdfs/HftpFileSystem.java

@@ -49,6 +49,7 @@ import org.apache.hadoop.fs.MD5MD5CRC32FileChecksum;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.hdfs.security.token.delegation.DelegationTokenIdentifier;
+import org.apache.hadoop.hdfs.security.token.delegation.DelegationTokenSelector;
 import org.apache.hadoop.hdfs.server.namenode.JspHelper;
 import org.apache.hadoop.hdfs.server.namenode.NameNode;
 import org.apache.hadoop.hdfs.server.namenode.StreamFile;
@@ -62,6 +63,7 @@ import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.security.token.Token;
 import org.apache.hadoop.security.token.TokenIdentifier;
 import org.apache.hadoop.security.token.TokenRenewer;
+import org.apache.hadoop.security.token.delegation.AbstractDelegationTokenSelector;
 import org.apache.hadoop.util.Progressable;
 import org.xml.sax.Attributes;
 import org.xml.sax.InputSource;
@@ -81,21 +83,28 @@ public class HftpFileSystem extends FileSystem {
     HttpURLConnection.setFollowRedirects(true);
   }
 
-  public static final int DEFAULT_PORT = 50470;
+  public static final int DEFAULT_PORT = 50070;
+  public static final int DEFAULT_SECURE_PORT = 50470;
   public static final Text TOKEN_KIND = new Text("HFTP delegation");
   
-  protected InetSocketAddress nnAddr;
   protected UserGroupInformation ugi; 
-  private String nnHttpUrl;
-  private Text hdfsServiceName;
   private URI hftpURI;
 
+  protected InetSocketAddress nnAddr;
+  protected InetSocketAddress nnSecureAddr;  
+
   public static final String HFTP_TIMEZONE = "UTC";
   public static final String HFTP_DATE_FORMAT = "yyyy-MM-dd'T'HH:mm:ssZ";
   private Token<?> delegationToken;
   private Token<?> renewToken;
   public static final String HFTP_SERVICE_NAME_KEY = "hdfs.service.host_";
 
+  private static final HftpDelegationTokenSelector hftpTokenSelector =
+      new HftpDelegationTokenSelector();
+
+  private static final DelegationTokenSelector hdfsTokenSelector =
+      new DelegationTokenSelector();
+
   public static final SimpleDateFormat getDateFormat() {
     final SimpleDateFormat df = new SimpleDateFormat(HFTP_DATE_FORMAT);
     df.setTimeZone(TimeZone.getTimeZone(HFTP_TIMEZONE));
@@ -111,97 +120,96 @@ public class HftpFileSystem extends FileSystem {
 
   @Override
   protected int getDefaultPort() {
-    return DEFAULT_PORT;
+    return getConf().getInt("dfs.http.port", DEFAULT_PORT);
+  }
+
+  protected int getDefaultSecurePort() {
+    return getConf().getInt("dfs.https.port", DEFAULT_SECURE_PORT);
+  }
+
+  protected InetSocketAddress getNamenodeAddr(URI uri) {
+    // use authority so user supplied uri can override port
+    return NetUtils.createSocketAddr(uri.getAuthority(), getDefaultPort());
+  }
+
+  protected InetSocketAddress getNamenodeSecureAddr(URI uri) {
+    // must only use the host and the configured https port
+    return NetUtils.makeSocketAddr(uri.getHost(), getDefaultSecurePort());
   }
 
   @Override
   public String getCanonicalServiceName() {
-    return SecurityUtil.buildDTServiceName(hftpURI, getDefaultPort());
+    // unlike other filesystems, hftp's service is the secure port, not the
+    // actual port in the uri
+    return SecurityUtil.buildTokenService(nnSecureAddr).toString();
   }
 
   @Override
   public void initialize(final URI name, final Configuration conf) 
   throws IOException {
-    super.initialize(name, conf);
     setConf(conf);
+    super.initialize(name, conf);
     this.ugi = UserGroupInformation.getCurrentUser();
-
-    nnAddr = NetUtils.createSocketAddr(name.toString());
-    StringBuilder sb = new StringBuilder();
-    sb.append(NetUtils.normalizeHostName(name.getHost()));
-    sb.append(":");
-    sb.append(conf.getInt("dfs.https.port", DEFAULT_PORT));
-    String tail = sb.toString();
-    nnHttpUrl = "https://" + tail;
+    this.nnAddr = getNamenodeAddr(name);
+    this.nnSecureAddr = getNamenodeSecureAddr(name);
+    this.hftpURI = createUri(name.getScheme(), nnAddr);
     
-    try {
-      hftpURI = new URI("hftp://" + tail);
-    } catch (URISyntaxException ue) {
-      throw new IOException("bad uri for hdfs", ue);
-    }
-    String key = HftpFileSystem.HFTP_SERVICE_NAME_KEY+
-        SecurityUtil.buildDTServiceName(name, DEFAULT_PORT);
-    LOG.debug("Trying to find DT for " + name + " using key=" + key + 
-        "; conf=" + conf.get(key, ""));
-    String nnServiceName = conf.get(key);
-    int nnPort = NameNode.DEFAULT_PORT;
-    if (nnServiceName != null) {
-      nnPort = NetUtils.createSocketAddr(nnServiceName, 
-                                         NameNode.DEFAULT_PORT).getPort();
-    }
-    sb = new StringBuilder("hdfs://");
-    sb.append(nnAddr.getHostName());
-    sb.append(":");
-    sb.append(nnPort);
-    try {
-      URI hdfsURI = new URI(sb.toString());
-      hdfsServiceName = new Text(SecurityUtil.buildDTServiceName(hdfsURI, 
-                                                                 nnPort));
-    } catch (URISyntaxException ue) {
-      throw new IOException("bad uri for hdfs", ue);
-    }
     if (UserGroupInformation.isSecurityEnabled()) {
-      String hftpServiceName = getCanonicalServiceName();
-      for (Token<? extends TokenIdentifier> t : ugi.getTokens()) {
-        Text kind = t.getKind();
-        if (DelegationTokenIdentifier.HDFS_DELEGATION_KIND.equals(kind)){
-          if (hdfsServiceName.equals(t.getService())) {
-            setDelegationToken(t);
-            break;
-          }
-        } else if (TOKEN_KIND.equals(kind)) {
-          if (hftpServiceName.equals(normalizeService(t.getService()
-                                       .toString()))) {
-            setDelegationToken(t);
-            break;
-          }
-        }
-      }
-      
+      Token<?> token = selectHftpDelegationToken();
+      if (token == null) {
+        token = selectHdfsDelegationToken();
+      }   
       //since we don't already have a token, go get one over https
-      if (delegationToken == null) {
-        Token<?> newToken = getDelegationToken(null);
-        if (newToken != null) {
-          setDelegationToken(newToken);
+      if (token == null) {
+        token = getDelegationToken(null);
+        // security might be disabled
+        if (token != null) {
+          setDelegationToken(token);
           renewer.addTokenToRenew(this);
-          LOG.debug("Created new DT for " + delegationToken.getService());
+          LOG.debug("Created new DT for " + token.getService());
         }
       } else {
-        LOG.debug("Found existing DT for " + delegationToken.getService());        
+        LOG.debug("Found existing DT for " + token.getService());        
       }
     }
   }
 
-  private String normalizeService(String service) {
-    int colonIndex = service.indexOf(':');
-    if (colonIndex == -1) {
-      throw new IllegalArgumentException("Invalid service for hftp token: " + 
-                                         service);
+  private Token<DelegationTokenIdentifier> selectHftpDelegationToken() {
+    Text serviceName = SecurityUtil.buildTokenService(nnSecureAddr);
+    return hftpTokenSelector.selectToken(serviceName, ugi.getTokens());      
+  }
+  
+  private Token<DelegationTokenIdentifier> selectHdfsDelegationToken() {
+    // this guesses the remote cluster's rpc service port.
+    // the current token design assumes it's the same as the local cluster's
+    // rpc port unless a config key is set.  there should be a way to automatic
+    // and correctly determine the value
+    String key = HftpFileSystem.HFTP_SERVICE_NAME_KEY+
+        SecurityUtil.buildTokenService(nnSecureAddr);
+    String nnServiceName = getConf().get(key);
+    LOG.debug("Trying to find DT for " + getUri() + " using key=" + key + 
+        "; conf=" + nnServiceName);
+    
+    int nnRpcPort = NameNode.DEFAULT_PORT;
+    if (nnServiceName != null) {
+      nnRpcPort = NetUtils.createSocketAddr(nnServiceName, nnRpcPort).getPort(); 
     }
-    String hostname = 
-        NetUtils.normalizeHostName(service.substring(0, colonIndex));
-    String port = service.substring(colonIndex + 1);
-    return hostname + ":" + port;
+    
+    InetSocketAddress addr = 
+        NetUtils.makeSocketAddr(nnAddr.getHostName(), nnRpcPort);
+    Text serviceName = SecurityUtil.buildTokenService(addr);
+    
+    return hdfsTokenSelector.selectToken(serviceName, ugi.getTokens());
+  }
+  
+  private static URI createUri(String scheme, InetSocketAddress addr) {
+    URI uri = null;
+    try {
+      uri = new URI(scheme, null, addr.getHostName(), addr.getPort(), null, null, null);
+    } catch (URISyntaxException ue) {
+      throw new IllegalArgumentException(ue);
+    }
+    return uri;
   }
 
   private <T extends TokenIdentifier> void setDelegationToken(Token<T> token) {
@@ -209,13 +217,18 @@ public class HftpFileSystem extends FileSystem {
     // emulate the 203 usage of the tokens
     // by setting the kind and service as if they were hdfs tokens
     delegationToken = new Token<T>(token);
+    // NOTE: the remote nn must be configured to use hdfs
     delegationToken.setKind(DelegationTokenIdentifier.HDFS_DELEGATION_KIND);
-    delegationToken.setService(hdfsServiceName);
+    // no need to change service because we aren't exactly sure what it
+    // should be.  we can guess, but it might be wrong if the local conf
+    // value is incorrect.  the service is a client side field, so the remote
+    // end does not care about the value
   }
 
   @Override
   public synchronized Token<?> getDelegationToken(final String renewer
                                                   ) throws IOException {
+    final String nnHttpUrl = createUri("https", nnSecureAddr).toString();
     try {
       //Renew TGT if needed
       ugi.checkTGTAndReloginFromKeytab();
@@ -246,12 +259,7 @@ public class HftpFileSystem extends FileSystem {
 
   @Override
   public URI getUri() {
-    try {
-      return new URI("hftp", null, nnAddr.getHostName(), nnAddr.getPort(),
-                     null, null, null);
-    } catch (URISyntaxException e) {
-      return null;
-    } 
+    return hftpURI;
   }
   
   /**
@@ -808,10 +816,11 @@ public class HftpFileSystem extends FileSystem {
       // update the kerberos credentials, if they are coming from a keytab
       UserGroupInformation.getLoginUser().checkTGTAndReloginFromKeytab();
       // use https to renew the token
-      return 
-        DelegationTokenFetcher.renewDelegationToken
-        ("https://" + token.getService().toString(), 
-         (Token<DelegationTokenIdentifier>) token);
+      InetSocketAddress serviceAddr = SecurityUtil.getTokenServiceAddr(token);
+      return DelegationTokenFetcher.renewDelegationToken(
+          createUri("https", serviceAddr).toString(),
+          (Token<DelegationTokenIdentifier>) token
+      );
     }
 
     @SuppressWarnings("unchecked")
@@ -821,10 +830,20 @@ public class HftpFileSystem extends FileSystem {
       // update the kerberos credentials, if they are coming from a keytab
       UserGroupInformation.getLoginUser().checkTGTAndReloginFromKeytab();
       // use https to cancel the token
-      DelegationTokenFetcher.cancelDelegationToken
-        ("https://" + token.getService().toString(), 
-         (Token<DelegationTokenIdentifier>) token);
+      InetSocketAddress serviceAddr = SecurityUtil.getTokenServiceAddr(token);
+      DelegationTokenFetcher.cancelDelegationToken(
+          createUri("https", serviceAddr).toString(), 
+          (Token<DelegationTokenIdentifier>) token
+      );
     }
     
   }
+  
+  private static class HftpDelegationTokenSelector
+  extends AbstractDelegationTokenSelector<DelegationTokenIdentifier> {
+
+    public HftpDelegationTokenSelector() {
+      super(TOKEN_KIND);
+    }
+  }
 }

src/hdfs/org/apache/hadoop/hdfs/HsftpFileSystem.java

@@ -20,6 +20,7 @@ package org.apache.hadoop.hdfs;
 
 import java.io.IOException;
 import java.net.HttpURLConnection;
+import java.net.InetSocketAddress;
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.net.URL;
@@ -64,6 +65,16 @@ public class HsftpFileSystem extends HftpFileSystem {
         "ssl.client.keystore.type", "jks"));
   }
 
+  @Override
+  protected int getDefaultPort() {
+    return getDefaultSecurePort();
+  }
+
+  @Override
+  protected InetSocketAddress getNamenodeSecureAddr(URI uri) {
+    return getNamenodeAddr(uri);
+  }
+
   @Override
   protected HttpURLConnection openConnection(String path, String query)
       throws IOException {
@@ -80,16 +91,6 @@ public class HsftpFileSystem extends HftpFileSystem {
     }
   }
 
-  @Override
-  public URI getUri() {
-    try {
-      return new URI("hsftp", null, nnAddr.getHostName(), nnAddr.getPort(),
-                     null, null, null);
-    } catch (URISyntaxException e) {
-      return null;
-    } 
-  }
-
   /**
    * Dummy hostname verifier that is used to bypass hostname checking
    */

src/hdfs/org/apache/hadoop/hdfs/tools/DelegationTokenFetcher.java

@@ -24,6 +24,7 @@ import java.io.InputStream;
 import java.io.InputStreamReader;
 import java.io.PrintStream;
 import java.net.HttpURLConnection;
+import java.net.InetSocketAddress;
 import java.net.URL;
 import java.net.URLConnection;
 import java.security.PrivilegedExceptionAction;
@@ -42,7 +43,7 @@ import org.apache.hadoop.hdfs.server.namenode.CancelDelegationTokenServlet;
 import org.apache.hadoop.hdfs.server.namenode.GetDelegationTokenServlet;
 import org.apache.hadoop.hdfs.server.namenode.RenewDelegationTokenServlet;
 import org.apache.hadoop.io.IOUtils;
-import org.apache.hadoop.io.Text;
+import org.apache.hadoop.net.NetUtils;
 import org.apache.hadoop.security.Credentials;
 import org.apache.hadoop.security.SecurityUtil;
 import org.apache.hadoop.security.UserGroupInformation;
@@ -112,7 +113,7 @@ public class DelegationTokenFetcher {
       printUsage(System.err);
     }
     if (remaining.length != 1 || remaining[0].charAt(0) == '-') {
-      System.err.println("ERROR: Must specify exacltly one token file");
+      System.err.println("ERROR: Must specify exactly one token file");
       printUsage(System.err);
     }
     // default to using the local file system
@@ -175,6 +176,7 @@ public class DelegationTokenFetcher {
   static public Credentials getDTfromRemote(String nnAddr, 
                                             String renewer) throws IOException {
     DataInputStream dis = null;
+    InetSocketAddress serviceAddr = NetUtils.createSocketAddr(nnAddr);
 
     try {
       StringBuffer url = new StringBuffer();
@@ -197,9 +199,7 @@ public class DelegationTokenFetcher {
       ts.readFields(dis);
       for(Token<?> token: ts.getAllTokens()) {
         token.setKind(HftpFileSystem.TOKEN_KIND);
-        token.setService(new Text(SecurityUtil.buildDTServiceName
-                                   (remoteURL.toURI(), 
-                                    HftpFileSystem.DEFAULT_PORT)));
+        SecurityUtil.setTokenService(token, serviceAddr);
       }
       return ts;
     } catch (Exception e) {

src/hdfs/org/apache/hadoop/hdfs/web/WebHdfsFileSystem.java

@@ -65,9 +65,9 @@ import org.apache.hadoop.hdfs.web.resources.RecursiveParam;
 import org.apache.hadoop.hdfs.web.resources.RenewerParam;
 import org.apache.hadoop.hdfs.web.resources.ReplicationParam;
 import org.apache.hadoop.hdfs.web.resources.UserParam;
-import org.apache.hadoop.io.Text;
 import org.apache.hadoop.ipc.RemoteException;
 import org.apache.hadoop.security.AccessControlException;
+import org.apache.hadoop.security.SecurityUtil;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.security.authentication.client.AuthenticatedURL;
 import org.apache.hadoop.security.authentication.client.AuthenticationException;
@@ -105,16 +105,6 @@ public class WebHdfsFileSystem extends HftpFileSystem {
     this.workingDir = getHomeDirectory();
   }
 
-  @Override
-  public URI getUri() {
-    try {
-      return new URI(SCHEME, null, nnAddr.getHostName(), nnAddr.getPort(),
-          null, null, null);
-    } catch (URISyntaxException e) {
-      return null;
-    }
-  }
-
   @Override
   public Path getHomeDirectory() {
     return makeQualified(new Path("/user/" + ugi.getShortUserName()));
@@ -416,7 +406,7 @@ public class WebHdfsFileSystem extends HftpFileSystem {
     final HttpOpParam.Op op = GetOpParam.Op.GETDELEGATIONTOKEN;
     final Map<String, Object> m = run(op, null, new RenewerParam(renewer));
     final Token<DelegationTokenIdentifier> token = JsonUtil.toDelegationToken(m); 
-    token.setService(new Text(getCanonicalServiceName()));
+    SecurityUtil.setTokenService(token, nnAddr);
     return token;
   }
 

src/mapred/org/apache/hadoop/mapred/Child.java

@@ -25,27 +25,22 @@ import java.io.PrintStream;
 import java.net.InetSocketAddress;
 import java.security.PrivilegedExceptionAction;
 import java.util.Arrays;
-import java.util.List;
-import java.util.regex.Pattern;
-
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.fs.FSError;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.LocalDirAllocator;
 import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.io.Text;
 import org.apache.hadoop.ipc.RPC;
 import org.apache.hadoop.mapreduce.security.TokenCache;
 import org.apache.hadoop.mapreduce.security.token.JobTokenIdentifier;
 import org.apache.hadoop.mapreduce.security.token.JobTokenSecretManager;
 import org.apache.hadoop.mapreduce.server.tasktracker.JVMInfo;
-import org.apache.hadoop.mapreduce.server.tasktracker.Localizer;
-import org.apache.hadoop.metrics2.MetricsException;
 import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem;
 import org.apache.hadoop.metrics2.source.JvmMetricsSource;
 import org.apache.hadoop.net.NetUtils;
 import org.apache.hadoop.security.Credentials;
+import org.apache.hadoop.security.SecurityUtil;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.security.token.Token;
 import org.apache.hadoop.util.Shell;
@@ -77,7 +72,7 @@ class Child {
     final JobConf defaultConf = new JobConf();
     String host = args[0];
     int port = Integer.parseInt(args[1]);
-    final InetSocketAddress address = new InetSocketAddress(host, port);
+    final InetSocketAddress address = NetUtils.makeSocketAddr(host, port);
     final TaskAttemptID firstTaskid = TaskAttemptID.forName(args[2]);
     final String logLocation = args[3];
     final int SLEEP_LONGER_COUNT = 5;
@@ -100,8 +95,7 @@ class Child {
         "; from file=" + jobTokenFile);
     
     Token<JobTokenIdentifier> jt = TokenCache.getJobToken(credentials);
-    jt.setService(new Text(address.getAddress().getHostAddress() + ":"
-        + address.getPort()));
+    SecurityUtil.setTokenService(jt, address);
     UserGroupInformation current = UserGroupInformation.getCurrentUser();
     current.addToken(jt);
 

src/mapred/org/apache/hadoop/mapred/JobClient.java

@@ -493,8 +493,7 @@ public class JobClient extends Configured implements MRConstants, Tool  {
     @Override
     public long renew(Token<?> token, Configuration conf
                       ) throws IOException, InterruptedException {
-      InetSocketAddress addr = 
-          NetUtils.createSocketAddr(token.getService().toString());
+      InetSocketAddress addr = SecurityUtil.getTokenServiceAddr(token);
       JobSubmissionProtocol jt = createRPCProxy(addr, conf);
       return jt.renewDelegationToken((Token<DelegationTokenIdentifier>) token);
     }
@@ -503,8 +502,7 @@ public class JobClient extends Configured implements MRConstants, Tool  {
     @Override
     public void cancel(Token<?> token, Configuration conf
                        ) throws IOException, InterruptedException {
-      InetSocketAddress addr = 
-          NetUtils.createSocketAddr(token.getService().toString());
+      InetSocketAddress addr = SecurityUtil.getTokenServiceAddr(token);
       JobSubmissionProtocol jt = createRPCProxy(addr, conf);
       jt.cancelDelegationToken((Token<DelegationTokenIdentifier>) token);
     }

src/mapred/org/apache/hadoop/mapreduce/security/TokenCache.java

@@ -118,7 +118,6 @@ public class TokenCache {
         Token<?> token = fs.getDelegationToken(delegTokenRenewer);
         if (token != null) {
           Text fsNameText = new Text(fsName);
-          token.setService(fsNameText);
           credentials.addToken(fsNameText, token);
           LOG.info("Got dt for " + p + ";uri="+ fsName + 
                    ";t.service="+token.getService());

src/test/org/apache/hadoop/fs/TestLocalFileSystem.java

@@ -158,6 +158,6 @@ public class TestLocalFileSystem extends TestCase {
   public void testGetCanonicalServiceName() throws IOException {
     Configuration conf = new Configuration();
     FileSystem fs = FileSystem.getLocal(conf);
-    assertEquals(fs.getUri().toString(), fs.getCanonicalServiceName());
+    assertNull(fs.getCanonicalServiceName());
   }
 }

src/test/org/apache/hadoop/hdfs/MiniDFSCluster.java

@@ -415,6 +415,8 @@ public class MiniDFSCluster {
         NetUtils.addStaticResolution(hosts[i - curDatanodesNum], "localhost");
       }
       DataNode dn = DataNode.instantiateDataNode(dnArgs, dnConf);
+      //NOTE: the following is true if and only if:
+      //      hadoop.security.token.service.use_ip=true
       //since the HDFS does things based on IP:port, we need to add the mapping
       //for IP:port to rackId
       String ipAddr = dn.getSelfAddr().getAddress().getHostAddress();
@@ -842,7 +844,7 @@ public class MiniDFSCluster {
     if (nameNode == null) {
       return;
     }
-    InetSocketAddress addr = new InetSocketAddress("localhost",
+    InetSocketAddress addr = NetUtils.makeSocketAddr("localhost",
                                                    getNameNodePort());
     DFSClient client = new DFSClient(addr, conf);
 

src/test/org/apache/hadoop/hdfs/TestHftpFileSystem.java

@@ -0,0 +1,209 @@
+/**
+ * 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.hdfs;
+
+import static org.junit.Assert.assertEquals;
+
+import java.io.IOException;
+import java.net.URI;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.junit.Before;
+import org.junit.Test;
+
+public class TestHftpFileSystem {
+  
+  @Before
+  public void resetFileSystem() throws IOException {
+    // filesystem caching has a quirk/bug that it caches based on the user's
+    // given uri.  the result is if a filesystem is instantiated with no port,
+    // it gets the default port.  then if the default port is changed,
+    // and another filesystem is instantiated with no port, the prior fs
+    // is returned, not a new one using the changed port.  so let's flush
+    // the cache between tests...
+    FileSystem.closeAll();
+  }
+  
+  @Test
+  public void testHftpDefaultPorts() throws IOException {
+    Configuration conf = new Configuration();
+    URI uri = URI.create("hftp://localhost");
+    HftpFileSystem fs = (HftpFileSystem) FileSystem.get(uri, conf);
+
+    assertEquals(HftpFileSystem.DEFAULT_PORT, fs.getDefaultPort());
+    assertEquals(HftpFileSystem.DEFAULT_SECURE_PORT, fs.getDefaultSecurePort());
+
+    URI fsUri = fs.getUri();
+    assertEquals(uri.getHost(), fsUri.getHost());
+    assertEquals(HftpFileSystem.DEFAULT_PORT, fsUri.getPort());
+    
+    assertEquals(
+        "127.0.0.1:"+HftpFileSystem.DEFAULT_SECURE_PORT,
+        fs.getCanonicalServiceName()
+    );
+  }
+  
+  @Test
+  public void testHftpCustomDefaultPorts() throws IOException {
+    Configuration conf = new Configuration();
+    conf.setInt("dfs.http.port", 123);
+    conf.setInt("dfs.https.port", 456);
+
+    URI uri = URI.create("hftp://localhost");
+    HftpFileSystem fs = (HftpFileSystem) FileSystem.get(uri, conf);
+
+    assertEquals(123, fs.getDefaultPort());
+    assertEquals(456, fs.getDefaultSecurePort());
+    
+    URI fsUri = fs.getUri();
+    assertEquals(uri.getHost(), fsUri.getHost());
+    assertEquals(123, fsUri.getPort());
+    
+    assertEquals(
+        "127.0.0.1:456",
+        fs.getCanonicalServiceName()
+    );
+  }
+
+  @Test
+  public void testHftpCustomUriPortWithDefaultPorts() throws IOException {
+    Configuration conf = new Configuration();
+    URI uri = URI.create("hftp://localhost:123");
+    HftpFileSystem fs = (HftpFileSystem) FileSystem.get(uri, conf);
+
+    assertEquals(HftpFileSystem.DEFAULT_PORT, fs.getDefaultPort());
+    assertEquals(HftpFileSystem.DEFAULT_SECURE_PORT, fs.getDefaultSecurePort());
+
+    URI fsUri = fs.getUri();
+    assertEquals(uri.getHost(), fsUri.getHost());
+    assertEquals(uri.getPort(), fsUri.getPort());
+    
+    assertEquals(
+        "127.0.0.1:"+HftpFileSystem.DEFAULT_SECURE_PORT,
+        fs.getCanonicalServiceName()
+    );
+  }
+
+  @Test
+  public void testHftpCustomUriPortWithCustomDefaultPorts() throws IOException {
+    Configuration conf = new Configuration();
+    conf.setInt("dfs.http.port", 123);
+    conf.setInt("dfs.https.port", 456);
+
+    URI uri = URI.create("hftp://localhost:789");
+    HftpFileSystem fs = (HftpFileSystem) FileSystem.get(uri, conf);
+
+    assertEquals(123, fs.getDefaultPort());
+    assertEquals(456, fs.getDefaultSecurePort());
+    
+    URI fsUri = fs.getUri();
+    assertEquals(uri.getHost(), fsUri.getHost());
+    assertEquals(789, fsUri.getPort());
+    
+    assertEquals(
+        "127.0.0.1:456",
+        fs.getCanonicalServiceName()
+    );
+  }
+
+  ///
+
+  @Test
+  public void testHsftpDefaultPorts() throws IOException {
+    Configuration conf = new Configuration();
+    URI uri = URI.create("hsftp://localhost");
+    HsftpFileSystem fs = (HsftpFileSystem) FileSystem.get(uri, conf);
+
+    assertEquals(HsftpFileSystem.DEFAULT_SECURE_PORT, fs.getDefaultPort());
+    assertEquals(HsftpFileSystem.DEFAULT_SECURE_PORT, fs.getDefaultSecurePort());
+
+    URI fsUri = fs.getUri();
+    assertEquals(uri.getHost(), fsUri.getHost());
+    assertEquals(HsftpFileSystem.DEFAULT_SECURE_PORT, fsUri.getPort());
+    
+    assertEquals(
+        "127.0.0.1:"+HsftpFileSystem.DEFAULT_SECURE_PORT,
+        fs.getCanonicalServiceName()
+    );
+  }
+
+  @Test
+  public void testHsftpCustomDefaultPorts() throws IOException {
+    Configuration conf = new Configuration();
+    conf.setInt("dfs.http.port", 123);
+    conf.setInt("dfs.https.port", 456);
+
+    URI uri = URI.create("hsftp://localhost");
+    HsftpFileSystem fs = (HsftpFileSystem) FileSystem.get(uri, conf);
+
+    assertEquals(456, fs.getDefaultPort());
+    assertEquals(456, fs.getDefaultSecurePort());
+    
+    URI fsUri = fs.getUri();
+    assertEquals(uri.getHost(), fsUri.getHost());
+    assertEquals(456, fsUri.getPort());
+    
+    assertEquals(
+        "127.0.0.1:456",
+        fs.getCanonicalServiceName()
+    );
+  }
+
+  @Test
+  public void testHsftpCustomUriPortWithDefaultPorts() throws IOException {
+    Configuration conf = new Configuration();
+    URI uri = URI.create("hsftp://localhost:123");
+    HsftpFileSystem fs = (HsftpFileSystem) FileSystem.get(uri, conf);
+
+    assertEquals(HsftpFileSystem.DEFAULT_SECURE_PORT, fs.getDefaultPort());
+    assertEquals(HsftpFileSystem.DEFAULT_SECURE_PORT, fs.getDefaultSecurePort());
+
+    URI fsUri = fs.getUri();
+    assertEquals(uri.getHost(), fsUri.getHost());
+    assertEquals(uri.getPort(), fsUri.getPort());
+    
+    assertEquals(
+        "127.0.0.1:123",
+        fs.getCanonicalServiceName()
+    );
+  }
+
+  @Test
+  public void testHsftpCustomUriPortWithCustomDefaultPorts() throws IOException {
+    Configuration conf = new Configuration();
+    conf.setInt("dfs.http.port", 123);
+    conf.setInt("dfs.https.port", 456);
+
+    URI uri = URI.create("hsftp://localhost:789");
+    HsftpFileSystem fs = (HsftpFileSystem) FileSystem.get(uri, conf);
+
+    assertEquals(456, fs.getDefaultPort());
+    assertEquals(456, fs.getDefaultSecurePort());
+    
+    URI fsUri = fs.getUri();
+    assertEquals(uri.getHost(), fsUri.getHost());
+    assertEquals(789, fsUri.getPort());
+    
+    assertEquals(
+        "127.0.0.1:789",
+        fs.getCanonicalServiceName()
+    );
+  }
+}

src/test/org/apache/hadoop/hdfs/server/namenode/TestJspHelper.java

@@ -30,6 +30,7 @@ import org.apache.hadoop.hdfs.DFSConfigKeys;
 import org.apache.hadoop.hdfs.security.token.delegation.DelegationTokenIdentifier;
 import org.apache.hadoop.hdfs.server.namenode.NameNode;
 import org.apache.hadoop.io.Text;
+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;
@@ -81,13 +82,12 @@ public class TestJspHelper {
     UserGroupInformation.setConfiguration(conf);
 
     InetSocketAddress serviceAddr = NameNode.getAddress(conf);
-    Text tokenService = new Text(serviceAddr.getAddress().getHostAddress()
-        + ":" + serviceAddr.getPort());
+    Text tokenService = SecurityUtil.buildTokenService(serviceAddr);
 
     UserGroupInformation ugi = JspHelper.getUGI(request, conf);
     Token<? extends TokenIdentifier> tokenInUgi = ugi.getTokens().iterator()
         .next();
-    Assert.assertEquals(tokenInUgi.getService(), tokenService);
+    Assert.assertEquals(tokenService, tokenInUgi.getService());
   }
   
   @Test

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

@@ -24,9 +24,9 @@ import static org.junit.Assert.*;
 import java.io.DataInput;
 import java.io.DataOutput;
 import java.io.IOException;
+import java.lang.reflect.Method;
 import java.net.InetSocketAddress;
 import java.security.PrivilegedExceptionAction;
-import java.util.Collection;
 import java.util.Set;
 
 import junit.framework.Assert;
@@ -41,10 +41,10 @@ import org.apache.hadoop.net.NetUtils;
 import org.apache.hadoop.security.KerberosInfo;
 import org.apache.hadoop.security.token.SecretManager;
 import org.apache.hadoop.security.token.Token;
-import org.apache.hadoop.security.token.TokenIdentifier;
 import org.apache.hadoop.security.token.TokenInfo;
-import org.apache.hadoop.security.token.TokenSelector;
 import org.apache.hadoop.security.token.SecretManager.InvalidToken;
+import org.apache.hadoop.security.token.delegation.AbstractDelegationTokenIdentifier;
+import org.apache.hadoop.security.token.delegation.AbstractDelegationTokenSelector;
 import org.apache.hadoop.security.SaslInputStream;
 import org.apache.hadoop.security.SaslRpcClient;
 import org.apache.hadoop.security.SaslRpcServer;
@@ -54,11 +54,12 @@ import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.security.UserGroupInformation.AuthenticationMethod;
 
 import org.apache.log4j.Level;
+import org.junit.BeforeClass;
 import org.junit.Test;
 
 /** Unit tests for using Sasl over RPC. */
 public class TestSaslRPC {
-  private static final String ADDRESS = "0.0.0.0";
+  private static final String ADDRESS = "localhost";
 
   public static final Log LOG =
     LogFactory.getLog(TestSaslRPC.class);
@@ -85,7 +86,19 @@ public class TestSaslRPC {
     ((Log4JLogger) SecurityUtil.LOG).getLogger().setLevel(Level.ALL);
   }
 
-  public static class TestTokenIdentifier extends TokenIdentifier {
+  static Method setUseIpMethod;
+  
+  @BeforeClass
+  public static void exposeUseIp() throws Exception {
+    setUseIpMethod = SecurityUtil.class.getDeclaredMethod("setTokenServiceUseIp", boolean.class);
+    setUseIpMethod.setAccessible(true);
+  }
+  
+  private void setTokenServiceUseIp(boolean flag) throws Exception {
+    setUseIpMethod.invoke(setUseIpMethod.getClass(), flag);
+  }
+
+  public static class TestTokenIdentifier extends AbstractDelegationTokenIdentifier {
     private Text tokenid;
     private Text realUser;
     final static Text KIND_NAME = new Text("test.token");
@@ -152,22 +165,10 @@ public class TestSaslRPC {
     }
   }
 
-  public static class TestTokenSelector implements
-      TokenSelector<TestTokenIdentifier> {
-    @SuppressWarnings("unchecked")
-    @Override
-    public Token<TestTokenIdentifier> selectToken(Text service,
-        Collection<Token<? extends TokenIdentifier>> tokens) {
-      if (service == null) {
-        return null;
-      }
-      for (Token<? extends TokenIdentifier> token : tokens) {
-        if (TestTokenIdentifier.KIND_NAME.equals(token.getKind())
-            && service.equals(token.getService())) {
-          return (Token<TestTokenIdentifier>) token;
-        }
-      }
-      return null;
+  public static class TestTokenSelector extends
+  AbstractDelegationTokenSelector<TestTokenIdentifier> {
+    TestTokenSelector() {
+      super(TestTokenIdentifier.KIND_NAME);
     }
   }
   
@@ -354,8 +355,9 @@ public class TestSaslRPC {
     System.out.println("Test is successful.");
   }
   
-  @Test
-  public void testDigestAuthMethod() throws Exception {
+  public void testDigestAuthMethod(boolean useIp) throws Exception {
+    setTokenServiceUseIp(useIp);
+
     TestTokenSecretManager sm = new TestTokenSecretManager();
     Server server = RPC.getServer(
         new TestSaslImpl(), ADDRESS, 0, 5, true, conf, sm);
@@ -369,6 +371,19 @@ public class TestSaslRPC {
         sm);
     SecurityUtil.setTokenService(token, addr);
     LOG.info("Service IP address for token is " + token.getService());
+
+    InetSocketAddress tokenAddr = SecurityUtil.getTokenServiceAddr(token);
+    String expectedHost, gotHost;
+    if (useIp) {
+      expectedHost = addr.getAddress().getHostAddress();
+      gotHost = tokenAddr.getAddress().getHostAddress();
+    } else {
+      gotHost = tokenAddr.getHostName();
+      expectedHost = ADDRESS;
+    }
+    Assert.assertEquals(expectedHost, gotHost);
+    Assert.assertEquals(expectedHost+":"+addr.getPort(), token.getService().toString());
+    
     current.addToken(token);
 
     current.doAs(new PrivilegedExceptionAction<Object>() {
@@ -388,7 +403,17 @@ public class TestSaslRPC {
     });
     server.stop();
   }
+
+  @Test
+  public void testDigestAuthMethodIpBasedToken() throws Exception {
+    testDigestAuthMethod(true);
+  }
   
+  @Test
+  public void testDigestAuthMethodHostBasedToken() throws Exception {
+    testDigestAuthMethod(false);
+  }
+
   public static void main(String[] args) throws Exception {
     System.out.println("Testing Kerberos authentication over RPC");
     if (args.length != 2) {

src/test/org/apache/hadoop/mapreduce/security/TestTokenCache.java

@@ -53,7 +53,6 @@ import org.apache.hadoop.mapred.OutputCollector;
 import org.apache.hadoop.mapred.Reporter;
 import org.apache.hadoop.mapreduce.JobContext;
 import org.apache.hadoop.security.Credentials;
-import org.apache.hadoop.security.SecurityUtil;
 import org.apache.hadoop.security.token.Token;
 import org.apache.hadoop.security.token.TokenIdentifier;
 import org.apache.hadoop.tools.HadoopArchives;
@@ -266,9 +265,8 @@ public class TestTokenCache {
     Credentials credentials = new Credentials();
     TokenCache.obtainTokensForNamenodesInternal(credentials, new Path [] {p1, p2},
                                         jConf);
-    // this token is keyed by hostname:port key.
-    String fs_addr = 
-      SecurityUtil.buildDTServiceName(p1.toUri(), NameNode.DEFAULT_PORT); 
+    // this filesystem's token is keyed by the canonical service
+    String fs_addr = fs.getCanonicalServiceName();
     Token<DelegationTokenIdentifier> nnt =
       TokenCache.getDelegationToken(credentials, fs_addr);
 

src/test/org/apache/hadoop/net/TestNetUtils.java

@@ -17,15 +17,25 @@
  */
 package org.apache.hadoop.net;
 
+import org.junit.Before;
+import org.junit.BeforeClass;
 import org.junit.Test;
 import static org.junit.Assert.*;
 
+import java.net.InetAddress;
 import java.net.Socket;
 import java.net.ConnectException;
 import java.net.InetSocketAddress;
 import java.net.SocketException;
+import java.net.UnknownHostException;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
 
+import org.apache.commons.lang.StringUtils;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.net.NetUtils.QualifiedHostResolver;
 
 public class TestNetUtils {
 
@@ -59,4 +69,216 @@ public class TestNetUtils {
       assertTrue(se.getMessage().contains("Invalid argument"));
     }
   }
-}
+
+  static TestNetUtilsResolver resolver;
+  
+  @BeforeClass
+  public static void setupResolver() {
+    resolver = new TestNetUtilsResolver();
+    resolver.setSearchDomains("a.b", "b", "c");
+    resolver.addResolvedHost("host.a.b.", "1.1.1.1");
+    resolver.addResolvedHost("b-host.b.", "2.2.2.2");
+    resolver.addResolvedHost("simple.", "3.3.3.3");    
+    NetUtils.setHostResolver(resolver);
+  }
+  
+  @Before
+  public void resetResolver() {
+    resolver.reset();
+  }
+
+  // getByExactName
+  
+  private void verifyGetByExactNameSearch(String host, String ... searches) {
+    assertNull(resolver.getByExactName(host));
+    assertBetterArrayEquals(searches, resolver.getHostSearches());
+  }
+
+  @Test
+  public void testResolverGetByExactNameUnqualified() {
+    verifyGetByExactNameSearch("unknown", "unknown.");
+  }
+
+  @Test
+  public void testResolverGetByExactNameUnqualifiedWithDomain() {
+    verifyGetByExactNameSearch("unknown.domain", "unknown.domain.");
+  }
+
+  @Test
+  public void testResolverGetByExactNameQualified() {
+    verifyGetByExactNameSearch("unknown.", "unknown.");
+  }
+  
+  @Test
+  public void testResolverGetByExactNameQualifiedWithDomain() {
+    verifyGetByExactNameSearch("unknown.domain.", "unknown.domain.");
+  }
+
+  // getByNameWithSearch
+  
+  private void verifyGetByNameWithSearch(String host, String ... searches) {
+    assertNull(resolver.getByNameWithSearch(host));
+    assertBetterArrayEquals(searches, resolver.getHostSearches());
+  }
+
+  @Test
+  public void testResolverGetByNameWithSearchUnqualified() {
+    String host = "unknown";
+    verifyGetByNameWithSearch(host, host+".a.b.", host+".b.", host+".c.");
+  }
+
+  @Test
+  public void testResolverGetByNameWithSearchUnqualifiedWithDomain() {
+    String host = "unknown.domain";
+    verifyGetByNameWithSearch(host, host+".a.b.", host+".b.", host+".c.");
+  }
+
+  @Test
+  public void testResolverGetByNameWithSearchQualified() {
+    String host = "unknown.";
+    verifyGetByNameWithSearch(host, host);
+  }
+
+  @Test
+  public void testResolverGetByNameWithSearchQualifiedWithDomain() {
+    String host = "unknown.domain.";
+    verifyGetByNameWithSearch(host, host);
+  }
+
+  // getByName
+
+  private void verifyGetByName(String host, String ... searches) {
+    InetAddress addr = null;
+    try {
+      addr = resolver.getByName(host);
+    } catch (UnknownHostException e) {} // ignore
+    assertNull(addr);
+    assertBetterArrayEquals(searches, resolver.getHostSearches());
+  }
+  
+  @Test
+  public void testResolverGetByNameQualified() {
+    String host = "unknown.";
+    verifyGetByName(host, host);
+  }
+
+  @Test
+  public void testResolverGetByNameQualifiedWithDomain() {
+    verifyGetByName("unknown.domain.", "unknown.domain.");
+  }
+
+  @Test
+  public void testResolverGetByNameUnqualified() {
+    String host = "unknown";
+    verifyGetByName(host, host+".a.b.", host+".b.", host+".c.", host+".");
+  }
+
+  @Test
+  public void testResolverGetByNameUnqualifiedWithDomain() {
+    String host = "unknown.domain";
+    verifyGetByName(host, host+".", host+".a.b.", host+".b.", host+".c.");
+  }
+  
+  // resolving of hosts
+
+  private InetAddress verifyResolve(String host, String ... searches) {
+    InetAddress addr = null;
+    try {
+      addr = resolver.getByName(host);
+    } catch (UnknownHostException e) {} // ignore
+    assertNotNull(addr);
+    assertBetterArrayEquals(searches, resolver.getHostSearches());
+    return addr;
+  }
+
+  private void
+  verifyInetAddress(InetAddress addr, String host, String ip) {
+    assertNotNull(addr);
+    assertEquals(host, addr.getHostName());
+    assertEquals(ip, addr.getHostAddress());
+  }
+  
+  @Test
+  public void testResolverUnqualified() {
+    String host = "host";
+    InetAddress addr = verifyResolve(host, host+".a.b.");
+    verifyInetAddress(addr, "host.a.b", "1.1.1.1");
+  }
+
+  @Test
+  public void testResolverUnqualifiedWithDomain() {
+    String host = "host.a";
+    InetAddress addr = verifyResolve(host, host+".", host+".a.b.", host+".b.");
+    verifyInetAddress(addr, "host.a.b", "1.1.1.1");
+  }
+
+  @Test
+  public void testResolverUnqualifedFull() {
+    String host = "host.a.b";
+    InetAddress addr = verifyResolve(host, host+".");
+    verifyInetAddress(addr, host, "1.1.1.1");
+  }
+  
+  @Test
+  public void testResolverQualifed() {
+    String host = "host.a.b.";
+    InetAddress addr = verifyResolve(host, host);
+    verifyInetAddress(addr, host, "1.1.1.1");
+  }
+  
+  // localhost
+  
+  @Test
+  public void testResolverLoopback() {
+    String host = "Localhost";
+    InetAddress addr = verifyResolve(host); // no lookup should occur
+    verifyInetAddress(addr, "Localhost", "127.0.0.1");
+  }
+
+  @Test
+  public void testResolverIP() {
+    String host = "1.1.1.1";
+    InetAddress addr = verifyResolve(host); // no lookup should occur for ips
+    verifyInetAddress(addr, host, host);
+  }
+
+  //
+
+  static class TestNetUtilsResolver extends QualifiedHostResolver {
+    Map<String, InetAddress> resolvedHosts = new HashMap<String, InetAddress>();
+    List<String> hostSearches = new LinkedList<String>();
+    
+    void addResolvedHost(String host, String ip) {
+      InetAddress addr;
+      try {
+        addr = InetAddress.getByName(ip);
+        addr = InetAddress.getByAddress(host, addr.getAddress());
+      } catch (UnknownHostException e) {
+        throw new IllegalArgumentException("not an ip:"+ip);
+      }
+      resolvedHosts.put(host, addr);
+    }
+    
+    InetAddress getInetAddressByName(String host) throws UnknownHostException {
+      hostSearches.add(host);
+      if (!resolvedHosts.containsKey(host)) {
+        throw new UnknownHostException(host);
+      }
+      return resolvedHosts.get(host);
+    }
+
+    String[] getHostSearches() {
+      return hostSearches.toArray(new String[0]);
+    }
+
+    void reset() {
+      hostSearches.clear();
+    }
+  }
+  
+  private <T> void assertBetterArrayEquals(T[] expect, T[]got) {
+    String expectStr = StringUtils.join(expect, ", ");
+    String gotStr = StringUtils.join(got, ", ");
+    assertEquals(expectStr, gotStr);
+  }
+}

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

@@ -20,11 +20,20 @@ import static org.junit.Assert.*;
 
 import java.io.IOException;
 import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.URI;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.security.token.Token;
 import org.junit.Test;
 import org.mockito.Mockito;
 
 public class TestSecurityUtil {
+  public static final Log LOG = LogFactory.getLog(TestSecurityUtil.class);
+
   @Test
   public void isOriginalTGTReturnsCorrectValues() {
     assertTrue(SecurityUtil.isOriginalTGT("krbtgt/foo@foo"));
@@ -90,4 +99,213 @@ public class TestSecurityUtil {
     assertEquals(null,
         SecurityUtil.getHostFromPrincipal("service@realm"));
   }
+
+  @Test
+  public void testBuildDTServiceName() {
+    assertEquals("127.0.0.1:123",
+        SecurityUtil.buildDTServiceName(URI.create("test://LocalHost"), 123)
+    );
+    assertEquals("127.0.0.1:123",
+        SecurityUtil.buildDTServiceName(URI.create("test://LocalHost:123"), 456)
+    );
+    assertEquals("127.0.0.1:123",
+        SecurityUtil.buildDTServiceName(URI.create("test://127.0.0.1"), 123)
+    );
+    assertEquals("127.0.0.1:123",
+        SecurityUtil.buildDTServiceName(URI.create("test://127.0.0.1:123"), 456)
+    );
+  }
+  
+  @Test
+  public void testBuildTokenServiceSockAddr() {
+    assertEquals("127.0.0.1:123",
+        SecurityUtil.buildTokenService(new InetSocketAddress("LocalHost", 123)).toString()
+    );
+    assertEquals("127.0.0.1:123",
+        SecurityUtil.buildTokenService(new InetSocketAddress("127.0.0.1", 123)).toString()
+    );
+    // what goes in, comes out
+    assertEquals("127.0.0.1:123",
+        SecurityUtil.buildTokenService(NetUtils.createSocketAddr("127.0.0.1", 123)).toString()
+    );
+  }
+
+  @Test
+  public void testGoodHostsAndPorts() {
+    InetSocketAddress compare = NetUtils.makeSocketAddr("localhost", 123);
+    runGoodCases(compare, "localhost", 123);
+    runGoodCases(compare, "localhost:", 123);
+    runGoodCases(compare, "localhost:123", 456);
+  }
+  
+  void runGoodCases(InetSocketAddress addr, String host, int port) {
+    assertEquals(addr, NetUtils.createSocketAddr(host, port));
+    assertEquals(addr, NetUtils.createSocketAddr("hdfs://"+host, port));
+    assertEquals(addr, NetUtils.createSocketAddr("hdfs://"+host+"/path", port));
+  }
+  
+  @Test
+  public void testBadHostsAndPorts() {
+    runBadCases("", true);
+    runBadCases(":", false);
+    runBadCases("hdfs/", false);
+    runBadCases("hdfs:/", false);
+    runBadCases("hdfs://", true);
+  }
+  
+  void runBadCases(String prefix, boolean validIfPosPort) {
+    runBadPortPermutes(prefix, false);
+    runBadPortPermutes(prefix+"*", false);
+    runBadPortPermutes(prefix+"localhost", validIfPosPort);
+    runBadPortPermutes(prefix+"localhost:-1", false);
+    runBadPortPermutes(prefix+"localhost:-123", false);
+    runBadPortPermutes(prefix+"localhost:xyz", false);
+    runBadPortPermutes(prefix+"localhost/xyz", validIfPosPort);
+    runBadPortPermutes(prefix+"localhost/:123", validIfPosPort);
+    runBadPortPermutes(prefix+":123", false);
+    runBadPortPermutes(prefix+":xyz", false);
+  }
+
+  void runBadPortPermutes(String arg, boolean validIfPosPort) {
+    int ports[] = { -123, -1, 123 };
+    boolean bad = false;
+    try {
+      NetUtils.createSocketAddr(arg);
+    } catch (IllegalArgumentException e) {
+      bad = true;
+    } finally {
+      assertTrue("should be bad: '"+arg+"'", bad);
+    }
+    for (int port : ports) {
+      if (validIfPosPort && port > 0) continue;
+      
+      bad = false;
+      try {
+        NetUtils.createSocketAddr(arg, port);
+      } catch (IllegalArgumentException e) {
+        bad = true;
+      } finally {
+        assertTrue("should be bad: '"+arg+"' (default port:"+port+")", bad);
+      }
+    }
+  }
+
+  // check that the socket addr has:
+  // 1) the InetSocketAddress has the correct hostname, ie. exact host/ip given
+  // 2) the address is resolved, ie. has an ip
+  // 3,4) the socket's InetAddress has the same hostname, and the correct ip
+  // 5) the port is correct
+  private void
+  verifyValues(InetSocketAddress addr, String host, String ip, int port) {
+    assertTrue(!addr.isUnresolved());
+    // don't know what the standard resolver will return for hostname.
+    // should be host for host; host or ip for ip is ambiguous
+    if (!SecurityUtil.getTokenServiceUseIp()) {
+      assertEquals(host, addr.getHostName());
+      assertEquals(host, addr.getAddress().getHostName());
+    }
+    assertEquals(ip, addr.getAddress().getHostAddress());
+    assertEquals(port, addr.getPort());    
+  }
+
+  // check:
+  // 1) buildTokenService honors use_ip setting
+  // 2) setTokenService & getService works
+  // 3) getTokenServiceAddr decodes to the identical socket addr
+  private void
+  verifyTokenService(InetSocketAddress addr, String host, String ip, int port, boolean useIp) {
+    LOG.info("address:"+addr+" host:"+host+" ip:"+ip+" port:"+port);
+
+    SecurityUtil.setTokenServiceUseIp(useIp);
+    String serviceHost = useIp ? ip : host.toLowerCase();
+    
+    Token token = new Token();
+    Text service = new Text(serviceHost+":"+port);
+    
+    assertEquals(service, SecurityUtil.buildTokenService(addr));
+    SecurityUtil.setTokenService(token, addr);
+    assertEquals(service, token.getService());
+    
+    InetSocketAddress serviceAddr = SecurityUtil.getTokenServiceAddr(token);
+    assertNotNull(serviceAddr);
+    verifyValues(serviceAddr, serviceHost, ip, port);
+  }
+
+  // check:
+  // 1) socket addr is created with fields set as expected
+  // 2) token service with ips
+  // 3) token service with the given host or ip
+  private void
+  verifyAddress(InetSocketAddress addr, String host, String ip, int port) {
+    verifyValues(addr, host, ip, port);
+    LOG.info("test that token service uses ip");
+    verifyTokenService(addr, host, ip, port, true);    
+    LOG.info("test that token service uses host");
+    verifyTokenService(addr, host, ip, port, false);
+  }
+
+  // check:
+  // 1-4) combinations of host and port
+  // this will construct a socket addr, verify all the fields, build the
+  // service to verify the use_ip setting is honored, set the token service
+  // based on addr and verify the token service is set correctly, decode
+  // the token service and ensure all the fields of the decoded addr match
+  private void verifyServiceAddr(String host, String ip) {
+    InetSocketAddress addr;
+    int port = 123;
+
+    // test host, port tuple
+    LOG.info("test tuple ("+host+","+port+")");
+    addr = NetUtils.makeSocketAddr(host, port);
+    verifyAddress(addr, host, ip, port);
+
+    // test authority with no default port
+    LOG.info("test authority '"+host+":"+port+"'");
+    addr = NetUtils.createSocketAddr(host+":"+port);
+    verifyAddress(addr, host, ip, port);
+
+    // test authority with a default port, make sure default isn't used
+    LOG.info("test authority '"+host+":"+port+"' with ignored default port");
+    addr = NetUtils.createSocketAddr(host+":"+port, port+1);
+    verifyAddress(addr, host, ip, port);
+
+    // test host-only authority, using port as default port
+    LOG.info("test host:"+host+" port:"+port);
+    addr = NetUtils.createSocketAddr(host, port);
+    verifyAddress(addr, host, ip, port);
+  }
+
+  @Test
+  public void testSocketAddrWithName() {
+    String staticHost = "my";
+    NetUtils.addStaticResolution(staticHost, "localhost");
+    verifyServiceAddr("LocalHost", "127.0.0.1");
+  }
+
+  @Test
+  public void testSocketAddrWithIP() {
+    verifyServiceAddr("127.0.0.1", "127.0.0.1");
+  }
+
+  @Test
+  public void testSocketAddrWithNameToStaticName() {
+    String staticHost = "host1";
+    NetUtils.addStaticResolution(staticHost, "localhost");
+    verifyServiceAddr(staticHost, "127.0.0.1");
+  }
+
+  @Test
+  public void testSocketAddrWithNameToStaticIP() {
+    String staticHost = "host3";
+    NetUtils.addStaticResolution(staticHost, "255.255.255.255");
+    verifyServiceAddr(staticHost, "255.255.255.255");
+  }
+
+  // this is a bizarre case, but it's if a test tries to remap an ip address
+  @Test
+  public void testSocketAddrWithIPToStaticIP() {
+    String staticHost = "1.2.3.4";
+    NetUtils.addStaticResolution(staticHost, "255.255.255.255");
+    verifyServiceAddr(staticHost, "255.255.255.255");
+  }
 }