Package paramiko :: Module channel
[frames] | no frames]

Source Code for Module paramiko.channel

   1  # Copyright (C) 2003-2007  Robey Pointer <robey@lag.net> 
   2  # 
   3  # This file is part of paramiko. 
   4  # 
   5  # Paramiko is free software; you can redistribute it and/or modify it under the 
   6  # terms of the GNU Lesser General Public License as published by the Free 
   7  # Software Foundation; either version 2.1 of the License, or (at your option) 
   8  # any later version. 
   9  # 
  10  # Paramiko is distrubuted in the hope that it will be useful, but WITHOUT ANY 
  11  # WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR 
  12  # A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more 
  13  # details. 
  14  # 
  15  # You should have received a copy of the GNU Lesser General Public License 
  16  # along with Paramiko; if not, write to the Free Software Foundation, Inc., 
  17  # 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA. 
  18   
  19  """ 
  20  Abstraction for an SSH2 channel. 
  21  """ 
  22   
  23  import binascii 
  24  import sys 
  25  import time 
  26  import threading 
  27  import socket 
  28  import os 
  29   
  30  from paramiko.common import * 
  31  from paramiko import util 
  32  from paramiko.message import Message 
  33  from paramiko.ssh_exception import SSHException 
  34  from paramiko.file import BufferedFile 
  35  from paramiko.buffered_pipe import BufferedPipe, PipeTimeout 
  36  from paramiko import pipe 
  37   
  38   
  39  # lower bound on the max packet size we'll accept from the remote host 
  40  MIN_PACKET_SIZE = 1024 
  41   
  42   
