Merge branch '2.0' of github.com:paramiko/paramiko into 2.0

6 files changed, 282 insertions, 18 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/test_client.py b/tests/test_client.py
index f2f2ea45..9da6eaca 100644
--- a/tests/test_client.py
+++ b/tests/test_client.py
@@ -35,7 +35,7 @@ import time
 from tests.util import test_path
 
 import paramiko
-from paramiko.common import PY2
+from paramiko.py3compat import PY2, b
 from paramiko.ssh_exception import SSHException
 
 
@@ -141,6 +141,7 @@ class SSHClientTest (unittest.TestCase):
         self.assertTrue(self.ts.is_active())
         self.assertEqual('slowdive', self.ts.get_username())
         self.assertEqual(True, self.ts.is_authenticated())
+        self.assertEqual(False, self.tc.get_transport().gss_kex_used)
 
         # Command execution functions?
         stdin, stdout, stderr = self.tc.exec_command('yes')
@@ -366,3 +367,63 @@ class SSHClientTest (unittest.TestCase):
             password='pygmalion',
         )
         self._test_connection(**kwargs)
+
+    def test_9_auth_trickledown_gsskex(self):
+        """
+        Failed gssapi-keyex auth doesn't prevent subsequent key auth from succeeding
+        """
+        if not paramiko.GSS_AUTH_AVAILABLE:
+            return  # for python 2.6 lacks skipTest
+        kwargs = dict(
+            gss_kex=True,
+            key_filename=[test_path('test_rsa.key')],
+        )
+        self._test_connection(**kwargs)
+
+    def test_10_auth_trickledown_gssauth(self):
+        """
+        Failed gssapi-with-mic auth doesn't prevent subsequent key auth from succeeding
+        """
+        if not paramiko.GSS_AUTH_AVAILABLE:
+            return  # for python 2.6 lacks skipTest
+        kwargs = dict(
+            gss_auth=True,
+            key_filename=[test_path('test_rsa.key')],
+        )
+        self._test_connection(**kwargs)
+
+    def test_11_reject_policy(self):
+        """
+        verify that SSHClient's RejectPolicy works.
+        """
+        threading.Thread(target=self._run).start()
+
+        self.tc = paramiko.SSHClient()
+        self.tc.set_missing_host_key_policy(paramiko.RejectPolicy())
+        self.assertEqual(0, len(self.tc.get_host_keys()))
+        self.assertRaises(
+            paramiko.SSHException,
+            self.tc.connect,
+            password='pygmalion', **self.connect_kwargs
+        )
+
+    def test_12_reject_policy_gsskex(self):
+        """
+        verify that SSHClient's RejectPolicy works,
+        even if gssapi-keyex was enabled but not used.
+        """
+        # Test for a bug present in paramiko versions released before 2017-08-01
+        if not paramiko.GSS_AUTH_AVAILABLE:
+            return  # for python 2.6 lacks skipTest
+        threading.Thread(target=self._run).start()
+
+        self.tc = paramiko.SSHClient()
+        self.tc.set_missing_host_key_policy(paramiko.RejectPolicy())
+        self.assertEqual(0, len(self.tc.get_host_keys()))
+        self.assertRaises(
+            paramiko.SSHException,
+            self.tc.connect,
+            password='pygmalion',
+            gss_kex=True,
+             **self.connect_kwargs
+        )
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_sftp.py b/tests/test_sftp.py
index d3064fff..98a9cebb 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
@@ -817,6 +818,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_ssh_gss.py b/tests/test_ssh_gss.py
index 967b3b81..d8d05d2b 100644
--- a/tests/test_ssh_gss.py
+++ b/tests/test_ssh_gss.py
@@ -29,11 +29,13 @@ import unittest
 
 import paramiko
 
+from tests.util import test_path
+from tests.test_client import FINGERPRINTS
 
 class NullServer (paramiko.ServerInterface):
 
     def get_allowed_auths(self, username):
-        return 'gssapi-with-mic'
+        return 'gssapi-with-mic,publickey'
 
     def check_auth_gssapi_with_mic(self, username,
                                    gss_authenticated=paramiko.AUTH_FAILED,
@@ -45,6 +47,16 @@ class NullServer (paramiko.ServerInterface):
     def enable_auth_gssapi(self):
         return True
 
+    def check_auth_publickey(self, username, key):
+        try:
+            expected = FINGERPRINTS[key.get_name()]
+        except KeyError:
+            return paramiko.AUTH_FAILED
+        else:
+            if key.get_fingerprint() == expected:
+                return paramiko.AUTH_SUCCESSFUL
+        return paramiko.AUTH_FAILED
+
     def check_channel_request(self, kind, chanid):
         return paramiko.OPEN_SUCCEEDED
 
@@ -85,19 +97,21 @@ class GSSAuthTest(unittest.TestCase):
         server = NullServer()
         self.ts.start_server(self.event, server)
 
-    def test_1_gss_auth(self):
+    def _test_connection(self, **kwargs):
         """
-        Verify that Paramiko can handle SSHv2 GSS-API / SSPI authentication
-        (gssapi-with-mic) in client and server mode.
+        (Most) kwargs get passed directly into SSHClient.connect().
+
+        The exception is ... no exception yet
         """
         host_key = paramiko.RSAKey.from_private_key_file('tests/test_rsa.key')
         public_host_key = paramiko.RSAKey(data=host_key.asbytes())
 
         self.tc = paramiko.SSHClient()
-        self.tc.get_host_keys().add('[%s]:%d' % (self.hostname, self.port),
+        self.tc.set_missing_host_key_policy(paramiko.WarningPolicy())
+        self.tc.get_host_keys().add('[%s]:%d' % (self.addr, self.port),
                                     'ssh-rsa', public_host_key)
-        self.tc.connect(self.hostname, self.port, username=self.username,
-                        gss_auth=True)
+        self.tc.connect(hostname=self.addr, port=self.port, username=self.username, gss_host=self.hostname,
+                        gss_auth=True, **kwargs)
 
         self.event.wait(1.0)
         self.assert_(self.event.is_set())
@@ -120,3 +134,20 @@ class GSSAuthTest(unittest.TestCase):
         stdin.close()
         stdout.close()
         stderr.close()
+
+    def test_1_gss_auth(self):
+        """
+        Verify that Paramiko can handle SSHv2 GSS-API / SSPI authentication
+        (gssapi-with-mic) in client and server mode.
+        """
+        self._test_connection(allow_agent=False,
+                              look_for_keys=False)
+
+    def test_2_auth_trickledown(self):
+        """
+        Failed gssapi-with-mic auth doesn't prevent subsequent key auth from succeeding
+        """
+        self.hostname = "this_host_does_not_exists_and_causes_a_GSSAPI-exception"
+        self._test_connection(key_filename=[test_path('test_rsa.key')],
+                              allow_agent=False,
+                              look_for_keys=False)
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)