Support for management proxy and Python 2 on the target host

ansible/connection_plugins/qubes.py

@@ -42,9 +42,10 @@ class Connection(ConnectionBase):
     become_from_methods = frozenset(["sudo"])
     _management_proxy = None
 
-#    def set_host_overrides(self, host):
+    def set_host_overrides(self, host, hostvars):
-#        host_vars = combine_vars(host.get_group_vars(), host.get_vars())
+        self._management_proxy = hostvars.get("management_proxy", None)
-#        _management_proxy = host_vars.get("_management_proxy", None)
+        if self._management_proxy:
+            self.chroot = hostvars.get("inventory_hostname_short")
 
     def __init__(self, play_context, new_stdin, *args, **kwargs):
         super(Connection, self).__init__(play_context, new_stdin, *args, **kwargs)
@@ -64,9 +65,6 @@ class Connection(ConnectionBase):
         if not self.qrun:
             raise errors.AnsibleError("qrun command not found in PATH")
 
-        if self._management_proxy:
-            assert 0, "still do not know how to deal with management proxy"
-
     def _connect(self):
         """Connect to the host we've been initialized with"""
 
@@ -84,14 +82,11 @@ class Connection(ConnectionBase):
 
     def _produce_command(self, cmd):
         # FIXME
-        # proxy = ["--proxy=%s" % self._management_proxy] if self._management_proxy else []
+        proxy = ["--proxy=%s" % self._management_proxy] if self._management_proxy else []
         if isinstance(cmd, basestring):
             unsplit = shlex.split(cmd)
-            return [self.qrun, self.chroot] + unsplit
+            return [self.qrun] + proxy + [self.chroot] + unsplit
-        #if proxy:
+        local_cmd = [self.qrun] + proxy + [self.chroot] + cmd
-        #  local_cmd = [self.qrun] + proxy + [chroot] + cmd
-        #else:
-        local_cmd = [self.qrun, self.chroot] + cmd
         local_cmd = map(to_bytes, local_cmd)
         return local_cmd
 

bin/bombshell-client

@@ -9,7 +9,10 @@ import errno
 import fcntl
 import os
 import pipes
-import queue
+try:
+    import queue
+except ImportError:
+    import Queue as queue
 import select
 import signal
 import struct
@@ -38,17 +41,27 @@ def mutexfile(filepath):
     f.close()
 
 
+def unset_cloexec(fd):
+    old = fcntl.fcntl(fd, fcntl.F_GETFD)
+    fcntl.fcntl(fd, fcntl.F_SETFD, old & ~ fcntl.FD_CLOEXEC)
+
+
 def openfdforappend(fd):
+  f = None
   try:
-    return os.fdopen(fd, "ab", 0, closefd=False)
+    f = os.fdopen(fd, "ab", 0)
   except IOError as e:
     if e.errno != errno.ESPIPE:
       raise
-    return os.fdopen(fd, "wb", 0, closefd=False)
+    f = os.fdopen(fd, "wb", 0)
+  unset_cloexec(f.fileno())
+  return f
 
 
 def openfdforread(fd):
-  return os.fdopen(fd, "rb", 0, closefd=False)
+  f = os.fdopen(fd, "rb", 0)
+  unset_cloexec(f.fileno())
+  return f
 
 
 debug_lock = threading.Lock()
@@ -171,18 +184,12 @@ class Signaler(MyThread):
     logging.debug("End of signaler")
 
 
-def unblock(fobj):
-  if hasattr(os, "set_blocking"):
-    return os.set_blocking(fobj.fileno(), False)
-  fl = fcntl.fcntl(fobj, fcntl.F_GETFL)
-  fcntl.fcntl(fobj, fcntl.F_SETFL, fl | os.O_NONBLOCK)
-
-
 def write(dst, buffer, l):
     alreadywritten = 0
     mv = memoryview(buffer)[:l]
     while len(mv):
