Hadoop集群的交互式用户的永久Kerberos票证

3 回答

• 0

  您只需要求用户将 --principal 和 --keytab 参数添加到他们的Spark作业中 . 然后Spark（实际上是YARN）代码会自动为您更新门票 . 我们有使用这种方法运行数周的工作 .

  参见例如https://spark.apache.org/docs/latest/security.html#yarn-mode

  对于像Spark Streaming应用程序这样长时间运行的应用程序能够写入HDFS，可以通过--principal和--keytab参数分别将主体和keytab传递给spark-submit . 传入的密钥表将通过Hadoop分布式缓存复制到运行Application Master的计算机上（安全地 - 如果YARN配置了SSL并启用了HDFS加密） . 将使用此主体和密钥表定期更新Kerberos登录，并且将定期生成HDFS所需的委派令牌，以便应用程序可以继续写入HDFS .

  当Yarn更新Kerberos票证时，您可以在Spark驱动程序日志中看到 .

  回复于 2024-04-20T17:52:55+08:00
• 1

  如果您正在访问Hive / Hbase或需要kerberos票证的任何其他组件，那么在票证过期的情况下使您的spark代码重新登录 . 您必须更新票证才能使用keytab，而不是依赖于TGT已经存在于缓存中 . 这是通过使用Hadoop Security包中的UserGroupInformation类完成的 . 在火花作业中添加以下片段以便长时间运行 -

  val configuration = new Configuration
  configuration.addResource("/etc/hadoop/conf/hdfs-site.xml")
  UserGroupInformation.setConfiguration(configuration)
  
  UserGroupInformation.getCurrentUser.setAuthenticationMethod(AuthenticationMethod.KERBEROS)
  UserGroupInformation.loginUserFromKeytabAndReturnUGI(
    "hadoop.kerberos.principal", " path of hadoop.kerberos.keytab file")
    .doAs(new PrivilegedExceptionAction[Unit]() {
      @Override
      def run(): Unit = {
         //hbase/hive connection
        // logic
  
      }
    })
  

  上面我们指定服务主体的名称和我们生成的keytab文件的路径 . 只要该密钥表有效，我们的程序将对所有操作使用所需的服务主体，无论运行该程序的用户是否已经过身份验证并接收到TGT .

  如果除了spark之外没有其他组件访问，那么您不需要编写上面的代码 . 只需在spark spark命令中提供keytab和principal .

  spark-submit --master yarn-cluster --keytab "xxxxxx.keytab" --principal "svc-xxxx@xxxx.COM"  xxxx.jar
  

  回复于 2024-04-20T17:52:55+08:00
• 1

  我采用上面的建议使用--keytab参数在我提交给Spark的网格节点上指定一个自定义keytab . 我使用下面的脚本创建自己的每用户密钥表 . 它一直保持到用户更改密码 .

  请注意，该脚本简化了假设，即Kerberos领域与DNS域和定义用户的LDAP目录相同 . 这适用于我的设置，小心使用你的设置 . 它还希望用户成为该网格节点上的sudoers . 更精细的脚本可能会分离keytab的生成和安装 .

  #!/usr/bin/python2.7
  
  from __future__ import print_function
  
  import os
  import sys
  import stat
  import getpass
  import subprocess
  import collections
  import socket
  import tempfile
  
  def runSudo(cmd, pw):
      try:
          subprocess.check_call("echo '{}' | sudo -S -p '' {}".format(pw, cmd), shell = True)
          return True
      except subprocess.CalledProcessError:
          return False
  
  def testPassword(pw):
      subprocess.check_call("sudo -k", shell = True)
      if not runSudo("true", pw):
          print("Incorrect password for user {}".format(getpass.getuser()), file = sys.stderr)
          sys.exit(os.EX_NOINPUT)    
  
  class KeytabFile(object):
      def __init__(self, pw):
          self.userName = getpass.getuser()
          self.pw = pw
          self.targetPath = "/etc/security/keytabs/{}.headless.keytab".format(self.userName)
          self.tempFile = None
  
      KeytabEntry = collections.namedtuple("KeytabEntry", ("kvno", "principal", "encryption"))
  
      def LoadExistingKeytab(self):
          if not os.access(self.targetPath, os.R_OK):
  
              # Note: the assumption made here, that the Kerberos realm is same as the DNS domain,
              # may not hold in other setups
              domainName = ".".join(socket.getfqdn().split(".")[1:])
  
              encryptions = ("aes128-cts-hmac-sha1-96", "arcfour-hmac", "aes256-cts-hmac-sha1-96")
              return [
                  self.KeytabEntry(0, "@".join( (self.userName, domainName)), encryption)
                      for encryption in encryptions ]
  
          def parseLine(keytabLine):
              tokens = keytabLine.strip().split(" ")
              return self.KeytabEntry(int(tokens[0]), tokens[1], tokens[2].strip("()"))
  
          cmd ="klist -ek {} | tail -n+4".format(self.targetPath)
          entryLines = subprocess.check_output(cmd, shell = True).splitlines()
          return map(parseLine, entryLines)
  
      class KtUtil(subprocess.Popen):
          def __init__(self):
              subprocess.Popen.__init__(self, "ktutil",
                  stdin = subprocess.PIPE, stdout = subprocess.PIPE, stderr=subprocess.PIPE, shell = True)
  
          def SendLine(self, line, expectPrompt = True):
              self.stdin.write(bytes(line + "\n"))
              self.stdin.flush()
              if expectPrompt:
                  self.stdout.readline()
  
          def Quit(self):
              self.SendLine("quit", False)
              rc = self.wait()
              if rc != 0:
                  raise subprocess.CalledProcessError(rc, "ktutil")
  
  
      def InstallUpdatedKeytab(self):
          fd, tempKt = tempfile.mkstemp(suffix = ".keytab")
          os.close(fd)
          entries = self.LoadExistingKeytab()
          ktutil = self.KtUtil()
          for entry in entries:
              cmd = "add_entry -password -p {} -k {} -e {}".format(
                  entry.principal, entry.kvno + 1, entry.encryption)
  
              ktutil.SendLine(cmd)
              ktutil.SendLine(self.pw)
  
          os.unlink(tempKt)
          ktutil.SendLine("write_kt {}".format(tempKt))
          ktutil.Quit()
  
          if not runSudo("mv {} {}".format(tempKt, self.targetPath), self.pw):
              os.unlink(tempKt)
              print("Failed to install the keytab to {}.".format(self.targetPath), file = sys.stderr)
              sys.exit(os.EX_CANTCREAT)
  
          os.chmod(self.targetPath, stat.S_IRUSR)
          # TODO: Also change group to 'hadoop'
  
  if __name__ == '__main__':
  
      def main():
          userPass = getpass.getpass("Please enter your password: ")
          testPassword(userPass)
          kt = KeytabFile(userPass)
          kt.InstallUpdatedKeytab()
  
      main()
  

  回复于 2024-04-20T17:52:55+08:00