Python实现Kerberos用户的增删改查操作

唐可

发布时间：2026-10-14| 5人浏览过

原创

本文介绍如何使用Python的subprocess和paramiko库封装Kerberos命令行工具，实现本地及远程用户的增删改查。重点解决Kerberos认证下Hive连接的交互问题，特别是删除Principal时处理expect交互的逻辑，并提供完整的Python代码示例。

Python模拟Shell命令行接口

首先，需要模拟类似Shell命令行操作的接口，以便在Python中执行系统命令。

在Python环境中，安装subprocess用于本地命令执行，以及paramiko用于SSH远程连接。


#-*- coding: UTF-8 -*-
#!/usr/bin/python
import os, sys
import subprocess
import paramiko
import settings
 
class RunCmd(object):
 def __init__(self):
  self.cmd = 'ls'
 
 @staticmethod
 def local_run(cmd):
  print('start executing...')
  print('cmd is -------> %s' % str(cmd))
  s = subprocess.Popen(str(cmd), shell=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
  out, err = s.communicate()
  print("outinfo is -------> %s" % out)
  print("errinfo is -------> %s" % err)
  print('finish executing...')
  print('result:------> %s' % s.returncode)
  return s.returncode
 
 @staticmethod
 def remote_run(host, username, password, port, cmd):
  client = paramiko.SSHClient()
  client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
  client.connect(hostname=host, port=int(port), username=username, password=password, timeout=5)
  stdin, stdout, stderr = client.exec_command(cmd)
  result = stdout.read()
  client.close()
  return result
 
 @staticmethod
 def krb_run(cmd):
  print('krb_run start...')
  print('cmd is -------> %s' % str(cmd))
  result = RunCmd.remote_run(settings.KRB_HOST, settings.USERNAME, settings.PASSWORD, settings.PORT, cmd)
  print('result:------> %s' % result)
  print('krb_run finish...')
  return result

Kerberos命令封装与交互处理

接下来，将Kerberos常用的命令操作封装为接口，其他操作相对简单，但删除Principal需要处理交互式输入。


 def delete_user(self, username):
  cmd = r"""
   expect -c "
   set timeout 1;
   spawn kadmin.local -q \"delete_principal {principal}\" ;
   expect yes/no {{ send \"yes\r\" }} ;
   expect *\r
   expect \r
   expect eof
   "
  """.format(principal=username)
  RunCmd.krb_run(cmd)

Kerberos认证下的Hive连接

补充知识：Python操作有Kerberos认证的Hive库。

之前访问Hive都比较简单，直接用pyhive连接即可。

但是最近遇到了一个问题，Hive有了Kerberos认证。

最终经过各种尝试和灵感迸发，终于解决了这个问题，遂记录之。

完整代码实现

代码


from pyhive.hive import connect
con = connect(host='XXXX',port=10000,auth='KERBEROS',kerberos_service_name="hive")
cursor = con.cursor()
cursor.execute('select * from tmp.pricing_calculate_result_spark where time_id="201907171355" limit 10,1')
datas = cursor.fetchall()
print(datas)
cursor.close()
con.close()

端口和ip都换成自己的，auth和kerberos_service_name不要改

运行效果