43 -class Channel (object):
44 """ 45 A secure tunnel across an SSH L{Transport}. A Channel is meant to behave 46 like a socket, and has an API that should be indistinguishable from the 47 python socket API. 48 49 Because SSH2 has a windowing kind of flow control, if you stop reading data 50 from a Channel and its buffer fills up, the server will be unable to send 51 you any more data until you read some of it. (This won't affect other 52 channels on the same transport -- all channels on a single transport are 53 flow-controlled independently.) Similarly, if the server isn't reading 54 data you send, calls to L{send} may block, unless you set a timeout. This 55 is exactly like a normal network socket, so it shouldn't be too surprising. 56 """ 57
58 - def __init__(self, chanid):
59 """ 60 Create a new channel. The channel is not associated with any 61 particular session or L{Transport} until the Transport attaches it. 62 Normally you would only call this method from the constructor of a 63 subclass of L{Channel}. 64 65 @param chanid: the ID of this channel, as passed by an existing 66 L{Transport}. 67 @type chanid: int 68 """ 69 self.chanid = chanid 70 self.remote_chanid = 0 71 self.transport = None 72 self.active = False 73 self.eof_received = 0 74 self.eof_sent = 0 75 self.in_buffer = BufferedPipe() 76 self.in_stderr_buffer = BufferedPipe() 77 self.timeout = None 78 self.closed = False 79 self.ultra_debug = False 80 self.lock = threading.Lock() 81 self.out_buffer_cv = threading.Condition(self.lock) 82 self.in_window_size = 0 83 self.out_window_size = 0 84 self.in_max_packet_size = 0 85 self.out_max_packet_size = 0 86 self.in_window_threshold = 0 87 self.in_window_sofar = 0 88 self.status_event = threading.Event() 89 self._name = str(chanid) 90 self.logger = util.get_logger('paramiko.transport') 91 self._pipe = None 92 self.event = threading.Event() 93 self.combine_stderr = False 94 self.exit_status = -1 95 self.origin_addr = None
96
97 - def __del__(self):
98 try: 99 self.close() 100 except: 101 pass
102
103 - def __repr__(self):
104 """ 105 Return a string representation of this object, for debugging. 106 107 @rtype: str 108 """ 109 out = '<paramiko.Channel %d' % self.chanid 110 if self.closed: 111 out += ' (closed)' 112 elif self.active: 113 if self.eof_received: 114 out += ' (EOF received)' 115 if self.eof_sent: 116 out += ' (EOF sent)' 117 out += ' (open) window=%d' % (self.out_window_size) 118 if len(self.in_buffer) > 0: 119 out += ' in-buffer=%d' % (len(self.in_buffer),) 120 out += ' -> ' + repr(self.transport) 121 out += '>' 122 return out
123
124 - def get_pty(self, term='vt100', width=80, height=24):
125 """ 126 Request a pseudo-terminal from the server. This is usually used right 127 after creating a client channel, to ask the server to provide some 128 basic terminal semantics for a shell invoked with L{invoke_shell}. 129 It isn't necessary (or desirable) to call this method if you're going 130 to exectue a single command with L{exec_command}. 131 132 @param term: the terminal type to emulate (for example, C{'vt100'}) 133 @type term: str 134 @param width: width (in characters) of the terminal screen 135 @type width: int 136 @param height: height (in characters) of the terminal screen 137 @type height: int 138 139 @raise SSHException: if the request was rejected or the channel was 140 closed 141 """ 142 if self.closed or self.eof_received or self.eof_sent or not self.active: 143 raise SSHException('Channel is not open') 144 m = Message() 145 m.add_byte(chr(MSG_CHANNEL_REQUEST)) 146 m.add_int(self.remote_chanid) 147 m.add_string('pty-req') 148 m.add_boolean(True) 149 m.add_string(term) 150 m.add_int(width) 151 m.add_int(height) 152 # pixel height, width (usually useless) 153 m.add_int(0).add_int(0) 154 m.add_string('') 155 self.event.clear() 156 self.transport._send_user_message(m) 157 self._wait_for_event()
158
159 - def invoke_shell(self):
160 """ 161 Request an interactive shell session on this channel. If the server 162 allows it, the channel will then be directly connected to the stdin, 163 stdout, and stderr of the shell. 164 165 Normally you would call L{get_pty} before this, in which case the 166 shell will operate through the pty, and the channel will be connected 167 to the stdin and stdout of the pty. 168 169 When the shell exits, the channel will be closed and can't be reused. 170 You must open a new channel if you wish to open another shell. 171 172 @raise SSHException: if the request was rejected or the channel was 173 closed 174 """ 175 if self.closed or self.eof_received or self.eof_sent or not self.active: 176 raise SSHException('Channel is not open') 177 m = Message() 178 m.add_byte(chr(MSG_CHANNEL_REQUEST)) 179 m.add_int(self.remote_chanid) 180 m.add_string('shell') 181 m.add_boolean(1) 182 self.event.clear() 183 self.transport._send_user_message(m) 184 self._wait_for_event()
185
186 - def exec_command(self, command):
187 """ 188 Execute a command on the server. If the server allows it, the channel 189 will then be directly connected to the stdin, stdout, and stderr of 190 the command being executed. 191 192 When the command finishes executing, the channel will be closed and 193 can't be reused. You must open a new channel if you wish to execute 194 another command. 195 196 @param command: a shell command to execute. 197 @type command: str 198 199 @raise SSHException: if the request was rejected or the channel was 200 closed 201 """ 202 if self.closed or self.eof_received or self.eof_sent or not self.active: 203 raise SSHException('Channel is not open') 204 m = Message() 205 m.add_byte(chr(MSG_CHANNEL_REQUEST)) 206 m.add_int(self.remote_chanid) 207 m.add_string('exec') 208 m.add_boolean(True) 209 m.add_string(command) 210 self.event.clear() 211 self.transport._send_user_message(m) 212 self._wait_for_event()
213
214 - def invoke_subsystem(self, subsystem):
215 """ 216 Request a subsystem on the server (for example, C{sftp}). If the 217 server allows it, the channel will then be directly connected to the 218 requested subsystem. 219 220 When the subsystem finishes, the channel will be closed and can't be 221 reused. 222 223 @param subsystem: name of the subsystem being requested. 224 @type subsystem: str 225 226 @raise SSHException: if the request was rejected or the channel was 227 closed 228 """ 229 if self.closed or self.eof_received or self.eof_sent or not self.active: 230 raise SSHException('Channel is not open') 231 m = Message() 232 m.add_byte(chr(MSG_CHANNEL_REQUEST)) 233 m.add_int(self.remote_chanid) 234 m.add_string('subsystem') 235 m.add_boolean(True) 236 m.add_string(subsystem) 237 self.event.clear() 238 self.transport._send_user_message(m) 239 self._wait_for_event()
240
241 - def resize_pty(self, width=80, height=24):
242 """ 243 Resize the pseudo-terminal. This can be used to change the width and 244 height of the terminal emulation created in a previous L{get_pty} call. 245 246 @param width: new width (in characters) of the terminal screen 247 @type width: int 248 @param height: new height (in characters) of the terminal screen 249 @type height: int 250 251 @raise SSHException: if the request was rejected or the channel was 252 closed 253 """ 254 if self.closed or self.eof_received or self.eof_sent or not self.active: 255 raise SSHException('Channel is not open') 256 m = Message() 257 m.add_byte(chr(MSG_CHANNEL_REQUEST)) 258 m.add_int(self.remote_chanid) 259 m.add_string('window-change') 260 m.add_boolean(True) 261 m.add_int(width) 262 m.add_int(height) 263 m.add_int(0).add_int(0) 264 self.event.clear() 265 self.transport._send_user_message(m) 266 self._wait_for_event()
267
268 - def exit_status_ready(self):
269 """ 270 Return true if the remote process has exited and returned an exit 271 status. You may use this to poll the process status if you don't 272 want to block in L{recv_exit_status}. Note that the server may not 273 return an exit status in some cases (like bad servers). 274 275 @return: True if L{recv_exit_status} will return immediately 276 @rtype: bool 277 @since: 1.7.3 278 """ 279 return self.closed or self.status_event.isSet()
280
281 - def recv_exit_status(self):
282 """ 283 Return the exit status from the process on the server. This is 284 mostly useful for retrieving the reults of an L{exec_command}. 285 If the command hasn't finished yet, this method will wait until 286 it does, or until the channel is closed. If no exit status is 287 provided by the server, -1 is returned. 288 289 @return: the exit code of the process on the server. 290 @rtype: int 291 292 @since: 1.2 293 """ 294 while True: 295 if self.closed or self.status_event.isSet(): 296 break 297 self.status_event.wait(0.1) 298 return self.exit_status
299
300 - def send_exit_status(self, status):
301 """ 302 Send the exit status of an executed command to the client. (This 303 really only makes sense in server mode.) Many clients expect to 304 get some sort of status code back from an executed command after 305 it completes. 306 307 @param status: the exit code of the process 308 @type status: int 309 310 @since: 1.2 311 """ 312 # in many cases, the channel will not still be open here. 313 # that's fine. 314 m = Message() 315 m.add_byte(chr(MSG_CHANNEL_REQUEST)) 316 m.add_int(self.remote_chanid) 317 m.add_string('exit-status') 318 m.add_boolean(False) 319 m.add_int(status) 320 self.transport._send_user_message(m)
321
322 - def request_x11(self, screen_number=0, auth_protocol=None, auth_cookie=None, 323 single_connection=False, handler=None):
324 """ 325 Request an x11 session on this channel. If the server allows it, 326 further x11 requests can be made from the server to the client, 327 when an x11 application is run in a shell session. 328 329 From RFC4254:: 330 331 It is RECOMMENDED that the 'x11 authentication cookie' that is 332 sent be a fake, random cookie, and that the cookie be checked and 333 replaced by the real cookie when a connection request is received. 334 335 If you omit the auth_cookie, a new secure random 128-bit value will be 336 generated, used, and returned. You will need to use this value to 337 verify incoming x11 requests and replace them with the actual local 338 x11 cookie (which requires some knoweldge of the x11 protocol). 339 340 If a handler is passed in, the handler is called from another thread 341 whenever a new x11 connection arrives. The default handler queues up 342 incoming x11 connections, which may be retrieved using 343 L{Transport.accept}. The handler's calling signature is:: 344 345 handler(channel: Channel, (address: str, port: int)) 346 347 @param screen_number: the x11 screen number (0, 10, etc) 348 @type screen_number: int 349 @param auth_protocol: the name of the X11 authentication method used; 350 if none is given, C{"MIT-MAGIC-COOKIE-1"} is used 351 @type auth_protocol: str 352 @param auth_cookie: hexadecimal string containing the x11 auth cookie; 353 if none is given, a secure random 128-bit value is generated 354 @type auth_cookie: str 355 @param single_connection: if True, only a single x11 connection will be 356 forwarded (by default, any number of x11 connections can arrive 357 over this session) 358 @type single_connection: bool 359 @param handler: an optional handler to use for incoming X11 connections 360 @type handler: function 361 @return: the auth_cookie used 362 """ 363 if self.closed or self.eof_received or self.eof_sent or not self.active: 364 raise SSHException('Channel is not open') 365 if auth_protocol is None: 366 auth_protocol = 'MIT-MAGIC-COOKIE-1' 367 if auth_cookie is None: 368 auth_cookie = binascii.hexlify(self.transport.randpool.get_bytes(16)) 369 370 m = Message() 371 m.add_byte(chr(MSG_CHANNEL_REQUEST)) 372 m.add_int(self.remote_chanid) 373 m.add_string('x11-req') 374 m.add_boolean(True) 375 m.add_boolean(single_connection) 376 m.add_string(auth_protocol) 377 m.add_string(auth_cookie) 378 m.add_int(screen_number) 379 self.event.clear() 380 self.transport._send_user_message(m) 381 self._wait_for_event() 382 self.transport._set_x11_handler(handler) 383 return auth_cookie
384
385 - def get_transport(self):
386 """ 387 Return the L{Transport} associated with this channel. 388 389 @return: the L{Transport} that was used to create this channel. 390 @rtype: L{Transport} 391 """ 392 return self.transport
393
394 - def set_name(self, name):
395 """ 396 Set a name for this channel. Currently it's only used to set the name 397 of the channel in logfile entries. The name can be fetched with the 398 L{get_name} method. 399 400 @param name: new channel name 401 @type name: str 402 """ 403 self._name = name
404
405 - def get_name(self):
406 """ 407 Get the name of this channel that was previously set by L{set_name}. 408 409 @return: the name of this channel. 410 @rtype: str 411 """ 412 return self._name
413
414 - def get_id(self):
415 """ 416 Return the ID # for this channel. The channel ID is unique across 417 a L{Transport} and usually a small number. It's also the number 418 passed to L{ServerInterface.check_channel_request} when determining 419 whether to accept a channel request in server mode. 420 421 @return: the ID of this channel. 422 @rtype: int 423 """ 424 return self.chanid
425
426 - def set_combine_stderr(self, combine):
427 """ 428 Set whether stderr should be combined into stdout on this channel. 429 The default is C{False}, but in some cases it may be convenient to 430 have both streams combined. 431 432 If this is C{False}, and L{exec_command} is called (or C{invoke_shell} 433 with no pty), output to stderr will not show up through the L{recv} 434 and L{recv_ready} calls. You will have to use L{recv_stderr} and 435 L{recv_stderr_ready} to get stderr output. 436 437 If this is C{True}, data will never show up via L{recv_stderr} or 438 L{recv_stderr_ready}. 439 440 @param combine: C{True} if stderr output should be combined into 441 stdout on this channel. 442 @type combine: bool 443 @return: previous setting. 444 @rtype: bool 445 446 @since: 1.1 447 """ 448 data = '' 449 self.lock.acquire() 450 try: 451 old = self.combine_stderr 452 self.combine_stderr = combine 453 if combine and not old: 454 # copy old stderr buffer into primary buffer 455 data = self.in_stderr_buffer.empty() 456 finally: 457 self.lock.release() 458 if len(data) > 0: 459 self._feed(data) 460 return old
461 462 463 ### socket API 464 465
466 - def settimeout(self, timeout):
467 """ 468 Set a timeout on blocking read/write operations. The C{timeout} 469 argument can be a nonnegative float expressing seconds, or C{None}. If 470 a float is given, subsequent channel read/write operations will raise 471 a timeout exception if the timeout period value has elapsed before the 472 operation has completed. Setting a timeout of C{None} disables 473 timeouts on socket operations. 474 475 C{chan.settimeout(0.0)} is equivalent to C{chan.setblocking(0)}; 476 C{chan.settimeout(None)} is equivalent to C{chan.setblocking(1)}. 477 478 @param timeout: seconds to wait for a pending read/write operation 479 before raising C{socket.timeout}, or C{None} for no timeout. 480 @type timeout: float 481 """ 482 self.timeout = timeout
483
484 - def gettimeout(self):
485 """ 486 Returns the timeout in seconds (as a float) associated with socket 487 operations, or C{None} if no timeout is set. This reflects the last 488 call to L{setblocking} or L{settimeout}. 489 490 @return: timeout in seconds, or C{None}. 491 @rtype: float 492 """ 493 return self.timeout
494
495 - def setblocking(self, blocking):
496 """ 497 Set blocking or non-blocking mode of the channel: if C{blocking} is 0, 498 the channel is set to non-blocking mode; otherwise it's set to blocking 499 mode. Initially all channels are in blocking mode. 500 501 In non-blocking mode, if a L{recv} call doesn't find any data, or if a 502 L{send} call can't immediately dispose of the data, an error exception 503 is raised. In blocking mode, the calls block until they can proceed. 504 505 C{chan.setblocking(0)} is equivalent to C{chan.settimeout(0)}; 506 C{chan.setblocking(1)} is equivalent to C{chan.settimeout(None)}. 507 508 @param blocking: 0 to set non-blocking mode; non-0 to set blocking 509 mode. 510 @type blocking: int 511 """ 512 if blocking: 513 self.settimeout(None) 514 else: 515 self.settimeout(0.0)
516
517 - def getpeername(self):
518 """ 519 Return the address of the remote side of this Channel, if possible. 520 This is just a wrapper around C{'getpeername'} on the Transport, used 521 to provide enough of a socket-like interface to allow asyncore to work. 522 (asyncore likes to call C{'getpeername'}.) 523 524 @return: the address if the remote host, if known 525 @rtype: tuple(str, int) 526 """ 527 return self.transport.getpeername()
528
529 - def close(self):
530 """ 531 Close the channel. All future read/write operations on the channel 532 will fail. The remote end will receive no more data (after queued data 533 is flushed). Channels are automatically closed when their L{Transport} 534 is closed or when they are garbage collected. 535 """ 536 self.lock.acquire() 537 try: 538 # only close the pipe when the user explicitly closes the channel. 539 # otherwise they will get unpleasant surprises. (and do it before 540 # checking self.closed, since the remote host may have already 541 # closed the connection.) 542 if self._pipe is not None: 543 self._pipe.close() 544 self._pipe = None 545 546 if not self.active or self.closed: 547 return 548 msgs = self._close_internal() 549 finally: 550 self.lock.release() 551 for m in msgs: 552 if m is not None: 553 self.transport._send_user_message(m)
554
555 - def recv_ready(self):
556 """ 557 Returns true if data is buffered and ready to be read from this 558 channel. A C{False} result does not mean that the channel has closed; 559 it means you may need to wait before more data arrives. 560 561 @return: C{True} if a L{recv} call on this channel would immediately 562 return at least one byte; C{False} otherwise. 563 @rtype: boolean 564 """ 565 return self.in_buffer.read_ready()
566
567 - def recv(self, nbytes):
568 """ 569 Receive data from the channel. The return value is a string 570 representing the data received. The maximum amount of data to be 571 received at once is specified by C{nbytes}. If a string of length zero 572 is returned, the channel stream has closed. 573 574 @param nbytes: maximum number of bytes to read. 575 @type nbytes: int 576 @return: data. 577 @rtype: str 578 579 @raise socket.timeout: if no data is ready before the timeout set by 580 L{settimeout}. 581 """ 582 try: 583 out = self.in_buffer.read(nbytes, self.timeout) 584 except PipeTimeout, e: 585 raise socket.timeout() 586 587 ack = self._check_add_window(len(out)) 588 # no need to hold the channel lock when sending this 589 if ack > 0: 590 m = Message() 591 m.add_byte(chr(MSG_CHANNEL_WINDOW_ADJUST)) 592 m.add_int(self.remote_chanid) 593 m.add_int(ack) 594 self.transport._send_user_message(m) 595 596 return out
597
598 - def recv_stderr_ready(self):
599 """ 600 Returns true if data is buffered and ready to be read from this 601 channel's stderr stream. Only channels using L{exec_command} or 602 L{invoke_shell} without a pty will ever have data on the stderr 603 stream. 604 605 @return: C{True} if a L{recv_stderr} call on this channel would 606 immediately return at least one byte; C{False} otherwise. 607 @rtype: boolean 608 609 @since: 1.1 610 """ 611 return self.in_stderr_buffer.read_ready()
612
613 - def recv_stderr(self, nbytes):
614 """ 615 Receive data from the channel's stderr stream. Only channels using 616 L{exec_command} or L{invoke_shell} without a pty will ever have data 617 on the stderr stream. The return value is a string representing the 618 data received. The maximum amount of data to be received at once is 619 specified by C{nbytes}. If a string of length zero is returned, the 620 channel stream has closed. 621 622 @param nbytes: maximum number of bytes to read. 623 @type nbytes: int 624 @return: data. 625 @rtype: str 626 627 @raise socket.timeout: if no data is ready before the timeout set by 628 L{settimeout}. 629 630 @since: 1.1 631 """ 632 try: 633 out = self.in_stderr_buffer.read(nbytes, self.timeout) 634 except PipeTimeout, e: 635 raise socket.timeout() 636 637 ack = self._check_add_window(len(out)) 638 # no need to hold the channel lock when sending this 639 if ack > 0: 640 m = Message() 641 m.add_byte(chr(MSG_CHANNEL_WINDOW_ADJUST)) 642 m.add_int(self.remote_chanid) 643 m.add_int(ack) 644 self.transport._send_user_message(m) 645 646 return out
647
648 - def send_ready(self):
649 """ 650 Returns true if data can be written to this channel without blocking. 651 This means the channel is either closed (so any write attempt would 652 return immediately) or there is at least one byte of space in the 653 outbound buffer. If there is at least one byte of space in the 654 outbound buffer, a L{send} call will succeed immediately and return 655 the number of bytes actually written. 656 657 @return: C{True} if a L{send} call on this channel would immediately 658 succeed or fail 659 @rtype: boolean 660 """ 661 self.lock.acquire() 662 try: 663 if self.closed or self.eof_sent: 664 return True 665 return self.out_window_size > 0 666 finally: 667 self.lock.release()
668
669 - def send(self, s):
670 """ 671 Send data to the channel. Returns the number of bytes sent, or 0 if 672 the channel stream is closed. Applications are responsible for 673 checking that all data has been sent: if only some of the data was 674 transmitted, the application needs to attempt delivery of the remaining 675 data. 676 677 @param s: data to send 678 @type s: str 679 @return: number of bytes actually sent 680 @rtype: int 681 682 @raise socket.timeout: if no data could be sent before the timeout set 683 by L{settimeout}. 684 """ 685 size = len(s) 686 self.lock.acquire() 687 try: 688 size = self._wait_for_send_window(size) 689 if size == 0: 690 # eof or similar 691 return 0 692 m = Message() 693 m.add_byte(chr(MSG_CHANNEL_DATA)) 694 m.add_int(self.remote_chanid) 695 m.add_string(s[:size]) 696 self.transport._send_user_message(m) 697 finally: 698 self.lock.release() 699 return size
700
701 - def send_stderr(self, s):
702 """ 703 Send data to the channel on the "stderr" stream. This is normally 704 only used by servers to send output from shell commands -- clients 705 won't use this. Returns the number of bytes sent, or 0 if the channel 706 stream is closed. Applications are responsible for checking that all 707 data has been sent: if only some of the data was transmitted, the 708 application needs to attempt delivery of the remaining data. 709 710 @param s: data to send. 711 @type s: str 712 @return: number of bytes actually sent. 713 @rtype: int 714 715 @raise socket.timeout: if no data could be sent before the timeout set 716 by L{settimeout}. 717 718 @since: 1.1 719 """ 720 size = len(s) 721 self.lock.acquire() 722 try: 723 size = self._wait_for_send_window(size) 724 if size == 0: 725 # eof or similar 726 return 0 727 m = Message() 728 m.add_byte(chr(MSG_CHANNEL_EXTENDED_DATA)) 729 m.add_int(self.remote_chanid) 730 m.add_int(1) 731 m.add_string(s[:size]) 732 self.transport._send_user_message(m) 733 finally: 734 self.lock.release() 735 return size
736
737 - def sendall(self, s):
738 """ 739 Send data to the channel, without allowing partial results. Unlike 740 L{send}, this method continues to send data from the given string until 741 either all data has been sent or an error occurs. Nothing is returned. 742 743 @param s: data to send. 744 @type s: str 745 746 @raise socket.timeout: if sending stalled for longer than the timeout 747 set by L{settimeout}. 748 @raise socket.error: if an error occured before the entire string was 749 sent. 750 751 @note: If the channel is closed while only part of the data hase been 752 sent, there is no way to determine how much data (if any) was sent. 753 This is irritating, but identically follows python's API. 754 """ 755 while s: 756 if self.closed: 757 # this doesn't seem useful, but it is the documented behavior of Socket 758 raise socket.error('Socket is closed') 759 sent = self.send(s) 760 s = s[sent:] 761 return None
762
763 - def sendall_stderr(self, s):
764 """ 765 Send data to the channel's "stderr" stream, without allowing partial 766 results. Unlike L{send_stderr}, this method continues to send data 767 from the given string until all data has been sent or an error occurs. 768 Nothing is returned. 769 770 @param s: data to send to the client as "stderr" output. 771 @type s: str 772 773 @raise socket.timeout: if sending stalled for longer than the timeout 774 set by L{settimeout}. 775 @raise socket.error: if an error occured before the entire string was 776 sent. 777 778 @since: 1.1 779 """ 780 while s: 781 if self.closed: 782 raise socket.error('Socket is closed') 783 sent = self.send_stderr(s) 784 s = s[sent:] 785 return None
786
787 - def makefile(self, *params):
788 """ 789 Return a file-like object associated with this channel. The optional 790 C{mode} and C{bufsize} arguments are interpreted the same way as by 791 the built-in C{file()} function in python. 792 793 @return: object which can be used for python file I/O. 794 @rtype: L{ChannelFile} 795 """ 796 return ChannelFile(*([self] + list(params)))
797
798 - def makefile_stderr(self, *params):
799 """ 800 Return a file-like object associated with this channel's stderr 801 stream. Only channels using L{exec_command} or L{invoke_shell} 802 without a pty will ever have data on the stderr stream. 803 804 The optional C{mode} and C{bufsize} arguments are interpreted the 805 same way as by the built-in C{file()} function in python. For a 806 client, it only makes sense to open this file for reading. For a 807 server, it only makes sense to open this file for writing. 808 809 @return: object which can be used for python file I/O. 810 @rtype: L{ChannelFile} 811 812 @since: 1.1 813 """ 814 return ChannelStderrFile(*([self] + list(params)))
815
816 - def fileno(self):
817 """ 818 Returns an OS-level file descriptor which can be used for polling, but 819 but I{not} for reading or writing. This is primaily to allow python's 820 C{select} module to work. 821 822 The first time C{fileno} is called on a channel, a pipe is created to 823 simulate real OS-level file descriptor (FD) behavior. Because of this, 824 two OS-level FDs are created, which will use up FDs faster than normal. 825 (You won't notice this effect unless you have hundreds of channels 826 open at the same time.) 827 828 @return: an OS-level file descriptor 829 @rtype: int 830 831 @warning: This method causes channel reads to be slightly less 832 efficient. 833 """ 834 self.lock.acquire() 835 try: 836 if self._pipe is not None: 837 return self._pipe.fileno() 838 # create the pipe and feed in any existing data 839 self._pipe = pipe.make_pipe() 840 p1, p2 = pipe.make_or_pipe(self._pipe) 841 self.in_buffer.set_event(p1) 842 self.in_stderr_buffer.set_event(p2) 843 return self._pipe.fileno() 844 finally: 845 self.lock.release()
846
847 - def shutdown(self, how):
848 """ 849 Shut down one or both halves of the connection. If C{how} is 0, 850 further receives are disallowed. If C{how} is 1, further sends 851 are disallowed. If C{how} is 2, further sends and receives are 852 disallowed. This closes the stream in one or both directions. 853 854 @param how: 0 (stop receiving), 1 (stop sending), or 2 (stop 855 receiving and sending). 856 @type how: int 857 """ 858 if (how == 0) or (how == 2): 859 # feign "read" shutdown 860 self.eof_received = 1 861 if (how == 1) or (how == 2): 862 self.lock.acquire() 863 try: 864 m = self._send_eof() 865 finally: 866 self.lock.release() 867 if m is not None: 868 self.transport._send_user_message(m)
869
870 - def shutdown_read(self):
871 """ 872 Shutdown the receiving side of this socket, closing the stream in 873 the incoming direction. After this call, future reads on this 874 channel will fail instantly. This is a convenience method, equivalent 875 to C{shutdown(0)}, for people who don't make it a habit to 876 memorize unix constants from the 1970s. 877 878 @since: 1.2 879 """ 880 self.shutdown(0)
881
882 - def shutdown_write(self):
883 """ 884 Shutdown the sending side of this socket, closing the stream in 885 the outgoing direction. After this call, future writes on this 886 channel will fail instantly. This is a convenience method, equivalent 887 to C{shutdown(1)}, for people who don't make it a habit to 888 memorize unix constants from the 1970s. 889 890 @since: 1.2 891 """ 892 self.shutdown(1)
893 894 895 ### calls from Transport 896 897
898 - def _set_transport(self, transport):
901
902 - def _set_window(self, window_size, max_packet_size):
903 self.in_window_size = window_size 904 self.in_max_packet_size = max_packet_size 905 # threshold of bytes we receive before we bother to send a window update 906 self.in_window_threshold = window_size // 10 907 self.in_window_sofar = 0 908 self._log(DEBUG, 'Max packet in: %d bytes' % max_packet_size)
909
910 - def _set_remote_channel(self, chanid, window_size, max_packet_size):
911 self.remote_chanid = chanid 912 self.out_window_size = window_size 913 self.out_max_packet_size = max(max_packet_size, MIN_PACKET_SIZE) 914 self.active = 1 915 self._log(DEBUG, 'Max packet out: %d bytes' % max_packet_size)
916
917 - def _request_success(self, m):
918 self._log(DEBUG, 'Sesch channel %d request ok' % self.chanid) 919 self.event.set() 920 return
921
922 - def _request_failed(self, m):
923 self.lock.acquire() 924 try: 925 msgs = self._close_internal() 926 finally: 927 self.lock.release() 928 for m in msgs: 929 if m is not None: 930 self.transport._send_user_message(m)
931
932 - def _feed(self, m):
933 if type(m) is str: 934 # passed from _feed_extended 935 s = m 936 else: 937 s = m.get_string() 938 self.in_buffer.feed(s)
939
940 - def _feed_extended(self, m):
941 code = m.get_int() 942 s = m.get_string() 943 if code != 1: 944 self._log(ERROR, 'unknown extended_data type %d; discarding' % code) 945 return 946 if self.combine_stderr: 947 self._feed(s) 948 else: 949 self.in_stderr_buffer.feed(s)
950
951 - def _window_adjust(self, m):
952 nbytes = m.get_int() 953 self.lock.acquire() 954 try: 955 if self.ultra_debug: 956 self._log(DEBUG, 'window up %d' % nbytes) 957 self.out_window_size += nbytes 958 self.out_buffer_cv.notifyAll() 959 finally: 960 self.lock.release()
961
962 - def _handle_request(self, m):
963 key = m.get_string() 964 want_reply = m.get_boolean() 965 server = self.transport.server_object 966 ok = False 967 if key == 'exit-status': 968 self.exit_status = m.get_int() 969 self.status_event.set() 970 ok = True 971 elif key == 'xon-xoff': 972 # ignore 973 ok = True 974 elif key == 'pty-req': 975 term = m.get_string() 976 width = m.get_int() 977 height = m.get_int() 978 pixelwidth = m.get_int() 979 pixelheight = m.get_int() 980 modes = m.get_string() 981 if server is None: 982 ok = False 983 else: 984 ok = server.check_channel_pty_request(self, term, width, height, pixelwidth, 985 pixelheight, modes) 986 elif key == 'shell': 987 if server is None: 988 ok = False 989 else: 990 ok = server.check_channel_shell_request(self) 991 elif key == 'exec': 992 cmd = m.get_string() 993 if server is None: 994 ok = False 995 else: 996 ok = server.check_channel_exec_request(self, cmd) 997 elif key == 'subsystem': 998 name = m.get_string() 999 if server is None: 1000 ok = False 1001 else: 1002 ok = server.check_channel_subsystem_request(self, name) 1003 elif key == 'window-change': 1004 width = m.get_int() 1005 height = m.get_int() 1006 pixelwidth = m.get_int() 1007 pixelheight = m.get_int() 1008 if server is None: 1009 ok = False 1010 else: 1011 ok = server.check_channel_window_change_request(self, width, height, pixelwidth, 1012 pixelheight) 1013 elif key == 'x11-req': 1014 single_connection = m.get_boolean() 1015 auth_proto = m.get_string() 1016 auth_cookie = m.get_string() 1017 screen_number = m.get_int() 1018 if server is None: 1019 ok = False 1020 else: 1021 ok = server.check_channel_x11_request(self, single_connection, 1022 auth_proto, auth_cookie, screen_number) 1023 else: 1024 self._log(DEBUG, 'Unhandled channel request "%s"' % key) 1025 ok = False 1026 if want_reply: 1027 m = Message() 1028 if ok: 1029 m.add_byte(chr(MSG_CHANNEL_SUCCESS)) 1030 else: 1031 m.add_byte(chr(MSG_CHANNEL_FAILURE)) 1032 m.add_int(self.remote_chanid) 1033 self.transport._send_user_message(m)
1034
1035 - def _handle_eof(self, m):
1036 self.lock.acquire() 1037 try: 1038 if not self.eof_received: 1039 self.eof_received = True 1040 self.in_buffer.close() 1041 self.in_stderr_buffer.close() 1042 if self._pipe is not None: 1043 self._pipe.set_forever() 1044 finally: 1045 self.lock.release() 1046 self._log(DEBUG, 'EOF received (%s)', self._name)
1047
1048 - def _handle_close(self, m):
1049 self.lock.acquire() 1050 try: 1051 msgs = self._close_internal() 1052 self.transport._unlink_channel(self.chanid) 1053 finally: 1054 self.lock.release() 1055 for m in msgs: 1056 if m is not None: 1057 self.transport._send_user_message(m)
1058 1059 1060 ### internals... 1061 1062
1063 - def _log(self, level, msg, *args):
1064 self.logger.log(level, "[chan " + self._name + "] " + msg, *args)
1065
1066 - def _wait_for_event(self):
1067 while True: 1068 self.event.wait(0.1) 1069 if self.event.isSet(): 1070 return 1071 if self.closed: 1072 e = self.transport.get_exception() 1073 if e is None: 1074 e = SSHException('Channel closed.') 1075 raise e 1076 return
1077
1078 - def _set_closed(self):
1079 # you are holding the lock. 1080 self.closed = True 1081 self.in_buffer.close() 1082 self.in_stderr_buffer.close() 1083 self.out_buffer_cv.notifyAll() 1084 if self._pipe is not None: 1085 self._pipe.set_forever()
1086
1087 - def _send_eof(self):
1088 # you are holding the lock. 1089 if self.eof_sent: 1090 return None 1091 m = Message() 1092 m.add_byte(chr(MSG_CHANNEL_EOF)) 1093 m.add_int(self.remote_chanid) 1094 self.eof_sent = True 1095 self._log(DEBUG, 'EOF sent (%s)', self._name) 1096 return m
1097
1098 - def _close_internal(self):
1099 # you are holding the lock. 1100 if not self.active or self.closed: 1101 return None, None 1102 m1 = self._send_eof() 1103 m2 = Message() 1104 m2.add_byte(chr(MSG_CHANNEL_CLOSE)) 1105 m2.add_int(self.remote_chanid) 1106 self._set_closed() 1107 # can't unlink from the Transport yet -- the remote side may still 1108 # try to send meta-data (exit-status, etc) 1109 return m1, m2
1110 1121
1122 - def _check_add_window(self, n):
1123 self.lock.acquire() 1124 try: 1125 if self.closed or self.eof_received or not self.active: 1126 return 0 1127 if self.ultra_debug: 1128 self._log(DEBUG, 'addwindow %d' % n) 1129 self.in_window_sofar += n 1130 if self.in_window_sofar <= self.in_window_threshold: 1131 return 0 1132 if self.ultra_debug: 1133 self._log(DEBUG, 'addwindow send %d' % self.in_window_sofar) 1134 out = self.in_window_sofar 1135 self.in_window_sofar = 0 1136 return out 1137 finally: 1138 self.lock.release()
1139
1140 - def _wait_for_send_window(self, size):
1141 """ 1142 (You are already holding the lock.) 1143 Wait for the send window to open up, and allocate up to C{size} bytes 1144 for transmission. If no space opens up before the timeout, a timeout 1145 exception is raised. Returns the number of bytes available to send 1146 (may be less than requested). 1147 """ 1148 # you are already holding the lock 1149 if self.closed or self.eof_sent: 1150 return 0 1151 if self.out_window_size == 0: 1152 # should we block? 1153 if self.timeout == 0.0: 1154 raise socket.timeout() 1155 # loop here in case we get woken up but a different thread has filled the buffer 1156 timeout = self.timeout 1157 while self.out_window_size == 0: 1158 if self.closed or self.eof_sent: 1159 return 0 1160 then = time.time() 1161 self.out_buffer_cv.wait(timeout) 1162 if timeout != None: 1163 timeout -= time.time() - then 1164 if timeout <= 0.0: 1165 raise socket.timeout() 1166 # we have some window to squeeze into 1167 if self.closed or self.eof_sent: 1168 return 0 1169 if self.out_window_size < size: 1170 size = self.out_window_size 1171 if self.out_max_packet_size - 64 < size: 1172 size = self.out_max_packet_size - 64 1173 self.out_window_size -= size 1174 if self.ultra_debug: 1175 self._log(DEBUG, 'window down to %d' % self.out_window_size) 1176 return size
1177 1178
1179 -class ChannelFile (BufferedFile):
1180 """ 1181 A file-like wrapper around L{Channel}. A ChannelFile is created by calling 1182 L{Channel.makefile}. 1183 1184 @bug: To correctly emulate the file object created from a socket's 1185 C{makefile} method, a L{Channel} and its C{ChannelFile} should be able 1186 to be closed or garbage-collected independently. Currently, closing 1187 the C{ChannelFile} does nothing but flush the buffer. 1188 """ 1189
1190 - def __init__(self, channel, mode = 'r', bufsize = -1):
1191 self.channel = channel 1192 BufferedFile.__init__(self) 1193 self._set_mode(mode, bufsize)
1194
1195 - def __repr__(self):
1196 """ 1197 Returns a string representation of this object, for debugging. 1198 1199 @rtype: str 1200 """ 1201 return '<paramiko.ChannelFile from ' + repr(self.channel) + '>'
1202
1203 - def _read(self, size):
1204 return self.channel.recv(size)
1205
1206 - def _write(self, data):
1207 self.channel.sendall(data) 1208 return len(data)
1209 1210
1211 -class ChannelStderrFile (ChannelFile):
1212 - def __init__(self, channel, mode = 'r', bufsize = -1):
1213 ChannelFile.__init__(self, channel, mode, bufsize)
1214
1215 - def _read(self, size):
1216 return self.channel.recv_stderr(size)
1217
1218 - def _write(self, data):
1219 self.channel.sendall_stderr(data) 1220 return len(data)
1221 1222 1223 # vim: set shiftwidth=4 expandtab : 1224