dorian
/
paramiko
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

[project @ robey@master-shake.local-20060127030114-b6810c4c29964a4e] 

code from mike looijmans for a windows interactive shell -- while i'm mucking around in here, clean up the demos a bit too
This commit is contained in:
Robey Pointer 2006-01-26 19:01:14 -08:00
parent 7058195e38
commit 0efdc8af5f
4 changed files with 151 additions and 115 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

109
demo.py → demos/demo.py
View File

@ -1,4 +1,4 @@
#!/usr/bin/python
#!/usr/bin/env python
# Copyright (C) 2003-2005 Robey Pointer <robey@lag.net>
#
@ -19,38 +19,49 @@
# 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA.
# ----- WINDOWS USERS PLEASE NOTE -----
# This demo won't work on Windows because it uses pseudo-terminals, which
# are a posix-only feature. check out the README file for a simpler demo.
import base64
import getpass
import os
import select
import socket
import sys
import threading
import time
import traceback
import sys, os, socket, threading, getpass, time, base64, select, termios, tty, traceback
import paramiko
import interactive
##### utility functions
def agent_auth(transport, username):
"""
Attempt to authenticate to the given transport using any of the private
keys available from an SSH agent.
"""
def agent_auth(username, t, event):
agent = paramiko.Agent()
agent_keys = agent.get_keys()
if len(agent_keys) > 0:
for key in agent_keys:
print 'Trying ssh-agent key %s' % paramiko.util.hexify(key.get_fingerprint()),
t.auth_publickey(username, key, event)
event.wait(10)
if t.is_authenticated():
print '... success!'
return
if len(agent_keys) == 0:
return
for key in agent_keys:
print 'Trying ssh-agent key %s' % paramiko.util.hexify(key.get_fingerprint()),
try:
transport.auth_publickey(username, key)
print '... success!'
return
except SSHException:
print '... nope.'
def manual_auth(username, hostname, event):
def manual_auth(username, hostname):
default_auth = 'p'
auth = raw_input('Auth by (p)assword, (r)sa key, or (d)ss key? [%s] ' % default_auth)
if len(auth) == 0:
auth = default_auth
if auth == 'r':
default_path = os.environ['HOME'] + '/.ssh/id_rsa'
default_path = os.path.join(os.environ['HOME'], '.ssh', 'id_rsa')
path = raw_input('RSA key [%s]: ' % default_path)
if len(path) == 0:
path = default_path
@ -59,9 +70,9 @@ def manual_auth(username, hostname, event):
except paramiko.PasswordRequiredException:
password = getpass.getpass('RSA key password: ')
key = paramiko.RSAKey.from_private_key_file(path, password)
t.auth_publickey(username, key, event)
t.auth_publickey(username, key)
elif auth == 'd':
default_path = os.environ['HOME'] + '/.ssh/id_dsa'
default_path = os.path.join(os.environ['HOME'], '.ssh', 'id_dsa')
path = raw_input('DSS key [%s]: ' % default_path)
if len(path) == 0:
path = default_path
@ -70,18 +81,15 @@ def manual_auth(username, hostname, event):
except paramiko.PasswordRequiredException:
password = getpass.getpass('DSS key password: ')
key = paramiko.DSSKey.from_private_key_file(path, password)
t.auth_publickey(username, key, event)
t.auth_publickey(username, key)
else:
pw = getpass.getpass('Password for %s@%s: ' % (username, hostname))
t.auth_password(username, pw, event)
t.auth_password(username, pw)
##### main demo
# setup logging
paramiko.util.log_to_file('demo.log')
username = ''
if len(sys.argv) > 1:
hostname = sys.argv[1]
@ -107,15 +115,11 @@ except Exception, e:
sys.exit(1)
try:
event = threading.Event()
t = paramiko.Transport(sock)
t.start_client(event)
# print repr(t)
event.wait(15)
t.start_client()
if not t.is_active():
print '*** SSH negotiation failed.'
sys.exit(1)
# print repr(t)
try:
keys = paramiko.util.load_host_keys(os.path.expanduser('~/.ssh/known_hosts'))
@ -137,8 +141,6 @@ try:
else:
print '*** Host key OK.'
event.clear()
# get username
if username == '':
default_username = getpass.getuser()
@ -146,14 +148,9 @@ try:
if len(username) == 0:
username = default_username
agent_auth(username, t, event)
# ask for what kind of authentication to try
agent_auth(t, username)
if not t.is_authenticated():
manual_auth(username, hostname, event)
event.wait(10)
# print repr(t)
manual_auth(username, hostname)
if not t.is_authenticated():
print '*** Authentication failed. :('
t.close()
@ -164,38 +161,7 @@ try:
chan.invoke_shell()
print '*** Here we go!'
print
try:
oldtty = termios.tcgetattr(sys.stdin)
tty.setraw(sys.stdin.fileno())
tty.setcbreak(sys.stdin.fileno())
chan.settimeout(0.0)
while 1:
r, w, e = select.select([chan, sys.stdin], [], [])
if chan in r:
try:
x = chan.recv(1024)
if len(x) == 0:
print
print '*** EOF\r\n',
break
sys.stdout.write(x)
sys.stdout.flush()
except socket.timeout:
pass
if sys.stdin in r:
# FIXME: reading 1 byte at a time is incredibly dumb.
x = sys.stdin.read(1)
if len(x) == 0:
print
print '*** Bye.\r\n',
break
chan.send(x)
finally:
termios.tcsetattr(sys.stdin, termios.TCSADRAIN, oldtty)
interactive.interactive_shell(chan)
chan.close()
t.close()
@ -208,3 +174,4 @@ except Exception, e:
pass
sys.exit(1)

