gaschz/ansible-qubes
1
0
Fork 0
mirror of https://github.com/Rudd-O/ansible-qubes.git synced 2025-06-06 18:08:31 +02:00
Code Issues Packages Projects Releases Wiki Activity

Code reformat and quality improvement.

Browse Source
This commit is contained in:
Manuel Amador (Rudd-O) 2022-07-12 05:47:00 +00:00
parent f6c623e5db
commit 167a82bac8
1 changed files with 325 additions and 277 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

602
bin/bombshell-client
View File

@ -3,16 +3,19 @@
import base64 import base64
import pickle import pickle
import contextlib import contextlib
import ctypes
import ctypes.util
import errno import errno
import fcntl import fcntl
import os import os
import pipes
try: try:
import queue from shlex import quote
except ImportError: except ImportError:
import Queue as queue from pipes import quote # noqa
try:
from queue import Queue
except ImportError:
from Queue import Queue # noqa
import select import select
import signal import signal
import struct import struct
@ -24,7 +27,7 @@ import time
import traceback import traceback
MAX_MUX_READ = 128*1024 # 64*1024*1024 MAX_MUX_READ = 128 * 1024 # 64*1024*1024
PACKLEN = 8 PACKLEN = 8
PACKFORMAT = "!HbIx" PACKFORMAT = "!HbIx"
@ -43,42 +46,48 @@ def mutexfile(filepath):
def unset_cloexec(fd): def unset_cloexec(fd):
old = fcntl.fcntl(fd, fcntl.F_GETFD) old = fcntl.fcntl(fd, fcntl.F_GETFD)
fcntl.fcntl(fd, fcntl.F_SETFD, old & ~ fcntl.FD_CLOEXEC) fcntl.fcntl(fd, fcntl.F_SETFD, old & ~fcntl.FD_CLOEXEC)
def openfdforappend(fd): def openfdforappend(fd):
f = None f = None
try: try:
f = os.fdopen(fd, "ab", 0) f = os.fdopen(fd, "ab", 0)
except IOError as e: except IOError as e:
if e.errno != errno.ESPIPE: if e.errno != errno.ESPIPE:
raise raise
f = os.fdopen(fd, "wb", 0) f = os.fdopen(fd, "wb", 0)
unset_cloexec(f.fileno()) unset_cloexec(f.fileno())
return f return f
def openfdforread(fd): def openfdforread(fd):
f = os.fdopen(fd, "rb", 0) f = os.fdopen(fd, "rb", 0)
unset_cloexec(f.fileno()) unset_cloexec(f.fileno())
return f return f
debug_lock = threading.Lock() debug_lock = threading.Lock()
debug_enabled = False debug_enabled = False
_startt = time.time() _startt = time.time()
class LoggingEmu():
class LoggingEmu:
def __init__(self, prefix): def __init__(self, prefix):
self.prefix = prefix self.prefix = prefix
syslog.openlog("bombshell-client.%s" % self.prefix) syslog.openlog("bombshell-client.%s" % self.prefix)
def debug(self, *a, **kw): def debug(self, *a, **kw):
if not debug_enabled: if not debug_enabled:
return return
self._print(syslog.LOG_DEBUG, *a, **kw) self._print(syslog.LOG_DEBUG, *a, **kw)
def info(self, *a, **kw): def info(self, *a, **kw):
self._print(syslog.LOG_INFO, *a, **kw) self._print(syslog.LOG_INFO, *a, **kw)
def error(self, *a, **kw): def error(self, *a, **kw):
self._print(syslog.LOG_ERR, *a, **kw) self._print(syslog.LOG_ERR, *a, **kw)
def _print(self, prio, *a, **kw): def _print(self, prio, *a, **kw):
debug_lock.acquire() debug_lock.acquire()
global _startt global _startt
@ -88,108 +97,126 @@ class LoggingEmu():
string = a[0] string = a[0]
else: else:
string = a[0] % a[1:] string = a[0] % a[1:]
syslog.syslog(prio, ("%.3f " % deltat) + threading.current_thread().name + ": " + string) n = threading.current_thread().name
syslog.syslog(
prio,
("%.3f " % deltat) + n + ": " + string,
)
finally: finally:
debug_lock.release() debug_lock.release()
logging = None logging = None
def send_confirmation(chan, retval, errmsg): def send_confirmation(chan, retval, errmsg):
chan.write(struct.pack("!H", retval)) chan.write(struct.pack("!H", retval))
l = len(errmsg) ln = len(errmsg)
assert l < 1<<32 assert ln < 1 << 32
chan.write(struct.pack("!I", l)) chan.write(struct.pack("!I", ln))
chan.write(errmsg) chan.write(errmsg)
chan.flush() chan.flush()
logging.debug("Sent confirmation on channel %s: %s %s", chan, retval, errmsg) logging.debug(
"Sent confirmation on channel %s: %s %s",
chan,
retval,
errmsg,
)
def recv_confirmation(chan): def recv_confirmation(chan):
logging.debug("Waiting for confirmation on channel %s", chan) logging.debug("Waiting for confirmation on channel %s", chan)
r = chan.read(2) r = chan.read(2)
if len(r) == 0: if len(r) == 0:
# This happens when the remote domain does not exist. # This happens when the remote domain does not exist.
r, errmsg = 125, "domain does not exist" r, errmsg = 125, "domain does not exist"
logging.debug("No confirmation: %s %s", r, errmsg) logging.debug("No confirmation: %s %s", r, errmsg)
return r, errmsg return r, errmsg
assert len(r) == 2, r assert len(r) == 2, r
r = struct.unpack("!H", r)[0] r = struct.unpack("!H", r)[0]
l = chan.read(4) lc = chan.read(4)
assert len(l) == 4, l assert len(lc) == 4, lc
l = struct.unpack("!I", l)[0] lu = struct.unpack("!I", lc)[0]
errmsg = chan.read(l) errmsg = chan.read(lu)
logging.debug("Received confirmation: %s %s", r, errmsg) logging.debug("Received confirmation: %s %s", r, errmsg)
return r, errmsg return r, errmsg
class MyThread(threading.Thread): class MyThread(threading.Thread):
def run(self):
def run(self): try:
try: self._run()
self._run() except Exception:
except Exception as e: n = threading.current_thread().name
logging.error("%s: unexpected exception", threading.current_thread().name) logging.error("%s: unexpected exception", n)
tb = traceback.format_exc() tb = traceback.format_exc()
logging.error("%s: traceback: %s", threading.current_thread().name, tb) logging.error("%s: traceback: %s", n, tb)
logging.error("%s: exiting program", threading.current_thread().name) logging.error("%s: exiting program", n)
os._exit(124) os._exit(124)
class SignalSender(MyThread): class SignalSender(MyThread):
def __init__(self, signals, sigqueue):
"""Handles signals by pushing them into a file-like object."""
threading.Thread.__init__(self)
self.daemon = True
self.queue = Queue()
self.sigqueue = sigqueue
for sig in signals:
signal.signal(sig, self.copy)
def __init__(self, signals, sigqueue): def copy(self, signum, frame):
"""Handles signals by pushing them into a file-like object.""" self.queue.put(signum)
threading.Thread.__init__(self) logging.debug("Signal %s pushed to queue", signum)
self.daemon = True
self.queue = queue.Queue()
self.sigqueue = sigqueue
for sig in signals:
signal.signal(sig, self.copy)
def copy(self, signum, frame): def _run(self):
self.queue.put(signum) while True:
logging.debug("Signal %s pushed to queue", signum) signum = self.queue.get()
logging.debug("Dequeued signal %s", signum)
def _run(self): if signum is None:
while True: break
signum = self.queue.get() assert signum > 0
logging.debug("Dequeued signal %s", signum) self.sigqueue.write(struct.pack("!H", signum))
if signum is None: self.sigqueue.flush()
break logging.debug("Wrote signal %s to remote end", signum)
assert signum > 0
self.sigqueue.write(struct.pack("!H", signum))
self.sigqueue.flush()
logging.debug("Wrote signal %s to remote end", signum)
class Signaler(MyThread): class Signaler(MyThread):
def __init__(self, process, sigqueue):
"""Reads integers from a file-like object and relays that as kill()."""
threading.Thread.__init__(self)
self.daemon = True
self.process = process
self.sigqueue = sigqueue
def __init__(self, process, sigqueue): def _run(self):
"""Reads integers from a file-like object and relays that as kill().""" while True:
threading.Thread.__init__(self) data = self.sigqueue.read(2)
self.daemon = True if len(data) == 0:
self.process = process logging.debug("Received no signal data")
self.sigqueue = sigqueue break
assert len(data) == 2
def _run(self): signum = struct.unpack("!H", data)[0]
while True: logging.debug(
data = self.sigqueue.read(2) "Received relayed signal %s, sending to process %s",
if len(data) == 0: signum,
logging.debug("Received no signal data") self.process.pid,
break )
assert len(data) == 2 try:
signum = struct.unpack("!H", data)[0] self.process.send_signal(signum)
logging.debug("Received relayed signal %s, sending to process %s", signum, self.process.pid) except BaseException as e:
try: logging.error(
self.process.send_signal(signum) "Failed to relay signal %s to process %s: %s",
except BaseException as e: signum,
logging.error("Failed to relay signal %s to process %s: %s", signum, self.process.pid, e) self.process.pid,
logging.debug("End of signaler") e,
)
logging.debug("End of signaler")
def write(dst, buffer, l): def write(dst, buffer, ln):
alreadywritten = 0 alreadywritten = 0
mv = memoryview(buffer)[:l] mv = memoryview(buffer)[:ln]
while len(mv): while len(mv):
dst.write(mv) dst.write(mv)
writtenthisloop = len(mv) writtenthisloop = len(mv)
@ -199,10 +226,10 @@ def write(dst, buffer, l):
alreadywritten = alreadywritten + writtenthisloop alreadywritten = alreadywritten + writtenthisloop
def copy(src, dst, buffer, l): def copy(src, dst, buffer, ln):
alreadyread = 0 alreadyread = 0
mv = memoryview(buffer)[:l] mv = memoryview(buffer)[:ln]
assert len(mv) == l, "Buffer object is too small: %s %s" % (len(mv), l) assert len(mv) == ln, "Buffer object is too small: %s %s" % (len(mv), ln)
while len(mv): while len(mv):
_, _, _ = select.select([src], (), ()) _, _, _ = select.select([src], (), ())
readthisloop = src.readinto(mv) readthisloop = src.readinto(mv)
@ -210,220 +237,241 @@ def copy(src, dst, buffer, l):
raise Exception("copy: Failed to read any bytes") raise Exception("copy: Failed to read any bytes")
mv = mv[readthisloop:] mv = mv[readthisloop:]
alreadyread = alreadyread + readthisloop alreadyread = alreadyread + readthisloop
return write(dst, buffer, l) return write(dst, buffer, ln)
class DataMultiplexer(MyThread): class DataMultiplexer(MyThread):
def __init__(self, sources, sink):
threading.Thread.__init__(self)
self.daemon = True
self.sources = dict((s, num) for num, s in enumerate(sources))
self.sink = sink
def __init__(self, sources, sink): def _run(self):
threading.Thread.__init__(self) logging.debug(
self.daemon = True "mux: Started with sources %s and sink %s", self.sources, self.sink
self.sources = dict((s,num) for num, s in enumerate(sources)) )
self.sink = sink buffer = bytearray(MAX_MUX_READ)
while self.sources:
def _run(self): sources, _, x = select.select(
logging.debug("mux: Started with sources %s and sink %s", self.sources, self.sink) (s for s in self.sources), (), (s for s in self.sources)
buffer = bytearray(MAX_MUX_READ) )
while self.sources: assert not x, x
sources, _, x = select.select((s for s in self.sources), (), (s for s in self.sources)) for s in sources:
assert not x, x n = self.sources[s]
for s in sources: logging.debug("mux: Source %s (%s) is active", n, s)
n = self.sources[s] readthisloop = s.readinto(buffer)
logging.debug("mux: Source %s (%s) is active", n, s) if readthisloop == 0:
readthisloop = s.readinto(buffer) logging.debug(
if readthisloop == 0: "mux: Received no bytes from source %s, signaling"
logging.debug("mux: Received no bytes from source %s, signaling peer to close corresponding source", n) " peer to close corresponding source",
del self.sources[s] n,
header = struct.pack(PACKFORMAT, n, False, 0) )
self.sink.write(header) del self.sources[s]
continue header = struct.pack(PACKFORMAT, n, False, 0)
l = readthisloop self.sink.write(header)
header = struct.pack(PACKFORMAT, n, True, l) continue
self.sink.write(header) ln = readthisloop
write(self.sink, buffer, l) header = struct.pack(PACKFORMAT, n, True, ln)
logging.debug("mux: End of data multiplexer") self.sink.write(header)
write(self.sink, buffer, ln)
logging.debug("mux: End of data multiplexer")
class DataDemultiplexer(MyThread): class DataDemultiplexer(MyThread):
def __init__(self, source, sinks):
threading.Thread.__init__(self)
self.daemon = True
self.sinks = dict(enumerate(sinks))
self.source = source
def __init__(self, source, sinks): def _run(self):
threading.Thread.__init__(self) logging.debug(
self.daemon = True "demux: Started with source %s and sinks %s",
self.sinks = dict(enumerate(sinks)) self.source,
self.source = source self.sinks,
)
def _run(self): buffer = bytearray(MAX_MUX_READ)
logging.debug("demux: Started with source %s and sinks %s", self.source, self.sinks) while self.sinks:
buffer = bytearray(MAX_MUX_READ) r, _, x = select.select([self.source], (), [self.source])
while self.sinks: assert not x, x
r, _, x = select.select([self.source], (), [self.source]) for s in r:
assert not x, x header = s.read(PACKLEN)
for s in r: if header == "":
header = s.read(PACKLEN) logging.debug(
if header == "": "demux: Received no bytes from source, closing sinks",
logging.debug("demux: Received no bytes from source, closing all sinks") )
for sink in self.sinks.values(): for sink in self.sinks.values():
sink.close() sink.close()
self.sinks = [] self.sinks = []
break break
n, active, l = struct.unpack(PACKFORMAT, header) n, active, ln = struct.unpack(PACKFORMAT, header)
if not active: if not active:
logging.debug("demux: Source %s now inactive, closing corresponding sink %s", s, self.sinks[n]) logging.debug(
self.sinks[n].close() "demux: Source %s inactive, closing matching sink %s",
del self.sinks[n] s,
else: self.sinks[n],
copy(self.source, self.sinks[n], buffer, l) )
logging.debug("demux: End of data demultiplexer") self.sinks[n].close()
del self.sinks[n]
else:
copy(self.source, self.sinks[n], buffer, ln)
logging.debug("demux: End of data demultiplexer")
def main_master(): def main_master():
global logging global logging
logging = LoggingEmu("master") logging = LoggingEmu("master")
logging.info("Started with arguments: %s", sys.argv[1:]) logging.info("Started with arguments: %s", sys.argv[1:])
global debug_enabled global debug_enabled
args = sys.argv[1:] args = sys.argv[1:]
if args[0] == "-d": if args[0] == "-d":
args = args[1:] args = args[1:]
debug_enabled = True debug_enabled = True
remote_vm = args[0] remote_vm = args[0]
remote_command = args[1:] remote_command = args[1:]
assert remote_command assert remote_command
def anypython(exe): def anypython(exe):
return "` test -x %s && echo %s || echo python`" % (pipes.quote(exe), return "` test -x %s && echo %s || echo python`" % (
pipes.quote(exe)) quote(exe),
quote(exe),
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()))
with mutexfile(os.path.expanduser("~/.bombshell-lock")):
try:
p = subprocess.Popen(
["qrexec-client-vm", remote_vm, "qubes.VMShell"],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
close_fds=True,
preexec_fn=os.setpgrp,
bufsize=0,
) )
except OSError as e:
logging.error("cannot launch qrexec-client-vm: %s", e)
return 127
logging.debug("Writing the helper text into the other side") remote_helper_text = b"exec "
p.stdin.write(remote_helper_text) remote_helper_text += bytes(anypython(sys.executable), "utf-8")
p.stdin.flush() remote_helper_text += bytes(" -u -c ", "utf-8")
remote_helper_text += bytes(
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"
confirmation, errmsg = recv_confirmation(p.stdout) saved_stderr = openfdforappend(os.dup(sys.stderr.fileno()))
if confirmation != 0:
logging.error("remote: %s", errmsg)
return confirmation
handled_signals = ( with mutexfile(os.path.expanduser("~/.bombshell-lock")):
signal.SIGINT, try:
signal.SIGABRT, p = subprocess.Popen(
signal.SIGALRM, ["qrexec-client-vm", remote_vm, "qubes.VMShell"],
signal.SIGTERM, stdin=subprocess.PIPE,
signal.SIGUSR1, stdout=subprocess.PIPE,
signal.SIGUSR2, close_fds=True,
signal.SIGTSTP, preexec_fn=os.setpgrp,
signal.SIGCONT, bufsize=0,
) )
read_signals, write_signals = pairofpipes() except OSError as e:
signaler = SignalSender(handled_signals, write_signals) logging.error("cannot launch qrexec-client-vm: %s", e)
signaler.name = "master signaler" return 127
signaler.start()
muxer = DataMultiplexer([sys.stdin, read_signals], p.stdin) logging.debug("Writing the helper text into the other side")
muxer.name = "master multiplexer" p.stdin.write(remote_helper_text)
muxer.start() p.stdin.flush()
demuxer = DataDemultiplexer(p.stdout, [sys.stdout, saved_stderr]) confirmation, errmsg = recv_confirmation(p.stdout)
demuxer.name = "master demultiplexer" if confirmation != 0:
demuxer.start() logging.error("remote: %s", errmsg)
return confirmation
retval = p.wait() handled_signals = (
logging.info("Return code %s for qubes.VMShell proxy", retval) signal.SIGINT,
demuxer.join() signal.SIGABRT,
logging.info("Ending bombshell") signal.SIGALRM,
return retval signal.SIGTERM,
signal.SIGUSR1,
signal.SIGUSR2,
signal.SIGTSTP,
signal.SIGCONT,
)
read_signals, write_signals = pairofpipes()
signaler = SignalSender(handled_signals, write_signals)
signaler.name = "master signaler"
signaler.start()
muxer = DataMultiplexer([sys.stdin, read_signals], p.stdin)
muxer.name = "master multiplexer"
muxer.start()
demuxer = DataDemultiplexer(p.stdout, [sys.stdout, saved_stderr])
demuxer.name = "master demultiplexer"
demuxer.start()
retval = p.wait()
logging.info("Return code %s for qubes.VMShell proxy", retval)
demuxer.join()
logging.info("Ending bombshell")
return retval
def pairofpipes(): def pairofpipes():
read, write = os.pipe() read, write = os.pipe()
return os.fdopen(read, "rb", 0), os.fdopen(write, "wb", 0) return os.fdopen(read, "rb", 0), os.fdopen(write, "wb", 0)
def main_remote(): def main_remote():
global logging global logging
logging = LoggingEmu("remote") logging = LoggingEmu("remote")
logging.info("Started with arguments: %s", sys.argv[1:]) logging.info("Started with arguments: %s", sys.argv[1:])
global debug_enabled global debug_enabled
if "-d" in sys.argv[1:]: if "-d" in sys.argv[1:]:
debug_enabled = True debug_enabled = True
cmd = sys.argv[2] cmd = sys.argv[2]
else: else:
cmd = sys.argv[1] cmd = sys.argv[1]
cmd = pickle.loads(base64.b64decode(cmd)) cmd = pickle.loads(base64.b64decode(cmd))
logging.debug("Received command: %s", cmd) logging.debug("Received command: %s", cmd)
nicecmd = " ".join(pipes.quote(a) for a in cmd) nicecmd = " ".join(quote(a) for a in cmd)
try: try:
p = subprocess.Popen( p = subprocess.Popen(
cmd, cmd,
# ["strace", "-s4096", "-ff"] + cmd, # ["strace", "-s4096", "-ff"] + cmd,
stdin = subprocess.PIPE, stdin=subprocess.PIPE,
stdout = subprocess.PIPE, stdout=subprocess.PIPE,
stderr = subprocess.PIPE, stderr=subprocess.PIPE,
close_fds=True, close_fds=True,
bufsize=0, bufsize=0,
) )
send_confirmation(sys.stdout, 0, b"") send_confirmation(sys.stdout, 0, b"")
except OSError as e: except OSError as e:
msg = "cannot execute %s: %s" % (nicecmd, e) msg = "cannot execute %s: %s" % (nicecmd, e)
logging.error(msg) logging.error(msg)
send_confirmation(sys.stdout, 127, bytes(msg, "utf-8")) send_confirmation(sys.stdout, 127, bytes(msg, "utf-8"))
sys.exit(0) sys.exit(0)
except BaseException as e: except BaseException as e:
msg = "cannot execute %s: %s" % (nicecmd, e) msg = "cannot execute %s: %s" % (nicecmd, e)
logging.error(msg) logging.error(msg)
send_confirmation(sys.stdout, 126, bytes(msg, "utf-8")) send_confirmation(sys.stdout, 126, bytes(msg, "utf-8"))
sys.exit(0) sys.exit(0)
signals_read, signals_written = pairofpipes() signals_read, signals_written = pairofpipes()
signaler = Signaler(p, signals_read) signaler = Signaler(p, signals_read)
signaler.name = "remote signaler" signaler.name = "remote signaler"
signaler.start() signaler.start()
demuxer = DataDemultiplexer(sys.stdin, [p.stdin, signals_written]) demuxer = DataDemultiplexer(sys.stdin, [p.stdin, signals_written])
demuxer.name = "remote demultiplexer" demuxer.name = "remote demultiplexer"
demuxer.start() demuxer.start()
muxer = DataMultiplexer([p.stdout, p.stderr], sys.stdout) muxer = DataMultiplexer([p.stdout, p.stderr], sys.stdout)
muxer.name = "remote multiplexer" muxer.name = "remote multiplexer"
muxer.start() muxer.start()
logging.info("Started %s", nicecmd) logging.info("Started %s", nicecmd)
retval = p.wait() retval = p.wait()
logging.info("Return code %s for %s", retval, nicecmd) logging.info("Return code %s for %s", retval, nicecmd)
muxer.join() muxer.join()
logging.info("Ending bombshell") logging.info("Ending bombshell")
return retval return retval
sys.stdin = openfdforread(sys.stdin.fileno()) sys.stdin = openfdforread(sys.stdin.fileno())