diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/__init__.py | 36 | ||||
| -rw-r--r-- | tests/stub_sftp.py | 14 | ||||
| -rw-r--r-- | tests/test_auth.py | 19 | ||||
| -rw-r--r-- | tests/test_client.py | 30 | ||||
| -rwxr-xr-x | tests/test_file.py | 57 | ||||
| -rw-r--r-- | tests/test_pkey.py | 35 | ||||
| -rwxr-xr-x | tests/test_sftp.py | 63 | ||||
| -rw-r--r-- | tests/test_transport.py | 69 |
8 files changed, 303 insertions, 20 deletions
diff --git a/tests/__init__.py b/tests/__init__.py index e69de29b..8878f14d 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -0,0 +1,36 @@ +# Copyright (C) 2017 Martin Packman <gzlist@googlemail.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., +# 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA. + +"""Base classes and helpers for testing paramiko.""" + +import unittest + +from paramiko.py3compat import ( + builtins, + ) + + +def skipUnlessBuiltin(name): + """Skip decorated test if builtin name does not exist.""" + if getattr(builtins, name, None) is None: + skip = getattr(unittest, "skip", None) + if skip is None: + # Python 2.6 pseudo-skip + return lambda func: None + return skip("No builtin " + repr(name)) + return lambda func: func diff --git a/tests/stub_sftp.py b/tests/stub_sftp.py index 334af561..0d673091 100644 --- a/tests/stub_sftp.py +++ b/tests/stub_sftp.py @@ -24,7 +24,7 @@ import os import sys from paramiko import ( ServerInterface, SFTPServerInterface, SFTPServer, SFTPAttributes, - SFTPHandle, SFTP_OK, AUTH_SUCCESSFUL, OPEN_SUCCEEDED, + SFTPHandle, SFTP_OK, SFTP_FAILURE, AUTH_SUCCESSFUL, OPEN_SUCCEEDED, ) from paramiko.common import o666 @@ -141,12 +141,24 @@ class StubSFTPServer (SFTPServerInterface): def rename(self, oldpath, newpath): oldpath = self._realpath(oldpath) newpath = self._realpath(newpath) + if os.path.exists(newpath): + return SFTP_FAILURE try: os.rename(oldpath, newpath) except OSError as e: return SFTPServer.convert_errno(e.errno) return SFTP_OK + def posix_rename(self, oldpath, newpath): + oldpath = self._realpath(oldpath) + newpath = self._realpath(newpath) + try: + os.rename(oldpath, newpath) + except OSError as e: + return SFTPServer.convert_errno(e.errno) + return SFTP_OK + + def mkdir(self, path, attr): path = self._realpath(path) try: diff --git a/tests/test_auth.py b/tests/test_auth.py index 96f7611c..e78397c6 100644 --- a/tests/test_auth.py +++ b/tests/test_auth.py @@ -23,6 +23,7 @@ Some unit tests for authenticating over a Transport. import sys import threading import unittest +from time import sleep from paramiko import ( Transport, ServerInterface, RSAKey, DSSKey, BadAuthenticationType, @@ -74,6 +75,9 @@ class NullServer (ServerInterface): return AUTH_SUCCESSFUL if username == 'bad-server': raise Exception("Ack!") + if username == 'unresponsive-server': + sleep(5) + return AUTH_SUCCESSFUL return AUTH_FAILED def check_auth_publickey(self, username, key): @@ -233,3 +237,18 @@ class AuthTest (unittest.TestCase): except: etype, evalue, etb = sys.exc_info() self.assertTrue(issubclass(etype, AuthenticationException)) + + def test_9_auth_non_responsive(self): + """ + verify that authentication times out if server takes to long to + respond (or never responds). + """ + self.tc.auth_timeout = 1 # 1 second, to speed up test + self.start_server() + self.tc.connect() + try: + remain = self.tc.auth_password('unresponsive-server', 'hello') + except: + etype, evalue, etb = sys.exc_info() + self.assertTrue(issubclass(etype, AuthenticationException)) + self.assertTrue('Authentication timeout' in str(evalue)) diff --git a/tests/test_client.py b/tests/test_client.py index 3a9001e2..bddaf4bc 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -36,7 +36,7 @@ from tests.util import test_path import paramiko from paramiko.common import PY2 -from paramiko.ssh_exception import SSHException +from paramiko.ssh_exception import SSHException, AuthenticationException FINGERPRINTS = { @@ -61,6 +61,9 @@ class NullServer (paramiko.ServerInterface): def check_auth_password(self, username, password): if (username == 'slowdive') and (password == 'pygmalion'): return paramiko.AUTH_SUCCESSFUL + if (username == 'slowdive') and (password == 'unresponsive-server'): + time.sleep(5) + return paramiko.AUTH_SUCCESSFUL return paramiko.AUTH_FAILED def check_auth_publickey(self, username, key): @@ -294,13 +297,10 @@ class SSHClientTest (unittest.TestCase): verify that when an SSHClient is collected, its transport (and the transport's packetizer) is closed. """ - # Unclear why this is borked on Py3, but it is, and does not seem worth - # pursuing at the moment. Skipped on PyPy because it fails on travis - # for unknown reasons, works fine locally. - # XXX: It's the release of the references to e.g packetizer that fails - # in py3... - if not PY2 or platform.python_implementation() == "PyPy": + # Skipped on PyPy because it fails on travis for unknown reasons + if platform.python_implementation() == "PyPy": return + threading.Thread(target=self._run).start() self.tc = paramiko.SSHClient() @@ -318,8 +318,8 @@ class SSHClientTest (unittest.TestCase): del self.tc # force a collection to see whether the SSHClient object is deallocated - # correctly. 2 GCs are needed to make sure it's really collected on - # PyPy + # 2 GCs are needed on PyPy, time is needed for Python 3 + time.sleep(0.3) gc.collect() gc.collect() @@ -384,6 +384,18 @@ class SSHClientTest (unittest.TestCase): ) self._test_connection(**kwargs) + def test_9_auth_timeout(self): + """ + verify that the SSHClient has a configurable auth timeout + """ + # Connect with a half second auth timeout + self.assertRaises( + AuthenticationException, + self._test_connection, + password='unresponsive-server', + auth_timeout=0.5, + ) + def test_update_environment(self): """ Verify that environment variables can be set by the client. diff --git a/tests/test_file.py b/tests/test_file.py index 7fab6985..b33ecd51 100755 --- a/tests/test_file.py +++ b/tests/test_file.py @@ -21,10 +21,14 @@ Some unit tests for the BufferedFile abstraction. """ import unittest -from paramiko.file import BufferedFile -from paramiko.common import linefeed_byte, crlf, cr_byte import sys +from paramiko.common import linefeed_byte, crlf, cr_byte +from paramiko.file import BufferedFile +from paramiko.py3compat import BytesIO + +from tests import skipUnlessBuiltin + class LoopbackFile (BufferedFile): """ @@ -33,19 +37,16 @@ class LoopbackFile (BufferedFile): def __init__(self, mode='r', bufsize=-1): BufferedFile.__init__(self) self._set_mode(mode, bufsize) - self.buffer = bytes() + self.buffer = BytesIO() + self.offset = 0 def _read(self, size): - if len(self.buffer) == 0: - return None - if size > len(self.buffer): - size = len(self.buffer) - data = self.buffer[:size] - self.buffer = self.buffer[size:] + data = self.buffer.getvalue()[self.offset:self.offset+size] + self.offset += len(data) return data def _write(self, data): - self.buffer += data + self.buffer.write(data) return len(data) @@ -187,6 +188,42 @@ class BufferedFileTest (unittest.TestCase): self.assertEqual(data, b'hello') f.close() + def test_write_bad_type(self): + with LoopbackFile('wb') as f: + self.assertRaises(TypeError, f.write, object()) + + def test_write_unicode_as_binary(self): + text = u"\xa7 why is writing text to a binary file allowed?\n" + with LoopbackFile('rb+') as f: + f.write(text) + self.assertEqual(f.read(), text.encode("utf-8")) + + @skipUnlessBuiltin('memoryview') + def test_write_bytearray(self): + with LoopbackFile('rb+') as f: + f.write(bytearray(12)) + self.assertEqual(f.read(), 12 * b"\0") + + @skipUnlessBuiltin('buffer') + def test_write_buffer(self): + data = 3 * b"pretend giant block of data\n" + offsets = range(0, len(data), 8) + with LoopbackFile('rb+') as f: + for offset in offsets: + f.write(buffer(data, offset, 8)) + self.assertEqual(f.read(), data) + + @skipUnlessBuiltin('memoryview') + def test_write_memoryview(self): + data = 3 * b"pretend giant block of data\n" + offsets = range(0, len(data), 8) + with LoopbackFile('rb+') as f: + view = memoryview(data) + for offset in offsets: + f.write(view[offset:offset+8]) + self.assertEqual(f.read(), data) + + if __name__ == '__main__': from unittest import main main() diff --git a/tests/test_pkey.py b/tests/test_pkey.py index a26ff170..6e589915 100644 --- a/tests/test_pkey.py +++ b/tests/test_pkey.py @@ -113,6 +113,25 @@ TEST_KEY_BYTESTR_3 = '\x00\x00\x00\x07ssh-rsa\x00\x00\x00\x01#\x00\x00\x00\x00ÓŹ class KeyTest(unittest.TestCase): + + def setUp(self): + pass + + def tearDown(self): + pass + + def assert_keyfile_is_encrypted(self, keyfile): + """ + A quick check that filename looks like an encrypted key. + """ + with open(keyfile, "r") as fh: + self.assertEqual( + fh.readline()[:-1], + "-----BEGIN RSA PRIVATE KEY-----" + ) + self.assertEqual(fh.readline()[:-1], "Proc-Type: 4,ENCRYPTED") + self.assertEqual(fh.readline()[0:10], "DEK-Info: ") + def test_1_generate_key_bytes(self): key = util.generate_key_bytes(md5, x1234, 'happy birthday', 30) exp = b'\x61\xE1\xF2\x72\xF4\xC1\xC4\x56\x15\x86\xBD\x32\x24\x98\xC0\xE9\x24\x67\x27\x80\xF4\x7B\xB3\x7D\xDA\x7D\x54\x01\x9E\x64' @@ -419,6 +438,7 @@ class KeyTest(unittest.TestCase): # When the bug under test exists, this will ValueError. try: key.write_private_key_file(newfile, password=newpassword) + self.assert_keyfile_is_encrypted(newfile) # Verify the inner key data still matches (when no ValueError) key2 = RSAKey(filename=newfile, password=newpassword) self.assertEqual(key, key2) @@ -437,3 +457,18 @@ class KeyTest(unittest.TestCase): ) self.assertNotEqual(key1.asbytes(), key2.asbytes()) + + def test_keyfile_is_actually_encrypted(self): + # Read an existing encrypted private key + file_ = test_path('test_rsa_password.key') + password = 'television' + newfile = file_ + '.new' + newpassword = 'radio' + key = RSAKey(filename=file_, password=password) + # Write out a newly re-encrypted copy with a new password. + # When the bug under test exists, this will ValueError. + try: + key.write_private_key_file(newfile, password=newpassword) + self.assert_keyfile_is_encrypted(newfile) + finally: + os.remove(newfile) diff --git a/tests/test_sftp.py b/tests/test_sftp.py index d3064fff..b3c7bf98 100755 --- a/tests/test_sftp.py +++ b/tests/test_sftp.py @@ -35,6 +35,7 @@ from tempfile import mkstemp import paramiko from paramiko.py3compat import PY2, b, u, StringIO from paramiko.common import o777, o600, o666, o644 +from tests import skipUnlessBuiltin from tests.stub_sftp import StubServer, StubSFTPServer from tests.loop import LoopSocket from tests.util import test_path @@ -276,6 +277,39 @@ class SFTPTest (unittest.TestCase): except: pass + + def test_5a_posix_rename(self): + """Test posix-rename@openssh.com protocol extension.""" + try: + # first check that the normal rename works as specified + with sftp.open(FOLDER + '/a', 'w') as f: + f.write('one') + sftp.rename(FOLDER + '/a', FOLDER + '/b') + with sftp.open(FOLDER + '/a', 'w') as f: + f.write('two') + try: + sftp.rename(FOLDER + '/a', FOLDER + '/b') + self.assertTrue(False, 'no exception when rename-ing onto existing file') + except (OSError, IOError): + pass + + # now check with the posix_rename + sftp.posix_rename(FOLDER + '/a', FOLDER + '/b') + with sftp.open(FOLDER + '/b', 'r') as f: + data = u(f.read()) + self.assertEqual('two', data, "Contents of renamed file not the same as original file") + + finally: + try: + sftp.remove(FOLDER + '/a') + except: + pass + try: + sftp.remove(FOLDER + '/b') + except: + pass + + def test_6_folder(self): """ create a temporary folder, verify that we can create a file in it, then @@ -817,6 +851,35 @@ class SFTPTest (unittest.TestCase): sftp_attributes = SFTPAttributes() self.assertEqual(str(sftp_attributes), "?--------- 1 0 0 0 (unknown date) ?") + @skipUnlessBuiltin('buffer') + def test_write_buffer(self): + """Test write() using a buffer instance.""" + data = 3 * b'A potentially large block of data to chunk up.\n' + try: + with sftp.open('%s/write_buffer' % FOLDER, 'wb') as f: + for offset in range(0, len(data), 8): + f.write(buffer(data, offset, 8)) + + with sftp.open('%s/write_buffer' % FOLDER, 'rb') as f: + self.assertEqual(f.read(), data) + finally: + sftp.remove('%s/write_buffer' % FOLDER) + + @skipUnlessBuiltin('memoryview') + def test_write_memoryview(self): + """Test write() using a memoryview instance.""" + data = 3 * b'A potentially large block of data to chunk up.\n' + try: + with sftp.open('%s/write_memoryview' % FOLDER, 'wb') as f: + view = memoryview(data) + for offset in range(0, len(data), 8): + f.write(view[offset:offset+8]) + + with sftp.open('%s/write_memoryview' % FOLDER, 'rb') as f: + self.assertEqual(f.read(), data) + finally: + sftp.remove('%s/write_memoryview' % FOLDER) + if __name__ == '__main__': SFTPTest.init_loopback() diff --git a/tests/test_transport.py b/tests/test_transport.py index c426cef1..3e352919 100644 --- a/tests/test_transport.py +++ b/tests/test_transport.py @@ -43,6 +43,7 @@ from paramiko.common import ( ) from paramiko.py3compat import bytes from paramiko.message import Message +from tests import skipUnlessBuiltin from tests.loop import LoopSocket from tests.util import test_path @@ -858,3 +859,71 @@ class TransportTest(unittest.TestCase): self.assertEqual([chan], r) self.assertEqual([], w) self.assertEqual([], e) + + def test_channel_send_misc(self): + """ + verify behaviours sending various instances to a channel + """ + self.setup_test_server() + text = u"\xa7 slice me nicely" + with self.tc.open_session() as chan: + schan = self.ts.accept(1.0) + if schan is None: + self.fail("Test server transport failed to accept") + sfile = schan.makefile() + + # TypeError raised on non string or buffer type + self.assertRaises(TypeError, chan.send, object()) + self.assertRaises(TypeError, chan.sendall, object()) + + # sendall() accepts a unicode instance + chan.sendall(text) + expected = text.encode("utf-8") + self.assertEqual(sfile.read(len(expected)), expected) + + @skipUnlessBuiltin('buffer') + def test_channel_send_buffer(self): + """ + verify sending buffer instances to a channel + """ + self.setup_test_server() + data = 3 * b'some test data\n whole' + with self.tc.open_session() as chan: + schan = self.ts.accept(1.0) + if schan is None: + self.fail("Test server transport failed to accept") + sfile = schan.makefile() + + # send() accepts buffer instances + sent = 0 + while sent < len(data): + sent += chan.send(buffer(data, sent, 8)) + self.assertEqual(sfile.read(len(data)), data) + + # sendall() accepts a buffer instance + chan.sendall(buffer(data)) + self.assertEqual(sfile.read(len(data)), data) + + @skipUnlessBuiltin('memoryview') + def test_channel_send_memoryview(self): + """ + verify sending memoryview instances to a channel + """ + self.setup_test_server() + data = 3 * b'some test data\n whole' + with self.tc.open_session() as chan: + schan = self.ts.accept(1.0) + if schan is None: + self.fail("Test server transport failed to accept") + sfile = schan.makefile() + + # send() accepts memoryview slices + sent = 0 + view = memoryview(data) + while sent < len(view): + sent += chan.send(view[sent:sent+8]) + self.assertEqual(sfile.read(len(data)), data) + + # sendall() accepts a memoryview instance + chan.sendall(memoryview(data)) + self.assertEqual(sfile.read(len(data)), data) |