-        writtenthisloop = dst.write(mv)
+        dst.write(mv)
+        writtenthisloop = len(mv)
         if writtenthisloop is None or writtenthisloop < 1:
             raise Exception("copy: Failed to write any bytes")
         mv = mv[writtenthisloop:]
@@ -210,14 +217,13 @@ class DataMultiplexer(MyThread):
     self.setDaemon(True)
     self.sources = dict((s,num) for num, s in enumerate(sources))
     self.sink = sink
-    for s in sources: unblock(s)
 
   def _run(self):
     logging.debug("mux: Started with sources %s and sink %s", self.sources, self.sink)
     buffer = bytearray(MAX_MUX_READ)
-    sources, _, x = select.select((s for s in self.sources), (), (s for s in self.sources))
+    while self.sources:
-    assert not x, x
+      sources, _, x = select.select((s for s in self.sources), (), (s for s in self.sources))
-    while sources:
+      assert not x, x
       for s in sources:
         n = self.sources[s]
         logging.debug("mux: Source %s (%s) is active", n, s)
@@ -232,10 +238,6 @@ class DataMultiplexer(MyThread):
         header = struct.pack(PACKFORMAT, n, True, l)
         self.sink.write(header)
         write(self.sink, buffer, l)
-      if not self.sources:
-        break
-      sources, _, x = select.select((s for s in self.sources), (), (s for s in self.sources))
-      assert not x, x
     logging.debug("mux: End of data multiplexer")
 
 
@@ -246,7 +248,6 @@ class DataDemultiplexer(MyThread):
     self.setDaemon(True)
     self.sinks = dict(enumerate(sinks))
     self.source = source
-    unblock(source)
 
   def _run(self):
     logging.debug("demux: Started with source %s and sinks %s", self.source, self.sinks)
@@ -288,12 +289,17 @@ def main_master():
   remote_command = args[1:]
   assert remote_command
 
-  remote_helper_text = b"exec %s -u -c %s%s %s\n" % (
+  def anypython(exe):
-      bytes(sys.executable, "utf-8"),
+      return "` test -x %s && echo %s || echo python`" % (pipes.quote(exe),
-      bytes(pipes.quote(open(__file__, "r").read()), "ascii"),
+                                                          pipes.quote(exe))
-      (b" -d" if debug_enabled else b""),
+
-      base64.b64encode(pickle.dumps(remote_command)),
+  remote_helper_text = b"exec "
-  )
+  remote_helper_text += bytes(anypython(sys.executable), "utf-8")
+  remote_helper_text += bytes(" -u -c ", "utf-8")
+  remote_helper_text += bytes(pipes.quote(open(__file__, "r").read()), "ascii")
+  remote_helper_text += b" -d " if debug_enabled else b" "
+  remote_helper_text += base64.b64encode(pickle.dumps(remote_command, 2))
+  remote_helper_text += b"\n"
 
   saved_stderr = openfdforappend(os.dup(sys.stderr.fileno()))
 

bin/qrun

@@ -33,9 +33,11 @@ if remotehost:
     therest_template = ("test -x ./.bombshell-client || "
                         "python -c 'import os; file(\"./.bombshell-client\", \"wb\").write(\"%s\".decode(\"hex_codec\")); os.chmod(\"./.bombshell-client\", 0700)' || "
                         "exit 127 ;"
-                        "export BOMBSHELL_DEBUG=%s ;"
+                        "./.bombshell-client %s %s %s")
-                        "./.bombshell-client %s %s")
+    therest = therest_template % (poop,
-    therest = therest_template % (poop, pipes.quote(os.getenv("BOMBSHELL_DEBUG")), pipes.quote(host), args)
+                                  "-d" if os.getenv("BOMBSHELL_DEBUG") else "",
+                                  pipes.quote(host),
+                                  args)
     cmd = [
         'ssh',
         '-o', 'BatchMode yes',