HDFS-1150. Verify datanodes' identities to clients in secure clusters. (jghoman)

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

.eclipse.templates/.classpath

@@ -15,6 +15,7 @@
 	<classpathentry kind="lib" path="build/ivy/lib/Hadoop-Hdfs/common/commons-cli-1.2.jar"/>
 	<classpathentry kind="lib" path="build/ivy/lib/Hadoop-Hdfs/common/commons-codec-1.4.jar"/>
 	<classpathentry kind="lib" path="build/ivy/lib/Hadoop-Hdfs/common/commons-el-1.0.jar"/>
+  <classpathentry kind="lib" path="build/ivy/lib/Hadoop-Hdfs/common/commons-daemon-1.0.1.jar" />
 	<classpathentry kind="lib" path="build/ivy/lib/Hadoop-Hdfs/common/commons-httpclient-3.1.jar"/>
 	<classpathentry kind="lib" path="build/ivy/lib/Hadoop-Hdfs/common/commons-logging-1.1.1.jar"/>
 	<classpathentry kind="lib" path="build/ivy/lib/Hadoop-Hdfs/common/commons-net-1.4.1.jar"/>

CHANGES.txt

@@ -24,6 +24,9 @@ Trunk (unreleased changes)
     HDFS-1023. Allow http server to start as regular principal if https 
     principal not defined. (jghoman)
 
+    HDFS-1150. Verify datanodes' identities to clients in secure clusters.
+    (jghoman)
+
   IMPROVEMENTS
 
     HDFS-1096. fix for prev. commit. (boryas)

bin/hdfs

@@ -46,6 +46,20 @@ fi
 COMMAND=$1
 shift
 
+# Determine if we're starting a secure datanode, and if so, redefine appropriate variables
+if [ "$COMMAND" == "datanode" ] && [ "$EUID" -eq 0 ] && [ -n "$HADOOP_SECURE_DN_USER" ]; then
+  if [ -n "$HADOOP_SECURE_DN_PID_DIR" ]; then
+    HADOOP_PID_DIR=$HADOOP_SECURE_DN_PID_DIR
+  fi
+
+  if [ -n "$HADOOP_SECURE_DN_LOG_DIR" ]; then
+    HADOOP_LOG_DIR=$HADOOP_SECURE_DN_LOG_DIR
+  fi
+ 
+  HADOOP_IDENT_STRING=$HADOOP_SECURE_DN_USER
+  starting_secure_dn="true"
+fi
+
 if [ "$COMMAND" = "namenode" ] ; then
   CLASS='org.apache.hadoop.hdfs.server.namenode.NameNode'
   HADOOP_OPTS="$HADOOP_OPTS $HADOOP_NAMENODE_OPTS"
@@ -54,7 +68,11 @@ elif [ "$COMMAND" = "secondarynamenode" ] ; then
   HADOOP_OPTS="$HADOOP_OPTS $HADOOP_SECONDARYNAMENODE_OPTS"
 elif [ "$COMMAND" = "datanode" ] ; then
   CLASS='org.apache.hadoop.hdfs.server.datanode.DataNode'
-  HADOOP_OPTS="$HADOOP_OPTS $HADOOP_DATANODE_OPTS"
+  if [[ $EUID -eq 0 ]]; then
+    HADOOP_OPTS="$HADOOP_OPTS -jvm server $HADOOP_DATANODE_OPTS"
+  else
+    HADOOP_OPTS="$HADOOP_OPTS -server $HADOOP_DATANODE_OPTS"
+  fi
 elif [ "$COMMAND" = "dfs" ] ; then
   CLASS=org.apache.hadoop.fs.FsShell
   HADOOP_OPTS="$HADOOP_OPTS $HADOOP_CLIENT_OPTS"
@@ -110,4 +128,26 @@ if $cygwin; then
   CLASSPATH=`cygpath -p -w "$CLASSPATH"`
 fi
 export CLASSPATH=$CLASSPATH
