Imported Upstream version 1.7.4upstream/1.7.4

1 files changed, 263 insertions, 208 deletions

diff --git a/paramiko/channel.py b/paramiko/channel.py
index 8a00233..910a03c 100644
--- a/paramiko/channel.py
+++ b/paramiko/channel.py
@@ -1,4 +1,4 @@
-# Copyright (C) 2003-2005 Robey Pointer <robey@lag.net>
+# Copyright (C) 2003-2007  Robey Pointer <robey@lag.net>
 #
 # This file is part of paramiko.
 #
@@ -20,6 +20,7 @@
 Abstraction for an SSH2 channel.
 """
 
+import binascii
 import sys
 import time
 import threading
@@ -31,9 +32,14 @@ from paramiko import util
 from paramiko.message import Message
 from paramiko.ssh_exception import SSHException
 from paramiko.file import BufferedFile
+from paramiko.buffered_pipe import BufferedPipe, PipeTimeout
 from paramiko import pipe
 
 
+# lower bound on the max packet size we'll accept from the remote host
+MIN_PACKET_SIZE = 1024
+
+
 class Channel (object):
     """
     A secure tunnel across an SSH L{Transport}.  A Channel is meant to behave
@@ -49,9 +55,6 @@ class Channel (object):
     is exactly like a normal network socket, so it shouldn't be too surprising.
     """
 
-    # lower bound on the max packet size we'll accept from the remote host
-    MIN_PACKET_SIZE = 1024
-
     def __init__(self, chanid):
         """
         Create a new channel.  The channel is not associated with any
@@ -69,14 +72,12 @@ class Channel (object):
         self.active = False
         self.eof_received = 0
         self.eof_sent = 0
-        self.in_buffer = ''
-        self.in_stderr_buffer = ''
+        self.in_buffer = BufferedPipe()
+        self.in_stderr_buffer = BufferedPipe()
         self.timeout = None
         self.closed = False
         self.ultra_debug = False
         self.lock = threading.Lock()
-        self.in_buffer_cv = threading.Condition(self.lock)
-        self.in_stderr_buffer_cv = threading.Condition(self.lock)
         self.out_buffer_cv = threading.Condition(self.lock)
         self.in_window_size = 0
         self.out_window_size = 0
@@ -85,15 +86,19 @@ class Channel (object):
         self.in_window_threshold = 0
         self.in_window_sofar = 0
         self.status_event = threading.Event()
-        self.name = str(chanid)
-        self.logger = util.get_logger('paramiko.chan.' + str(chanid))
-        self.pipe = None
+        self._name = str(chanid)
+        self.logger = util.get_logger('paramiko.transport')
+        self._pipe = None
         self.event = threading.Event()
         self.combine_stderr = False
         self.exit_status = -1
+        self.origin_addr = None
     
     def __del__(self):
-        self.close()
+        try:
+            self.close()
+        except:
+            pass
         
     def __repr__(self):
         """
@@ -124,14 +129,15 @@ class Channel (object):
         It isn't necessary (or desirable) to call this method if you're going
         to exectue a single command with L{exec_command}.
 
-        @param term: the terminal type to emulate (for example, C{'vt100'}).
+        @param term: the terminal type to emulate (for example, C{'vt100'})
         @type term: str
         @param width: width (in characters) of the terminal screen
         @type width: int
         @param height: height (in characters) of the terminal screen
         @type height: int
-        @return: C{True} if the operation succeeded; C{False} if not.
-        @rtype: bool
+        
+        @raise SSHException: if the request was rejected or the channel was
+            closed
         """
         if self.closed or self.eof_received or self.eof_sent or not self.active:
             raise SSHException('Channel is not open')
@@ -148,12 +154,7 @@ class Channel (object):
         m.add_string('')
         self.event.clear()
         self.transport._send_user_message(m)
-        while True:
-            self.event.wait(0.1)
-            if self.closed:
-                return False
-            if self.event.isSet():
-                return True
+        self._wait_for_event()
 
     def invoke_shell(self):
         """
@@ -168,8 +169,8 @@ class Channel (object):
         When the shell exits, the channel will be closed and can't be reused.
         You must open a new channel if you wish to open another shell.
         
-        @return: C{True} if the operation succeeded; C{False} if not.
-        @rtype: bool
+        @raise SSHException: if the request was rejected or the channel was
+            closed
         """
         if self.closed or self.eof_received or self.eof_sent or not self.active:
             raise SSHException('Channel is not open')
