zoukankan html css js c++ java

paramiko模块

paramiko模块，基于SSH用于连接远程服务器并执行相关操作。

一、安装

pip3 install paramiko

二、使用

（1）SSHClient

　　用于连接远程服务器并执行基本命令

基于用户名密码连接：

import paramiko
   
# 创建SSH对象
ssh = paramiko.SSHClient()
# 允许连接不在know_hosts文件中的主机
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
# 连接服务器
ssh.connect(hostname='c1.salt.com', port=22, username='wupeiqi', password='123')
   
# 执行命令
stdin, stdout, stderr = ssh.exec_command('ls')
# 获取命令结果
result = stdout.read()
   
# 关闭连接
ssh.close()

SSHClient封装Transport

import paramiko

transport = paramiko.Transport(('192.168.147.147', 22))
transport.connect(username='root', password='centos')

ssh = paramiko.SSHClient()
ssh._transport = transport

stdin, stdout, stderr = ssh.exec_command('df')
print(stdout.read())
transport.close()

基于公钥密钥连接：

import paramiko
  
private_key = paramiko.RSAKey.from_private_key_file('/home/auto/.ssh/id_rsa')
  
# 创建SSH对象
ssh = paramiko.SSHClient()
# 允许连接不在know_hosts文件中的主机
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
# 连接服务器
ssh.connect(hostname='c1.salt.com', port=22, username='wupeiqi', pkey=private_key)
  
# 执行命令
stdin, stdout, stderr = ssh.exec_command('df')
# 获取命令结果
result = stdout.read()
  
# 关闭连接
ssh.close()

SSHClient封装Transport

import paramiko

private_key = paramiko.RSAKey.from_private_key_file('id_rsa')

transport = paramiko.Transport(('192.168.147.147', 22))
transport.connect(username='root', pkey=private_key)

ssh = paramiko.SSHClient()
ssh._transport = transport

stdin, stdout, stderr = ssh.exec_command('df')

transport.close()

（2）SFTPClient(只能用Transport)

　　用于连接远程服务器并执行上传下载

基于用户名密码上传下载：

import paramiko
  
transport = paramiko.Transport(('hostname',22))
transport.connect(username='wupeiqi',password='123')
  
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('remove_path', 'local_path')
  
transport.close()

基于公钥密钥上传下载：

import paramiko
  
private_key = paramiko.RSAKey.from_private_key_file('/home/auto/.ssh/id_rsa')
  
transport = paramiko.Transport(('hostname', 22))
transport.connect(username='wupeiqi', 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('remove_path', 'local_path')
  
transport.close()

应用场景：修改啊haproxy配置文件的时候，先在本地生成，远程主机配置文件备份，配置文件再上传到远程主机。这样每操作一次需要建立一次连接，断开连接，能否建立一个连接，通过这一个连接进行命令操作和文件操作呢？

当然可以，就用Transport

import paramiko
class SSH(object):
    def __init__(self,host,port,user,pwd):
        self.host = host
        self.port = port
        self.user = user
        self.pwd = pwd
        self.transport = None
    def connect(self):
        self.transport = paramiko.Transport(self.host,self.port)
        self.transport.connect(username=self.user,password=self.pwd)

    def cmd(self,cmd):
        ssh = paramiko.SSHClient()
        ssh._transport = self.transport
        stdin,stdout,stderr = ssh.exec_command(cmd)
        return stdout.read()

    def download(self,server_path,local_path):
        sftp = paramiko.SFTPClient.from_transport(self.transport)
        sftp.get(server_path,local_path)

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

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


obj = SSH('192.168.147.147',22,'root','centos')
obj.connect()
#只需要建立一个连接，操作命令 上传文件都是利用这个连接
a = obj.cmd('ls')
print(a.decode())
obj.close()

下面通过paramiko和sqlalchemy可以做个堡垒机

堡垒机

堡垒机执行流程：

管理员为用户在服务器上创建账号（将公钥放置服务器，或者使用用户名密码）
用户登陆堡垒机，输入堡垒机用户名密码，显示当前用户管理的服务器列表
用户选择服务器，并自动登陆
执行操作并同时将用户操作记录

注：用户在这台堡垒机上只能看到服务器列表并进行选择，并不能看到堡垒机机器本身的内容，在退出远程服务器时也不能停留在堡垒机上

配置.brashrc实现ssh登陆后自动执行脚本，如：/usr/bin/python /home/hongpeng/ssh.py

表结构

#!/usr/bin/env python
# -*- coding:utf-8 -*-

from sqlalchemy import create_engine, and_, or_, func, Table
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, DateTime
from sqlalchemy.orm import sessionmaker, relationship

Base = declarative_base()  # 生成一个SqlORM 基类


class Host(Base):
    __tablename__ = 'host'
    id = Column(Integer, primary_key=True, autoincrement=True)
    hostname = Column(String(64), unique=True, nullable=False)
    ip_addr = Column(String(128), unique=True, nullable=False)
    port = Column(Integer, default=22)


class HostUser(Base):
    __tablename__ = 'host_user'
    id = Column(Integer, primary_key=True, autoincrement=True)
    username = Column(String(64), unique=True, nullable=False)
    AuthTypes = [
        ('p', 'SSH/Password'),
        ('r', 'SSH/KEY'),
    ]
    auth_type = Column(String(16))
    cert = Column(String(255))

    host_id = Column(Integer, ForeignKey('host.id'))

    __table_args__ = (
        UniqueConstraint('host_id', 'username', name='_host_username_uc'),
    )


class Group(Base):
    __tablename__ = 'group'
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(64), unique=True, nullable=False)