-exec "$JAVA" $JAVA_HEAP_MAX $HADOOP_OPTS $CLASS "$@"
+
+# Check to see if we should start a secure datanode
+if [ "$starting_secure_dn" = "true" ]; then
+  if [ "$HADOOP_PID_DIR" = "" ]; then
+    HADOOP_SECURE_DN_PID="/tmp/hadoop_secure_dn.pid"
+  else
+   HADOOP_SECURE_DN_PID="$HADOOP_PID_DIR/hadoop_secure_dn.pid"
+  fi
+
+  exec "$HADOOP_HOME/bin/jsvc" \
+           -Dproc_$COMMAND -outfile "$HADOOP_LOG_DIR/jsvc.out" \
+           -errfile "$HADOOP_LOG_DIR/jsvc.err" \
+           -pidfile "$HADOOP_SECURE_DN_PID" \
+           -nodetach \
+           -user "$HADOOP_SECURE_DN_USER" \
+            -cp "$CLASSPATH" \
+           $JAVA_HEAP_MAX $HADOOP_OPTS \
+           org.apache.hadoop.hdfs.server.datanode.SecureDataNodeStarter "$@"
+else
+  # run it
+  exec "$JAVA" -Dproc_$COMMAND $JAVA_HEAP_MAX $HADOOP_OPTS $CLASS "$@"
+fi
+

bin/start-dfs.sh

@@ -48,5 +48,10 @@ fi
 # start namenode after datanodes, to minimize time namenode is up w/o data
 # note: datanodes will log connection errors until namenode starts
 "$HADOOP_COMMON_HOME"/bin/hadoop-daemon.sh --config $HADOOP_CONF_DIR --script "$bin"/hdfs start namenode $nameStartOpt
+#
+if [ -n "$HADOOP_SECURE_DN_USER" ]; then
+  echo "Attempting to start secure cluster, skipping datanodes. Run start-secure-dns.sh as root to complete startup."
+else
 "$HADOOP_COMMON_HOME"/bin/hadoop-daemons.sh --config $HADOOP_CONF_DIR --script "$bin"/hdfs start datanode $dataStartOpt
+fi
 "$HADOOP_COMMON_HOME"/bin/hadoop-daemons.sh --config $HADOOP_CONF_DIR --hosts masters --script "$bin"/hdfs start secondarynamenode

bin/start-secure-dns.sh

@@ -0,0 +1,31 @@
+#!/usr/bin/env bash
+
+# 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.
+
+# Run as root to start secure datanodes in a security-enabled cluster.
+
+usage="Usage (run as root in order to start secure datanodes): start-secure-dns.sh"
+
+bin=`dirname "${BASH_SOURCE-$0}"`
+bin=`cd "$bin"; pwd`
+
+. "$bin"/hdfs-config.sh
+
+if [ "$EUID" -eq 0 ] && [ -n "$HADOOP_SECURE_DN_USER" ]; then
+  "$HADOOP_COMMON_HOME"/bin/hadoop-daemons.sh --config $HADOOP_CONF_DIR --script "$bin"/hdfs start datanode $dataStartOpt
+else
+  echo $usage
+fi

bin/stop-dfs.sh

@@ -24,5 +24,9 @@ bin=`cd "$bin"; pwd`
 . "$bin"/hdfs-config.sh
 
 "$HADOOP_COMMON_HOME"/bin/hadoop-daemon.sh --config $HADOOP_CONF_DIR --script "$bin"/hdfs stop namenode
-"$HADOOP_COMMON_HOME"/bin/hadoop-daemons.sh --config $HADOOP_CONF_DIR --script "$bin"/hdfs stop datanode
+if [ -n "$HADOOP_SECURE_DN_USER" ]; then
+  echo "Attempting to stop secure cluster, skipping datanodes. Run stop-secure-dns.sh as root to complete shutdown."
+else
+  "$HADOOP_COMMON_HOME"/bin/hadoop-daemons.sh --config $HADOOP_CONF_DIR --script "$bin"/hdfs stop datanode
+fi
 "$HADOOP_COMMON_HOME"/bin/hadoop-daemons.sh --config $HADOOP_CONF_DIR --hosts masters --script "$bin"/hdfs stop secondarynamenode

