15 OPENSSH_HAS_PERSIST = None
16 CONTROL_PATH = "yyy_ssh_ctrl_path"
18 if hasattr(os, "devnull"):
21 DEV_NULL = "/dev/null"
23 SHELL_SAFE = re.compile('^[-a-zA-Z0-9_=+:.,/]*$')
25 hostbyname_cache = dict()
29 Special value that when given to rspawn in stderr causes stderr to
30 redirect to whatever stdout was redirected to.
35 Process is still running
45 Process hasn't started running yet (this should be very rare)
48 def openssh_has_persist():
49 """ The ssh_config options ControlMaster and ControlPersist allow to
50 reuse a same network connection for multiple ssh sessions. In this
51 way limitations on number of open ssh connections can be bypassed.
52 However, older versions of openSSH do not support this feature.
53 This function is used to determine if ssh connection persist features
56 global OPENSSH_HAS_PERSIST
57 if OPENSSH_HAS_PERSIST is None:
58 proc = subprocess.Popen(["ssh","-v"],
59 stdout = subprocess.PIPE,
60 stderr = subprocess.STDOUT,
61 stdin = open("/dev/null","r") )
62 out,err = proc.communicate()
65 vre = re.compile(r'OpenSSH_(?:[6-9]|5[.][8-9]|5[.][1-9][0-9]|[1-9][0-9]).*', re.I)
66 OPENSSH_HAS_PERSIST = bool(vre.match(out))
67 return OPENSSH_HAS_PERSIST
70 """ Escapes strings so that they are safe to use as command-line
72 if SHELL_SAFE.match(s):
73 # safe string - no escaping needed
76 # unsafe string - escape
78 if (32 <= ord(c) < 127 or c in ('\r','\n','\t')) and c not in ("'",'"'):
81 return "'$'\\x%02x''" % (ord(c),)
82 s = ''.join(map(escp,s))
85 def eintr_retry(func):
86 """Retries a function invocation when a EINTR occurs"""
88 @functools.wraps(func)
90 retry = kw.pop("_retry", False)
91 for i in xrange(0 if retry else 4):
94 except (select.error, socket.error), args:
95 if args[0] == errno.EINTR:
100 if e.errno == errno.EINTR:
105 return func(*p, **kw)
108 def make_connkey(user, host, port, x11, agent):
109 # It is important to consider the x11 and agent forwarding
110 # parameters when creating the connection key since the parameters
111 # used for the first ssh connection will determine the
112 # parameters of all subsequent connections using the same key
113 x11 = 1 if x11 else 0
114 agent = 1 if agent else 0
116 connkey = repr((user, host, port, x11, agent)
117 ).encode("base64").strip().replace('/','.')
119 if len(connkey) > 60:
120 connkey = hashlib.sha1(connkey).hexdigest()
123 def make_control_path(user, host, port, x11, agent):
124 connkey = make_connkey(user, host, port, x11, agent)
125 return '/tmp/%s_%s' % ( CONTROL_PATH, connkey, )
127 def rexec(command, host, user,
138 err_on_timeout = True,
139 connect_timeout = 30,
142 Executes a remote command, returns ((stdout,stderr),process)
145 # Don't bother with localhost. Makes test easier
146 '-o', 'NoHostAuthenticationForLocalhost=yes',
147 # XXX: Possible security issue
148 # Avoid interactive requests to accept new host keys
149 '-o', 'StrictHostKeyChecking=no',
150 '-o', 'ConnectTimeout=%d' % (int(connect_timeout),),
151 '-o', 'ConnectionAttempts=3',
152 '-o', 'ServerAliveInterval=30',
153 '-o', 'TCPKeepAlive=yes',
156 if persistent and openssh_has_persist():
157 control_path = make_control_path(user, host, port, x11, agent)
159 '-o', 'ControlMaster=auto',
160 '-o', 'ControlPath=%s' % control_path,
161 '-o', 'ControlPersist=60' ])
165 args.append('-p%d' % port)
167 args.extend(('-i', identity))
177 for envkey, envval in env.iteritems():
178 export += '%s=%s ' % (envkey, envval)
179 command = export + command
182 command = "sudo " + command
186 for x in xrange(retry or 3):
187 # connects to the remote host and starts a remote connection
188 proc = subprocess.Popen(args,
189 stdout = subprocess.PIPE,
190 stdin = subprocess.PIPE,
191 stderr = subprocess.PIPE)
194 out, err = _communicate(proc, stdin, timeout, err_on_timeout)
196 if err.strip().startswith('ssh: ') or err.strip().startswith('mux_client_hello_exchange: '):
197 # SSH error, can safely retry
200 # Probably timed out or plain failed but can retry
203 except RuntimeError,e:
208 return ((out, err), proc)
210 def rcopy(source, dest,
215 Copies file from/to remote sites.
217 Source and destination should have the user and host encoded
220 Source can be a list of files to copy to a single destination,
221 in which case it is advised that the destination be a folder.
224 # Parse destination as <user>@<server>:<path>
225 if isinstance(dest, basestring) and ':' in dest:
226 remspec, path = dest.split(':',1)
227 elif isinstance(source, basestring) and ':' in source:
228 remspec, path = source.split(':',1)
230 raise ValueError, "Both endpoints cannot be local"
231 user, host = remspec.rsplit('@',1)
233 raw_string = r'''rsync -rlpcSz --timeout=900 '''
234 raw_string += r''' -e 'ssh -o BatchMode=yes '''
235 raw_string += r''' -o NoHostAuthenticationForLocalhost=yes '''
236 raw_string += r''' -o StrictHostKeyChecking=no '''
237 raw_string += r''' -o ConnectionAttempts=3 '''
239 if openssh_has_persist():
240 control_path = make_control_path(user, host, port, False, agent)
241 raw_string += r''' -o ControlMaster=auto '''
242 raw_string += r''' -o ControlPath=%s ''' % control_path
245 raw_string += r''' -A '''
248 raw_string += r''' -p %d ''' % port
251 raw_string += r''' -i "%s" ''' % identity
253 # closing -e 'ssh...'
254 raw_string += r''' ' '''
256 if isinstance(source,list):
257 source = ' '.join(source)
259 source = '"%s"' % source
261 raw_string += r''' %s ''' % source
262 raw_string += r''' %s ''' % dest
264 # connects to the remote host and starts a remote connection
265 proc = subprocess.Popen(raw_string,
267 stdout = subprocess.PIPE,
268 stdin = subprocess.PIPE,
269 stderr = subprocess.PIPE)
271 comm = proc.communicate()
272 eintr_retry(proc.wait)()
275 def rspawn(command, pidfile,
276 stdout = '/dev/null',
289 Spawn a remote command such that it will continue working asynchronously.
292 command: the command to run - it should be a single line.
294 pidfile: path of a (ideally unique to this task) pidfile for tracking the process.
296 stdout: path of a file to redirect standard output to - must be a string.
297 Defaults to /dev/null
298 stderr: path of a file to redirect standard error to - string or the special STDOUT value
299 to redirect to the same file stdout was redirected to. Defaults to STDOUT.
300 stdin: path of a file with input to be piped into the command's standard input
302 home: path of a folder to use as working directory - should exist, unless you specify create_home
304 create_home: if True, the home folder will be created first with mkdir -p
306 sudo: whether the command needs to be executed as root
308 host/port/user/agent/identity: see rexec
311 (stdout, stderr), process
313 Of the spawning process, which only captures errors at spawning time.
314 Usually only useful for diagnostics.
316 # Start process in a "daemonized" way, using nohup and heavy
317 # stdin/out redirection to avoid connection issues
321 stderr = ' ' + stderr
323 #XXX: ppid is always 1!!!
324 daemon_command = '{ { %(command)s > %(stdout)s 2>%(stderr)s < %(stdin)s & } ; echo $! 1 > %(pidfile)s ; }' % {
333 cmd = "%(create)s%(gohome)s rm -f %(pidfile)s ; %(sudo)s nohup bash -c '%(command)s' " % {
334 'command' : daemon_command,
336 'sudo' : 'sudo -S' if sudo else '',
339 'gohome' : 'cd %s ; ' % home if home else '',
340 'create' : 'mkdir -p %s ; ' % home if create_home else '',
343 (out,err), proc = rexec(
354 raise RuntimeError, "Failed to set up application: %s %s" % (out,err,)
356 return (out,err),proc
359 def rcheckpid(pidfile,
366 Check the pidfile of a process spawned with remote_spawn.
369 pidfile: the pidfile passed to remote_span
371 host/port/user/agent/identity: see rexec
375 A (pid, ppid) tuple useful for calling remote_status and remote_kill,
376 or None if the pidfile isn't valid yet (maybe the process is still starting).
379 (out,err),proc = rexec(
380 "cat %(pidfile)s" % {
395 return map(int,out.strip().split(' ',1))
397 # Ignore, many ways to fail that don't matter that much
401 def rstatus(pid, ppid,
408 Check the status of a process spawned with remote_spawn.
411 pid/ppid: pid and parent-pid of the spawned process. See remote_check_pid
413 host/port/user/agent/identity: see rexec
417 One of NOT_STARTED, RUNNING, FINISHED
421 (out,err),proc = rexec(
422 "ps --pid %(pid)d -o pid | grep -c %(pid)d ; true" % {
439 status = bool(int(out.strip()))
442 logging.warn("Error checking remote status:\n%s%s\n", out, err)
443 # Ignore, many ways to fail that don't matter that much
445 return RUNNING if status else FINISHED
458 Kill a process spawned with remote_spawn.
460 First tries a SIGTERM, and if the process does not end in 10 seconds,
464 pid/ppid: pid and parent-pid of the spawned process. See remote_check_pid
466 sudo: whether the command was run with sudo - careful killing like this.
468 host/port/user/agent/identity: see rexec
472 Nothing, should have killed the process
475 subkill = "$(ps --ppid %(pid)d -o pid h)" % { 'pid' : pid }
477 SUBKILL="%(subkill)s" ;
478 %(sudo)s kill -- -%(pid)d $SUBKILL || /bin/true
479 %(sudo)s kill %(pid)d $SUBKILL || /bin/true
480 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
482 if [ `ps --pid %(pid)d -o pid | grep -c %(pid)d` == '0' ]; then
485 %(sudo)s kill -- -%(pid)d $SUBKILL || /bin/true
486 %(sudo)s kill %(pid)d $SUBKILL || /bin/true
490 if [ `ps --pid %(pid)d -o pid | grep -c %(pid)d` != '0' ]; then
491 %(sudo)s kill -9 -- -%(pid)d $SUBKILL || /bin/true
492 %(sudo)s kill -9 %(pid)d $SUBKILL || /bin/true
496 cmd = "( %s ) >/dev/null 2>/dev/null </dev/null &" % (cmd,)
498 (out,err),proc = rexec(
502 'sudo' : 'sudo -S' if sudo else '',
512 # wait, don't leave zombies around
516 def _communicate(self, input, timeout=None, err_on_timeout=True):
519 stdout = None # Return
520 stderr = None # Return
524 if timeout is not None:
525 timelimit = time.time() + timeout
526 killtime = timelimit + 4
527 bailtime = timelimit + 4
530 # Flush stdio buffer. This might block, if the user has
531 # been writing to .stdin in an uncontrolled fashion.
534 write_set.append(self.stdin)
538 read_set.append(self.stdout)
541 read_set.append(self.stderr)
545 while read_set or write_set:
546 if timeout is not None:
547 curtime = time.time()
548 if timeout is None or curtime > timelimit:
549 if curtime > bailtime:
551 elif curtime > killtime:
552 signum = signal.SIGKILL
554 signum = signal.SIGTERM
556 os.kill(self.pid, signum)
559 select_timeout = timelimit - curtime + 0.1
563 if select_timeout > 1.0:
567 rlist, wlist, xlist = select.select(read_set, write_set, [], select_timeout)
568 except select.error,e:
574 if not rlist and not wlist and not xlist and self.poll() is not None:
575 # timeout and process exited, say bye
578 if self.stdin in wlist:
579 # When select has indicated that the file is writable,
580 # we can write up to PIPE_BUF bytes without risk
581 # blocking. POSIX defines PIPE_BUF >= 512
582 bytes_written = os.write(self.stdin.fileno(), buffer(input, input_offset, 512))
583 input_offset += bytes_written
584 if input_offset >= len(input):
586 write_set.remove(self.stdin)
588 if self.stdout in rlist:
589 data = os.read(self.stdout.fileno(), 1024)
592 read_set.remove(self.stdout)
595 if self.stderr in rlist:
596 data = os.read(self.stderr.fileno(), 1024)
599 read_set.remove(self.stderr)
602 # All data exchanged. Translate lists into strings.
603 if stdout is not None:
604 stdout = ''.join(stdout)
605 if stderr is not None:
606 stderr = ''.join(stderr)
608 # Translate newlines, if requested. We cannot let the file
609 # object do the translation: It is based on stdio, which is
610 # impossible to combine with select (unless forcing no
612 if self.universal_newlines and hasattr(file, 'newlines'):
614 stdout = self._translate_newlines(stdout)
616 stderr = self._translate_newlines(stderr)
618 if killed and err_on_timeout:
619 errcode = self.poll()
620 raise RuntimeError, ("Operation timed out", errcode, stdout, stderr)
626 return (stdout, stderr)