#!/usr/bin/env python # Copyright (C) - Robey Pointer <robeypointer@gmail.com>
#
# This file is part of paramiko.
#
# Paramiko is free software; you can redistribute it and/or modify it under the
# terms of the GNU Lesser General Public License as published by the Free
# Software Foundation; either version 2.1 of the License, or (at your option)
# any later version.
#
# Paramiko is distributed in the hope that it will be useful, but WITHOUT ANY
# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
# A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more
# details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with Paramiko; if not, write to the Free Software Foundation, Inc.,
# Temple Place, Suite , Boston, MA - USA. import base64
from binascii import hexlify
import getpass
import os
import select
import socket
import sys
import time
import traceback
from paramiko.py3compat import input import paramiko dic_iplist = {
'172.16.230.151':'',
'172.16.230.130':'Admin2015',
'172.16.230.223':'Admin2015'
} try:
import interactive
except ImportError:
from . import interactive def agent_auth(transport, username):
"""
Attempt to authenticate to the given transport using any of the private
keys available from an SSH agent.
""" agent = paramiko.Agent()
agent_keys = agent.get_keys()
if len(agent_keys) == :
return for key in agent_keys:
print('Trying ssh-agent key %s' % hexlify(key.get_fingerprint()))
try:
transport.auth_publickey(username, key)
print('... success!')
return
except paramiko.SSHException:
print('... nope.') def manual_auth(username, hostname,pw):
t.auth_password(username, pw) # setup logging
paramiko.util.log_to_file('demo.log') username = ''
if len(sys.argv) > :
hostname = sys.argv[]
if hostname.find('@') >= :
username, hostname = hostname.split('@')
else:
for num,key in enumerate(dic_iplist.keys()):
print num,key
chooies = input('chooise number: ')
if chooies.isdigit():
chooies = int(chooies)
hostname = dic_iplist.keys()[chooies] #ipaddr
password = dic_iplist[hostname] #password if len(hostname) == :
print('*** Hostname required.')
sys.exit()
port =
if hostname.find(':') >= :
hostname, portstr = hostname.split(':')
port = int(portstr) # now connect
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect((hostname, port))
except Exception as e:
print('*** Connect failed: ' + str(e))
traceback.print_exc()
sys.exit() try:
t = paramiko.Transport(sock)
try:
t.start_client()
except paramiko.SSHException:
print('*** SSH negotiation failed.')
sys.exit() try:
keys = paramiko.util.load_host_keys(os.path.expanduser('~/.ssh/known_hosts'))
except IOError:
try:
keys = paramiko.util.load_host_keys(os.path.expanduser('~/ssh/known_hosts'))
except IOError:
print('*** Unable to open host keys file')
keys = {} # check server's host key -- this is important.
key = t.get_remote_server_key()
if hostname not in keys:
print('*** WARNING: Unknown host key!')
elif key.get_name() not in keys[hostname]:
print('*** WARNING: Unknown host key!')
elif keys[hostname][key.get_name()] != key:
print('*** WARNING: Host key has changed!!!')
sys.exit()
else:
print('*** Host key OK.') # get username
if username == '':
default_username = getpass.getuser() if default_username == 'root':
username = default_username
else:
username = 'devuser' agent_auth(t, username)
if not t.is_authenticated():
manual_auth(username, hostname,password)
if not t.is_authenticated():
print('*** Authentication failed. :(')
t.close()
sys.exit() chan = t.open_session()
chan.get_pty()
chan.invoke_shell()
print('*** Here we go!\n')
interactive.interactive_shell(chan,default_username,hostname,username)
chan.close()
t.close() except Exception as e:
print('*** Caught exception: ' + str(e.__class__) + ': ' + str(e))
traceback.print_exc()
try:
t.close()
except:
pass
sys.exit()
# Copyright (C) - Robey Pointer <robeypointer@gmail.com>
#
# This file is part of paramiko.
#
# Paramiko is free software; you can redistribute it and/or modify it under the
# terms of the GNU Lesser General Public License as published by the Free
# Software Foundation; either version 2.1 of the License, or (at your option)
# any later version.
#
# Paramiko is distributed in the hope that it will be useful, but WITHOUT ANY
# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
# A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more
# details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with Paramiko; if not, write to the Free Software Foundation, Inc.,
# Temple Place, Suite , Boston, MA - USA. import socket
import sys
from paramiko.py3compat import u
import time
import os fiedir = '/tmp'
logfile = 'history_log.txt' # windows does not have termios...
try:
import termios
import tty
has_termios = True
except ImportError:
has_termios = False def interactive_shell(chan,default_username,hostname,username):
if has_termios:
posix_shell(chan,default_username,hostname,username)
else:
windows_shell(chan,default_username,hostname,username) def posix_shell(chan,default_username,hostname,username):
import select oldtty = termios.tcgetattr(sys.stdin)
try:
tty.setraw(sys.stdin.fileno())
tty.setcbreak(sys.stdin.fileno())
chan.settimeout(0.0)
res_list = []
file_dir = os.path.join(fiedir,logfile)
with open(file_dir,'ab+') as f:
while True:
r, w, e = select.select([chan, sys.stdin], [], [])
if chan in r:
try:
x = u(chan.recv())
if len(x) == :
sys.stdout.write('\r\n*** EOF\r\n')
break
sys.stdout.write(x)
sys.stdout.flush()
except socket.timeout:
pass
if sys.stdin in r:
x = sys.stdin.read()
res_list.append(x)
if x == '\r':
cmd =''.join(res_list).replace('\r','\n') c_time = time.strftime('%Y-%m-%d %H:%M:%S')
filename = '%s %s %s %s %s'%(c_time,default_username,username,hostname,cmd)
#filename = '%s %s'%(c_time,cmd)
f.write(filename)
res_list = []
if len(x) == :
break
chan.send(x) finally:
termios.tcsetattr(sys.stdin, termios.TCSADRAIN, oldtty) # thanks to Mike Looijmans for this code
def windows_shell(chan):
import threading 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()
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()
if not d:
break
chan.send(d)
except EOFError:
# user hit ^Z or F6
pass
为了安全，需要在用户名的环境变量加载demo.py这个脚本
vim ~/.bashrc
python /home/feng/data/paramiko-master/demos/demo.py
logout
登陆结果如下，退出后，直接退出终端