bin/stop-secure-dns.sh

@@ -0,0 +1,31 @@
+#!/usr/bin/env bash
+
+# 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.
+
+# Run as root to start secure datanodes in a security-enabled cluster.
+
+usage="Usage (run as root in order to stop secure datanodes): stop-secure-dns.sh"
+
+bin=`dirname "${BASH_SOURCE-$0}"`
+bin=`cd "$bin"; pwd`
+
+. "$bin"/hdfs-config.sh
+
+if [ "$EUID" -eq 0 ] && [ -n "$HADOOP_SECURE_DN_USER" ]; then
+  "$HADOOP_COMMON_HOME"/bin/hadoop-daemons.sh --config $HADOOP_CONF_DIR --script "$bin"/hdfs stop datanode
+else
+  echo $usage
+fi

build.xml

@@ -139,6 +139,12 @@
   <property name="patch.cmd" value="patch"/>
   <property name="make.cmd" value="make"/>
 	
+	<!-- jsvc properties set here -->
+  <property name="jsvc.build.dir" value="${build.dir}/jsvc" />
+  <property name="jsvc.install.dir" value="${dist.dir}/bin" /> 
+  <property name="jsvc.location" value="http://apache.org/dist/commons/daemon/binaries/1.0.2/linux/commons-daemon-1.0.2-bin-linux-i386.tar.gz" />
+  <property name="jsvc.dest.name" value="jsvc.tar.gz" />
+	
   <!-- IVY properties set here -->
   <property name="ivy.dir" location="ivy" />
   <loadproperties srcfile="${ivy.dir}/libraries.properties"/>
@@ -919,7 +925,7 @@
   <!-- ================================================================== -->
   <!--                                                                    -->
   <!-- ================================================================== -->
-  <target name="package" depends="compile, jar, javadoc, docs, api-report, jar-test, ant-tasks"
+  <target name="package" depends="compile, jar, javadoc, docs, api-report, jar-test, ant-tasks, jsvc"
 	  description="Build distribution">
     <mkdir dir="${dist.dir}"/>
     <mkdir dir="${dist.dir}/lib"/>
@@ -1017,7 +1023,7 @@
     </macro_tar>
   </target>
 
-  <target name="bin-package" depends="compile, jar, jar-test, ant-tasks" 
+  <target name="bin-package" depends="compile, jar, jar-test, ant-tasks, jsvc" 
 		description="assembles artifacts for binary target">
     <mkdir dir="${dist.dir}"/>
     <mkdir dir="${dist.dir}/lib"/>
@@ -1621,4 +1627,15 @@
     </echo>
   </target>
 
+	<target name="jsvc" >
+    <mkdir dir="${jsvc.build.dir}" />
+    <get src="${jsvc.location}" dest="${jsvc.build.dir}/${jsvc.dest.name}" />
+
+    <untar compression="gzip" src="${jsvc.build.dir}/${jsvc.dest.name}" dest="${jsvc.build.dir}" />
+
+    <copy file="${jsvc.build.dir}/jsvc" todir="${jsvc.install.dir}" verbose="true" />
+    <chmod perm="ugo+x" type="file">
+      <fileset file="${jsvc.install.dir}/jsvc"/>
+    </chmod>
+ </target>
 </project>

ivy.xml

@@ -59,6 +59,7 @@
     <dependency org="org.apache.hadoop" name="hadoop-common" rev="${hadoop-common.version}" conf="common->default"/>
     <dependency org="org.apache.hadoop" name="hadoop-common-instrumented" rev="${hadoop-common.version}" conf="system->default"/>
     <dependency org="commons-logging" name="commons-logging" rev="${commons-logging.version}" conf="common->master"/>
