Imported Upstream version 1.15.0upstream/1.15.0

1 files changed, 96 insertions, 56 deletions

diff --git a/tests/test_transport.py b/tests/test_transport.py
index 485a18e..50b1d86 100644
--- a/tests/test_transport.py
+++ b/tests/test_transport.py
@@ -20,22 +20,27 @@
 Some unit tests for the ssh2 protocol in Transport.
 """
 
+from __future__ import with_statement
+
 from binascii import hexlify
 import select
 import socket
 import time
 import threading
 import random
+import unittest
 
 from paramiko import Transport, SecurityOptions, ServerInterface, RSAKey, DSSKey, \
     SSHException, ChannelException
 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
+from paramiko.common import MSG_KEXINIT, cMSG_CHANNEL_WINDOW_ADJUST, \
+                            MIN_PACKET_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
-from tests.util import ParamikoTest, test_path
+from tests.util import test_path
 
 
 LONG_BANNER = """\
@@ -55,7 +60,7 @@ class NullServer (ServerInterface):
     paranoid_did_password = False
     paranoid_did_public_key = False
     paranoid_key = DSSKey.from_private_key_file(test_path('test_dss.key'))
-    
+
     def get_allowed_auths(self, username):
         if username == 'slowdive':
             return 'publickey,password'
@@ -78,24 +83,24 @@ class NullServer (ServerInterface):
 
     def check_channel_shell_request(self, channel):
         return True
-    
+
     def check_global_request(self, kind, msg):
         self._global_request = kind
         return False
-    
+
     def check_channel_x11_request(self, channel, single_connection, auth_protocol, auth_cookie, screen_number):
         self._x11_single_connection = single_connection
         self._x11_auth_protocol = auth_protocol
         self._x11_auth_cookie = auth_cookie
         self._x11_screen_number = screen_number
         return True
-    
+
     def check_port_forward_request(self, addr, port):
         self._listen = socket.socket()
         self._listen.bind(('127.0.0.1', 0))
         self._listen.listen(1)
         return self._listen.getsockname()[1]
-    
+
     def cancel_port_forward_request(self, addr, port):
         self._listen.close()
         self._listen = None
@@ -105,7 +110,7 @@ class NullServer (ServerInterface):
         return OPEN_SUCCEEDED
 
 
-class TransportTest(ParamikoTest):
+class TransportTest(unittest.TestCase):
     def setUp(self):
         self.socks = LoopSocket()
         self.sockc = LoopSocket()
@@ -123,12 +128,12 @@ class TransportTest(ParamikoTest):
         host_key = RSAKey.from_private_key_file(test_path('test_rsa.key'))
         public_host_key = RSAKey(data=host_key.asbytes())
         self.ts.add_server_key(host_key)
-        
+
         if client_options is not None:
             client_options(self.tc.get_security_options())
         if server_options is not None:
             server_options(self.ts.get_security_options())
-        
+
         event = threading.Event()
         self.server = NullServer()
         self.assertTrue(not event.isSet())
@@ -155,7 +160,7 @@ class TransportTest(ParamikoTest):
             self.assertTrue(False)
         except TypeError:
             pass
-            
+
     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'
@@ -208,7 +213,7 @@ class TransportTest(ParamikoTest):
         event.wait(1.0)
         self.assertTrue(event.isSet())
         self.assertTrue(self.ts.is_active())
-        
+
     def test_4_special(self):
         """
         verify that the client can demand odd handshake settings, and can
@@ -222,7 +227,7 @@ class TransportTest(ParamikoTest):
         self.assertEqual('aes256-cbc', self.tc.remote_cipher)
         self.assertEqual(12, self.tc.packetizer.get_mac_size_out())
         self.assertEqual(12, self.tc.packetizer.get_mac_size_in())
-        
+
         self.tc.send_ignore(1024)
         self.tc.renegotiate_keys()
         self.ts.send_ignore(1024)
@@ -236,7 +241,7 @@ class TransportTest(ParamikoTest):
         self.tc.set_keepalive(1)
         time.sleep(2)
         self.assertEqual('keepalive@lag.net', self.server._global_request)
-        
+
     def test_6_exec_command(self):
         """
         verify that exec_command() does something reasonable.