@@ -180,12 +181,7 @@ class Channel (object):
         m.add_boolean(1)
         self.event.clear()
         self.transport._send_user_message(m)
-        while True:
-            self.event.wait(0.1)
-            if self.closed:
-                return False
-            if self.event.isSet():
-                return True
+        self._wait_for_event()
 
     def exec_command(self, command):
         """
@@ -199,8 +195,9 @@ class Channel (object):
 
         @param command: a shell command to execute.
         @type command: str
-        @return: C{True} if the operation succeeded; C{False} if not.
-        @rtype: bool
+
+        @raise SSHException: if the request was rejected or the channel was
+            closed
         """
         if self.closed or self.eof_received or self.eof_sent or not self.active:
             raise SSHException('Channel is not open')
@@ -208,16 +205,11 @@ class Channel (object):
         m.add_byte(chr(MSG_CHANNEL_REQUEST))
         m.add_int(self.remote_chanid)
         m.add_string('exec')
-        m.add_boolean(1)
+        m.add_boolean(True)
         m.add_string(command)
         self.event.clear()
         self.transport._send_user_message(m)
-        while True:
-            self.event.wait(0.1)
-            if self.closed:
-                return False
-            if self.event.isSet():
-                return True
+        self._wait_for_event()
 
     def invoke_subsystem(self, subsystem):
         """
@@ -230,8 +222,9 @@ class Channel (object):
 
         @param subsystem: name of the subsystem being requested.
         @type subsystem: str
-        @return: C{True} if the operation succeeded; C{False} if not.
-        @rtype: bool
+
+        @raise SSHException: if the request was rejected or the channel was
+            closed
         """
         if self.closed or self.eof_received or self.eof_sent or not self.active:
             raise SSHException('Channel is not open')
@@ -239,16 +232,11 @@ class Channel (object):
         m.add_byte(chr(MSG_CHANNEL_REQUEST))
         m.add_int(self.remote_chanid)
         m.add_string('subsystem')
-        m.add_boolean(1)
+        m.add_boolean(True)
         m.add_string(subsystem)
         self.event.clear()
         self.transport._send_user_message(m)
-        while True:
-            self.event.wait(0.1)
-            if self.closed:
-                return False
-            if self.event.isSet():
-                return True
+        self._wait_for_event()
 
     def resize_pty(self, width=80, height=24):
         """
@@ -259,8 +247,9 @@ class Channel (object):
         @type width: int
         @param height: new height (in characters) of the terminal screen
         @type height: int
-        @return: C{True} if the operation succeeded; C{False} if not.
-        @rtype: bool
+
+        @raise SSHException: if the request was rejected or the channel was
+            closed
         """
         if self.closed or self.eof_received or self.eof_sent or not self.active:
             raise SSHException('Channel is not open')
@@ -268,19 +257,27 @@ class Channel (object):
         m.add_byte(chr(MSG_CHANNEL_REQUEST))
         m.add_int(self.remote_chanid)
         m.add_string('window-change')
-        m.add_boolean(1)
+        m.add_boolean(True)
         m.add_int(width)
         m.add_int(height)
         m.add_int(0).add_int(0)
         self.event.clear()
         self.transport._send_user_message(m)
-        while True:
-            self.event.wait(0.1)
-            if self.closed:
-                return False
-            if self.event.isSet():
-                return True
+        self._wait_for_event()
 
+    def exit_status_ready(self):
+        """
+        Return true if the remote process has exited and returned an exit
+        status. You may use this to poll the process status if you don't
+        want to block in L{recv_exit_status}. Note that the server may not
+        return an exit status in some cases (like bad servers).
+        
+        @return: True if L{recv_exit_status} will return immediately
+        @rtype: bool
+        @since: 1.7.3
+        """
+        return self.closed or self.status_event.isSet()
+        
     def recv_exit_status(self):
         """
         Return the exit status from the process on the server.  This is
@@ -296,8 +293,9 @@ class Channel (object):
         """
         while True:
             if self.closed or self.status_event.isSet():