class UserProfile(Base):
    __tablename__ = 'user_profile'
    id = Column(Integer, primary_key=True, autoincrement=True)
    username = Column(String(64), unique=True, nullable=False)
    password = Column(String(255), nullable=False)


class Group2UserProfile(Base):
    __tablename__ = 'group_2_user_profile'
    id = Column(Integer, primary_key=True, autoincrement=True)
    user_profile_id = Column(Integer, ForeignKey('user_profile.id'))
    group_id = Column(Integer, ForeignKey('group.id'))
    __table_args__ = (
        UniqueConstraint('user_profile_id', 'group_id', name='ux_user_group'),
    )


class Group2HostUser(Base):
    __tablename__ = 'group_2_host_user'
    id = Column(Integer, primary_key=True, autoincrement=True)
    host_user_id = Column(Integer, ForeignKey('host_user.id'))
    group_id = Column(Integer, ForeignKey('group.id'))
    __table_args__ = (
        UniqueConstraint('group_id', 'host_user_id', name='ux_group_host_user'),
    )


class UserProfile2HostUser(Base):
    __tablename__ = 'user_profile_2_host_user'
    id = Column(Integer, primary_key=True, autoincrement=True)
    host_user_id = Column(Integer, ForeignKey('host_user.id'))
    user_profile_id = Column(Integer, ForeignKey('user_profile.id'))
    __table_args__ = (
        UniqueConstraint('user_profile_id', 'host_user_id', name='ux_user_host_user'),
    )


class AuditLog(Base):
    __tablename__ = 'audit_log'
    id = Column(Integer, primary_key=True, autoincrement=True)

    action_choices2 = [
        (u'cmd', u'CMD'),
        (u'login', u'Login'),
        (u'logout', u'Logout'),
    ]
    action_type = Column(String(16))
    cmd = Column(String(255))
    date = Column(DateTime)
    user_profile_id = Column(Integer, ForeignKey('user_profile.id'))
    host_user_id = Column(Integer, ForeignKey('host_user.id'))

实现过程

1、前戏

import paramiko
import sys
import os
import socket
import select
import getpass
 
tran = paramiko.Transport(('10.211.55.4', 22,))
tran.start_client()
tran.auth_password('wupeiqi', '123')
 
# 打开一个通道
chan = tran.open_session()
# 获取一个终端
chan.get_pty()
# 激活器
chan.invoke_shell()
 
#########
# 利用sys.stdin,肆意妄为执行操作
# 用户在终端输入内容，并将内容发送至远程服务器
# 远程服务器执行命令，并将结果返回
# 用户终端显示内容
#########
 
chan.close()
tran.close()

2、肆意妄为（一）

import paramiko
import sys
import os
import socket
import select
import getpass
from paramiko.py3compat import u
 
tran = paramiko.Transport(('10.211.55.4', 22,))
tran.start_client()
tran.auth_password('wupeiqi', '123')
 
# 打开一个通道
chan = tran.open_session()
# 获取一个终端
chan.get_pty()
# 激活器
chan.invoke_shell()
 