@@ -250,7 +255,7 @@ class TransportTest(ParamikoTest):
             self.assertTrue(False)
         except SSHException:
             pass
-        
+
         chan = self.tc.open_session()
         chan.exec_command('yes')
         schan = self.ts.accept(1.0)
@@ -264,7 +269,7 @@ class TransportTest(ParamikoTest):
         f = chan.makefile_stderr()
         self.assertEqual('This is on stderr.\n', f.readline())
         self.assertEqual('', f.readline())
-        
+
         # now try it with combined stdout/stderr
         chan = self.tc.open_session()
         chan.exec_command('yes')
@@ -273,11 +278,27 @@ class TransportTest(ParamikoTest):
         schan.send_stderr('This is on stderr.\n')
         schan.close()
 
-        chan.set_combine_stderr(True)        
+        chan.set_combine_stderr(True)
         f = chan.makefile()
         self.assertEqual('Hello there.\n', f.readline())
         self.assertEqual('This is on stderr.\n', f.readline())
         self.assertEqual('', f.readline())
+        
+    def test_6a_channel_can_be_used_as_context_manager(self):
+        """
+        verify that exec_command() does something reasonable.
+        """
+        self.setup_test_server()
+
+        with self.tc.open_session() as chan:
+            with self.ts.accept(1.0) as schan:
+                chan.exec_command('yes')
+                schan.send('Hello there.\n')
+                schan.close()
+
+                f = chan.makefile()
+                self.assertEqual('Hello there.\n', f.readline())
+                self.assertEqual('', f.readline())
 
     def test_7_invoke_shell(self):
         """
@@ -320,7 +341,7 @@ class TransportTest(ParamikoTest):
         schan.shutdown_write()
         schan.send_exit_status(23)
         schan.close()
-        
+
         f = chan.makefile()
         self.assertEqual('Hello there.\n', f.readline())
         self.assertEqual('', f.readline())
@@ -342,14 +363,14 @@ class TransportTest(ParamikoTest):
         chan.invoke_shell()
         schan = self.ts.accept(1.0)
 
-        # nothing should be ready        
+        # nothing should be ready
         r, w, e = select.select([chan], [], [], 0.1)
         self.assertEqual([], r)
         self.assertEqual([], w)
         self.assertEqual([], e)
-        
+
         schan.send('hello\n')
-        
+
         # something should be ready now (give it 1 second to appear)
         for i in range(10):
             r, w, e = select.select([chan], [], [], 0.1)
@@ -361,7 +382,7 @@ class TransportTest(ParamikoTest):
         self.assertEqual([], e)
 
         self.assertEqual(b'hello\n', chan.recv(6))
-        
+
         # and, should be dead again now
         r, w, e = select.select([chan], [], [], 0.1)
         self.assertEqual([], r)
@@ -369,7 +390,7 @@ class TransportTest(ParamikoTest):
         self.assertEqual([], e)
 
         schan.close()
-        
+
         # detect eof?
         for i in range(10):
             r, w, e = select.select([chan], [], [], 0.1)
@@ -380,14 +401,14 @@ class TransportTest(ParamikoTest):
         self.assertEqual([], w)
         self.assertEqual([], e)
         self.assertEqual(bytes(), chan.recv(16))
-        
+
         # make sure the pipe is still open for now...
         p = chan._pipe
         self.assertEqual(False, p._closed)
         chan.close()
         # ...and now is closed.
         self.assertEqual(True, p._closed)
-   
+
     def test_B_renegotiate(self):
         """
         verify that a transport can correctly renegotiate mid-stream.
@@ -402,7 +423,7 @@ class TransportTest(ParamikoTest):
         for i in range(20):
             chan.send('x' * 1024)
         chan.close()
-        
+
         # allow a few seconds for the rekeying to complete
         for i in range(50):
             if self.tc.H != self.tc.session_id:
@@ -441,28 +462,28 @@ class TransportTest(ParamikoTest):
         chan = self.tc.open_session()
         chan.exec_command('yes')
         schan = self.ts.accept(1.0)
-        
+
         requested = []
         def handler(c, addr_port):
             addr, port = addr_port
             requested.append((addr, port))
             self.tc._queue_incoming_channel(c)
