Paramiko模块

通过ssh远程链接服务器并执行响应的操作，类似于XShell

ps:ansible批量管理服务器工具，底层用的就是paramiko模块

安装

pip3 install paramiko

基本使用

远程链接服务器的方式

• 用户名和密码
• 公钥私钥的方式

paramiko上面两种方式都支持

链接--用户名和密码的方式

import paramiko


# 创建SSH对象
ssh = paramiko.SSHClient()
# 允许链接不在know_hosts文件中的主机
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

# 链接服务器  用户名和密码的方式
ssh.connect(hostname=‘10.0.0.200‘,port=22,username=‘root‘,password=‘123456‘)

# 执行命令
stdin, stdout, stderr = ssh.exec_command(‘ls /‘)
"""
stdin可以用来输入额外的命令 
stdout,stderr  正确和错误的返回结果
"""
res = stdout.read()
print(res.decode(‘utf-8‘))


# 关闭链接
ssh.close()

连接--公钥私钥的方式

"""
首先你得生成你本地的公钥和私钥
以mac为例
1.生成
ssh-keygen -t rsa
2.将公钥拷贝到远程服务器
ssh-copy-id -i ~/.ssh/id_rsa.pub 用户名@服务器地址
3.查看私钥
cat ~/.ssh/id_rsa
"""
# 公钥和私钥(先讲公钥保存到服务器上)
import paramiko

# 读取本地私钥
private_key = paramiko.RSAKey.from_private_key_file(‘a.txt‘)

# 创建SSH对象
ssh = paramiko.SSHClient()
# 允许连接不在know_hosts文件中的主机
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
# 连接服务器
ssh.connect(hostname=‘172.16.219.173‘, port=22, username=‘root‘, pkey=private_key)

# 执行命令
stdin, stdout, stderr = ssh.exec_command(‘ls /‘)
# 获取命令结果
result = stdout.read()
print(result.decode(‘utf-8‘))
# 关闭连接
ssh.close()

上传下载文件

# 用户名和密码的方式
import paramiko

# 用户名和密码
transport = paramiko.Transport((‘10.0.0.200‘, 22))
transport.connect(username=‘root‘, password=‘123456‘)

sftp = paramiko.SFTPClient.from_transport(transport)

# 上传文件
sftp.put("a.txt", ‘/data/b.txt‘)  # 注意上传文件到远程某个文件下 文件目录必须存在

# 下载文件
# sftp.get(‘远程服务器文件路径‘, ‘本地文件路径‘)  # 将远程文件下载到本地并重新命令
transport.close()


# 公钥私钥的方式
import paramiko
private_key = paramiko.RSAKey.from_private_key_file(‘a.txt‘)
transport = paramiko.Transport((‘172.16.219.173‘, 22))
transport.connect(username=‘root‘, pkey=private_key)
sftp = paramiko.SFTPClient.from_transport(transport)
# 将location.py 上传至服务器 /tmp/test.py
# sftp.put(‘/tmp/location.py‘, ‘/tmp/test.py‘)

# 将remove_path 下载到本地 local_path
sftp.get(‘/data/b.txt‘, ‘hahaha.txt‘)
transport.close()

paramiko代码封装

import paramiko


class SSHProxy(object):
    def __init__(self, hostname, port, username, password):
        self.hostname = hostname
        self.port = port
        self.username = username
        self.password = password
        self.transport = None

    def open(self):  # 给对象赋值一个上传下载文件对象连接
        self.transport = paramiko.Transport((self.hostname, self.port))
        self.transport.connect(username=self.username, password=self.password)

    def command(self, cmd):  # 正常执行命令的连接  至此对象内容就既有执行命令的连接又有上传下载链接
        ssh = paramiko.SSHClient()
        ssh._transport = self.transport

        stdin, stdout, stderr = ssh.exec_command(cmd)
        result = stdout.read()
        return result

    def upload(self, local_path, remote_path):
        sftp = paramiko.SFTPClient.from_transport(self.transport)
        sftp.put(local_path, remote_path)
        sftp.close()

    def close(self):
        self.transport.close()

    # with开始时自动触发
    def __enter__(self):
        # print(‘hahaha‘)
        self.open()
        return self

    # with代码块允许结束之后自动触发
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.close()

面试题

"""
题目:请在Context类中添加代码完成该类的实现
class Context:
	pass

with Context() as ctx:
	ctx.do_something()
"""
class Context:
  def __enter__(self,*args,**kwargs):
    return self
  
  def __exit__(self,*args,**kwargs):
    pass
  
  def do_something(self):
    pass
  
with Context() as ctx:
	ctx.do_something()