+    <dependency org="commons-daemon" name="commons-daemon" rev="${commons-daemon.version}" conf="common->default" />
     <dependency org="log4j" name="log4j" rev="${log4j.version}" conf="common->master"/>
     <dependency org="org.apache.hadoop" name="avro" rev="${avro.version}" conf="common->default">
       <exclude module="ant"/>

ivy/libraries.properties

@@ -22,6 +22,8 @@ checkstyle.version=4.2
 commons-cli.version=1.2
 commons-cli2.version=2.0-mahout
 commons-collections.version=3.1
+commons-daemon.version=1.0.1
+commons-httpclient.version=3.0.1
 commons-lang.version=2.5
 commons-logging.version=1.1.1
 commons-logging-api.version=1.1

src/contrib/hdfsproxy/src/java/org/apache/hadoop/hdfsproxy/ProxyHttpServer.java

@@ -46,7 +46,7 @@ public class ProxyHttpServer extends HttpServer {
   }
 
   /** {@inheritDoc} */
-  protected Connector createBaseListener(Configuration conf)
+  public Connector createBaseListener(Configuration conf)
       throws IOException {
     final String sAddr;
     if (null == (sAddr = conf.get("proxy.http.test.listener.addr"))) {

src/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java

@@ -18,6 +18,8 @@
 package org.apache.hadoop.hdfs.server.datanode;
 
 
+import static org.apache.hadoop.hdfs.server.common.Util.now;
+
 import java.io.BufferedOutputStream;
 import java.io.DataOutputStream;
 import java.io.File;
@@ -54,7 +56,9 @@ import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.LocalFileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.hdfs.DFSConfigKeys;
 import org.apache.hadoop.hdfs.HDFSPolicyProvider;
+import org.apache.hadoop.hdfs.HdfsConfiguration;
 import org.apache.hadoop.hdfs.protocol.Block;
 import org.apache.hadoop.hdfs.protocol.BlockListAsLongs;
 import org.apache.hadoop.hdfs.protocol.ClientDatanodeProtocol;
@@ -74,9 +78,9 @@ import org.apache.hadoop.hdfs.server.common.IncorrectVersionException;
 import org.apache.hadoop.hdfs.server.common.JspHelper;
 import org.apache.hadoop.hdfs.server.common.Storage;
 import org.apache.hadoop.hdfs.server.common.Util;
-import static org.apache.hadoop.hdfs.server.common.Util.now;
 import org.apache.hadoop.hdfs.server.common.HdfsConstants.ReplicaState;
 import org.apache.hadoop.hdfs.server.common.HdfsConstants.StartupOption;
+import org.apache.hadoop.hdfs.server.datanode.SecureDataNodeStarter.SecureResources;
 import org.apache.hadoop.hdfs.server.datanode.metrics.DataNodeMetrics;
 import org.apache.hadoop.hdfs.server.namenode.FSNamesystem;
 import org.apache.hadoop.hdfs.server.namenode.FileChecksumServlets;
@@ -94,11 +98,6 @@ import org.apache.hadoop.hdfs.server.protocol.NamespaceInfo;
 import org.apache.hadoop.hdfs.server.protocol.ReplicaRecoveryInfo;
 import org.apache.hadoop.hdfs.server.protocol.UpgradeCommand;
 import org.apache.hadoop.hdfs.server.protocol.BlockRecoveryCommand.RecoveringBlock;
-import org.apache.hadoop.hdfs.DFSConfigKeys;
-import org.apache.hadoop.hdfs.DFSUtil;
-import org.apache.hadoop.hdfs.HdfsConfiguration;
-import org.apache.hadoop.security.token.Token;
-import org.apache.hadoop.security.token.TokenIdentifier;
 import org.apache.hadoop.http.HttpServer;
 import org.apache.hadoop.io.IOUtils;
 import org.apache.hadoop.ipc.RPC;
@@ -109,6 +108,8 @@ import org.apache.hadoop.net.NetUtils;
 import org.apache.hadoop.security.SecurityUtil;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.security.authorize.ServiceAuthorizationManager;
+import org.apache.hadoop.security.token.Token;
+import org.apache.hadoop.security.token.TokenIdentifier;
 import org.apache.hadoop.util.Daemon;
 import org.apache.hadoop.util.DiskChecker;
 import org.apache.hadoop.util.GenericOptionsParser;
@@ -228,16 +229,28 @@ public class DataNode extends Configured
   // For InterDataNodeProtocol
   public Server ipcServer;
 
+  private SecureResources secureResources = null;  
+  
   /**
    * Create the DataNode given a configuration and an array of dataDirs.
    * 'dataDirs' is where the blocks are stored.
    */
   DataNode(final Configuration conf, 
            final AbstractList<File> dataDirs) throws IOException {
+    this(conf, dataDirs, null);
+  }
+  
+  /**
+   * Start a Datanode with specified server sockets for secure environments
+   * where they are run with privileged ports and injected from a higher
+   * level of capability
+   */
+  DataNode(final Configuration conf,
+           final AbstractList<File> dataDirs, final SecureResources resources) throws IOException {  
     this(conf, dataDirs, (DatanodeProtocol)RPC.waitForProxy(DatanodeProtocol.class,
                        DatanodeProtocol.versionID,
                        NameNode.getServiceAddress(conf, true), 
-                       conf));
+                       conf), resources);
   }
   
   /**
@@ -246,13 +259,13 @@ public class DataNode extends Configured
    */
   DataNode(final Configuration conf, 
            final AbstractList<File> dataDirs,
-           final DatanodeProtocol namenode) throws IOException {
+           final DatanodeProtocol namenode, final SecureResources resources) throws IOException {
     super(conf);
 
     DataNode.setDataNode(this);
     
     try {
-      startDataNode(conf, dataDirs, namenode);
+      startDataNode(conf, dataDirs, namenode, resources);
     } catch (IOException ie) {
       shutdown();
      throw ie;
@@ -271,8 +284,14 @@ public class DataNode extends Configured
    */
   void startDataNode(Configuration conf, 
                      AbstractList<File> dataDirs,
-                     DatanodeProtocol namenode
+                     DatanodeProtocol namenode, SecureResources resources
                      ) throws IOException {
+    if(UserGroupInformation.isSecurityEnabled() && resources == null)
+      throw new RuntimeException("Cannot start secure cluster without " +
+      "privileged resources.");
+
+    this.secureResources = resources;
+    
     // use configured nameserver & interface to get local hostname
     if (conf.get(DFSConfigKeys.DFS_DATANODE_HOST_NAME_KEY) != null) {
       machineName = conf.get(DFSConfigKeys.DFS_DATANODE_HOST_NAME_KEY);   
@@ -294,8 +313,7 @@ public class DataNode extends Configured
                                              true);
     this.writePacketSize = conf.getInt(DFSConfigKeys.DFS_CLIENT_WRITE_PACKET_SIZE_KEY, 
                                        DFSConfigKeys.DFS_CLIENT_WRITE_PACKET_SIZE_DEFAULT);
-    InetSocketAddress socAddr = NetUtils.createSocketAddr(
-        conf.get("dfs.datanode.address", "0.0.0.0:50010"));
+    InetSocketAddress socAddr = DataNode.getStreamingAddr(conf);
     int tmpPort = socAddr.getPort();
     storage = new DataStorage();
     // construct registration
@@ -336,10 +354,15 @@ public class DataNode extends Configured
     }
 
       
-    // find free port
-    ServerSocket ss = (socketWriteTimeout > 0) ? 
+    // find free port or use privileged port provide
+    ServerSocket ss;
+    if(secureResources == null) {
+      ss = (socketWriteTimeout > 0) ? 
           ServerSocketChannel.open().socket() : new ServerSocket();
-    Server.bind(ss, socAddr, 0);
+          Server.bind(ss, socAddr, 0);
+    } else {
+      ss = resources.getStreamingSocket();
+    }
     ss.setReceiveBufferSize(DEFAULT_DATA_SOCKET_SIZE); 
     // adjust machine name with the actual port
     tmpPort = ss.getLocalPort();
@@ -379,12 +402,14 @@ public class DataNode extends Configured
     }
 
     //create a servlet to serve full-file content
-    InetSocketAddress infoSocAddr = NetUtils.createSocketAddr(
-        conf.get("dfs.datanode.http.address", "0.0.0.0:50075"));
+    InetSocketAddress infoSocAddr = DataNode.getInfoAddr(conf);
     String infoHost = infoSocAddr.getHostName();
     int tmpInfoPort = infoSocAddr.getPort();
-    this.infoServer = new HttpServer("datanode", infoHost, tmpInfoPort,
-        tmpInfoPort == 0, conf);
+    this.infoServer = (secureResources == null) 
+       ? new HttpServer("datanode", infoHost, tmpInfoPort, tmpInfoPort == 0, 
+           conf)
+       : new HttpServer("datanode", infoHost, tmpInfoPort, tmpInfoPort == 0,
+           conf, secureResources.getListener());
     LOG.debug("Datanode listening on " + infoHost + ":" + tmpInfoPort);
     if (conf.getBoolean("dfs.https.enable", false)) {
       boolean needClientAuth = conf.getBoolean(DFSConfigKeys.DFS_CLIENT_HTTPS_NEED_AUTH_KEY,
@@ -441,6 +466,14 @@ public class DataNode extends Configured
     }
   }
 
+  /**
+   * Determine the http server's effective addr
+   */
+  public static InetSocketAddress getInfoAddr(Configuration conf) {
+    return NetUtils.createSocketAddr(
+        conf.get("dfs.datanode.http.address", "0.0.0.0:50075"));
+  }
+  
   /**
    * Creates either NIO or regular depending on socketWriteTimeout.
    */
@@ -1373,6 +1406,15 @@ public class DataNode extends Configured
    */
   public static DataNode instantiateDataNode(String args[],
                                       Configuration conf) throws IOException {
+    return instantiateDataNode(args, conf, null);
+  }
+  
+  /** Instantiate a single datanode object, along with its secure resources. 
+   * This must be run by invoking{@link DataNode#runDatanodeDaemon(DataNode)} 
+   * subsequently. 
+   */
+  public static DataNode instantiateDataNode(String args [], Configuration conf,
+      SecureResources resources) throws IOException {
     if (conf == null)
       conf = new HdfsConfiguration();
     
@@ -1394,10 +1436,10 @@ public class DataNode extends Configured
     Collection<URI> dataDirs = getStorageDirs(conf);
     dnThreadName = "DataNode: [" +
                     StringUtils.uriToString(dataDirs.toArray(new URI[0])) + "]";
-    UserGroupInformation.setConfiguration(conf);
+   UserGroupInformation.setConfiguration(conf);
     SecurityUtil.login(conf, DFSConfigKeys.DFS_DATANODE_KEYTAB_FILE_KEY,
         DFSConfigKeys.DFS_DATANODE_USER_NAME_KEY);
-    return makeInstance(dataDirs, conf);
+    return makeInstance(dataDirs, conf, resources);
   }
 
   static Collection<URI> getStorageDirs(Configuration conf) {
@@ -1411,7 +1453,16 @@ public class DataNode extends Configured
    */
   public static DataNode createDataNode(String args[],
                                  Configuration conf) throws IOException {
-    DataNode dn = instantiateDataNode(args, conf);
+    return createDataNode(args, conf, null);
+  }
+  
+  /** Instantiate & Start a single datanode daemon and wait for it to finish.
+   *  If this thread is specifically interrupted, it will stop waiting.
+   */
+  @InterfaceAudience.Private
+  public static DataNode createDataNode(String args[], Configuration conf,
+      SecureResources resources) throws IOException {
+    DataNode dn = instantiateDataNode(args, conf, resources);
     runDatanodeDaemon(dn);
     return dn;
   }
@@ -1431,12 +1482,13 @@ public class DataNode extends Configured
    * @param dataDirs List of directories, where the new DataNode instance should
    * keep its files.
    * @param conf Configuration instance to use.
+   * @param resources Secure resources needed to run under Kerberos
    * @return DataNode instance for given list of data dirs and conf, or null if
    * no directory from this directory list can be created.
    * @throws IOException
    */
-  static DataNode makeInstance(Collection<URI> dataDirs, Configuration conf)
-    throws IOException {
+  static DataNode makeInstance(Collection<URI> dataDirs, Configuration conf,
+      SecureResources resources) throws IOException {
     LocalFileSystem localFS = FileSystem.getLocal(conf);
     FsPermission permission = new FsPermission(
         conf.get(DFSConfigKeys.DFS_DATANODE_DATA_DIR_PERMISSION_KEY,
@@ -1444,7 +1496,7 @@ public class DataNode extends Configured
     ArrayList<File> dirs = getDataDirsFromURIs(dataDirs, localFS, permission);
 
     if (dirs.size() > 0) {
-      return new DataNode(conf, dirs);
+      return new DataNode(conf, dirs, resources);
     }
     LOG.error("All directories in "
         + DFSConfigKeys.DFS_DATANODE_DATA_DIR_KEY + " are invalid.");
@@ -1551,12 +1603,11 @@ public class DataNode extends Configured
     return data;
   }
 
-  /**
-   */
-  public static void main(String args[]) {
+
+  public static void secureMain(String args[], SecureResources resources) {
     try {
       StringUtils.startupShutdownMessage(DataNode.class, args, LOG);
-      DataNode datanode = createDataNode(args, null);
+      DataNode datanode = createDataNode(args, null, resources);
       if (datanode != null)
         datanode.join();
     } catch (Throwable e) {
@@ -1564,6 +1615,10 @@ public class DataNode extends Configured
       System.exit(-1);
     }
   }
+  
+  public static void main(String args[]) {
+    secureMain(args, null);
+  }
 
   public Daemon recoverBlocks(final Collection<RecoveringBlock> blocks) {
     Daemon d = new Daemon(threadGroup, new Runnable() {
@@ -1827,4 +1882,10 @@ public class DataNode extends Configured
 
     return data.getReplicaVisibleLength(block);
   }
+  
+  // Determine a Datanode's streaming address
+  public static InetSocketAddress getStreamingAddr(Configuration conf) {
+    return NetUtils.createSocketAddr(
+        conf.get("dfs.datanode.address", "0.0.0.0:50010"));
+  }
 }

src/java/org/apache/hadoop/hdfs/server/datanode/SecureDataNodeStarter.java

@@ -0,0 +1,111 @@
+/**
+ * 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.server.datanode;
+
+import static org.apache.hadoop.fs.CommonConfigurationKeys.HADOOP_SECURITY_AUTHENTICATION;
+
+import java.net.InetSocketAddress;
+import java.net.ServerSocket;
+import java.nio.channels.ServerSocketChannel;
+
+import org.apache.commons.daemon.Daemon;
+import org.apache.commons.daemon.DaemonContext;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdfs.server.common.HdfsConstants;
+import org.apache.hadoop.http.HttpServer;
+import org.mortbay.jetty.nio.SelectChannelConnector;
+
+/**
+ * Utility class to start a datanode in a secure cluster, first obtaining 
+ * privileged resources before main startup and handing them to the datanode.
+ */
+public class SecureDataNodeStarter implements Daemon {
+  /**
+   * Stash necessary resources needed for datanode operation in a secure env.
+   */
+  public static class SecureResources {
+    private final ServerSocket streamingSocket;
+    private final SelectChannelConnector listener;
+    public SecureResources(ServerSocket streamingSocket,
+        SelectChannelConnector listener) {
+
+      this.streamingSocket = streamingSocket;
+      this.listener = listener;
+    }
+
+    public ServerSocket getStreamingSocket() { return streamingSocket; }
+
+    public SelectChannelConnector getListener() { return listener; }
+  }
+  
+  private String [] args;
+  private SecureResources resources;
+  
+  @Override
+  public void init(DaemonContext context) throws Exception {
+    System.err.println("Initializing secure datanode resources");
+    // We should only start up a secure datanode in a Kerberos-secured cluster
+    Configuration conf = new Configuration(); // Skip UGI method to not log in
+    if(!conf.get(HADOOP_SECURITY_AUTHENTICATION).equals("kerberos"))
+      throw new RuntimeException("Cannot start secure datanode in unsecure cluster");
+    
+    // Stash command-line arguments for regular datanode
+    args = context.getArguments();
+    
+    // Obtain secure port for data streaming to datanode
+    InetSocketAddress socAddr = DataNode.getStreamingAddr(conf);
+    int socketWriteTimeout = conf.getInt("dfs.datanode.socket.write.timeout",
+        HdfsConstants.WRITE_TIMEOUT);
+    
+    ServerSocket ss = (socketWriteTimeout > 0) ? 
+        ServerSocketChannel.open().socket() : new ServerSocket();
+    ss.bind(socAddr, 0);
+    
+    // Check that we got the port we need
+    if(ss.getLocalPort() != socAddr.getPort())
+      throw new RuntimeException("Unable to bind on specified streaming port in secure " +
+      		"context. Needed " + socAddr.getPort() + ", got " + ss.getLocalPort());
+
+    // Obtain secure listener for web server
+    SelectChannelConnector listener = 
+                   (SelectChannelConnector)HttpServer.createDefaultChannelConnector();
+    InetSocketAddress infoSocAddr = DataNode.getInfoAddr(conf);
+    listener.setHost(infoSocAddr.getHostName());
+    listener.setPort(infoSocAddr.getPort());
+    // Open listener here in order to bind to port as root
+    listener.open(); 
+    if(listener.getPort() != infoSocAddr.getPort())
+      throw new RuntimeException("Unable to bind on specified info port in secure " +
+          "context. Needed " + socAddr.getPort() + ", got " + ss.getLocalPort());
+    System.err.println("Successfully obtained privileged resources (streaming port = "
+        + ss + " ) (http listener port = " + listener.getConnection() +")");
+    
+    if(ss.getLocalPort() >= 1023 || listener.getPort() >= 1023)
+      throw new RuntimeException("Cannot start secure datanode with unprivileged ports");
+    
+    resources = new SecureResources(ss, listener);
+  }
+
+  @Override
+  public void start() throws Exception {
+    System.err.println("Starting regular datanode initialization");
+    DataNode.secureMain(args, resources);
+  }
+  
+  @Override public void destroy() { /* Nothing to do */ }
+  @Override public void stop() throws Exception { /* Nothing to do */ }
+}

src/test/unit/org/apache/hadoop/hdfs/server/datanode/TestBlockRecovery.java

@@ -101,7 +101,7 @@ public class TestBlockRecovery {
     when(namenode.sendHeartbeat(any(DatanodeRegistration.class), anyLong(), 
         anyLong(), anyLong(), anyInt(), anyInt())).thenReturn(
             new DatanodeCommand[0]);
-    dn = new DataNode(conf, dirs, namenode);
+    dn = new DataNode(conf, dirs, namenode, null);
   }
 
   /**