diff options
author | Jeff Forcier <jeff@bitprophet.org> | 2017-06-06 12:31:57 -0700 |
---|---|---|
committer | Jeff Forcier <jeff@bitprophet.org> | 2017-06-06 12:31:57 -0700 |
commit | a2da21d46bb9a441dbb8da570262bb424e1f9450 (patch) | |
tree | 8e1c7bdd402fb640f75b061bc2051f5fe6eba676 /tests | |
parent | 79fcbdad812cc3be39afbf8375c11e0581eeb86e (diff) | |
parent | d285b80ecb6102b0ad501b74d02e04d61e8ec632 (diff) |
Merge branch '2.0' into 667-int
Diffstat (limited to 'tests')
-rw-r--r-- | tests/loop.py | 2 | ||||
-rw-r--r-- | tests/stub_sftp.py | 8 | ||||
-rw-r--r-- | tests/test_auth.py | 7 | ||||
-rw-r--r-- | tests/test_client.py | 8 | ||||
-rw-r--r-- | tests/test_ecdsa_256.key (renamed from tests/test_ecdsa.key) | 0 | ||||
-rw-r--r-- | tests/test_ecdsa_384.key | 6 | ||||
-rw-r--r-- | tests/test_ecdsa_521.key | 7 | ||||
-rw-r--r-- | tests/test_ecdsa_password_256.key (renamed from tests/test_ecdsa_password.key) | 0 | ||||
-rw-r--r-- | tests/test_ecdsa_password_384.key | 9 | ||||
-rw-r--r-- | tests/test_ecdsa_password_521.key | 10 | ||||
-rw-r--r-- | tests/test_gssapi.py | 8 | ||||
-rw-r--r-- | tests/test_hostkeys.py | 12 | ||||
-rw-r--r-- | tests/test_packetizer.py | 12 | ||||
-rw-r--r-- | tests/test_pkey.py | 199 | ||||
-rwxr-xr-x | tests/test_sftp.py | 2 | ||||
-rw-r--r-- | tests/test_ssh_gss.py | 4 | ||||
-rw-r--r-- | tests/test_transport.py | 40 | ||||
-rw-r--r-- | tests/test_util.py | 80 |
18 files changed, 354 insertions, 60 deletions
diff --git a/tests/loop.py b/tests/loop.py index 4f5dc163..e805ad96 100644 --- a/tests/loop.py +++ b/tests/loop.py @@ -37,9 +37,11 @@ class LoopSocket (object): self.__cv = threading.Condition(self.__lock) self.__timeout = None self.__mate = None + self._closed = False def close(self): self.__unlink() + self._closed = True try: self.__lock.acquire() self.__in_buffer = bytes() diff --git a/tests/stub_sftp.py b/tests/stub_sftp.py index 24380ba1..334af561 100644 --- a/tests/stub_sftp.py +++ b/tests/stub_sftp.py @@ -22,8 +22,10 @@ A stub SFTP server for loopback SFTP testing. import os import sys -from paramiko import ServerInterface, SFTPServerInterface, SFTPServer, SFTPAttributes, \ - SFTPHandle, SFTP_OK, AUTH_SUCCESSFUL, OPEN_SUCCEEDED +from paramiko import ( + ServerInterface, SFTPServerInterface, SFTPServer, SFTPAttributes, + SFTPHandle, SFTP_OK, AUTH_SUCCESSFUL, OPEN_SUCCEEDED, +) from paramiko.common import o666 @@ -55,7 +57,7 @@ class StubSFTPHandle (SFTPHandle): class StubSFTPServer (SFTPServerInterface): # assume current folder is a fine root - # (the tests always create and eventualy delete a subfolder, so there shouldn't be any mess) + # (the tests always create and eventually delete a subfolder, so there shouldn't be any mess) ROOT = os.getcwd() def _realpath(self, path): diff --git a/tests/test_auth.py b/tests/test_auth.py index 23517790..96f7611c 100644 --- a/tests/test_auth.py +++ b/tests/test_auth.py @@ -24,9 +24,10 @@ import sys import threading import unittest -from paramiko import Transport, ServerInterface, RSAKey, DSSKey, \ - BadAuthenticationType, InteractiveQuery, \ - AuthenticationException +from paramiko import ( + Transport, ServerInterface, RSAKey, DSSKey, BadAuthenticationType, + InteractiveQuery, AuthenticationException, +) from paramiko import AUTH_FAILED, AUTH_PARTIALLY_SUCCESSFUL, AUTH_SUCCESSFUL from paramiko.py3compat import u from tests.loop import LoopSocket diff --git a/tests/test_client.py b/tests/test_client.py index f42d79d9..9c5761d6 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -182,7 +182,7 @@ class SSHClientTest (unittest.TestCase): """ verify that SSHClient works with an ECDSA key. """ - self._test_connection(key_filename=test_path('test_ecdsa.key')) + self._test_connection(key_filename=test_path('test_ecdsa_256.key')) def test_3_multiple_key_files(self): """ @@ -199,8 +199,8 @@ class SSHClientTest (unittest.TestCase): for attempt, accept in ( (['rsa', 'dss'], ['dss']), # Original test #3 (['dss', 'rsa'], ['dss']), # Ordering matters sometimes, sadly - (['dss', 'rsa', 'ecdsa'], ['dss']), # Try ECDSA but fail - (['rsa', 'ecdsa'], ['ecdsa']), # ECDSA success + (['dss', 'rsa', 'ecdsa_256'], ['dss']), # Try ECDSA but fail + (['rsa', 'ecdsa_256'], ['ecdsa']), # ECDSA success ): try: self._test_connection( @@ -357,7 +357,7 @@ class SSHClientTest (unittest.TestCase): # NOTE: re #387, re #394 # If pkey module used within Client._auth isn't correctly handling auth # errors (e.g. if it allows things like ValueError to bubble up as per - # midway thru #394) client.connect() will fail (at key load step) + # midway through #394) client.connect() will fail (at key load step) # instead of succeeding (at password step) kwargs = dict( # Password-protected key whose passphrase is not 'pygmalion' (it's diff --git a/tests/test_ecdsa.key b/tests/test_ecdsa_256.key index 42d44734..42d44734 100644 --- a/tests/test_ecdsa.key +++ b/tests/test_ecdsa_256.key diff --git a/tests/test_ecdsa_384.key b/tests/test_ecdsa_384.key new file mode 100644 index 00000000..796bf417 --- /dev/null +++ b/tests/test_ecdsa_384.key @@ -0,0 +1,6 @@ +-----BEGIN EC PRIVATE KEY----- +MIGkAgEBBDBDdO8IXvlLJgM7+sNtPl7tI7FM5kzuEUEEPRjXIPQM7mISciwJPBt+ +y43EuG8nL4mgBwYFK4EEACKhZANiAAQWxom0C1vQAGYhjdoREMVmGKBWlisDdzyk +mgyUjKpiJ9WfbIEVLsPGP8OdNjhr1y/8BZNIts+dJd6VmYw+4HzB+4F+U1Igs8K0 +JEvh59VNkvWheViadDXCM2MV8Nq+DNg= +-----END EC PRIVATE KEY----- diff --git a/tests/test_ecdsa_521.key b/tests/test_ecdsa_521.key new file mode 100644 index 00000000..b87dc90f --- /dev/null +++ b/tests/test_ecdsa_521.key @@ -0,0 +1,7 @@ +-----BEGIN EC PRIVATE KEY----- +MIHcAgEBBEIAprQtAS3OF6iVUkT8IowTHWicHzShGgk86EtuEXvfQnhZFKsWm6Jo +iqAr1yEaiuI9LfB3Xs8cjuhgEEfbduYr/f6gBwYFK4EEACOhgYkDgYYABACaOaFL +ZGuxa5AW16qj6VLypFbLrEWrt9AZUloCMefxO8bNLjK/O5g0rAVasar1TnyHE9qj +4NwzANZASWjQNbc4MAG8vzqezFwLIn/kNyNTsXNfqEko9OgHZknlj2Z79dwTJcRA +L4QLcT5aND0EHZLB2fAUDXiWIb2j4rg1mwPlBMiBXA== +-----END EC PRIVATE KEY----- diff --git a/tests/test_ecdsa_password.key b/tests/test_ecdsa_password_256.key index eb7910ed..eb7910ed 100644 --- a/tests/test_ecdsa_password.key +++ b/tests/test_ecdsa_password_256.key diff --git a/tests/test_ecdsa_password_384.key b/tests/test_ecdsa_password_384.key new file mode 100644 index 00000000..eba33c14 --- /dev/null +++ b/tests/test_ecdsa_password_384.key @@ -0,0 +1,9 @@ +-----BEGIN EC PRIVATE KEY----- +Proc-Type: 4,ENCRYPTED +DEK-Info: AES-128-CBC,7F7B5DBE4CE040D822441AFE7A023A1D + +y/d6tGonAXYgJniQoFCdto+CuT1y1s41qzwNLN9YdNq/+R/dtQvZAaOuGtHJRFE6 +wWabhY1bSjavVPT2z1Zw1jhDJX5HGrf9LDoyORKtUWtUJoUvGdYLHbcg8Q+//WRf +R0A01YuSw1SJX0a225S1aRcsDAk1k5F8EMb8QzSSDgjAOI8ldQF35JI+ofNSGjgS +BPOlorQXTJxDOGmokw/Wql6MbhajXKPO39H2Z53W88U= +-----END EC PRIVATE KEY----- diff --git a/tests/test_ecdsa_password_521.key b/tests/test_ecdsa_password_521.key new file mode 100644 index 00000000..5986b930 --- /dev/null +++ b/tests/test_ecdsa_password_521.key @@ -0,0 +1,10 @@ +-----BEGIN EC PRIVATE KEY----- +Proc-Type: 4,ENCRYPTED +DEK-Info: AES-128-CBC,AEB2DE62C65D1A88C4940A3476B2F10A + +5kNk/FFPbHa0402QTrgpIT28uirJ4Amvb2/ryOEyOCe0NPbTLCqlQekj2RFYH2Un +pgCLUDkelKQv4pyuK8qWS7R+cFjE/gHHCPUWkK3djZUC8DKuA9lUKeQIE+V1vBHc +L5G+MpoYrPgaydcGx/Uqnc/kVuZx1DXLwrGGtgwNROVBtmjXC9EdfeXHLL1y0wvH +paNgacJpUtgqJEmiehf7eL/eiReegG553rZK3jjfboGkREUaKR5XOgamiKUtgKoc +sMpImVYCsRKd/9RI+VOqErZaEvy/9j0Ye3iH32wGOaA= +-----END EC PRIVATE KEY----- diff --git a/tests/test_gssapi.py b/tests/test_gssapi.py index 96c268d9..bc220108 100644 --- a/tests/test_gssapi.py +++ b/tests/test_gssapi.py @@ -104,9 +104,11 @@ class GSSAPITest(unittest.TestCase): status = gss_srv_ctxt.verify_mic(mic_msg, mic_token) self.assertEquals(0, status) else: - gss_flags = sspicon.ISC_REQ_INTEGRITY |\ - sspicon.ISC_REQ_MUTUAL_AUTH |\ - sspicon.ISC_REQ_DELEGATE + gss_flags = ( + sspicon.ISC_REQ_INTEGRITY | + sspicon.ISC_REQ_MUTUAL_AUTH | + sspicon.ISC_REQ_DELEGATE + ) # Initialize a GSS-API context. target_name = "host/" + socket.getfqdn(targ_name) gss_ctxt = sspi.ClientAuth("Kerberos", diff --git a/tests/test_hostkeys.py b/tests/test_hostkeys.py index 2bdcad9c..2c7ceeb9 100644 --- a/tests/test_hostkeys.py +++ b/tests/test_hostkeys.py @@ -115,3 +115,15 @@ class HostKeysTest (unittest.TestCase): self.assertEqual(b'7EC91BB336CB6D810B124B1353C32396', fp) fp = hexlify(hostdict['secure.example.com']['ssh-dss'].get_fingerprint()).upper() self.assertEqual(b'4478F0B9A23CC5182009FF755BC1D26C', fp) + + def test_delitem(self): + hostdict = paramiko.HostKeys('hostfile.temp') + target = 'happy.example.com' + entry = hostdict[target] # will KeyError if not present + del hostdict[target] + try: + entry = hostdict[target] + except KeyError: + pass # Good + else: + assert False, "Entry was not deleted from HostKeys on delitem!" diff --git a/tests/test_packetizer.py b/tests/test_packetizer.py index ccfe26bd..02173292 100644 --- a/tests/test_packetizer.py +++ b/tests/test_packetizer.py @@ -20,6 +20,7 @@ Some unit tests for the ssh2 protocol in Transport. """ +import sys import unittest from hashlib import sha1 @@ -34,7 +35,6 @@ from paramiko.common import byte_chr, zero_byte x55 = byte_chr(0x55) x1f = byte_chr(0x1f) - class PacketizerTest (unittest.TestCase): def test_1_write(self): @@ -85,6 +85,8 @@ class PacketizerTest (unittest.TestCase): self.assertEqual(900, m.get_int()) def test_3_closed(self): + if sys.platform.startswith("win"): # no SIGALRM on windows + return rsock = LoopSocket() wsock = LoopSocket() rsock.link(wsock) @@ -112,9 +114,13 @@ class PacketizerTest (unittest.TestCase): import signal class TimeoutError(Exception): - pass + def __init__(self, error_message): + if hasattr(errno, 'ETIME'): + self.message = os.sterror(errno.ETIME) + else: + self.messaage = error_message - def timeout(seconds=1, error_message=os.strerror(errno.ETIME)): + def timeout(seconds=1, error_message='Timer expired'): def decorator(func): def _handle_timeout(signum, frame): raise TimeoutError(error_message) diff --git a/tests/test_pkey.py b/tests/test_pkey.py index ec128140..24d78c3e 100644 --- a/tests/test_pkey.py +++ b/tests/test_pkey.py @@ -1,3 +1,4 @@ +# -*- coding: utf-8 -*- # Copyright (C) 2003-2009 Robey Pointer <robeypointer@gmail.com> # # This file is part of paramiko. @@ -24,20 +25,25 @@ import unittest import os from binascii import hexlify from hashlib import md5 +import base64 from paramiko import RSAKey, DSSKey, ECDSAKey, Message, util -from paramiko.py3compat import StringIO, byte_chr, b, bytes +from paramiko.py3compat import StringIO, byte_chr, b, bytes, PY2 from tests.util import test_path # from openssh's ssh-keygen PUB_RSA = 'ssh-rsa AAAAB3NzaC1yc2EAAAABIwAAAIEA049W6geFpmsljTwfvI1UmKWWJPNFI74+vNKTk4dmzkQY2yAMs6FhlvhlI8ysU4oj71ZsRYMecHbBbxdN79+JRFVYTKaLqjwGENeTd+yv4q+V2PvZv3fLnzApI3l7EJCqhWwJUHJ1jAkZzqDx0tyOL4uoZpww3nmE0kb3y21tH4c=' PUB_DSS = 'ssh-dss AAAAB3NzaC1kc3MAAACBAOeBpgNnfRzr/twmAQRu2XwWAp3CFtrVnug6s6fgwj/oLjYbVtjAy6pl/h0EKCWx2rf1IetyNsTxWrniA9I6HeDj65X1FyDkg6g8tvCnaNB8Xp/UUhuzHuGsMIipRxBxw9LF608EqZcj1E3ytktoW5B5OcjrkEoz3xG7C+rpIjYvAAAAFQDwz4UnmsGiSNu5iqjn3uTzwUpshwAAAIEAkxfFeY8P2wZpDjX0MimZl5wkoFQDL25cPzGBuB4OnB8NoUk/yjAHIIpEShw8V+LzouMK5CTJQo5+Ngw3qIch/WgRmMHy4kBq1SsXMjQCte1So6HBMvBPIW5SiMTmjCfZZiw4AYHK+B/JaOwaG9yRg2Ejg4Ok10+XFDxlqZo8Y+wAAACARmR7CCPjodxASvRbIyzaVpZoJ/Z6x7dAumV+ysrV1BVYd0lYukmnjO1kKBWApqpH1ve9XDQYN8zgxM4b16L21kpoWQnZtXrY3GZ4/it9kUgyB7+NwacIBlXa8cMDL7Q/69o0d54U0X/NeX5QxuYR6OMJlrkQB7oiW/P/1mwjQgE=' -PUB_ECDSA = 'ecdsa-sha2-nistp256 AAAAE2VjZHNhLXNoYTItbmlzdHAyNTYAAAAIbmlzdHAyNTYAAABBBJSPZm3ZWkvk/Zx8WP+fZRZ5/NBBHnGQwR6uIC6XHGPDIHuWUzIjAwA0bzqkOUffEsbLe+uQgKl5kbc/L8KA/eo=' +PUB_ECDSA_256 = 'ecdsa-sha2-nistp256 AAAAE2VjZHNhLXNoYTItbmlzdHAyNTYAAAAIbmlzdHAyNTYAAABBBJSPZm3ZWkvk/Zx8WP+fZRZ5/NBBHnGQwR6uIC6XHGPDIHuWUzIjAwA0bzqkOUffEsbLe+uQgKl5kbc/L8KA/eo=' +PUB_ECDSA_384 = 'ecdsa-sha2-nistp384 AAAAE2VjZHNhLXNoYTItbmlzdHAzODQAAAAIbmlzdHAzODQAAABhBBbGibQLW9AAZiGN2hEQxWYYoFaWKwN3PKSaDJSMqmIn1Z9sgRUuw8Y/w502OGvXL/wFk0i2z50l3pWZjD7gfMH7gX5TUiCzwrQkS+Hn1U2S9aF5WJp0NcIzYxXw2r4M2A==' +PUB_ECDSA_521 = 'ecdsa-sha2-nistp521 AAAAE2VjZHNhLXNoYTItbmlzdHA1MjEAAAAIbmlzdHA1MjEAAACFBACaOaFLZGuxa5AW16qj6VLypFbLrEWrt9AZUloCMefxO8bNLjK/O5g0rAVasar1TnyHE9qj4NwzANZASWjQNbc4MAG8vzqezFwLIn/kNyNTsXNfqEko9OgHZknlj2Z79dwTJcRAL4QLcT5aND0EHZLB2fAUDXiWIb2j4rg1mwPlBMiBXA==' FINGER_RSA = '1024 60:73:38:44:cb:51:86:65:7f:de:da:a2:2b:5a:57:d5' FINGER_DSS = '1024 44:78:f0:b9:a2:3c:c5:18:20:09:ff:75:5b:c1:d2:6c' -FINGER_ECDSA = '256 25:19:eb:55:e6:a1:47:ff:4f:38:d2:75:6f:a5:d5:60' +FINGER_ECDSA_256 = '256 25:19:eb:55:e6:a1:47:ff:4f:38:d2:75:6f:a5:d5:60' +FINGER_ECDSA_384 = '384 c1:8d:a0:59:09:47:41:8e:a8:a6:07:01:29:23:b4:65' +FINGER_ECDSA_521 = '521 44:58:22:52:12:33:16:0e:ce:0e:be:2c:7c:7e:cc:1e' SIGNED_RSA = '20:d7:8a:31:21:cb:f7:92:12:f2:a4:89:37:f5:78:af:e6:16:b6:25:b9:97:3d:a2:cd:5f:ca:20:21:73:4c:ad:34:73:8f:20:77:28:e2:94:15:08:d8:91:40:7a:85:83:bf:18:37:95:dc:54:1a:9b:88:29:6c:73:ca:38:b4:04:f1:56:b9:f2:42:9d:52:1b:29:29:b4:4f:fd:c9:2d:af:47:d2:40:76:30:f3:63:45:0c:d9:1d:43:86:0f:1c:70:e2:93:12:34:f3:ac:c5:0a:2f:14:50:66:59:f1:88:ee:c1:4a:e9:d1:9c:4e:46:f0:0e:47:6f:38:74:f1:44:a8' RSA_PRIVATE_OUT = """\ @@ -73,7 +79,7 @@ h9pT9XHqn+1rZ4bK+QGA -----END DSA PRIVATE KEY----- """ -ECDSA_PRIVATE_OUT = """\ +ECDSA_PRIVATE_OUT_256 = """\ -----BEGIN EC PRIVATE KEY----- MHcCAQEEIKB6ty3yVyKEnfF/zprx0qwC76MsMlHY4HXCnqho2eKioAoGCCqGSM49 AwEHoUQDQgAElI9mbdlaS+T9nHxY/59lFnn80EEecZDBHq4gLpccY8Mge5ZTMiMD @@ -81,8 +87,30 @@ ADRvOqQ5R98Sxst765CAqXmRtz8vwoD96g== -----END EC PRIVATE KEY----- """ +ECDSA_PRIVATE_OUT_384 = """\ +-----BEGIN EC PRIVATE KEY----- +MIGkAgEBBDBDdO8IXvlLJgM7+sNtPl7tI7FM5kzuEUEEPRjXIPQM7mISciwJPBt+ +y43EuG8nL4mgBwYFK4EEACKhZANiAAQWxom0C1vQAGYhjdoREMVmGKBWlisDdzyk +mgyUjKpiJ9WfbIEVLsPGP8OdNjhr1y/8BZNIts+dJd6VmYw+4HzB+4F+U1Igs8K0 +JEvh59VNkvWheViadDXCM2MV8Nq+DNg= +-----END EC PRIVATE KEY----- +""" + +ECDSA_PRIVATE_OUT_521 = """\ +-----BEGIN EC PRIVATE KEY----- +MIHcAgEBBEIAprQtAS3OF6iVUkT8IowTHWicHzShGgk86EtuEXvfQnhZFKsWm6Jo +iqAr1yEaiuI9LfB3Xs8cjuhgEEfbduYr/f6gBwYFK4EEACOhgYkDgYYABACaOaFL +ZGuxa5AW16qj6VLypFbLrEWrt9AZUloCMefxO8bNLjK/O5g0rAVasar1TnyHE9qj +4NwzANZASWjQNbc4MAG8vzqezFwLIn/kNyNTsXNfqEko9OgHZknlj2Z79dwTJcRA +L4QLcT5aND0EHZLB2fAUDXiWIb2j4rg1mwPlBMiBXA== +-----END EC PRIVATE KEY----- +""" + x1234 = b'\x01\x02\x03\x04' +TEST_KEY_BYTESTR_2 = '\x00\x00\x00\x07ssh-rsa\x00\x00\x00\x01#\x00\x00\x00\x81\x00\xd3\x8fV\xea\x07\x85\xa6k%\x8d<\x1f\xbc\x8dT\x98\xa5\x96$\xf3E#\xbe>\xbc\xd2\x93\x93\x87f\xceD\x18\xdb \x0c\xb3\xa1a\x96\xf8e#\xcc\xacS\x8a#\xefVlE\x83\x1epv\xc1o\x17M\xef\xdf\x89DUXL\xa6\x8b\xaa<\x06\x10\xd7\x93w\xec\xaf\xe2\xaf\x95\xd8\xfb\xd9\xbfw\xcb\x9f0)#y{\x10\x90\xaa\x85l\tPru\x8c\t\x19\xce\xa0\xf1\xd2\xdc\x8e/\x8b\xa8f\x9c0\xdey\x84\xd2F\xf7\xcbmm\x1f\x87' +TEST_KEY_BYTESTR_3 = '\x00\x00\x00\x07ssh-rsa\x00\x00\x00\x01#\x00\x00\x00\x00ӏV\x07k%<\x1fT$E#>ғfD\x18 \x0cae#̬S#VlE\x1epvo\x17M߉DUXL<\x06\x10דw\u2bd5ٿw˟0)#y{\x10l\tPru\t\x19Π\u070e/f0yFmm\x1f' + class KeyTest (unittest.TestCase): @@ -205,43 +233,72 @@ class KeyTest (unittest.TestCase): msg.rewind() self.assertTrue(key.verify_ssh_sig(b'jerri blank', msg)) - def test_10_load_ecdsa(self): - key = ECDSAKey.from_private_key_file(test_path('test_ecdsa.key')) + def test_C_generate_ecdsa(self): + key = ECDSAKey.generate() + msg = key.sign_ssh_data(b'jerri blank') + msg.rewind() + self.assertTrue(key.verify_ssh_sig(b'jerri blank', msg)) + self.assertEqual(key.get_bits(), 256) + self.assertEqual(key.get_name(), 'ecdsa-sha2-nistp256') + + key = ECDSAKey.generate(bits=256) + msg = key.sign_ssh_data(b'jerri blank') + msg.rewind() + self.assertTrue(key.verify_ssh_sig(b'jerri blank', msg)) + self.assertEqual(key.get_bits(), 256) + self.assertEqual(key.get_name(), 'ecdsa-sha2-nistp256') + + key = ECDSAKey.generate(bits=384) + msg = key.sign_ssh_data(b'jerri blank') + msg.rewind() + self.assertTrue(key.verify_ssh_sig(b'jerri blank', msg)) + self.assertEqual(key.get_bits(), 384) + self.assertEqual(key.get_name(), 'ecdsa-sha2-nistp384') + + key = ECDSAKey.generate(bits=521) + msg = key.sign_ssh_data(b'jerri blank') + msg.rewind() + self.assertTrue(key.verify_ssh_sig(b'jerri blank', msg)) + self.assertEqual(key.get_bits(), 521) + self.assertEqual(key.get_name(), 'ecdsa-sha2-nistp521') + + def test_10_load_ecdsa_256(self): + key = ECDSAKey.from_private_key_file(test_path('test_ecdsa_256.key')) self.assertEqual('ecdsa-sha2-nistp256', key.get_name()) - exp_ecdsa = b(FINGER_ECDSA.split()[1].replace(':', '')) + exp_ecdsa = b(FINGER_ECDSA_256.split()[1].replace(':', '')) my_ecdsa = hexlify(key.get_fingerprint()) self.assertEqual(exp_ecdsa, my_ecdsa) - self.assertEqual(PUB_ECDSA.split()[1], key.get_base64()) + self.assertEqual(PUB_ECDSA_256.split()[1], key.get_base64()) self.assertEqual(256, key.get_bits()) s = StringIO() key.write_private_key(s) - self.assertEqual(ECDSA_PRIVATE_OUT, s.getvalue()) + self.assertEqual(ECDSA_PRIVATE_OUT_256, s.getvalue()) s.seek(0) key2 = ECDSAKey.from_private_key(s) self.assertEqual(key, key2) - def test_11_load_ecdsa_password(self): - key = ECDSAKey.from_private_key_file(test_path('test_ecdsa_password.key'), b'television') + def test_11_load_ecdsa_password_256(self): + key = ECDSAKey.from_private_key_file(test_path('test_ecdsa_password_256.key'), b'television') self.assertEqual('ecdsa-sha2-nistp256', key.get_name()) - exp_ecdsa = b(FINGER_ECDSA.split()[1].replace(':', '')) + exp_ecdsa = b(FINGER_ECDSA_256.split()[1].replace(':', '')) my_ecdsa = hexlify(key.get_fingerprint()) self.assertEqual(exp_ecdsa, my_ecdsa) - self.assertEqual(PUB_ECDSA.split()[1], key.get_base64()) + self.assertEqual(PUB_ECDSA_256.split()[1], key.get_base64()) self.assertEqual(256, key.get_bits()) - def test_12_compare_ecdsa(self): + def test_12_compare_ecdsa_256(self): # verify that the private & public keys compare equal - key = ECDSAKey.from_private_key_file(test_path('test_ecdsa.key')) + key = ECDSAKey.from_private_key_file(test_path('test_ecdsa_256.key')) self.assertEqual(key, key) pub = ECDSAKey(data=key.asbytes()) self.assertTrue(key.can_sign()) self.assertTrue(not pub.can_sign()) self.assertEqual(key, pub) - def test_13_sign_ecdsa(self): + def test_13_sign_ecdsa_256(self): # verify that the rsa private key can sign and verify - key = ECDSAKey.from_private_key_file(test_path('test_ecdsa.key')) + key = ECDSAKey.from_private_key_file(test_path('test_ecdsa_256.key')) msg = key.sign_ssh_data(b'ice weasels') self.assertTrue(type(msg) is Message) msg.rewind() @@ -255,6 +312,109 @@ class KeyTest (unittest.TestCase): pub = ECDSAKey(data=key.asbytes()) self.assertTrue(pub.verify_ssh_sig(b'ice weasels', msg)) + def test_14_load_ecdsa_384(self): + key = ECDSAKey.from_private_key_file(test_path('test_ecdsa_384.key')) + self.assertEqual('ecdsa-sha2-nistp384', key.get_name()) + exp_ecdsa = b(FINGER_ECDSA_384.split()[1].replace(':', '')) + my_ecdsa = hexlify(key.get_fingerprint()) + self.assertEqual(exp_ecdsa, my_ecdsa) + self.assertEqual(PUB_ECDSA_384.split()[1], key.get_base64()) + self.assertEqual(384, key.get_bits()) + + s = StringIO() + key.write_private_key(s) + self.assertEqual(ECDSA_PRIVATE_OUT_384, s.getvalue()) + s.seek(0) + key2 = ECDSAKey.from_private_key(s) + self.assertEqual(key, key2) + + def test_15_load_ecdsa_password_384(self): + key = ECDSAKey.from_private_key_file(test_path('test_ecdsa_password_384.key'), b'television') + self.assertEqual('ecdsa-sha2-nistp384', key.get_name()) + exp_ecdsa = b(FINGER_ECDSA_384.split()[1].replace(':', '')) + my_ecdsa = hexlify(key.get_fingerprint()) + self.assertEqual(exp_ecdsa, my_ecdsa) + self.assertEqual(PUB_ECDSA_384.split()[1], key.get_base64()) + self.assertEqual(384, key.get_bits()) + + def test_16_compare_ecdsa_384(self): + # verify that the private & public keys compare equal + key = ECDSAKey.from_private_key_file(test_path('test_ecdsa_384.key')) + self.assertEqual(key, key) + pub = ECDSAKey(data=key.asbytes()) + self.assertTrue(key.can_sign()) + self.assertTrue(not pub.can_sign()) + self.assertEqual(key, pub) + + def test_17_sign_ecdsa_384(self): + # verify that the rsa private key can sign and verify + key = ECDSAKey.from_private_key_file(test_path('test_ecdsa_384.key')) + msg = key.sign_ssh_data(b'ice weasels') + self.assertTrue(type(msg) is Message) + msg.rewind() + self.assertEqual('ecdsa-sha2-nistp384', msg.get_text()) + # ECDSA signatures, like DSS signatures, tend to be different + # each time, so we can't compare against a "known correct" + # signature. + # Even the length of the signature can change. + + msg.rewind() + pub = ECDSAKey(data=key.asbytes()) + self.assertTrue(pub.verify_ssh_sig(b'ice weasels', msg)) + + def test_18_load_ecdsa_521(self): + key = ECDSAKey.from_private_key_file(test_path('test_ecdsa_521.key')) + self.assertEqual('ecdsa-sha2-nistp521', key.get_name()) + exp_ecdsa = b(FINGER_ECDSA_521.split()[1].replace(':', '')) + my_ecdsa = hexlify(key.get_fingerprint()) + self.assertEqual(exp_ecdsa, my_ecdsa) + self.assertEqual(PUB_ECDSA_521.split()[1], key.get_base64()) + self.assertEqual(521, key.get_bits()) + + s = StringIO() + key.write_private_key(s) + # Different versions of OpenSSL (SSLeay versions 0x1000100f and + # 0x1000207f for instance) use different apparently valid (as far as + # ssh-keygen is concerned) padding. So we can't check the actual value + # of the pem encoded key. + s.seek(0) + key2 = ECDSAKey.from_private_key(s) + self.assertEqual(key, key2) + + def test_19_load_ecdsa_password_521(self): + key = ECDSAKey.from_private_key_file(test_path('test_ecdsa_password_521.key'), b'television') + self.assertEqual('ecdsa-sha2-nistp521', key.get_name()) + exp_ecdsa = b(FINGER_ECDSA_521.split()[1].replace(':', '')) + my_ecdsa = hexlify(key.get_fingerprint()) + self.assertEqual(exp_ecdsa, my_ecdsa) + self.assertEqual(PUB_ECDSA_521.split()[1], key.get_base64()) + self.assertEqual(521, key.get_bits()) + + def test_20_compare_ecdsa_521(self): + # verify that the private & public keys compare equal + key = ECDSAKey.from_private_key_file(test_path('test_ecdsa_521.key')) + self.assertEqual(key, key) + pub = ECDSAKey(data=key.asbytes()) + self.assertTrue(key.can_sign()) + self.assertTrue(not pub.can_sign()) + self.assertEqual(key, pub) + + def test_21_sign_ecdsa_521(self): + # verify that the rsa private key can sign and verify + key = ECDSAKey.from_private_key_file(test_path('test_ecdsa_521.key')) + msg = key.sign_ssh_data(b'ice weasels') + self.assertTrue(type(msg) is Message) + msg.rewind() + self.assertEqual('ecdsa-sha2-nistp521', msg.get_text()) + # ECDSA signatures, like DSS signatures, tend to be different + # each time, so we can't compare against a "known correct" + # signature. + # Even the length of the signature can change. + + msg.rewind() + pub = ECDSAKey(data=key.asbytes()) + self.assertTrue(pub.verify_ssh_sig(b'ice weasels', msg)) + def test_salt_size(self): # Read an existing encrypted private key file_ = test_path('test_rsa_password.key') @@ -271,3 +431,8 @@ class KeyTest (unittest.TestCase): self.assertEqual(key, key2) finally: os.remove(newfile) + + def test_stringification(self): + key = RSAKey.from_private_key_file(test_path('test_rsa.key')) + comparable = TEST_KEY_BYTESTR_2 if PY2 else TEST_KEY_BYTESTR_3 + self.assertEqual(str(key), comparable) diff --git a/tests/test_sftp.py b/tests/test_sftp.py index e4c2c3a3..d3064fff 100755 --- a/tests/test_sftp.py +++ b/tests/test_sftp.py @@ -413,7 +413,7 @@ class SFTPTest (unittest.TestCase): def test_A_readline_seek(self): """ create a text file and write a bunch of text into it. then count the lines - in the file, and seek around to retreive particular lines. this should + in the file, and seek around to retrieve particular lines. this should verify that read buffering and 'tell' work well together, and that read buffering is reset on 'seek'. """ diff --git a/tests/test_ssh_gss.py b/tests/test_ssh_gss.py index e20d348f..967b3b81 100644 --- a/tests/test_ssh_gss.py +++ b/tests/test_ssh_gss.py @@ -43,9 +43,7 @@ class NullServer (paramiko.ServerInterface): return paramiko.AUTH_FAILED def enable_auth_gssapi(self): - UseGSSAPI = True - GSSAPICleanupCredentials = True - return UseGSSAPI + return True def check_channel_request(self, kind, chanid): return paramiko.OPEN_SUCCEEDED diff --git a/tests/test_transport.py b/tests/test_transport.py index 5069e5b0..c426cef1 100644 --- a/tests/test_transport.py +++ b/tests/test_transport.py @@ -31,13 +31,16 @@ import random from hashlib import sha1 import unittest -from paramiko import Transport, SecurityOptions, ServerInterface, RSAKey, DSSKey, \ - SSHException, ChannelException, Packetizer +from paramiko import ( + Transport, SecurityOptions, ServerInterface, RSAKey, DSSKey, SSHException, + ChannelException, Packetizer, +) from paramiko import AUTH_FAILED, AUTH_SUCCESSFUL from paramiko import OPEN_SUCCEEDED, OPEN_FAILED_ADMINISTRATIVELY_PROHIBITED -from paramiko.common import MSG_KEXINIT, cMSG_CHANNEL_WINDOW_ADJUST, \ - MIN_PACKET_SIZE, MIN_WINDOW_SIZE, MAX_WINDOW_SIZE, \ - DEFAULT_WINDOW_SIZE, DEFAULT_MAX_PACKET_SIZE +from paramiko.common import ( + MSG_KEXINIT, cMSG_CHANNEL_WINDOW_ADJUST, MIN_PACKET_SIZE, MIN_WINDOW_SIZE, + MAX_WINDOW_SIZE, DEFAULT_WINDOW_SIZE, DEFAULT_MAX_PACKET_SIZE, +) from paramiko.py3compat import bytes from paramiko.message import Message from tests.loop import LoopSocket @@ -162,6 +165,15 @@ class TransportTest(unittest.TestCase): except TypeError: pass + def test_1b_security_options_reset(self): + o = self.tc.get_security_options() + # should not throw any exceptions + o.ciphers = o.ciphers + o.digests = o.digests + o.key_types = o.key_types + o.kex = o.kex + o.compression = o.compression + def test_2_compute_key(self): self.tc.K = 123281095979686581523377256114209720774539068973101330872763622971399429481072519713536292772709507296759612401802191955568143056534122385270077606457721553469730659233569339356140085284052436697480759510519672848743794433460113118986816826624865291116513647975790797391795651716378444844877749505443714557929 self.tc.H = b'\x0C\x83\x07\xCD\xE6\x85\x6F\xF3\x0B\xA9\x36\x84\xEB\x0F\x04\xC2\x52\x0E\x9E\xD3' @@ -828,3 +840,21 @@ class TransportTest(unittest.TestCase): hostkey=public_host_key, username='slowdive', password='pygmalion') + + def test_M_select_after_close(self): + """ + verify that select works when a channel is already closed. + """ + self.setup_test_server() + chan = self.tc.open_session() + chan.invoke_shell() + schan = self.ts.accept(1.0) + schan.close() + + # give client a moment to receive close notification + time.sleep(0.1) + + r, w, e = select.select([chan], [], [], 0.1) + self.assertEqual([chan], r) + self.assertEqual([], w) + self.assertEqual([], e) diff --git a/tests/test_util.py b/tests/test_util.py index a6a2c30b..7880e156 100644 --- a/tests/test_util.py +++ b/tests/test_util.py @@ -66,7 +66,7 @@ from paramiko import * class UtilTest(unittest.TestCase): - def test_1_import(self): + def test_import(self): """ verify that all the classes can be imported from paramiko. """ @@ -104,7 +104,7 @@ class UtilTest(unittest.TestCase): self.assertTrue('SSHConfig' in symbols) self.assertTrue('util' in symbols) - def test_2_parse_config(self): + def test_parse_config(self): global test_config_file f = StringIO(test_config_file) config = paramiko.util.parse_ssh_config(f) @@ -114,7 +114,7 @@ class UtilTest(unittest.TestCase): {'host': ['*'], 'config': {'crazy': 'something dumb'}}, {'host': ['spoo.example.com'], 'config': {'crazy': 'something else'}}]) - def test_3_host_config(self): + def test_host_config(self): global test_config_file f = StringIO(test_config_file) config = paramiko.util.parse_ssh_config(f) @@ -141,12 +141,12 @@ class UtilTest(unittest.TestCase): values ) - def test_4_generate_key_bytes(self): + def test_generate_key_bytes(self): x = paramiko.util.generate_key_bytes(sha1, b'ABCDEFGH', 'This is my secret passphrase.', 64) hex = ''.join(['%02x' % byte_ord(c) for c in x]) self.assertEqual(hex, '9110e2f6793b69363e58173e9436b13a5a4b339005741d5c680e505f57d871347b4239f14fb5c46e857d5e100424873ba849ac699cea98d729e57b3e84378e8b') - def test_5_host_keys(self): + def test_host_keys(self): with open('hostfile.temp', 'w') as f: f.write(test_hosts_file) try: @@ -159,7 +159,7 @@ class UtilTest(unittest.TestCase): finally: os.unlink('hostfile.temp') - def test_7_host_config_expose_issue_33(self): + def test_host_config_expose_issue_33(self): test_config_file = """ Host www13.* Port 22 @@ -178,7 +178,7 @@ Host * {'hostname': host, 'port': '22'} ) - def test_8_eintr_retry(self): + def test_eintr_retry(self): self.assertEqual('foo', paramiko.util.retry_on_signal(lambda: 'foo')) # Variables that are set by raises_intr @@ -203,7 +203,7 @@ Host * self.assertRaises(AssertionError, lambda: paramiko.util.retry_on_signal(raises_other_exception)) - def test_9_proxycommand_config_equals_parsing(self): + def test_proxycommand_config_equals_parsing(self): """ ProxyCommand should not split on equals signs within the value. """ @@ -222,7 +222,7 @@ Host equals-delimited 'foo bar=biz baz' ) - def test_10_proxycommand_interpolation(self): + def test_proxycommand_interpolation(self): """ ProxyCommand should perform interpolation on the value """ @@ -248,7 +248,20 @@ Host * val ) - def test_11_host_config_test_negation(self): + def test_proxycommand_tilde_expansion(self): + """ + Tilde (~) should be expanded inside ProxyCommand + """ + config = paramiko.util.parse_ssh_config(StringIO(""" +Host test + ProxyCommand ssh -F ~/.ssh/test_config bastion nc %h %p +""")) + self.assertEqual( + 'ssh -F %s/.ssh/test_config bastion nc test 22' % os.path.expanduser('~'), + host_config('test', config)['proxycommand'] + ) + + def test_host_config_test_negation(self): test_config_file = """ Host www13.* !*.example.com Port 22 @@ -270,7 +283,7 @@ Host * {'hostname': host, 'port': '8080'} ) - def test_12_host_config_test_proxycommand(self): + def test_host_config_test_proxycommand(self): test_config_file = """ Host proxy-with-equal-divisor-and-space ProxyCommand = foo=bar @@ -298,7 +311,7 @@ ProxyCommand foo=bar:%h-%p values ) - def test_11_host_config_test_identityfile(self): + def test_host_config_test_identityfile(self): test_config_file = """ IdentityFile id_dsa0 @@ -328,7 +341,7 @@ IdentityFile id_dsa22 values ) - def test_12_config_addressfamily_and_lazy_fqdn(self): + def test_config_addressfamily_and_lazy_fqdn(self): """ Ensure the code path honoring non-'all' AddressFamily doesn't asplode """ @@ -344,13 +357,13 @@ IdentityFile something_%l_using_fqdn self.assertEqual(32767, paramiko.util.clamp_value(32767, 32765, 32769)) self.assertEqual(32769, paramiko.util.clamp_value(32767, 32770, 32769)) - def test_13_config_dos_crlf_succeeds(self): + def test_config_dos_crlf_succeeds(self): config_file = StringIO("host abcqwerty\r\nHostName 127.0.0.1\r\n") config = paramiko.SSHConfig() config.parse(config_file) self.assertEqual(config.lookup("abcqwerty")["hostname"], "127.0.0.1") - def test_14_get_hostnames(self): + def test_get_hostnames(self): f = StringIO(test_config_file) config = paramiko.util.parse_ssh_config(f) self.assertEqual(config.get_hostnames(), set(['*', '*.example.com', 'spoo.example.com'])) @@ -462,9 +475,10 @@ Host param3 parara safe_has_bytes = safe_string(has_bytes) expected_bytes = b("has %07%03 bytes") err = "{0!r} != {1!r}" - assert safe_vanilla == vanilla, err.format(safe_vanilla, vanilla) - assert safe_has_bytes == expected_bytes, \ - err.format(safe_has_bytes, expected_bytes) + msg = err.format(safe_vanilla, vanilla) + assert safe_vanilla == vanilla, msg + msg = err.format(safe_has_bytes, expected_bytes) + assert safe_has_bytes == expected_bytes, msg def test_proxycommand_none_issue_418(self): test_config_file = """ @@ -485,3 +499,33 @@ Host proxycommand-with-equals-none paramiko.util.lookup_ssh_host_config(host, config), values ) + + def test_proxycommand_none_masking(self): + # Re: https://github.com/paramiko/paramiko/issues/670 + source_config = """ +Host specific-host + ProxyCommand none + +Host other-host + ProxyCommand other-proxy + +Host * + ProxyCommand default-proxy +""" + config = paramiko.SSHConfig() + config.parse(StringIO(source_config)) + # When bug is present, the full stripping-out of specific-host's + # ProxyCommand means it actually appears to pick up the default + # ProxyCommand value instead, due to cascading. It should (for + # backwards compatibility reasons in 1.x/2.x) appear completely blank, + # as if the host had no ProxyCommand whatsoever. + # Threw another unrelated host in there just for sanity reasons. + self.assertFalse('proxycommand' in config.lookup('specific-host')) + self.assertEqual( + config.lookup('other-host')['proxycommand'], + 'other-proxy' + ) + self.assertEqual( + config.lookup('some-random-host')['proxycommand'], + 'default-proxy' + ) |