-                return self.exit_status
+                break
             self.status_event.wait(0.1)
+        return self.exit_status
 
     def send_exit_status(self, status):
         """
@@ -317,10 +315,73 @@ class Channel (object):
         m.add_byte(chr(MSG_CHANNEL_REQUEST))
         m.add_int(self.remote_chanid)
         m.add_string('exit-status')
-        m.add_boolean(0)
+        m.add_boolean(False)
         m.add_int(status)
         self.transport._send_user_message(m)
+    
+    def request_x11(self, screen_number=0, auth_protocol=None, auth_cookie=None,
+                    single_connection=False, handler=None):
+        """
+        Request an x11 session on this channel.  If the server allows it,
+        further x11 requests can be made from the server to the client,
+        when an x11 application is run in a shell session.
+        
+        From RFC4254::
+
+            It is RECOMMENDED that the 'x11 authentication cookie' that is
+            sent be a fake, random cookie, and that the cookie be checked and
+            replaced by the real cookie when a connection request is received.
         
+        If you omit the auth_cookie, a new secure random 128-bit value will be
+        generated, used, and returned.  You will need to use this value to
+        verify incoming x11 requests and replace them with the actual local
+        x11 cookie (which requires some knoweldge of the x11 protocol).
+        
+        If a handler is passed in, the handler is called from another thread
+        whenever a new x11 connection arrives.  The default handler queues up
+        incoming x11 connections, which may be retrieved using
+        L{Transport.accept}.  The handler's calling signature is::
+        
+            handler(channel: Channel, (address: str, port: int))
+        
+        @param screen_number: the x11 screen number (0, 10, etc)
+        @type screen_number: int
+        @param auth_protocol: the name of the X11 authentication method used;
+            if none is given, C{"MIT-MAGIC-COOKIE-1"} is used
+        @type auth_protocol: str
+        @param auth_cookie: hexadecimal string containing the x11 auth cookie;
+            if none is given, a secure random 128-bit value is generated
+        @type auth_cookie: str
+        @param single_connection: if True, only a single x11 connection will be
+            forwarded (by default, any number of x11 connections can arrive
+            over this session)
+        @type single_connection: bool
+        @param handler: an optional handler to use for incoming X11 connections
+        @type handler: function
+        @return: the auth_cookie used
+        """
+        if self.closed or self.eof_received or self.eof_sent or not self.active:
+            raise SSHException('Channel is not open')
+        if auth_protocol is None:
+            auth_protocol = 'MIT-MAGIC-COOKIE-1'
+        if auth_cookie is None:
+            auth_cookie = binascii.hexlify(self.transport.randpool.get_bytes(16))
+
+        m = Message()
+        m.add_byte(chr(MSG_CHANNEL_REQUEST))
+        m.add_int(self.remote_chanid)
+        m.add_string('x11-req')
+        m.add_boolean(True)
+        m.add_boolean(single_connection)
+        m.add_string(auth_protocol)
+        m.add_string(auth_cookie)
+        m.add_int(screen_number)
+        self.event.clear()
+        self.transport._send_user_message(m)
+        self._wait_for_event()
+        self.transport._set_x11_handler(handler)
+        return auth_cookie
+
     def get_transport(self):
         """
         Return the L{Transport} associated with this channel.
@@ -333,14 +394,13 @@ class Channel (object):
     def set_name(self, name):
         """
         Set a name for this channel.  Currently it's only used to set the name
-        of the log level used for debugging.  The name can be fetched with the
+        of the channel in logfile entries.  The name can be fetched with the
         L{get_name} method.
 
-        @param name: new channel name.
+        @param name: new channel name
         @type name: str
         """
-        self.name = name
-        self.logger = util.get_logger(self.transport.get_log_channel() + '.' + self.name)
+        self._name = name
 
     def get_name(self):
         """
@@ -349,7 +409,7 @@ class Channel (object):
         @return: the name of this channel.
         @rtype: str
         """
-        return self.name
+        return self._name
 
     def get_id(self):
         """
@@ -360,8 +420,6 @@ class Channel (object):
 
         @return: the ID of this channel.
         @rtype: int
-
-        @since: ivysaur
         """
         return self.chanid
     
