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