-            
+
         self.assertEqual(None, getattr(self.server, '_x11_screen_number', None))
         cookie = chan.request_x11(0, single_connection=True, handler=handler)
         self.assertEqual(0, self.server._x11_screen_number)
         self.assertEqual('MIT-MAGIC-COOKIE-1', self.server._x11_auth_protocol)
         self.assertEqual(cookie, self.server._x11_auth_cookie)
         self.assertEqual(True, self.server._x11_single_connection)
-        
+
         x11_server = self.ts.open_x11_channel(('localhost', 6093))
         x11_client = self.tc.accept()
         self.assertEqual('localhost', requested[0][0])
         self.assertEqual(6093, requested[0][1])
-        
+
         x11_server.send('hello')
         self.assertEqual(b'hello', x11_client.recv(5))
-        
+
         x11_server.close()
         x11_client.close()
         chan.close()
@@ -477,13 +498,13 @@ class TransportTest(ParamikoTest):
         chan = self.tc.open_session()
         chan.exec_command('yes')
         schan = self.ts.accept(1.0)
-        
+
         requested = []
         def handler(c, origin_addr_port, server_addr_port):
             requested.append(origin_addr_port)
             requested.append(server_addr_port)
             self.tc._queue_incoming_channel(c)
-            
+
         port = self.tc.request_port_forward('127.0.0.1', 0, handler)
         self.assertEqual(port, self.server._listen.getsockname()[1])
 
@@ -492,14 +513,14 @@ class TransportTest(ParamikoTest):
         ss, _ = self.server._listen.accept()
         sch = self.ts.open_forwarded_tcpip_channel(ss.getsockname(), ss.getpeername())
         cch = self.tc.accept()
-        
+
         sch.send('hello')
         self.assertEqual(b'hello', cch.recv(5))
         sch.close()
         cch.close()
         ss.close()
         cs.close()
-        
+
         # now cancel it.
         self.tc.cancel_port_forward('127.0.0.1', port)
         self.assertTrue(self.server._listen is None)
@@ -513,7 +534,7 @@ class TransportTest(ParamikoTest):
         chan = self.tc.open_session()
         chan.exec_command('yes')
         schan = self.ts.accept(1.0)
-        
+
         # open a port on the "server" that the client will ask to forward to.
         greeting_server = socket.socket()
         greeting_server.bind(('127.0.0.1', 0))
@@ -524,13 +545,13 @@ class TransportTest(ParamikoTest):
         sch = self.ts.accept(1.0)
         cch = socket.socket()
         cch.connect(self.server._tcpip_dest)
-        
+
         ss, _ = greeting_server.accept()
         ss.send(b'Hello!\n')
         ss.close()
         sch.send(cch.recv(8192))
         sch.close()
-        
+
         self.assertEqual(b'Hello!\n', cs.recv(7))
         cs.close()
 
@@ -544,14 +565,14 @@ class TransportTest(ParamikoTest):
         chan.invoke_shell()
         schan = self.ts.accept(1.0)
 
-        # nothing should be ready        
+        # nothing should be ready
         r, w, e = select.select([chan], [], [], 0.1)
         self.assertEqual([], r)
         self.assertEqual([], w)
         self.assertEqual([], e)
-        
+
         schan.send_stderr('hello\n')
-        
+
         # something should be ready now (give it 1 second to appear)
         for i in range(10):
             r, w, e = select.select([chan], [], [], 0.1)
@@ -563,7 +584,7 @@ class TransportTest(ParamikoTest):
         self.assertEqual([], e)
 
         self.assertEqual(b'hello\n', chan.recv_stderr(6))
-        
+
         # and, should be dead again now
         r, w, e = select.select([chan], [], [], 0.1)
         self.assertEqual([], r)
@@ -585,12 +606,13 @@ class TransportTest(ParamikoTest):
         self.assertEqual(chan.send_ready(), True)
         total = 0
         K = '*' * 1024
-        while total < 1024 * 1024:
+        limit = 1+(64 * 2 ** 15)
+        while total < limit:
             chan.send(K)
             total += len(K)
             if not chan.send_ready():
                 break
-        self.assertTrue(total < 1024 * 1024)
+        self.assertTrue(total < limit)
 
         schan.close()
         chan.close()