@@ -394,8 +452,7 @@ class Channel (object):
             self.combine_stderr = combine
             if combine and not old:
                 # copy old stderr buffer into primary buffer
-                data = self.in_stderr_buffer
-                self.in_stderr_buffer = ''
+                data = self.in_stderr_buffer.empty()
         finally:
             self.lock.release()
         if len(data) > 0:
@@ -419,7 +476,7 @@ class Channel (object):
         C{chan.settimeout(None)} is equivalent to C{chan.setblocking(1)}.
 
         @param timeout: seconds to wait for a pending read/write operation
-        before raising C{socket.timeout}, or C{None} for no timeout.
+            before raising C{socket.timeout}, or C{None} for no timeout.
         @type timeout: float
         """
         self.timeout = timeout
@@ -439,17 +496,19 @@ class Channel (object):
         """
         Set blocking or non-blocking mode of the channel: if C{blocking} is 0,
         the channel is set to non-blocking mode; otherwise it's set to blocking
-        mode.  Initially all channels are in blocking mode.
+        mode. Initially all channels are in blocking mode.
 
         In non-blocking mode, if a L{recv} call doesn't find any data, or if a
         L{send} call can't immediately dispose of the data, an error exception
-        is raised.  In blocking mode, the calls block until they can proceed.
+        is raised. In blocking mode, the calls block until they can proceed. An
+        EOF condition is considered "immediate data" for L{recv}, so if the
+        channel is closed in the read direction, it will never block.
 
         C{chan.setblocking(0)} is equivalent to C{chan.settimeout(0)};
         C{chan.setblocking(1)} is equivalent to C{chan.settimeout(None)}.
 
         @param blocking: 0 to set non-blocking mode; non-0 to set blocking
-        mode.
+            mode.
         @type blocking: int
         """
         if blocking:
@@ -457,6 +516,18 @@ class Channel (object):
         else:
             self.settimeout(0.0)
 
+    def getpeername(self):
+        """
+        Return the address of the remote side of this Channel, if possible.
+        This is just a wrapper around C{'getpeername'} on the Transport, used
+        to provide enough of a socket-like interface to allow asyncore to work.
+        (asyncore likes to call C{'getpeername'}.)
+
+        @return: the address if the remote host, if known
+        @rtype: tuple(str, int)
+        """
+        return self.transport.getpeername()
+
     def close(self):
         """
         Close the channel.  All future read/write operations on the channel
@@ -466,15 +537,17 @@ class Channel (object):
         """
         self.lock.acquire()
         try:
+            # only close the pipe when the user explicitly closes the channel.
+            # otherwise they will get unpleasant surprises.  (and do it before
+            # checking self.closed, since the remote host may have already
+            # closed the connection.)
+            if self._pipe is not None:
+                self._pipe.close()
+                self._pipe = None
+
             if not self.active or self.closed:
                 return
             msgs = self._close_internal()
-
-            # only close the pipe when the user explicitly closes the channel.
-            # otherwise they will get unpleasant surprises.
-            if self.pipe is not None:
-                self.pipe.close()
-                self.pipe = None
         finally:
             self.lock.release()
         for m in msgs:
@@ -491,13 +564,7 @@ class Channel (object):
             return at least one byte; C{False} otherwise.
         @rtype: boolean
         """
-        self.lock.acquire()
-        try:
-            if len(self.in_buffer) == 0:
-                return False
-            return True
-        finally:
-            self.lock.release()
+        return self.in_buffer.read_ready()
 
     def recv(self, nbytes):
         """
@@ -514,38 +581,12 @@ class Channel (object):
         @raise socket.timeout: if no data is ready before the timeout set by
             L{settimeout}.
         """
-        out = ''
-        self.lock.acquire()
         try:
