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

import paramiko
import select   #无法在windows下操作
import sys
t = paramiko.Transport((‘10.10.50.31‘, 22))
t.start_client()
#基于公钥验证
##default_path = os.path.join(os.environ[‘HOME‘], ‘.ssh‘, ‘id_rsa‘)
#key = paramiko.RSAKey.from_private_key_file(‘/root/.ssh/id_rsa‘)
#t.auth_publickey(‘root‘,key)
#基于用户密码验证
t.auth_password(‘root‘, ‘123456‘)

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

#########
# 利用sys.stdin,肆意妄为执行操作
# 用户在终端输入内容，并将内容发送至远程服务器
# 远程服务器执行命令，并将结果返回
# 用户终端显示内容
while True:
    # 监视用户输入和服务器返回数据
# sys.stdin 处理用户输入
# chan 是之前创建的通道，用于接收服务器返回信息
r_list,w_list,error = select.select([chan,sys.stdin,],[],[],1)
    if chan in r_list:                  #接收命令执行后返回的结果
try:
            x = chan.recv(1024)
            if len(x) == 0:
                print ‘\r\n*** EOF\r\n‘,
break
sys.stdout.write(x)
            sys.stdout.flush()
        except socket.timeout:
            pass
    if sys.stdin in r_list:             #用户输入
inp = sys.stdin.readline()
        chan.sendall(inp)               #把用户输入命令发送执行
#########
chan.close()
t.close()

import paramiko
import select   #无法在windows下操作
import sys
import termios
import tty

t = paramiko.Transport((‘10.10.50.31‘, 22))
t.start_client()
t.auth_password(‘root‘, ‘123456‘)

chan = t.open_session()
chan.get_pty()
chan.invoke_shell()

#########
oldtty = termios.tcgetattr(sys.stdin)
try:
    tty.setraw(sys.stdin.fileno())
    chan.settimeout(0.0)
    f  = open(‘record.log‘,‘a‘)
    records = []
    while True:
        r_list,w_list,error = select.select([chan,sys.stdin,],[],[],1)
        if chan in r_list:
            try:
                x = chan.recv(1024)
                if len(x) == 0:
                    print ‘\r\n*** EOF\r\n‘,
f.close()   #异常中断，或者exit退出时保存
break
sys.stdout.write(x)
                sys.stdout.flush()
            except socket.timeout:
                pass
        if sys.stdin in r_list:
            inp = sys.stdin.read(1)
            records.append(inp)  #把个字符键入列表暂时保存 [‘l‘,‘s‘]
if len(inp) == 0:
                break
            if inp == ‘\t‘:     #tab键不记录
pass
            if inp == ‘\r‘:     #每次回车
cmd = ‘‘.join(records).replace(‘\r‘,‘\n‘) #[ls]+换行
f.write(cmd)
                records = []    #每次记录一个条命令
chan.sendall(inp)
finally:
    termios.tcsetattr(sys.stdin,termios.TCSADRAIN,oldtty)
#########
chan.close()
t.close()

import paramiko
import sys
import threading

t = paramiko.Transport((‘10.10.50.31‘, 22))
t.start_client()
t.auth_password(‘root‘, ‘123456‘)

chan = t.open_session()
chan.get_pty()
chan.invoke_shell()

#########
sys.stdout.write("Line-buffered terminal emulation. Press F6 or ^Z to send EOF.\r\n\r\n")
def writeall(sock):
    while True:
        data = sock.recv(256)
        if not data:
            sys.stdout.write(‘\r\n*** EOF ***\r\n\r\n‘)
            sys.stdout.flush()
            break
sys.stdout.write(data)
        sys.stdout.flush()

writer = threading.Thread(target=writeall, args=(chan,))
writer.start()

try:
    while True:
        d = sys.stdin.read(1)
        if not d:
            break
chan.send(d)
except EOFError:
    # user hit ^Z or F6
pass
#########
chan.close()
t.close()