@@ -599,10 +621,10 @@ class TransportTest(ParamikoTest):
     def test_I_rekey_deadlock(self):
         """
         Regression test for deadlock when in-transit messages are received after MSG_KEXINIT is sent
-        
+
         Note: When this test fails, it may leak threads.
         """
-        
+
         # Test for an obscure deadlocking bug that can occur if we receive
         # certain messages while initiating a key exchange.
         #
@@ -619,7 +641,7 @@ class TransportTest(ParamikoTest):
         #      NeedRekeyException.
         #   4. In response to NeedRekeyException, the transport thread sends
         #      MSG_KEXINIT to the remote host.
-        # 
+        #
         # On the remote host (using any SSH implementation):
         #   5. The MSG_CHANNEL_DATA is received, and MSG_CHANNEL_WINDOW_ADJUST is sent.
         #   6. The MSG_KEXINIT is received, and a corresponding MSG_KEXINIT is sent.
@@ -654,7 +676,7 @@ class TransportTest(ParamikoTest):
                 self.done_event = done_event
                 self.watchdog_event = threading.Event()
                 self.last = None
-            
+
             def run(self):
                 try:
                     for i in range(1, 1+self.iterations):
@@ -666,7 +688,7 @@ class TransportTest(ParamikoTest):
                 finally:
                     self.done_event.set()
                     self.watchdog_event.set()
-        
+
         class ReceiveThread(threading.Thread):
             def __init__(self, chan, done_event):
                 threading.Thread.__init__(self, None, None, self.__class__.__name__)
@@ -674,7 +696,7 @@ class TransportTest(ParamikoTest):
                 self.chan = chan
                 self.done_event = done_event
                 self.watchdog_event = threading.Event()
-            
+
             def run(self):
                 try:
                     while not self.done_event.isSet():
@@ -687,10 +709,10 @@ class TransportTest(ParamikoTest):
                 finally:
                     self.done_event.set()
                     self.watchdog_event.set()
-        
+
         self.setup_test_server()
         self.ts.packetizer.REKEY_BYTES = 2048
-        
+
         chan = self.tc.open_session()
         chan.exec_command('yes')
         schan = self.ts.accept(1.0)
@@ -712,7 +734,7 @@ class TransportTest(ParamikoTest):
                 self._send_message(m2)
             return _negotiate_keys(self, m)
         self.tc._handler_table[MSG_KEXINIT] = _negotiate_keys_wrapper
-        
+
         # Parameters for the test
         iterations = 500    # The deadlock does not happen every time, but it
                             # should after many iterations.
@@ -724,12 +746,12 @@ class TransportTest(ParamikoTest):
         # Start the sending thread
         st = SendThread(schan, iterations, done_event)
         st.start()
-        
+
         # Start the receiving thread
         rt = ReceiveThread(chan, done_event)
         rt.start()
 
-        # Act as a watchdog timer, checking 
+        # Act as a watchdog timer, checking
         deadlocked = False
         while not deadlocked and not done_event.isSet():
             for event in (st.watchdog_event, rt.watchdog_event):
@@ -740,7 +762,7 @@ class TransportTest(ParamikoTest):
                     deadlocked = True
                     break
                 event.clear()
-        
+
         # Tell the threads to stop (if they haven't already stopped).  Note
         # that if one or more threads are deadlocked, they might hang around
         # forever (until the process exits).
@@ -752,3 +774,21 @@ class TransportTest(ParamikoTest):
         # Close the channels
         schan.close()
         chan.close()
+
+    def test_J_sanitze_packet_size(self):
+        """
+        verify that we conform to the rfc of packet and window sizes.
+        """
+        for val, correct in [(32767, MIN_PACKET_SIZE),
+                             (None, DEFAULT_MAX_PACKET_SIZE),
+                             (2**32, MAX_WINDOW_SIZE)]:
+            self.assertEqual(self.tc._sanitize_packet_size(val), correct)
+
+    def test_K_sanitze_window_size(self):
+        """
+        verify that we conform to the rfc of packet and window sizes.
+        """
+        for val, correct in [(32767, MIN_PACKET_SIZE),
+                             (None, DEFAULT_WINDOW_SIZE),
+                             (2**32, MAX_WINDOW_SIZE)]:
+            self.assertEqual(self.tc._sanitize_window_size(val), correct)