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

[project @ Arch-1:robey@lag.net--2003-public%secsh--dev--1.0--patch-94] 

start testing Transport
the beginnings of tests for Transport.  only the bare minimum is there right
now.

also started doc'ing things up to ivysaur.
This commit is contained in:
Robey Pointer 2004-10-20 16:52:51 +00:00
parent 2939b6936b
commit 6caf15b425
8 changed files with 334 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
Makefile
View File

@ -7,6 +7,7 @@
# fearow (23apr04)
# gyarados (31may04)
# horsea (27jun04)
# ivysaur (20oct04)
release:
python ./setup.py sdist --formats=zip

29
README
View File

@ -1,5 +1,5 @@
paramiko 0.9
"horsea" release, 27 jun 2004
"ivysaur" release, 20 oct 2004
Copyright (c) 2003-2004 Robey Pointer <robey@lag.net>
@ -93,6 +93,10 @@ which hopefully demonstrates how you can use the paramiko library.
a simpler example is in demo_simple.py, which is a copy of the demo client
that uses the simpler "connect" method call (new with 0.9-doduo).
a demo for windows is in demo_windows.py. it executes 'ls' on the remote
server and prints the results, avoiding terminal i/o and select() (which
are missing in windows).
there's also now a demo server (demo_server.py) which listens on port 2200
and accepts a login (robey/foo) and pretends to be a BBS, just to demonstrate
how to perform the server side of things.
@ -100,12 +104,12 @@ how to perform the server side of things.
*** USE
the demo clients (demo.py & demo_simple.py) and the demo server
(demo_server.py) are probably the best example of how to use this package.
there is also a lot of documentation, generated with epydoc, in the doc/
folder. point your browser there. seriously, do it. mad props to epydoc,
which actually motivated me to write more documentation than i ever would
have before.
the demo clients (demo.py, demo_simple.py, and demo_windows.py) and the demo
server (demo_server.py) are probably the best example of how to use this
package. there is also a lot of documentation, generated with epydoc, in the
doc/ folder. point your browser there. seriously, do it. mad props to
epydoc, which actually motivated me to write more documentation than i ever
would have before.
there are also unit tests here:
$ python ./test.py
@ -118,6 +122,16 @@ the best and easiest examples of how to use the SFTP class.
highlights of what's new in each release:
v0.9 IVYSAUR
* new ServerInterface class for implementing server policy, so it's no longer
necessary to subclass Transport
* some bugfixes for re-keying an active session
* Transport.get_security_options() allows fine-tuned control over the crypto
negotiation on a new session
* .......?
v0.9 HORSEA
* fixed a lockup that could happen if the channel was closed while the send
window was full
@ -157,4 +171,3 @@ v0.9 FEAROW
* server mode needs better documentation
* sftp server mode
ivysaur?

104
demo_windows.py Normal file
View File