-            if len(self.in_buffer) == 0:
-                if self.closed or self.eof_received:
-                    return out
-                # should we block?
-                if self.timeout == 0.0:
-                    raise socket.timeout()
-                # loop here in case we get woken up but a different thread has grabbed everything in the buffer
-                timeout = self.timeout
-                while (len(self.in_buffer) == 0) and not self.closed and not self.eof_received:
-                    then = time.time()
-                    self.in_buffer_cv.wait(timeout)
-                    if timeout != None:
-                        timeout -= time.time() - then
-                        if timeout <= 0.0:
-                            raise socket.timeout()
-            # something in the buffer and we have the lock
-            if len(self.in_buffer) <= nbytes:
-                out = self.in_buffer
-                self.in_buffer = ''
-                if self.pipe is not None:
-                    # clear the pipe, since no more data is buffered
-                    self.pipe.clear()
-            else:
-                out = self.in_buffer[:nbytes]
-                self.in_buffer = self.in_buffer[nbytes:]
-            ack = self._check_add_window(len(out))
-        finally:
-            self.lock.release()
+            out = self.in_buffer.read(nbytes, self.timeout)
+        except PipeTimeout, e:
+            raise socket.timeout()
 
+        ack = self._check_add_window(len(out))
         # no need to hold the channel lock when sending this
         if ack > 0:
             m = Message()
@@ -569,13 +610,7 @@ class Channel (object):
         
         @since: 1.1
         """
-        self.lock.acquire()
-        try:
-            if len(self.in_stderr_buffer) == 0:
-                return False
-            return True
-        finally:
-            self.lock.release()
+        return self.in_stderr_buffer.read_ready()
 
     def recv_stderr(self, nbytes):
         """
@@ -596,36 +631,43 @@ class Channel (object):
         
         @since: 1.1
         """
-        out = ''
+        try:
+            out = self.in_stderr_buffer.read(nbytes, self.timeout)
+        except PipeTimeout, e:
+            raise socket.timeout()
+            
+        ack = self._check_add_window(len(out))
+        # no need to hold the channel lock when sending this
+        if ack > 0:
+            m = Message()
+            m.add_byte(chr(MSG_CHANNEL_WINDOW_ADJUST))
+            m.add_int(self.remote_chanid)
+            m.add_int(ack)
+            self.transport._send_user_message(m)
+
+        return out
+
+    def send_ready(self):
+        """
+        Returns true if data can be written to this channel without blocking.
+        This means the channel is either closed (so any write attempt would
+        return immediately) or there is at least one byte of space in the 
+        outbound buffer. If there is at least one byte of space in the
+        outbound buffer, a L{send} call will succeed immediately and return
+        the number of bytes actually written.
+        
+        @return: C{True} if a L{send} call on this channel would immediately
+            succeed or fail
+        @rtype: boolean
+        """
         self.lock.acquire()
         try:
-            if len(self.in_stderr_buffer) == 0:
-                if self.closed or self.eof_received:
-                    return out
-                # should we block?
-                if self.timeout == 0.0:
-                    raise socket.timeout()
-                # loop here in case we get woken up but a different thread has grabbed everything in the buffer
-                timeout = self.timeout
-                while (len(self.in_stderr_buffer) == 0) and not self.closed and not self.eof_received:
-                    then = time.time()
-                    self.in_stderr_buffer_cv.wait(timeout)
-                    if timeout != None:
-                        timeout -= time.time() - then
-                        if timeout <= 0.0:
-                            raise socket.timeout()
-            # something in the buffer and we have the lock
-            if len(self.in_stderr_buffer) <= nbytes:
-                out = self.in_stderr_buffer
-                self.in_stderr_buffer = ''
-            else:
-                out = self.in_stderr_buffer[:nbytes]
-                self.in_stderr_buffer = self.in_stderr_buffer[nbytes:]
-            self._check_add_window(len(out))
+            if self.closed or self.eof_sent:
+                return True
+            return self.out_window_size > 0
         finally:
             self.lock.release()
-        return out
-
+    
     def send(self, s):
         """
         Send data to the channel.  Returns the number of bytes sent, or 0 if
@@ -634,9 +676,9 @@ class Channel (object):
         transmitted, the application needs to attempt delivery of the remaining
         data.
 
-        @param s: data to send.
+        @param s: data to send
         @type s: str
-        @return: number of bytes actually sent.
+        @return: number of bytes actually sent
         @rtype: int
 
         @raise socket.timeout: if no data could be sent before the timeout set
@@ -653,9 +695,11 @@ class Channel (object):
             m.add_byte(chr(MSG_CHANNEL_DATA))
             m.add_int(self.remote_chanid)
             m.add_string(s[:size])
-            self.transport._send_user_message(m)
         finally:
             self.lock.release()