while True:
    # 监视用户输入和服务器返回数据
    # sys.stdin 处理用户输入
    # chan 是之前创建的通道，用于接收服务器返回信息
    readable, writeable, error = select.select([chan, sys.stdin, ],[],[],1)
    if chan in readable:
        try:
            x = u(chan.recv(1024))
            if len(x) == 0:
                print('
*** EOF
')
                break
            sys.stdout.write(x)
            sys.stdout.flush()
        except socket.timeout:
            pass
    if sys.stdin in readable:
        inp = sys.stdin.readline()
        chan.sendall(inp)
 
chan.close()
tran.close()

上面的代码因为监听用户输入的时候是当用户回车之后才把这一整行发过去，所以不能tab补全，sys.stdin用户输入，chan远程返回的socket连接

3、肆意妄为（二）

import paramiko
import sys
import os
import socket
import select
import getpass
import termios
import tty
from paramiko.py3compat import u
 
tran = paramiko.Transport(('10.211.55.4', 22,))
tran.start_client()
tran.auth_password('wupeiqi', '123')
 
# 打开一个通道
chan = tran.open_session()
# 获取一个终端
chan.get_pty()
# 激活器
chan.invoke_shell()
 
 
# 获取原tty属性
oldtty = termios.tcgetattr(sys.stdin)
try:
    # 为tty设置新属性
    # 默认当前tty设备属性：
    #   输入一行回车，执行
    #   CTRL+C 进程退出，遇到特殊字符，特殊处理。
 
    # 这是为原始模式，不认识所有特殊符号
    # 放置特殊字符应用在当前终端，如此设置，将所有的用户输入均发送到远程服务器
    tty.setraw(sys.stdin.fileno())
    chan.settimeout(0.0)
 
    while True:
        # 监视 用户输入 和 远程服务器返回数据（socket）
        # 阻塞，直到句柄可读
        r, w, e = select.select([chan, sys.stdin], [], [], 1)
        if chan in r:
            try:
                x = u(chan.recv(1024))
                if len(x) == 0:
                    print('
*** EOF
')
                    break
                sys.stdout.write(x)
                sys.stdout.flush()
            except socket.timeout:
                pass
        if sys.stdin in r:
            x = sys.stdin.read(1)
            if len(x) == 0:
                break
            chan.send(x)
 
finally:
    # 重新设置终端属性
    termios.tcsetattr(sys.stdin, termios.TCSADRAIN, oldtty)
 
 
chan.close()
tran.close()
 

这段代码是用户输入一个发一个，可以tab补全。

基于这个可以完成堡垒机的功能，只是paramiko有个bug，在使用历史命令的时候得按两次上键。

记录操作日志

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import paramiko
import sys
import os
import socket
import getpass
import termios
import tty
import select
from paramiko.py3compat import u


def interactive_shell(chan):
    # 获取原tty属性
    oldtty = termios.tcgetattr(sys.stdin)
    try:
        # 为tty设置新属性
        # 默认当前tty设备属性：
        # 输入一行回车，执行
        # CTRL+C 进程退出，遇到特殊字符，特殊处理。

        # 这是为原始模式，不认识所有特殊符号
        # 放置特殊字符应用在当前终端，如此设置，将所有的用户输入均发送到远程服务器
        tty.setraw(sys.stdin.fileno())
        tty.setcbreak(sys.stdin.fileno())
        chan.settimeout(0.0)

        log = open('handle.log', 'a+', encoding='utf-8')
        flag = False　　#没有按tab键
        temp_list = []

        while True:
            r, w, e = select.select([chan, sys.stdin], [], [])
            if chan in r:
                try:
                    x = u(chan.recv(1024))
                    if len(x) == 0:
                        sys.stdout.write('
*** EOF
')
                        break
                    # 如果用户上一次点击的是tab键，则获取返回的内容写入在记录中
                    if flag:
                        if x.startswith('
'):#换行符，tab补全显示多个的时候
                            pass
                        else:
                            temp_list.append(x)
                        flag = False
                    sys.stdout.write(x)
                    sys.stdout.flush()
                except socket.timeout:
                    pass
            if sys.stdin in r:
                # 读取用户在终端数据每一个字符
                x = sys.stdin.read(1)
                if len(x) == 0:
                    break
                # 如果用户点击TAB键
                if x == '	':
                    flag = True
                else:
                    # 未点击TAB键，则将每个操作字符记录添加到列表中，以便之后写入文件
                    temp_list.append(x)

                # 如果用户敲回车，则将操作记录写入文件
                if x == '
':
                    log.write(''.join(temp_list))
                    log.flush()
                    temp_list.clear()
                chan.send(x)

    finally:
        # 重新设置终端属性
        termios.tcsetattr(sys.stdin, termios.TCSADRAIN, oldtty)


def run():
    db_dict = {
        'c1.salt.com': {
            'root': {'user': 'root', 'auth': 'r', "cert": 'key路径'},
            'alex': {'user': 'alex', 'auth': 'p', "cert": '密码'},
        },
        'c2.salt.com': {
            'root': {'user': 'root', 'auth': 'r', "cert": 'key路径'},
            'alex': {'user': 'alex', 'auth': 'p', "cert": '密码'},
        },

    }

    for row in db_dict.keys():
        print(row)

    hostname = input('请选择主机: ')
    tran = paramiko.Transport((hostname, 22,))
    tran.start_client()

    for item in db_dict[hostname].keys():
        print(item)

    username = input('请输入用户: ')

    user_dict = db_dict[hostname][username]
    if username['auth'] == 'r':
        key = paramiko.RSAKey.from_private_key_file(user_dict['cert'])
        tran.auth_publickey(username, key)
    else:
        pw = user_dict['cert']
        tran.auth_password(username, pw)

    # 打开一个通道
    chan = tran.open_session()
    # 获取一个终端
    chan.get_pty()
    # 激活器
    chan.invoke_shell()

    interactive_shell(chan)

    chan.close()
    tran.close()


if __name__ == '__main__':
    run()

查看全文

相关阅读:
Java多线程问题
 pattern-matching as an expression without a prior match -scala
从Zero到Hero，OpenAI重磅发布深度强化学习资源
 What-does-git-remote-and-origin-mean
flink source code
如何生成ExecutionGraph及物理执行图
 rocketmq 源码
 Flink source task 源码分析
 flink 获取上传的Jar源码
 fileupload

原文地址：https://www.cnblogs.com/hongpeng0209/p/6289347.html