@ -0,0 +1,104 @@
#!/usr/bin/python
# This demo is like demo_simple.py, but it doesn't try to use select()
# to poll the ssh channel for reading, so it can be used on Windows.
# It logs into a shell, executes "ls", prints out the results, and
# exits.
import sys, os, base64, getpass, socket, traceback
import paramiko
##### utility functions
def load_host_keys():
filename = os.environ['HOME'] + '/.ssh/known_hosts'
keys = {}
try:
f = open(filename, 'r')
except Exception, e:
print '*** Unable to open host keys file (%s)' % filename
return
for line in f:
keylist = line.split(' ')
if len(keylist) != 3:
continue
hostlist, keytype, key = keylist
hosts = hostlist.split(',')
for host in hosts:
if not keys.has_key(host):
keys[host] = {}
if keytype == 'ssh-rsa':
keys[host][keytype] = paramiko.RSAKey(data=base64.decodestring(key))
elif keytype == 'ssh-dss':
keys[host][keytype] = paramiko.DSSKey(data=base64.decodestring(key))
f.close()
return keys
# setup logging
paramiko.util.log_to_file('demo.log')
# get hostname
username = ''
if len(sys.argv) > 1:
hostname = sys.argv[1]
if hostname.find('@') >= 0:
username, hostname = hostname.split('@')
else:
hostname = raw_input('Hostname: ')
if len(hostname) == 0:
print '*** Hostname required.'
sys.exit(1)
port = 22
if hostname.find(':') >= 0:
hostname, portstr = hostname.split(':')
port = int(portstr)
# get username
if username == '':
default_username = getpass.getuser()
username = raw_input('Username [%s]: ' % default_username)
if len(username) == 0:
username = default_username
password = getpass.getpass('Password for %s@%s: ' % (username, hostname))
# get host key, if we know one
hostkeytype = None
hostkey = None
hkeys = load_host_keys()
if hkeys.has_key(hostname):
hostkeytype = hkeys[hostname].keys()[0]
hostkey = hkeys[hostname][hostkeytype]
print 'Using host key of type %s' % hostkeytype
# now, connect and use paramiko Transport to negotiate SSH2 across the connection
try:
t = paramiko.Transport((hostname, port))
t.connect(username=username, password=password, hostkey=hostkey)
chan = t.open_session()
chan.get_pty()
print '*** Here we go!'
print
print '>>> ls'
chan.exec_command('ps auxww')
f = chan.makefile('r+')
for line in f:
print line.strip('\n')
chan.close()
t.close()
except Exception, e:
print '*** Caught exception: ' + str(e.__class__) + ': ' + str(e)
traceback.print_exc()
try:
t.close()
except:
pass
sys.exit(1)

6
paramiko/__init__.py
View File

