فهرست منبع

HUE-219. Finish securing Thrift connections (Hue->JT, Hue->DN, DN->NN)

Todd Lipcon 15 سال پیش
والد
کامیت
cf23e7f21e

+ 0 - 1
apps/filebrowser/src/filebrowser/views_test.py

@@ -112,7 +112,6 @@ def test_view_gz():
   try:
     c = make_logged_in_client()
     cluster.fs.setuser(cluster.superuser)
-
     if cluster.fs.isdir("/test-gz-filebrowser"):
       cluster.fs.rmtree('/test-gz-filebrowser/')
 

+ 26 - 12
desktop/core/src/desktop/lib/thrift_util.py

@@ -108,7 +108,7 @@ class ConnectionPooler(object):
           q = Queue.Queue(self.poolsize)
           self.pooldict[(conf.host, conf.port)] = q
           for i in xrange(self.poolsize):
-            client = construct_client(conf)
+            client = construct_superclient(conf)
             client.CID = i
             q.put(client, False)
       finally:
@@ -146,28 +146,42 @@ class ConnectionPooler(object):
     """
     self.pooldict[(host, port)].put(client)
 
-def construct_client(conf):
+def construct_superclient(conf):
   """
   Constructs a thrift client, lazily.
   """
+  service, protocol, transport = connect_to_thrift(conf)
+  return SuperClient(service, transport, timeout_seconds=conf.timeout_seconds)
+
 
-  def sasl_factory():
-    saslc = sasl.Client()
-    saslc.setAttr("host", conf.host)
-    saslc.setAttr("service", conf.kerberos_principal)
-    saslc.init()
-    return saslc
+def connect_to_thrift(conf):
+  """
+  Connect to a thrift endpoint as determined by the 'conf' parameter.
+  Note that this does *not* open the transport.
 