0
demo_server.py → demos/demo_server.py
View File

58
demo_simple.py → demos/demo_simple.py
View File

@ -1,4 +1,4 @@
#!/usr/bin/python
#!/usr/bin/env python
# Copyright (C) 2003-2005 Robey Pointer <robey@lag.net>
#
@ -19,13 +19,15 @@
# 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA.
# ----- WINDOWS USERS PLEASE NOTE -----
# This demo won't work on Windows because it uses pseudo-terminals, which
# are a posix-only feature. check out the README file for a simpler demo.
import base64
import getpass
import os
import socket
import sys
import traceback
import sys, os, base64, getpass, socket, traceback, termios, tty, select
import paramiko
import interactive
# setup logging
@ -61,17 +63,17 @@ password = getpass.getpass('Password for %s@%s: ' % (username, hostname))
hostkeytype = None
hostkey = None
try:
hkeys = paramiko.util.load_host_keys(os.path.expanduser('~/.ssh/known_hosts'))
host_keys = paramiko.util.load_host_keys(os.path.expanduser('~/.ssh/known_hosts'))
except IOError:
try:
hkeys = paramiko.util.load_host_keys(os.path.expanduser('~/ssh/known_hosts'))
host_keys = paramiko.util.load_host_keys(os.path.expanduser('~/ssh/known_hosts'))
except IOError:
print '*** Unable to open host keys file'
hkeys = {}
host_keys = {}
if hkeys.has_key(hostname):
hostkeytype = hkeys[hostname].keys()[0]
hostkey = hkeys[hostname][hostkeytype]
if host_keys.has_key(hostname):
hostkeytype = host_keys[hostname].keys()[0]
hostkey = host_keys[hostname][hostkeytype]
print 'Using host key of type %s' % hostkeytype
@ -84,37 +86,7 @@ try:
chan.invoke_shell()
print '*** Here we go!'
print
try:
oldtty = termios.tcgetattr(sys.stdin)
tty.setraw(sys.stdin.fileno())
tty.setcbreak(sys.stdin.fileno())
chan.settimeout(0.0)
while True:
r, w, e = select.select([chan, sys.stdin], [], [])
if chan in r:
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:
# FIXME: reading 1 byte at a time is incredibly dumb.
x = sys.stdin.read(1)
if len(x) == 0:
print
print '*** Bye.\r\n',
break
chan.send(x)
finally:
termios.tcsetattr(sys.stdin, termios.TCSADRAIN, oldtty)
interactive.interactive_shell(chan)
chan.close()
t.close()

97
demos/interactive.py Normal file
View File

@ -0,0 +1,97 @@
# Copyright (C) 2003-2005 Robey Pointer <robey@lag.net>
#
# 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 distrubuted 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.,
# 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA.
import socket
import sys
# windows does not have termios...
try:
import termios
import tty
has_termios = True
except ImportError:
has_termios = False
def interactive_shell(chan):
if has_termios:
posix_shell(chan)
else:
windows_shell(chan)
def posix_shell(chan):
import select
oldtty = termios.tcgetattr(sys.stdin)
try:
tty.setraw(sys.stdin.fileno())
tty.setcbreak(sys.stdin.fileno())
chan.settimeout(0.0)
while True:
r, w, e = select.select([chan, sys.stdin], [], [])
if chan in r:
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:
x = sys.stdin.read(1)
if len(x) == 0:
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(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