@ -48,7 +48,7 @@ released under the GNU Lesser General Public License (LGPL).
Website: U{http://www.lag.net/~robey/paramiko/}
@version: 0.9 (horsea)
@version: 0.9 (ivysaur)
@author: Robey Pointer
@contact: robey@lag.net
@license: GNU Lesser General Public License (LGPL)
@ -61,8 +61,8 @@ if sys.version_info < (2, 2):
__author__ = "Robey Pointer <robey@lag.net>"
__date__ = "31 May 2004"
__version__ = "0.9-horsea"
__date__ = "20 Oct 2004"
__version__ = "0.9-ivysaur"
__license__ = "GNU Lesser General Public License (LGPL)"

4
setup.py
View File

@ -13,13 +13,13 @@ Required packages:
'''
setup(name = "paramiko",
version = "0.9-horsea",
version = "0.9-ivysaur",
description = "SSH2 protocol library",
author = "Robey Pointer",
author_email = "robey@lag.net",
url = "http://www.lag.net/~robey/paramiko/",
packages = [ 'paramiko' ],
download_url = 'http://www.lag.net/~robey/paramiko/paramiko-0.9-horsea.zip',
download_url = 'http://www.lag.net/~robey/paramiko/paramiko-0.9-ivysaur.zip',
license = 'LGPL',
platforms = 'Posix; MacOS X; Windows',
classifiers = [ 'Development Status :: 3 - Alpha',

7
test.py
View File

@ -31,7 +31,7 @@ sys.path.append('tests/')
from test_message import MessageTest
from test_file import BufferedFileTest
from test_pkey import KeyTest
#from test_transport import TransportTest
from test_transport import TransportTest
from test_sftp import SFTPTest
default_host = 'localhost'
@ -54,6 +54,8 @@ parser.add_option('-K', '--sftp-key', dest='keyfile', type='string', default=def
parser.add_option('-P', '--sftp-passwd', dest='password', type='string', default=default_passwd,
metavar='<password>',
help='(optional) password to unlock the private key for sftp tests')
parser.add_option('--no-pkey', action='store_false', dest='use_pkey', default=True,
help='skip RSA/DSS private key tests (which can take a while)')
options, args = parser.parse_args()
if len(args) > 0:
@ -68,8 +70,9 @@ paramiko.util.log_to_file('test.log')
suite = unittest.TestSuite()
suite.addTest(unittest.makeSuite(MessageTest))
suite.addTest(unittest.makeSuite(BufferedFileTest))
if options.use_pkey:
suite.addTest(unittest.makeSuite(KeyTest))
#suite.addTest(unittest.makeSuite(TransportTest))
suite.addTest(unittest.makeSuite(TransportTest))
if options.use_sftp:
suite.addTest(unittest.makeSuite(SFTPTest))
unittest.TextTestRunner(verbosity=2).run(suite)

104
tests/loop.py Normal file
View File

@ -0,0 +1,104 @@
#!/usr/bin/python
# Copyright (C) 2003-2004 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 threading, socket
class LoopSocket (object):
"""
A LoopSocket looks like a normal socket, but all data written to it is
delivered on the read-end of another LoopSocket, and vice versa, It's
like a software "socketpair".
"""
def __init__(self):
self.__in_buffer = ''
self.__lock = threading.Lock()
self.__cv = threading.Condition(self.__lock)
self.__timeout = None
self.__mate = None
def close(self):
self.__unlink()
try:
self.__lock.acquire()
self.__in_buffer = ''
finally:
self.__lock.release()
def send(self, data):
if self.__mate is None:
# EOF
raise EOFError()
self.__mate.__feed(data)
return len(data)
def recv(self, n):
self.__lock.acquire()
try:
if self.__mate is None:
# EOF
return ''
if len(self.__in_buffer) == 0:
self.__cv.wait(self.__timeout)
if len(self.__in_buffer) == 0:
raise socket.timeout
if n < self.__in_buffer:
out = self.__in_buffer[:n]
self.__in_buffer = self.__in_buffer[n:]
else:
out = self.__in_buffer
self.__in_buffer = ''
return out
finally:
self.__lock.release()
def settimeout(self, n):
self.__timeout = n
def link(self, other):
self.__mate = other
self.__mate.__mate = self
def __feed(self, data):
self.__lock.acquire()
try:
self.__in_buffer += data
self.__cv.notifyAll()
finally:
self.__lock.release()
def __unlink(self):
m = None
self.__lock.acquire()
try:
if self.__mate is not None:
m = self.__mate
self.__mate = None
finally:
self.__lock.release()
if m is not None:
m.__unlink()

93
tests/test_transport.py Normal file
View File

@ -0,0 +1,93 @@
#!/usr/bin/python
# Copyright (C) 2003-2004 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.
"""
Some unit tests for the ssh2 protocol in Transport.
"""
import unittest, threading
from paramiko import Transport, SecurityOptions, ServerInterface, RSAKey, DSSKey
from paramiko import AUTH_FAILED, AUTH_SUCCESSFUL
from loop import LoopSocket
class NullServer (ServerInterface):
def get_allowed_auths(self, username):
return 'publickey'
def check_auth_password(self, username, password):
if (username == 'slowdive') and (password == 'pygmalion'):
return AUTH_SUCCESSFUL
return AUTH_FAILED
class TransportTest (unittest.TestCase):
def setUp(self):
self.socks = LoopSocket()
self.sockc = LoopSocket()
self.sockc.link(self.socks)
self.tc = Transport(self.sockc)
self.ts = Transport(self.socks)
def tearDown(self):
self.tc.close()
self.ts.close()
self.socks.close()
self.sockc.close()
def test_1_security_options(self):
o = self.tc.get_security_options()
self.assertEquals(type(o), SecurityOptions)
self.assert_(('aes256-cbc', 'blowfish-cbc') != o.ciphers)
o.ciphers = ('aes256-cbc', 'blowfish-cbc')
self.assertEquals(('aes256-cbc', 'blowfish-cbc'), o.ciphers)
try:
o.ciphers = ('aes256-cbc', 'made-up-cipher')
self.assert_(False)
except ValueError:
pass
try:
o.ciphers = 23
self.assert_(False)
except TypeError:
pass
def test_2_simple(self):
"""
verify that we can establish an ssh link with ourselves across the
loopback sockets. this is hardly "simple" but it's simpler than the
later tests. :)
"""
host_key = RSAKey.from_private_key_file('tests/test_rsa.key')
public_host_key = RSAKey(data=str(host_key))
self.ts.add_server_key(host_key)
event = threading.Event()
server = NullServer()
self.assert_(not event.isSet())
self.ts.start_server(event, server)
self.tc.ultra_debug = True
self.tc.connect(hostkey=public_host_key,
username='slowdive', password='pygmalion')
event.wait(1.0)
self.assert_(event.isSet())
self.assert_(self.ts.is_active())