001/**
002 * Licensed to the Apache Software Foundation (ASF) under one or more
003 * contributor license agreements. See the NOTICE file distributed with this
004 * work for additional information regarding copyright ownership. The ASF
005 * licenses this file to you under the Apache License, Version 2.0 (the
006 * "License"); you may not use this file except in compliance with the License.
007 * You may obtain a copy of the License at
008 * 
009 * http://www.apache.org/licenses/LICENSE-2.0
010 * 
011 * Unless required by applicable law or agreed to in writing, software
012 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
013 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
014 * License for the specific language governing permissions and limitations under
015 * the License.
016 */
017package org.apache.hadoop.hdfs.server.datanode;
018
019import com.google.common.annotations.VisibleForTesting;
020import org.apache.commons.daemon.Daemon;
021import org.apache.commons.daemon.DaemonContext;
022import org.apache.hadoop.conf.Configuration;
023import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
024import org.apache.hadoop.hdfs.DFSConfigKeys;
025import org.apache.hadoop.hdfs.DFSUtil;
026import org.apache.hadoop.hdfs.HdfsConfiguration;
027import org.apache.hadoop.hdfs.protocol.HdfsConstants;
028import org.apache.hadoop.http.HttpConfig;
029import org.apache.hadoop.security.SecurityUtil;
030import org.apache.hadoop.security.UserGroupInformation;
031
032import java.net.InetSocketAddress;
033import java.net.ServerSocket;
034import java.nio.channels.ServerSocketChannel;
035
036/**
037 * Utility class to start a datanode in a secure cluster, first obtaining 
038 * privileged resources before main startup and handing them to the datanode.
039 */
040public class SecureDataNodeStarter implements Daemon {
041  /**
042   * Stash necessary resources needed for datanode operation in a secure env.
043   */
044  public static class SecureResources {
045    private final ServerSocket streamingSocket;
046    private final ServerSocketChannel httpServerSocket;
047    public SecureResources(ServerSocket streamingSocket, ServerSocketChannel httpServerSocket) {
048      this.streamingSocket = streamingSocket;
049      this.httpServerSocket = httpServerSocket;
050    }
051
052    public ServerSocket getStreamingSocket() { return streamingSocket; }
053
054    public ServerSocketChannel getHttpServerChannel() {
055      return httpServerSocket;
056    }
057  }
058  
059  private String [] args;
060  private SecureResources resources;
061
062  @Override
063  public void init(DaemonContext context) throws Exception {
064    System.err.println("Initializing secure datanode resources");
065    // Create a new HdfsConfiguration object to ensure that the configuration in
066    // hdfs-site.xml is picked up.
067    Configuration conf = new HdfsConfiguration();
068    
069    // Stash command-line arguments for regular datanode
070    args = context.getArguments();
071    resources = getSecureResources(conf);
072  }
073
074  @Override
075  public void start() throws Exception {
076    System.err.println("Starting regular datanode initialization");
077    DataNode.secureMain(args, resources);
078  }
079
080  @Override public void destroy() {}
081  @Override public void stop() throws Exception { /* Nothing to do */ }
082
083  /**
084   * Acquire privileged resources (i.e., the privileged ports) for the data
085   * node. The privileged resources consist of the port of the RPC server and
086   * the port of HTTP (not HTTPS) server.
087   */
088  @VisibleForTesting
089  public static SecureResources getSecureResources(Configuration conf)
090      throws Exception {
091    HttpConfig.Policy policy = DFSUtil.getHttpPolicy(conf);
092    boolean isSecure = UserGroupInformation.isSecurityEnabled();
093
094    // Obtain secure port for data streaming to datanode
095    InetSocketAddress streamingAddr  = DataNode.getStreamingAddr(conf);
096    int socketWriteTimeout = conf.getInt(
097        DFSConfigKeys.DFS_DATANODE_SOCKET_WRITE_TIMEOUT_KEY,
098        HdfsConstants.WRITE_TIMEOUT);
099    int backlogLength = conf.getInt(
100        CommonConfigurationKeysPublic.IPC_SERVER_LISTEN_QUEUE_SIZE_KEY,
101        CommonConfigurationKeysPublic.IPC_SERVER_LISTEN_QUEUE_SIZE_DEFAULT);
102
103    ServerSocket ss = (socketWriteTimeout > 0) ? 
104        ServerSocketChannel.open().socket() : new ServerSocket();
105    ss.bind(streamingAddr, backlogLength);
106
107    // Check that we got the port we need
108    if (ss.getLocalPort() != streamingAddr.getPort()) {
109      throw new RuntimeException(
110          "Unable to bind on specified streaming port in secure "
111              + "context. Needed " + streamingAddr.getPort() + ", got "
112              + ss.getLocalPort());
113    }
114
115    if (!SecurityUtil.isPrivilegedPort(ss.getLocalPort()) && isSecure) {
116      throw new RuntimeException(
117        "Cannot start secure datanode with unprivileged RPC ports");
118    }
119
120    System.err.println("Opened streaming server at " + streamingAddr);
121
122    // Bind a port for the web server. The code intends to bind HTTP server to
123    // privileged port only, as the client can authenticate the server using
124    // certificates if they are communicating through SSL.
125    final ServerSocketChannel httpChannel;
126    if (policy.isHttpEnabled()) {
127      httpChannel = ServerSocketChannel.open();
128      InetSocketAddress infoSocAddr = DataNode.getInfoAddr(conf);
129      httpChannel.socket().bind(infoSocAddr);
130      InetSocketAddress localAddr = (InetSocketAddress) httpChannel.socket()
131        .getLocalSocketAddress();
132
133      if (localAddr.getPort() != infoSocAddr.getPort()) {
134        throw new RuntimeException("Unable to bind on specified info port in secure " +
135            "context. Needed " + streamingAddr.getPort() + ", got " + ss.getLocalPort());
136      }
137      System.err.println("Successfully obtained privileged resources (streaming port = "
138          + ss + " ) (http listener port = " + localAddr.getPort() +")");
139
140      if (localAddr.getPort() > 1023 && isSecure) {
141        throw new RuntimeException(
142            "Cannot start secure datanode with unprivileged HTTP ports");
143      }
144      System.err.println("Opened info server at " + infoSocAddr);
145    } else {
146      httpChannel = null;
147    }
148
149    return new SecureResources(ss, httpChannel);
150  }
151
152}