-  logging.info("service: %s   host: %s" % (conf.kerberos_principal, conf.host))
+  Returns a tuple of (service, protocol, transport)
+  """
   sock = TSocket(conf.host, conf.port)
   if conf.timeout_seconds:
     # Thrift trivia: You can do this after the fact with
     # self.wrapped.transport._TBufferedTransport__trans.setTimeout(seconds*1000)
     sock.setTimeout(conf.timeout_seconds*1000.0)
-  transport = TSaslClientTransport(sasl_factory, "GSSAPI", sock)
-  protocol = TBinaryProtocolAccelerated(transport)
+  if conf.use_sasl:
+    def sasl_factory():
+      saslc = sasl.Client()
+      saslc.setAttr("host", conf.host)
+      saslc.setAttr("service", conf.kerberos_principal)
+      saslc.init()
+      return saslc
+
+    transport = TSaslClientTransport(sasl_factory, "GSSAPI", sock)
+  else:
+    transport = TBufferedTransport(sock)
+
+  protocol = TBinaryProtocol(transport)
   service = conf.klass(protocol)
-  return SuperClient(service, transport, timeout_seconds=conf.timeout_seconds)
+  return service, protocol, transport
+
 
 _connection_pool = ConnectionPooler()
 

+ 1 - 1
desktop/libs/hadoop/java/src/java/org/apache/hadoop/thriftfs/HadoopThriftAuthBridge.java

@@ -199,7 +199,7 @@ class HadoopThriftAuthBridge {
         }
         TSaslServerTransport saslTrans = (TSaslServerTransport)trans;
         String authId = saslTrans.getSaslServer().getAuthorizationID();
-        System.err.println("AUTH ID ======>" + authId);
+        LOG.debug("Authenticated  " + authId + " for Thrift call ");
 
         UserGroupInformation clientUgi = UserGroupInformation.createRemoteUser(authId);
 

+ 10 - 0
desktop/libs/hadoop/java/src/java/org/apache/hadoop/thriftfs/ThriftUtils.java

@@ -25,10 +25,12 @@ import java.util.Map;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdfs.DFSConfigKeys;
 import org.apache.hadoop.hdfs.protocol.DatanodeID;
 import org.apache.hadoop.hdfs.protocol.LocatedBlock;
 import org.apache.hadoop.hdfs.server.namenode.NameNode;
 import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.thriftfs.api.Block;
 import org.apache.hadoop.thriftfs.api.Constants;
 import org.apache.hadoop.thriftfs.api.ContentSummary;
@@ -185,6 +187,14 @@ public class ThriftUtils {
     }
 
     TTransport t = new TSocket(addr.getHostName(), addr.getPort());
+    if (UserGroupInformation.isSecurityEnabled()) {
+      t = new HadoopThriftAuthBridge.Client()
+        .createClientTransport(
+          conf.get(DFSConfigKeys.DFS_NAMENODE_USER_NAME_KEY),
+          addr.getHostName(),
+          "KERBEROS", t);
+    }
+
     t.open();
     TProtocol p = new TBinaryProtocol(t);
     return new Namenode.Client(p);

+ 1 - 2
desktop/libs/hadoop/src/hadoop/cluster.py

@@ -41,8 +41,7 @@ def _make_filesystem(identifier):
 
 def _make_mrcluster(identifier):
   cluster_conf = conf.MR_CLUSTERS[identifier]
-  return LiveJobTracker(cluster_conf.JT_HOST.get(),
-                        cluster_conf.JT_THRIFT_PORT.get())
+  return LiveJobTracker.from_conf(cluster_conf)
 
 FS_CACHE = None
 def get_hdfs(identifier="default"):

+ 9 - 2
desktop/libs/hadoop/src/hadoop/conf.py

@@ -115,7 +115,9 @@ HDFS_CLUSTERS = UnspecifiedConfigSection(
                             type=int),
       NN_HDFS_PORT=Config("hdfs_port", help="Hadoop IPC port for the name node", default=8020,
                             type=int),
-      NN_KERBEROS_PRINCIPAL=Config("kerberos_principal", help="Kerberos principal for NameNode",
+      NN_KERBEROS_PRINCIPAL=Config("nn_kerberos_principal", help="Kerberos principal for NameNode",
+                                   default="hdfs", type=str),
+      DN_KERBEROS_PRINCIPAL=Config("nn_kerberos_principal", help="Kerberos principal for DataNode",
                                    default="hdfs", type=str),
       SECURITY_ENABLED=Config("security_enabled", help="Is running with Kerberos authentication",
                               default=False, type=bool),
@@ -131,7 +133,12 @@ MR_CLUSTERS = UnspecifiedConfigSection(
     members=dict(
       JT_HOST=Config("jobtracker_host", help="IP for JobTracker"),
       JT_THRIFT_PORT=Config("thrift_port", help="Thrift port for JobTracker", default=9290,
-                            type=int))))
+                            type=int),
+      JT_KERBEROS_PRINCIPAL=Config("jt_kerberos_principal", help="Kerberos principal for JobTracker",
+                                   default="mapred", type=str),
+      SECURITY_ENABLED=Config("security_enabled", help="Is running with Kerberos authentication",
+                              default=False, type=bool))
+))
 
 
 def config_validator():

+ 21 - 11
desktop/libs/hadoop/src/hadoop/fs/hadoopfs.py

@@ -26,6 +26,7 @@ import random
 import stat as statconsts
 import subprocess
 import sys
+import time
 import urlparse
 import threading
 
@@ -175,7 +176,8 @@ class HadoopFileSystem(object):
   """
 
   def __init__(self, host, thrift_port, hdfs_port=8020,
-               kerberos_principal="hdfs",
+               nn_kerberos_principal="hdfs",
+               dn_kerberos_principal="hdfs",
                security_enabled=False,
                hadoop_bin_path="hadoop"):
     """
@@ -190,7 +192,8 @@ class HadoopFileSystem(object):
     self.thrift_port = thrift_port
     self.hdfs_port = hdfs_port
     self.security_enabled = security_enabled
-    self.kerberos_principal = kerberos_principal
+    self.nn_kerberos_principal = nn_kerberos_principal
+    self.dn_kerberos_principal = dn_kerberos_principal
     self.hadoop_bin_path = hadoop_bin_path
     self._resolve_hadoop_path()
 
@@ -198,7 +201,7 @@ class HadoopFileSystem(object):
       Namenode.Client, host, thrift_port,
       service_name="HDFS Namenode HUE Plugin",
       use_sasl=security_enabled,
-      kerberos_principal=kerberos_principal,
+      kerberos_principal=nn_kerberos_principal,
       timeout_seconds=NN_THRIFT_TIMEOUT)
 
     # The file systems are cached globally.  We store
@@ -214,7 +217,8 @@ class HadoopFileSystem(object):
                thrift_port=fs_config.NN_THRIFT_PORT.get(),
                hdfs_port=fs_config.NN_HDFS_PORT.get(),
                security_enabled=fs_config.SECURITY_ENABLED.get(),
-               kerberos_principal=fs_config.NN_KERBEROS_PRINCIPAL.get(),
+               nn_kerberos_principal=fs_config.NN_KERBEROS_PRINCIPAL.get(),
+               dn_kerberos_principal=fs_config.DN_KERBEROS_PRINCIPAL.get(),
                hadoop_bin_path=hadoop_bin_path)
 
 
@@ -567,14 +571,20 @@ class HadoopFileSystem(object):
     return ret
 
   def _connect_dn(self, node):
-    sock = TSocket.TSocket(node.host, node.thriftPort)
-    sock.setTimeout(int(DN_THRIFT_TIMEOUT * 1000))
-    transport = TTransport.TBufferedTransport(sock)
-    protocol = TBinaryProtocol.TBinaryProtocol(transport)
-    client = Datanode.Client(protocol)
+    dn_conf = thrift_util.ConnectionConfig(
+      Datanode.Client,
+      node.host,
+      node.thriftPort,
+      "HDFS Datanode Thrift",
+      use_sasl=self.security_enabled,
+      kerberos_principal=self.dn_kerberos_principal,
+      timeout_seconds=DN_THRIFT_TIMEOUT)
+
+    service, protocol, transport = \
+        thrift_util.connect_to_thrift(dn_conf)
     transport.open()
-    client.close = lambda: transport.close()
-    return client
+    service.close = lambda: transport.close()
+    return service
 
   @staticmethod
   def _unpack_stat(stat):

+ 15 - 2
desktop/libs/hadoop/src/hadoop/job_tracker.py

@@ -46,7 +46,7 @@ def test_jt_configuration(cluster):
     return err
 
   try:
-    jt = LiveJobTracker(cluster.JT_HOST.get(), cluster.JT_THRIFT_PORT.get())
+    jt = LiveJobTracker.from_conf(cluster)
     jt.runtime_info()
   except TTransport.TTransportException:
     msg = 'Failed to contact JobTracker plugin at %s:%s.' % \
@@ -63,10 +63,14 @@ class LiveJobTracker(object):
   In particular, if Thrift returns None for anything, this will throw.
   """
 
-  def __init__(self, host, thrift_port):
+  def __init__(self, host, thrift_port,
+               security_enabled=False,
+               kerberos_principal="mapred"):
     self.client = thrift_util.get_client(
       Jobtracker.Client, host, thrift_port,
       service_name="Hadoop MR JobTracker HUE Plugin",
+      use_sasl=security_enabled,
+      kerberos_principal=kerberos_principal,
       timeout_seconds=JT_THRIFT_TIMEOUT)
     self.host = host
     self.thrift_port = thrift_port
@@ -76,6 +80,15 @@ class LiveJobTracker(object):
     self.thread_local = threading.local()
     self.setuser(DEFAULT_USER)
 
+  @classmethod
+  def from_conf(cls, conf):
+    return cls(
+      conf.JT_HOST.get(),
+      conf.JT_THRIFT_PORT.get(),
+      security_enabled=conf.SECURITY_ENABLED.get(),
+      kerberos_principal=conf.JT_KERBEROS_PRINCIPAL.get())
+
+
   def thriftjobid_from_string(self, jobid):
     """The jobid looks like this: job_201001301455_0001"""
     _, tid, jid = jobid.split("_")