src/nepi/util/sshfuncs.py

   1 #
   2 #    NEPI, a framework to manage network experiments
   3 #    Copyright (C) 2013 INRIA
   4 #
   5 #    This program is free software: you can redistribute it and/or modify
   6 #    it under the terms of the GNU General Public License as published by
   7 #    the Free Software Foundation, either version 3 of the License, or
   8 #    (at your option) any later version.
   9 #
  10 #    This program is distributed in the hope that it will be useful,
  11 #    but WITHOUT ANY WARRANTY; without even the implied warranty of
  12 #    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13 #    GNU General Public License for more details.
  14 #
  15 #    You should have received a copy of the GNU General Public License
  16 #    along with this program.  If not, see <http://www.gnu.org/licenses/>.
  17 #
  18 # Author: Alina Quereilhac <alina.quereilhac@inria.fr>
  19 #         Claudio Freire <claudio-daniel.freire@inria.fr>
  20
  21 ## TODO: This code needs reviewing !!!
  22
  23 import base64
  24 import errno
  25 import hashlib
  26 import logging
  27 import os
  28 import os.path
  29 import re
  30 import select
  31 import signal
  32 import socket
  33 import subprocess
  34 import threading
  35 import time
  36 import tempfile
  37
  38 logger = logging.getLogger("sshfuncs")
  39
  40 def log(msg, level, out = None, err = None):
  41     if out:
  42         msg += " - OUT: %s " % out
  43
  44     if err:
  45         msg += " - ERROR: %s " % err
  46
  47     logger.log(level, msg)
  48
  49 if hasattr(os, "devnull"):
  50     DEV_NULL = os.devnull
  51 else:
  52     DEV_NULL = "/dev/null"
  53
  54 SHELL_SAFE = re.compile('^[-a-zA-Z0-9_=+:.,/]*$')
  55
  56 class STDOUT:
  57     """
  58     Special value that when given to rspawn in stderr causes stderr to
  59     redirect to whatever stdout was redirected to.
  60     """
  61
  62 class ProcStatus:
  63     """
  64     Codes for status of remote spawned process
  65     """
  66     # Process is still running
  67     RUNNING = 1
  68
  69     # Process is finished
  70     FINISHED = 2
  71
  72     # Process hasn't started running yet (this should be very rare)
  73     NOT_STARTED = 3
  74
  75 hostbyname_cache = dict()
  76 hostbyname_cache_lock = threading.Lock()
  77
  78 def gethostbyname(host):
  79     global hostbyname_cache
  80     global hostbyname_cache_lock
  81
  82     hostbyname = hostbyname_cache.get(host)
  83     if not hostbyname:
  84         with hostbyname_cache_lock:
  85             hostbyname = socket.gethostbyname(host)
  86             hostbyname_cache[host] = hostbyname
  87
  88             msg = " Added hostbyname %s - %s " % (host, hostbyname)
  89             log(msg, logging.DEBUG)
  90
  91     return hostbyname
  92
  93 OPENSSH_HAS_PERSIST = None
  94
  95 def openssh_has_persist():
  96     """ The ssh_config options ControlMaster and ControlPersist allow to
  97     reuse a same network connection for multiple ssh sessions. In this
  98     way limitations on number of open ssh connections can be bypassed.
  99     However, older versions of openSSH do not support this feature.
 100     This function is used to determine if ssh connection persist features
 101     can be used.
 102     """
 103     global OPENSSH_HAS_PERSIST
 104     if OPENSSH_HAS_PERSIST is None:
 105         proc = subprocess.Popen(["ssh","-v"],
 106             stdout = subprocess.PIPE,
 107             stderr = subprocess.STDOUT,
 108             stdin = open("/dev/null","r") )
 109         out,err = proc.communicate()
 110         proc.wait()
 111
 112         vre = re.compile(r'OpenSSH_(?:[6-9]|5[.][8-9]|5[.][1-9][0-9]|[1-9][0-9]).*', re.I)
 113         OPENSSH_HAS_PERSIST = bool(vre.match(out))
 114     return OPENSSH_HAS_PERSIST
 115
 116 def make_server_key_args(server_key, host, port):
 117     """ Returns a reference to a temporary known_hosts file, to which
 118     the server key has been added.
 119
 120     Make sure to hold onto the temp file reference until the process is
 121     done with it
 122
 123     :param server_key: the server public key
 124     :type server_key: str
 125
 126     :param host: the hostname
 127     :type host: str
 128
 129     :param port: the ssh port
 130     :type port: str
 131
 132     """
 133     if port is not None:
 134         host = '%s:%s' % (host, str(port))
 135
 136     # Create a temporary server key file
 137     tmp_known_hosts = tempfile.NamedTemporaryFile()
 138
 139     hostbyname = gethostbyname(host)
 140
 141     # Add the intended host key
 142     tmp_known_hosts.write('%s,%s %s\n' % (host, hostbyname, server_key))
 143
 144     # If we're not in strict mode, add user-configured keys
 145     if os.environ.get('NEPI_STRICT_AUTH_MODE',"").lower() not in ('1','true','on'):
 146         user_hosts_path = '%s/.ssh/known_hosts' % (os.environ.get('HOME',""),)
 147         if os.access(user_hosts_path, os.R_OK):
 148             f = open(user_hosts_path, "r")
 149             tmp_known_hosts.write(f.read())
 150             f.close()
 151
 152     tmp_known_hosts.flush()
 153
 154     return tmp_known_hosts
 155
 156 def make_control_path(agent, forward_x11):
 157     ctrl_path = "/tmp/nepi_ssh"
 158
 159     if agent:
 160         ctrl_path +="_a"
 161
 162     if forward_x11:
 163         ctrl_path +="_x"
 164
 165     ctrl_path += "-%r@%h:%p"
 166
 167     return ctrl_path
 168
 169 def shell_escape(s):
 170     """ Escapes strings so that they are safe to use as command-line
 171     arguments """
 172     if SHELL_SAFE.match(s):
 173         # safe string - no escaping needed
 174         return s
 175     else:
 176         # unsafe string - escape
 177         def escp(c):
 178             if (32 <= ord(c) < 127 or c in ('\r','\n','\t')) and c not in ("'",'"'):
 179                 return c
 180             else:
 181                 return "'$'\\x%02x''" % (ord(c),)
 182         s = ''.join(map(escp,s))
 183         return "'%s'" % (s,)
 184
 185 def eintr_retry(func):
 186     """Retries a function invocation when a EINTR occurs"""
 187     import functools
 188     @functools.wraps(func)
 189     def rv(*p, **kw):
 190         retry = kw.pop("_retry", False)
 191         for i in xrange(0 if retry else 4):
 192             try:
 193                 return func(*p, **kw)
 194             except (select.error, socket.error), args:
 195                 if args[0] == errno.EINTR:
 196                     continue
 197                 else:
 198                     raise
 199             except OSError, e:
 200                 if e.errno == errno.EINTR:
 201                     continue
 202                 else:
 203                     raise
 204         else:
 205             return func(*p, **kw)
 206     return rv
 207
 208 def rexec(command, host, user,
 209         port = None,
 210         gwuser = None,
 211         gw = None,
 212         agent = True,
 213         sudo = False,
 214         identity = None,
 215         server_key = None,
 216         env = None,
 217         tty = False,
 218         connect_timeout = 30,
 219         retry = 3,
 220         persistent = True,
 221         forward_x11 = False,
 222         blocking = True,
 223         strict_host_checking = True):
 224     """
 225     Executes a remote command, returns ((stdout,stderr),process)
 226     """
 227
 228     tmp_known_hosts = None
 229     if not gw:
 230         hostip = gethostbyname(host)
 231     else: hostip = None
 232
 233     args = ['ssh', '-C',
 234             # Don't bother with localhost. Makes test easier
 235             '-o', 'NoHostAuthenticationForLocalhost=yes',
 236             '-o', 'ConnectTimeout=%d' % (int(connect_timeout),),
 237             '-o', 'ConnectionAttempts=3',
 238             '-o', 'ServerAliveInterval=30',
 239             '-o', 'TCPKeepAlive=yes',
 240             '-o', 'Batchmode=yes',
 241             '-l', user, hostip or host]
 242
 243     if persistent and openssh_has_persist():
 244         args.extend([
 245             '-o', 'ControlMaster=auto',
 246             '-o', 'ControlPath=%s' % (make_control_path(agent, forward_x11),),
 247             '-o', 'ControlPersist=60' ])
 248
 249     if not strict_host_checking:
 250         # Do not check for Host key. Unsafe.
 251         args.extend(['-o', 'StrictHostKeyChecking=no'])
 252
 253     if gw:
 254         proxycommand = _proxy_command(gw, gwuser, identity)
 255         args.extend(['-o', proxycommand])
 256
 257     if agent:
 258         args.append('-A')
 259
 260     if port:
 261         args.append('-p%d' % port)
 262
 263     if identity:
 264         identity = os.path.expanduser(identity)
 265         args.extend(('-i', identity))
 266
 267     if tty:
 268         args.append('-t')
 269         args.append('-t')
 270
 271     if forward_x11:
 272         args.append('-X')
 273
 274     if server_key:
 275         # Create a temporary server key file
 276         tmp_known_hosts = make_server_key_args(server_key, host, port)
 277         args.extend(['-o', 'UserKnownHostsFile=%s' % (tmp_known_hosts.name,)])
 278
 279     if sudo:
 280         command = "sudo " + command
 281
 282     args.append(command)
 283
 284     log_msg = " rexec - host %s - command %s " % (str(host), " ".join(map(str, args)))
 285
 286     stdout = stderr = stdin = subprocess.PIPE
 287     if forward_x11:
 288         stdout = stderr = stdin = None
 289
 290     return _retry_rexec(args, log_msg,
 291             stderr = stderr,
 292             stdin = stdin,
 293             stdout = stdout,
 294             env = env,
 295             retry = retry,
 296             tmp_known_hosts = tmp_known_hosts,
 297             blocking = blocking)
 298
 299 def rcopy(source, dest,
 300         port = None,
 301         gwuser = None,
 302         gw = None,
 303         recursive = False,
 304         identity = None,
 305         server_key = None,
 306         retry = 3,
 307         strict_host_checking = True):
 308     """
 309     Copies from/to remote sites.
 310
 311     Source and destination should have the user and host encoded
 312     as per scp specs.
 313
 314     Source can be a list of files to copy to a single destination,
 315     (in which case it is advised that the destination be a folder),
 316     or a single file in a string.
 317     """
 318
 319     # Parse destination as <user>@<server>:<path>
 320     if isinstance(dest, str) and ':' in dest:
 321         remspec, path = dest.split(':',1)
 322     elif isinstance(source, str) and ':' in source:
 323         remspec, path = source.split(':',1)
 324     else:
 325         raise ValueError, "Both endpoints cannot be local"
 326     user,host = remspec.rsplit('@',1)
 327
 328     # plain scp
 329     tmp_known_hosts = None
 330
 331     args = ['scp', '-q', '-p', '-C',
 332             # Speed up transfer using blowfish cypher specification which is
 333             # faster than the default one (3des)
 334             '-c', 'blowfish',
 335             # Don't bother with localhost. Makes test easier
 336             '-o', 'NoHostAuthenticationForLocalhost=yes',
 337             '-o', 'ConnectTimeout=60',
 338             '-o', 'ConnectionAttempts=3',
 339             '-o', 'ServerAliveInterval=30',
 340             '-o', 'TCPKeepAlive=yes' ]
 341
 342     if port:
 343         args.append('-P%d' % port)
 344
 345     if gw:
 346         proxycommand = _proxy_command(gw, gwuser, identity)
 347         args.extend(['-o', proxycommand])
 348
 349     if recursive:
 350         args.append('-r')
 351
 352     if identity:
 353         identity = os.path.expanduser(identity)
 354         args.extend(('-i', identity))
 355
 356     if server_key:
 357         # Create a temporary server key file
 358         tmp_known_hosts = make_server_key_args(server_key, host, port)
 359         args.extend(['-o', 'UserKnownHostsFile=%s' % (tmp_known_hosts.name,)])
 360
 361     if not strict_host_checking:
 362         # Do not check for Host key. Unsafe.
 363         args.extend(['-o', 'StrictHostKeyChecking=no'])
 364
 365     if isinstance(source, list):
 366         args.extend(source)
 367     else:
 368         if openssh_has_persist():
 369             args.extend([
 370                 '-o', 'ControlMaster=auto',
 371                 '-o', 'ControlPath=%s' % (make_control_path(False, False),)
 372                 ])
 373         args.append(source)
 374
 375     if isinstance(dest, list):
 376         args.extend(dest)
 377     else:
 378         args.append(dest)
 379
 380     log_msg = " rcopy - host %s - command %s " % (str(host), " ".join(map(str, args)))
 381
 382     return _retry_rexec(args, log_msg, env = None, retry = retry,
 383             tmp_known_hosts = tmp_known_hosts,
 384             blocking = True)
 385
 386 def rspawn(command, pidfile,
 387         stdout = '/dev/null',
 388         stderr = STDOUT,
 389         stdin = '/dev/null',
 390         home = None,
 391         create_home = False,
 392         sudo = False,
 393         host = None,
 394         port = None,
 395         user = None,
 396         gwuser = None,
 397         gw = None,
 398         agent = None,
 399         identity = None,
 400         server_key = None,
 401         tty = False,
 402         strict_host_checking = True):
 403     """
 404     Spawn a remote command such that it will continue working asynchronously in
 405     background.
 406
 407         :param command: The command to run, it should be a single line.
 408         :type command: str
 409
 410         :param pidfile: Path to a file where to store the pid and ppid of the
 411                         spawned process
 412         :type pidfile: str
 413
 414         :param stdout: Path to file to redirect standard output.
 415                        The default value is /dev/null
 416         :type stdout: str
 417
 418         :param stderr: Path to file to redirect standard error.
 419                        If the special STDOUT value is used, stderr will
 420                        be redirected to the same file as stdout
 421         :type stderr: str
 422
 423         :param stdin: Path to a file with input to be piped into the command's standard input
 424         :type stdin: str
 425
 426         :param home: Path to working directory folder.
 427                     It is assumed to exist unless the create_home flag is set.
 428         :type home: str
 429
 430         :param create_home: Flag to force creation of the home folder before
 431                             running the command
 432         :type create_home: bool
 433
 434         :param sudo: Flag forcing execution with sudo user
 435         :type sudo: bool
 436
 437         :rtype: tuple
 438
 439         (stdout, stderr), process
 440
 441         Of the spawning process, which only captures errors at spawning time.
 442         Usually only useful for diagnostics.
 443     """
 444     # Start process in a "daemonized" way, using nohup and heavy
 445     # stdin/out redirection to avoid connection issues
 446     if stderr is STDOUT:
 447         stderr = '&1'
 448     else:
 449         stderr = ' ' + stderr
 450
 451     daemon_command = '{ { %(command)s > %(stdout)s 2>%(stderr)s < %(stdin)s & } ; echo $! 1 > %(pidfile)s ; }' % {
 452         'command' : command,
 453         'pidfile' : shell_escape(pidfile),
 454         'stdout' : stdout,
 455         'stderr' : stderr,
 456         'stdin' : stdin,
 457     }
 458
 459     cmd = "%(create)s%(gohome)s rm -f %(pidfile)s ; %(sudo)s nohup bash -c %(command)s " % {
 460             'command' : shell_escape(daemon_command),
 461             'sudo' : 'sudo -S' if sudo else '',
 462             'pidfile' : shell_escape(pidfile),
 463             'gohome' : 'cd %s ; ' % (shell_escape(home),) if home else '',
 464             'create' : 'mkdir -p %s ; ' % (shell_escape(home),) if create_home and home else '',
 465         }
 466
 467     (out,err),proc = rexec(
 468         cmd,
 469         host = host,
 470         port = port,
 471         user = user,
 472         gwuser = gwuser,
 473         gw = gw,
 474         agent = agent,
 475         identity = identity,
 476         server_key = server_key,
 477         tty = tty,
 478         strict_host_checking = strict_host_checking ,
 479         )
 480
 481     if proc.wait():
 482         raise RuntimeError, "Failed to set up application on host %s: %s %s" % (host, out,err,)
 483
 484     return ((out, err), proc)
 485
 486 @eintr_retry
 487 def rgetpid(pidfile,
 488         host = None,
 489         port = None,
 490         user = None,
 491         gwuser = None,
 492         gw = None,
 493         agent = None,
 494         identity = None,
 495         server_key = None,
 496         strict_host_checking = True):
 497     """
 498     Returns the pid and ppid of a process from a remote file where the
 499     information was stored.
 500
 501         :param home: Path to directory where the pidfile is located
 502         :type home: str
 503
 504         :param pidfile: Name of file containing the pid information
 505         :type pidfile: str
 506
 507         :rtype: int
 508
 509         A (pid, ppid) tuple useful for calling rstatus and rkill,
 510         or None if the pidfile isn't valid yet (can happen when process is staring up)
 511
 512     """
 513     (out,err),proc = rexec(
 514         "cat %(pidfile)s" % {
 515             'pidfile' : pidfile,
 516         },
 517         host = host,
 518         port = port,
 519         user = user,
 520         gwuser = gwuser,
 521         gw = gw,
 522         agent = agent,
 523         identity = identity,
 524         server_key = server_key,
 525         strict_host_checking = strict_host_checking
 526         )
 527
 528     if proc.wait():
 529         return None
 530
 531     if out:
 532         try:
 533             return map(int,out.strip().split(' ',1))
 534         except:
 535             # Ignore, many ways to fail that don't matter that much
 536             return None
 537
 538 @eintr_retry
 539 def rstatus(pid, ppid,
 540         host = None,
 541         port = None,
 542         user = None,
 543         gwuser = None,
 544         gw = None,
 545         agent = None,
 546         identity = None,
 547         server_key = None,
 548         strict_host_checking = True):
 549     """
 550     Returns a code representing the the status of a remote process
 551
 552         :param pid: Process id of the process
 553         :type pid: int
 554
 555         :param ppid: Parent process id of process
 556         :type ppid: int
 557
 558         :rtype: int (One of NOT_STARTED, RUNNING, FINISHED)
 559
 560     """
 561     (out,err),proc = rexec(
 562         # Check only by pid. pid+ppid does not always work (especially with sudo)
 563         " (( ps --pid %(pid)d -o pid | grep -c %(pid)d && echo 'wait')  || echo 'done' ) | tail -n 1" % {
 564             'ppid' : ppid,
 565             'pid' : pid,
 566         },
 567         host = host,
 568         port = port,
 569         user = user,
 570         gwuser = gwuser,
 571         gw = gw,
 572         agent = agent,
 573         identity = identity,
 574         server_key = server_key,
 575         strict_host_checking = strict_host_checking
 576         )
 577
 578     if proc.wait():
 579         return ProcStatus.NOT_STARTED
 580
 581     status = False
 582     if err:
 583         if err.strip().find("Error, do this: mount -t proc none /proc") >= 0:
 584             status = True
 585     elif out:
 586         status = (out.strip() == 'wait')
 587     else:
 588         return ProcStatus.NOT_STARTED
 589     return ProcStatus.RUNNING if status else ProcStatus.FINISHED
 590
 591 @eintr_retry
 592 def rkill(pid, ppid,
 593         host = None,
 594         port = None,
 595         user = None,
 596         gwuser = None,
 597         gw = None,
 598         agent = None,
 599         sudo = False,
 600         identity = None,
 601         server_key = None,
 602         nowait = False,
 603         strict_host_checking = True):
 604     """
 605     Sends a kill signal to a remote process.
 606
 607     First tries a SIGTERM, and if the process does not end in 10 seconds,
 608     it sends a SIGKILL.
 609
 610         :param pid: Process id of process to be killed
 611         :type pid: int
 612
 613         :param ppid: Parent process id of process to be killed
 614         :type ppid: int
 615
 616         :param sudo: Flag indicating if sudo should be used to kill the process
 617         :type sudo: bool
 618
 619     """
 620     subkill = "$(ps --ppid %(pid)d -o pid h)" % { 'pid' : pid }
 621     cmd = """
 622 SUBKILL="%(subkill)s" ;
 623 %(sudo)s kill -- -%(pid)d $SUBKILL || /bin/true
 624 %(sudo)s kill %(pid)d $SUBKILL || /bin/true
 625 for x in 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 ; do
 626     sleep 0.2
 627     if [ `ps --pid %(pid)d -o pid | grep -c %(pid)d` == '0' ]; then
 628         break
 629     else
 630         %(sudo)s kill -- -%(pid)d $SUBKILL || /bin/true
 631         %(sudo)s kill %(pid)d $SUBKILL || /bin/true
 632     fi
 633     sleep 1.8
 634 done
 635 if [ `ps --pid %(pid)d -o pid | grep -c %(pid)d` != '0' ]; then
 636     %(sudo)s kill -9 -- -%(pid)d $SUBKILL || /bin/true
 637     %(sudo)s kill -9 %(pid)d $SUBKILL || /bin/true
 638 fi
 639 """
 640     if nowait:
 641         cmd = "( %s ) >/dev/null 2>/dev/null </dev/null &" % (cmd,)
 642
 643     (out,err),proc = rexec(
 644         cmd % {
 645             'ppid' : ppid,
 646             'pid' : pid,
 647             'sudo' : 'sudo -S' if sudo else '',
 648             'subkill' : subkill,
 649         },
 650         host = host,
 651         port = port,
 652         user = user,
 653         gwuser = gwuser,
 654         gw = gw,
 655         agent = agent,
 656         identity = identity,
 657         server_key = server_key,
 658         strict_host_checking = strict_host_checking
 659         )
 660
 661     # wait, don't leave zombies around
 662     proc.wait()
 663
 664     return (out, err), proc
 665
 666 def _retry_rexec(args,
 667         log_msg,
 668         stdout = subprocess.PIPE,
 669         stdin = subprocess.PIPE,
 670         stderr = subprocess.PIPE,
 671         env = None,
 672         retry = 3,
 673         tmp_known_hosts = None,
 674         blocking = True):
 675
 676     for x in xrange(retry):
 677         # connects to the remote host and starts a remote connection
 678         proc = subprocess.Popen(args,
 679                 env = env,
 680                 stdout = stdout,
 681                 stdin = stdin,
 682                 stderr = stderr)
 683
 684         # attach tempfile object to the process, to make sure the file stays
 685         # alive until the process is finished with it
 686         proc._known_hosts = tmp_known_hosts
 687
 688         # The argument block == False forces to rexec to return immediately,
 689         # without blocking
 690         try:
 691             err = out = " "
 692             if blocking:
 693                 #(out, err) = proc.communicate()
 694                 # The method communicate was re implemented for performance issues
 695                 # when using python subprocess communicate method the ssh commands
 696                 # last one minute each
 697                 out, err = _communicate(proc, input=None)
 698
 699             elif stdout:
 700                 out = proc.stdout.read()
 701                 if proc.poll() and stderr:
 702                     err = proc.stderr.read()
 703
 704             log(log_msg, logging.DEBUG, out, err)
 705
 706             if proc.poll():
 707                 skip = False
 708
 709                 if err.strip().startswith('ssh: ') or err.strip().startswith('mux_client_hello_exchange: '):
 710                     # SSH error, can safely retry
 711                     skip = True
 712                 elif retry:
 713                     # Probably timed out or plain failed but can retry
 714                     skip = True
 715
 716                 if skip:
 717                     t = x*2
 718                     msg = "SLEEPING %d ... ATEMPT %d - command %s " % (
 719                             t, x, " ".join(args))
 720                     log(msg, logging.DEBUG)
 721
 722                     time.sleep(t)
 723                     continue
 724             break
 725         except RuntimeError, e:
 726             msg = " rexec EXCEPTION - TIMEOUT -> %s \n %s" % ( e.args, log_msg )
 727             log(msg, logging.DEBUG, out, err)
 728
 729             if retry <= 0:
 730                 raise
 731             retry -= 1
 732
 733     return ((out, err), proc)
 734
 735 # POSIX
 736 # Don't remove. The method communicate was re implemented for performance issues
 737 def _communicate(proc, input, timeout=None, err_on_timeout=True):
 738     read_set = []
 739     write_set = []
 740     stdout = None # Return
 741     stderr = None # Return
 742
 743     killed = False
 744
 745     if timeout is not None:
 746         timelimit = time.time() + timeout
 747         killtime = timelimit + 4
 748         bailtime = timelimit + 4
 749
 750     if proc.stdin:
 751         # Flush stdio buffer.  This might block, if the user has
 752         # been writing to .stdin in an uncontrolled fashion.
 753         proc.stdin.flush()
 754         if input:
 755             write_set.append(proc.stdin)
 756         else:
 757             proc.stdin.close()
 758
 759     if proc.stdout:
 760         read_set.append(proc.stdout)
 761         stdout = []
 762
 763     if proc.stderr:
 764         read_set.append(proc.stderr)
 765         stderr = []
 766
 767     input_offset = 0
 768     while read_set or write_set:
 769         if timeout is not None:
 770             curtime = time.time()
 771             if timeout is None or curtime > timelimit:
 772                 if curtime > bailtime:
 773                     break
 774                 elif curtime > killtime:
 775                     signum = signal.SIGKILL
 776                 else:
 777                     signum = signal.SIGTERM
 778                 # Lets kill it
 779                 os.kill(proc.pid, signum)
 780                 select_timeout = 0.5
 781             else:
 782                 select_timeout = timelimit - curtime + 0.1
 783         else:
 784             select_timeout = 1.0
 785
 786         if select_timeout > 1.0:
 787             select_timeout = 1.0
 788
 789         try:
 790             rlist, wlist, xlist = select.select(read_set, write_set, [], select_timeout)
 791         except select.error,e:
 792             if e[0] != 4:
 793                 raise
 794             else:
 795                 continue
 796
 797         if not rlist and not wlist and not xlist and proc.poll() is not None:
 798             # timeout and process exited, say bye
 799             break
 800
 801         if proc.stdin in wlist:
 802             # When select has indicated that the file is writable,
 803             # we can write up to PIPE_BUF bytes without risk
 804             # blocking.  POSIX defines PIPE_BUF >= 512
 805             bytes_written = os.write(proc.stdin.fileno(),
 806                     buffer(input, input_offset, 512))
 807             input_offset += bytes_written
 808
 809             if input_offset >= len(input):
 810                 proc.stdin.close()
 811                 write_set.remove(proc.stdin)
 812
 813         if proc.stdout in rlist:
 814             data = os.read(proc.stdout.fileno(), 1024)
 815             if data == "":
 816                 proc.stdout.close()
 817                 read_set.remove(proc.stdout)
 818             stdout.append(data)
 819
 820         if proc.stderr in rlist:
 821             data = os.read(proc.stderr.fileno(), 1024)
 822             if data == "":
 823                 proc.stderr.close()
 824                 read_set.remove(proc.stderr)
 825             stderr.append(data)
 826
 827     # All data exchanged.  Translate lists into strings.
 828     if stdout is not None:
 829         stdout = ''.join(stdout)
 830     if stderr is not None:
 831         stderr = ''.join(stderr)
 832
 833     # Translate newlines, if requested.  We cannot let the file
 834     # object do the translation: It is based on stdio, which is
 835     # impossible to combine with select (unless forcing no
 836     # buffering).
 837     if proc.universal_newlines and hasattr(file, 'newlines'):
 838         if stdout:
 839             stdout = proc._translate_newlines(stdout)
 840         if stderr:
 841             stderr = proc._translate_newlines(stderr)
 842
 843     if killed and err_on_timeout:
 844         errcode = proc.poll()
 845         raise RuntimeError, ("Operation timed out", errcode, stdout, stderr)
 846     else:
 847         if killed:
 848             proc.poll()
 849         else:
 850             proc.wait()
 851         return (stdout, stderr)
 852
 853 def _proxy_command(gw, gwuser, gwidentity):
 854     """
 855     Constructs the SSH ProxyCommand option to add to the SSH command to connect
 856     via a proxy
 857         :param gw: SSH proxy hostname
 858         :type gw:  str
 859
 860         :param gwuser: SSH proxy username
 861         :type gwuser:  str
 862
 863         :param gwidentity: SSH proxy identity file
 864         :type gwidentity:  str
 865
 866
 867         :rtype: str
 868
 869         returns the SSH ProxyCommand option.
 870     """
 871
 872     proxycommand = 'ProxyCommand=ssh -q '
 873     if gwidentity:
 874         proxycommand += '-i %s ' % os.path.expanduser(gwidentity)
 875     if gwuser:
 876         proxycommand += '%s' % gwuser
 877     else:
 878         proxycommand += '%r'
 879     proxycommand += '@%s -W %%h:%%p' % gw
 880
 881     return proxycommand
 882