+        # Note: We release self.lock before calling _send_user_message.
+        # Otherwise, we can deadlock during re-keying.
+        self.transport._send_user_message(m)
         return size
 
     def send_stderr(self, s):
@@ -689,9 +733,11 @@ class Channel (object):
             m.add_int(self.remote_chanid)
             m.add_int(1)
             m.add_string(s[:size])
-            self.transport._send_user_message(m)
         finally:
             self.lock.release()
+        # Note: We release self.lock before calling _send_user_message.
+        # Otherwise, we can deadlock during re-keying.
+        self.transport._send_user_message(m)
         return size
 
     def sendall(self, s):
@@ -776,14 +822,14 @@ class Channel (object):
     def fileno(self):
         """
         Returns an OS-level file descriptor which can be used for polling, but
-        but I{not} for reading or writing).  This is primaily to allow python's
+        but I{not} for reading or writing.  This is primaily to allow python's
         C{select} module to work.
 
         The first time C{fileno} is called on a channel, a pipe is created to
         simulate real OS-level file descriptor (FD) behavior.  Because of this,
         two OS-level FDs are created, which will use up FDs faster than normal.
-        You won't notice this effect unless you open hundreds or thousands of
-        channels simultaneously, but it's still notable.
+        (You won't notice this effect unless you have hundreds of channels
+        open at the same time.)
 
         @return: an OS-level file descriptor
         @rtype: int
@@ -793,13 +839,14 @@ class Channel (object):
         """
         self.lock.acquire()
         try:
-            if self.pipe is not None:
-                return self.pipe.fileno()
+            if self._pipe is not None:
+                return self._pipe.fileno()
             # create the pipe and feed in any existing data
-            self.pipe = pipe.make_pipe()
-            if len(self.in_buffer) > 0:
-                self.pipe.set()
-            return self.pipe.fileno()
+            self._pipe = pipe.make_pipe()
+            p1, p2 = pipe.make_or_pipe(self._pipe)
+            self.in_buffer.set_event(p1)
+            self.in_stderr_buffer.set_event(p2)
+            return self._pipe.fileno()
         finally:
             self.lock.release()
 
@@ -856,7 +903,7 @@ class Channel (object):
 
     def _set_transport(self, transport):
         self.transport = transport
-        self.logger = util.get_logger(self.transport.get_log_channel() + '.' + self.name)
+        self.logger = util.get_logger(self.transport.get_log_channel())
 
     def _set_window(self, window_size, max_packet_size):
         self.in_window_size = window_size
@@ -869,7 +916,7 @@ class Channel (object):
     def _set_remote_channel(self, chanid, window_size, max_packet_size):
         self.remote_chanid = chanid
         self.out_window_size = window_size
-        self.out_max_packet_size = max(max_packet_size, self.MIN_PACKET_SIZE)
+        self.out_max_packet_size = max(max_packet_size, MIN_PACKET_SIZE)
         self.active = 1
         self._log(DEBUG, 'Max packet out: %d bytes' % max_packet_size)
 
@@ -894,16 +941,7 @@ class Channel (object):
             s = m
         else:
             s = m.get_string()
-        self.lock.acquire()
-        try:
-            if self.ultra_debug:
-                self._log(DEBUG, 'fed %d bytes' % len(s))
-            if self.pipe is not None:
-                self.pipe.set()
-            self.in_buffer += s
-	    self.in_buffer_cv.notifyAll()
-        finally:
-            self.lock.release()
+        self.in_buffer.feed(s)
 
     def _feed_extended(self, m):
         code = m.get_int()
@@ -912,15 +950,9 @@ class Channel (object):
             self._log(ERROR, 'unknown extended_data type %d; discarding' % code)
             return
         if self.combine_stderr:
-            return self._feed(s)
-        self.lock.acquire()
-        try:
-            if self.ultra_debug:
-                self._log(DEBUG, 'fed %d stderr bytes' % len(s))
-            self.in_stderr_buffer += s
-            self.in_stderr_buffer_cv.notifyAll()
-        finally:
-            self.lock.release()
+            self._feed(s)
+        else:
+            self.in_stderr_buffer.feed(s)
         
     def _window_adjust(self, m):
         nbytes = m.get_int()
