Python使用Paramiko实现服务器命令执行与文件传输
本文介绍了如何利用Python的Paramiko库实现远程服务器管理。通过封装SSH连接方法,结合RSA密钥认证,实现了安全的服务器连接。文章提供了批量执行远程命令和上传文件到服务器的完整代码示例,包含错误处理与日志记录机制,适用于运维自动化场景。
Python Paramiko远程管理方案概述
直接展示核心实现代码。
封装SSH连接与密钥认证
首先封装服务器连接逻辑,建立稳定的SSH会话。
@staticmethod
def connect(ip, server_user, server_port, server_path):
"""
连接服务器
:param :
:return:
"""
ssh = paramiko.SSHClient()
private_key = paramiko.RSAKey.from_private_key_file('{}.ssh/id_rsa'.format(server_path))
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
try:
ssh.connect(ip, port=server_port, username=server_user, pkey=private_key, timeout=5, allow_agent=True)
return ssh
except Exception as e:
app.logger.info('connect_error info is {}'.format(e.__repr__()))
return {
'result': 1,
'ip': ip,
'msg': e.__repr__(),
}
远程命令执行与状态监控
在确认连接成功后,向目标服务器发送并执行指定命令。
def fabric_run_cmd(self, ip, server_user, server_port, server_path, cmd):
"""
批量在服务器执行命令
:return:
"""
conn = self.connect(ip, server_user, server_port, server_path)
try:
if type(conn) == dict:
return conn
else:
stdin, stdout, stderr = conn.exec_command(cmd)
app.logger.info('fabric_run_cmd_stdout info is {}'.format(stdout.readlines()))
app.logger.info('fabric_run_cmd_stderr info is {}'.format(stderr.readlines()))
channel = stdout.channel
status = channel.recv_exit_status()
conn.close()
app.logger.info('fabric_run_cmd_status info is {}'.format(status))
if status == 0:
return {
'ip': ip,
'msg': '执行成功',
'result': 0
}
else:
return {
'ip': ip,
'msg': stderr.readlines(),
'result': 1
}
except Exception as e:
app.logger.info('fabric_run_cmd_error info is {}'.format(e.__repr__()))
return {
'ip': ip,
'msg': e.__repr__(),
'result': 1
}
文件上传至服务器的实现
利用SFTP通道将本地文件上传至远程服务器指定路径。
def fabric_put_file(self, ip, src_file, dst_file, server_user, server_port, server_path):
"""
发送文件到服务器
:return:
"""
conn = self.connect(ip, server_user, server_port, server_path)
if type(conn) == dict:
return conn
else:
try:
ftp = conn.open_sftp()
ftp.put(dst_file, src_file) # 发送文件之前先判断有没有目标文件夹 如果没有现在服务器上创建文件夹
ftp.close() # 在此判断文件是否发送成功
return {
'ip': ip,
'msg': '上传成功',
'result': 0
}
except Exception as e:
app.logger.info('fabric_put_file_error info is {}'.format(e.__repr__()))
return {
'ip': ip,
'msg': e.__repr__(),
'result': 1
}
受限于篇幅与具体业务场景,如有疏漏欢迎留言指正。
以上即为Python利用Paramiko执行服务器命令的完整方案,更多相关技术细节可参考本站其他文章。
免责声明:本文内容仅供参考,不代表本站观点。转载或引用请注明出处。