@@ -984,6 +1016,16 @@ class Channel (object):
             else:
                 ok = server.check_channel_window_change_request(self, width, height, pixelwidth,
                                                                 pixelheight)
+        elif key == 'x11-req':
+            single_connection = m.get_boolean()
+            auth_proto = m.get_string()
+            auth_cookie = m.get_string()
+            screen_number = m.get_int()
+            if server is None:
+                ok = False
+            else:
+                ok = server.check_channel_x11_request(self, single_connection,
+                                                      auth_proto, auth_cookie, screen_number)
         else:
             self._log(DEBUG, 'Unhandled channel request "%s"' % key)
             ok = False
@@ -1001,13 +1043,13 @@ class Channel (object):
         try:
             if not self.eof_received:
                 self.eof_received = True
-                self.in_buffer_cv.notifyAll()
-                self.in_stderr_buffer_cv.notifyAll()
-                if self.pipe is not None:
-                    self.pipe.set_forever()
+                self.in_buffer.close()
+                self.in_stderr_buffer.close()
+                if self._pipe is not None:
+                    self._pipe.set_forever()
         finally:
             self.lock.release()
-        self._log(DEBUG, 'EOF received')
+        self._log(DEBUG, 'EOF received (%s)', self._name)
 
     def _handle_close(self, m):
         self.lock.acquire()
@@ -1024,17 +1066,29 @@ class Channel (object):
     ###  internals...
 
 
-    def _log(self, level, msg):
-        self.logger.log(level, msg)
+    def _log(self, level, msg, *args):
+        self.logger.log(level, "[chan " + self._name + "] " + msg, *args)
+
+    def _wait_for_event(self):
+        while True:
+            self.event.wait(0.1)
+            if self.event.isSet():
+                return
+            if self.closed:
+                e = self.transport.get_exception()
+                if e is None:
+                    e = SSHException('Channel closed.')
+                raise e
+        return
 
     def _set_closed(self):
         # you are holding the lock.
         self.closed = True
-        self.in_buffer_cv.notifyAll()
-        self.in_stderr_buffer_cv.notifyAll()
+        self.in_buffer.close()
+        self.in_stderr_buffer.close()
         self.out_buffer_cv.notifyAll()
-        if self.pipe is not None:
-            self.pipe.set_forever()
+        if self._pipe is not None:
+            self._pipe.set_forever()
 
     def _send_eof(self):
         # you are holding the lock.
@@ -1044,7 +1098,7 @@ class Channel (object):
         m.add_byte(chr(MSG_CHANNEL_EOF))
         m.add_int(self.remote_chanid)
         self.eof_sent = True
-        self._log(DEBUG, 'EOF sent')
+        self._log(DEBUG, 'EOF sent (%s)', self._name)
         return m
 
     def _close_internal(self):
@@ -1072,19 +1126,22 @@ class Channel (object):
             self.lock.release()
 
     def _check_add_window(self, n):
-        # already holding the lock!
-        if self.closed or self.eof_received or not self.active:
-            return 0
-        if self.ultra_debug:
-            self._log(DEBUG, 'addwindow %d' % n)
-        self.in_window_sofar += n
-        if self.in_window_sofar <= self.in_window_threshold:
-            return 0
-        if self.ultra_debug:
-            self._log(DEBUG, 'addwindow send %d' % self.in_window_sofar)
-        out = self.in_window_sofar
-        self.in_window_sofar = 0
-        return out
+        self.lock.acquire()
+        try:
+            if self.closed or self.eof_received or not self.active:
+                return 0
+            if self.ultra_debug:
+                self._log(DEBUG, 'addwindow %d' % n)
+            self.in_window_sofar += n
+            if self.in_window_sofar <= self.in_window_threshold:
+                return 0
+            if self.ultra_debug:
+                self._log(DEBUG, 'addwindow send %d' % self.in_window_sofar)
+            out = self.in_window_sofar
+            self.in_window_sofar = 0
+            return out
+        finally:
+            self.lock.release()
 
     def _wait_for_send_window(self, size):
         """
@@ -1155,8 +1212,6 @@ class ChannelFile (BufferedFile):
     def _write(self, data):
         self.channel.sendall(data)
         return len(data)
-    
-    seek = BufferedFile.seek
 
 
 class ChannelStderrFile (ChannelFile):